From 3ec2fd804e8a30583290de28c12bf32f587dacd9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Fri, 14 Aug 2015 21:25:41 +0300 Subject: [PATCH] eth, eth/downloader: eth/62 downloader boilerplace + ancestor lookup --- eth/downloader/downloader.go | 273 ++++++++++++++++++++++++++---- eth/downloader/downloader_test.go | 70 +++++--- eth/downloader/peer.go | 36 +++- eth/handler.go | 21 ++- eth/peer.go | 27 +-- 5 files changed, 346 insertions(+), 81 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index b28879ee67..05b62d5ed4 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -19,6 +19,7 @@ package downloader import ( "errors" + "fmt" "math" "math/big" "sync" @@ -67,6 +68,7 @@ var ( errPendingQueue = errors.New("pending items in queue") errTimeout = errors.New("timeout") errEmptyHashSet = errors.New("empty hash set by peer") + errEmptyHeaderSet = errors.New("empty header set by peer") errPeersUnavailable = errors.New("no peers available or all peers tried for block download process") errAlreadyInPool = errors.New("hash already in pool") errInvalidChain = errors.New("retrieved hash chain is invalid") @@ -91,28 +93,37 @@ type chainInsertFn func(types.Blocks) (int, error) // peerDropFn is a callback type for dropping a peer detected as malicious. type peerDropFn func(id string) -type blockPack struct { - peerId string - blocks []*types.Block -} - +// hashPack is a batch of block hashes returned by a peer (eth/61). type hashPack struct { peerId string hashes []common.Hash } -type crossCheck struct { - expire time.Time - parent common.Hash +// blockPack is a batch of blocks returned by a peer (eth/61). +type blockPack struct { + peerId string + blocks []*types.Block +} + +// headerPack is a batch of block headers returned by a peer. +type headerPack struct { + peerId string + headers []*types.Header +} + +// bodyPack is a batch of block bodies returned by a peer. +type bodyPack struct { + peerId string + transactions [][]*types.Transaction + uncles [][]*types.Header } type Downloader struct { mux *event.TypeMux - queue *queue // Scheduler for selecting the hashes to download - peers *peerSet // Set of active peers from which download can proceed - checks map[common.Hash]*crossCheck // Pending cross checks to verify a hash chain - banned *set.Set // Set of hashes we've received and banned + queue *queue // Scheduler for selecting the hashes to download + peers *peerSet // Set of active peers from which download can proceed + banned *set.Set // Set of hashes we've received and banned interrupt int32 // Atomic boolean to signal termination @@ -137,9 +148,11 @@ type Downloader struct { // Channels newPeerCh chan *peer - hashCh chan hashPack // Channel receiving inbound hashes - blockCh chan blockPack // Channel receiving inbound blocks - processCh chan bool // Channel to signal the block fetcher of new or finished work + hashCh chan hashPack // [eth/61] Channel receiving inbound hashes + blockCh chan blockPack // [eth/61] Channel receiving inbound blocks + headerCh chan headerPack // [eth/62] Channel receiving inbound block headers + bodyCh chan bodyPack // [eth/62] Channel receiving inbound block bodies + processCh chan bool // Channel to signal the block fetcher of new or finished work cancelCh chan struct{} // Channel to cancel mid-flight syncs cancelLock sync.RWMutex // Lock to protect the cancel channel in delivers @@ -166,6 +179,8 @@ func New(mux *event.TypeMux, hasBlock hashCheckFn, getBlock blockRetrievalFn, he newPeerCh: make(chan *peer, 1), hashCh: make(chan hashPack, 1), blockCh: make(chan blockPack, 1), + headerCh: make(chan headerPack, 1), + bodyCh: make(chan bodyPack, 1), processCh: make(chan bool, 1), } // Inject all the known bad hashes @@ -206,7 +221,9 @@ func (d *Downloader) Synchronising() bool { // RegisterPeer injects a new download peer into the set of block source to be // used for fetching hashes and blocks from. -func (d *Downloader) RegisterPeer(id string, version int, head common.Hash, getRelHashes relativeHashFetcherFn, getAbsHashes absoluteHashFetcherFn, getBlocks blockFetcherFn) error { +func (d *Downloader) RegisterPeer(id string, version int, head common.Hash, + getRelHashes relativeHashFetcherFn, getAbsHashes absoluteHashFetcherFn, getBlocks blockFetcherFn, // eth/61 callbacks, remove when upgrading + getRelHeaders relativeHeaderFetcherFn, getAbsHeaders absoluteHeaderFetcherFn, getBlockBodies blockBodyFetcherFn) error { // If the peer wants to send a banned hash, reject if d.banned.Has(head) { glog.V(logger.Debug).Infoln("Register rejected, head hash banned:", id) @@ -214,7 +231,7 @@ func (d *Downloader) RegisterPeer(id string, version int, head common.Hash, getR } // Otherwise try to construct and register the peer glog.V(logger.Detail).Infoln("Registering peer", id) - if err := d.peers.Register(newPeer(id, version, head, getRelHashes, getAbsHashes, getBlocks)); err != nil { + if err := d.peers.Register(newPeer(id, version, head, getRelHashes, getAbsHashes, getBlocks, getRelHeaders, getAbsHeaders, getBlockBodies)); err != nil { glog.V(logger.Error).Infoln("Register failed:", err) return err } @@ -244,7 +261,7 @@ func (d *Downloader) Synchronise(id string, head common.Hash, td *big.Int) { case errBusy: glog.V(logger.Detail).Infof("Synchronisation already in progress") - case errTimeout, errBadPeer, errStallingPeer, errBannedHead, errEmptyHashSet, errPeersUnavailable, errInvalidChain, errCrossCheckFailed: + case errTimeout, errBadPeer, errStallingPeer, errBannedHead, errEmptyHashSet, errEmptyHeaderSet, errPeersUnavailable, errInvalidChain, errCrossCheckFailed: glog.V(logger.Debug).Infof("Removing peer %v: %v", id, err) d.dropPeer(id) @@ -285,7 +302,6 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int) error // Reset the queue and peer set to clean any internal leftover state d.queue.Reset() d.peers.Reset() - d.checks = make(map[common.Hash]*crossCheck) // Create cancel channel for aborting mid-flight d.cancelLock.Lock() @@ -324,13 +340,13 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e switch p.version { case eth61: // Old eth/61, use forward, concurrent hash and block retrieval algorithm - number, err := d.findAncestor(p) + number, err := d.findAncestor61(p) if err != nil { return err } errc := make(chan error, 2) - go func() { errc <- d.fetchHashes(p, td, number+1) }() - go func() { errc <- d.fetchBlocks(number + 1) }() + go func() { errc <- d.fetchHashes61(p, td, number+1) }() + go func() { errc <- d.fetchBlocks61(number + 1) }() // If any fetcher fails, cancel the other if err := <-errc; err != nil { @@ -340,6 +356,16 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e } return <-errc + case eth62: + // New eth/62, use forward, concurrent header and block body retrieval algorithm + fmt.Println("Looking for common ancestor:", p) + number, err := d.findAncestor(p) + if err != nil { + return err + } + fmt.Println("Common ancestor:", number) + return errBadPeer + default: // Something very wrong, stop right here glog.V(logger.Error).Infof("Unsupported eth protocol: %d", p.version) @@ -373,12 +399,12 @@ func (d *Downloader) Terminate() { d.cancel() } -// findAncestor tries to locate the common ancestor block of the local chain and +// findAncestor61 tries to locate the common ancestor block of the local chain and // a remote peers blockchain. In the general case when our node was in sync and // on the correct chain, checking the top N blocks should already get us a match. // In the rare scenario when we ended up on a long soft fork (i.e. none of the // head blocks match), we do a binary search to find the common ancestor. -func (d *Downloader) findAncestor(p *peer) (uint64, error) { +func (d *Downloader) findAncestor61(p *peer) (uint64, error) { glog.V(logger.Debug).Infof("%v: looking for common ancestor", p) // Request out head blocks to short circuit ancestor location @@ -422,6 +448,12 @@ func (d *Downloader) findAncestor(p *peer) (uint64, error) { case <-d.blockCh: // Out of bounds blocks received, ignore them + case <-d.headerCh: + // Out of bounds eth/62 block headers received, ignore them + + case <-d.bodyCh: + // Out of bounds eth/62 block bodies received, ignore them + case <-timeout: glog.V(logger.Debug).Infof("%v: head hash timeout", p) return 0, errTimeout @@ -476,6 +508,12 @@ func (d *Downloader) findAncestor(p *peer) (uint64, error) { case <-d.blockCh: // Out of bounds blocks received, ignore them + case <-d.headerCh: + // Out of bounds eth/62 block headers received, ignore them + + case <-d.bodyCh: + // Out of bounds eth/62 block bodies received, ignore them + case <-timeout: glog.V(logger.Debug).Infof("%v: search hash timeout", p) return 0, errTimeout @@ -485,9 +523,9 @@ func (d *Downloader) findAncestor(p *peer) (uint64, error) { return start, nil } -// fetchHashes keeps retrieving hashes from the requested number, until no more +// fetchHashes61 keeps retrieving hashes from the requested number, until no more // are returned, potentially throttling on the way. -func (d *Downloader) fetchHashes(p *peer, td *big.Int, from uint64) error { +func (d *Downloader) fetchHashes61(p *peer, td *big.Int, from uint64) error { glog.V(logger.Debug).Infof("%v: downloading hashes from #%d", p, from) // Create a timeout timer, and the associated hash fetcher @@ -573,10 +611,10 @@ func (d *Downloader) fetchHashes(p *peer, td *big.Int, from uint64) error { } } -// fetchBlocks iteratively downloads the scheduled hashes, taking any available +// fetchBlocks61 iteratively downloads the scheduled hashes, taking any available // peers, reserving a chunk of blocks for each, waiting for delivery and also // periodically checking for timeouts. -func (d *Downloader) fetchBlocks(from uint64) error { +func (d *Downloader) fetchBlocks61(from uint64) error { glog.V(logger.Debug).Infof("Downloading blocks from #%d", from) defer glog.V(logger.Debug).Infof("Block download terminated") @@ -718,6 +756,130 @@ func (d *Downloader) fetchBlocks(from uint64) error { } } +// findAncestor tries to locate the common ancestor block of the local chain and +// a remote peers blockchain. In the general case when our node was in sync and +// on the correct chain, checking the top N blocks should already get us a match. +// In the rare scenario when we ended up on a long soft fork (i.e. none of the +// head blocks match), we do a binary search to find the common ancestor. +func (d *Downloader) findAncestor(p *peer) (uint64, error) { + glog.V(logger.Debug).Infof("%v: looking for common ancestor", p) + + // Request our head blocks to short circuit ancestor location + head := d.headBlock().NumberU64() + from := int64(head) - int64(MaxHashFetch) + if from < 0 { + from = 0 + } + go p.getAbsHeaders(uint64(from), MaxHeaderFetch, 0, false) + + // Wait for the remote response to the head fetch + number, hash := uint64(0), common.Hash{} + timeout := time.After(hashTTL) + + for finished := false; !finished; { + select { + case <-d.cancelCh: + return 0, errCancelHashFetch + + case headerPack := <-d.headerCh: + // Discard anything not from the origin peer + if headerPack.peerId != p.id { + glog.V(logger.Debug).Infof("Received headers from incorrect peer(%s)", headerPack.peerId) + break + } + // Make sure the peer actually gave something valid + headers := headerPack.headers + if len(headers) == 0 { + glog.V(logger.Debug).Infof("%v: empty head hash set", p) + return 0, errEmptyHashSet + } + // Check if a common ancestor was found + finished = true + for i := len(headers) - 1; i >= 0; i-- { + if d.hasBlock(headers[i].Hash()) { + number, hash = headers[i].Number.Uint64(), headers[i].Hash() + break + } + } + + case <-d.bodyCh: + // Out of bounds block bodies received, ignore them + + case <-d.hashCh: + // Out of bounds eth/61 hashes received, ignore them + + case <-d.blockCh: + // Out of bounds eth/61 blocks received, ignore them + + case <-timeout: + glog.V(logger.Debug).Infof("%v: head hash timeout", p) + return 0, errTimeout + } + } + // If the head fetch already found an ancestor, return + if !common.EmptyHash(hash) { + glog.V(logger.Debug).Infof("%v: common ancestor: #%d [%x]", p, number, hash[:4]) + return number, nil + } + // Ancestor not found, we need to binary search over our chain + start, end := uint64(0), head + for start+1 < end { + // Split our chain interval in two, and request the hash to cross check + check := (start + end) / 2 + + timeout := time.After(hashTTL) + go p.getAbsHeaders(uint64(check), 1, 0, false) + + // Wait until a reply arrives to this request + for arrived := false; !arrived; { + select { + case <-d.cancelCh: + return 0, errCancelHashFetch + + case headerPack := <-d.headerCh: + // Discard anything not from the origin peer + if headerPack.peerId != p.id { + glog.V(logger.Debug).Infof("Received hashes from incorrect peer(%s)", headerPack.peerId) + break + } + // Make sure the peer actually gave something valid + headers := headerPack.headers + if len(headers) != 1 { + glog.V(logger.Debug).Infof("%v: invalid search header set (%d)", p, len(headers)) + return 0, errBadPeer + } + arrived = true + + // Modify the search interval based on the response + block := d.getBlock(headers[0].Hash()) + if block == nil { + end = check + break + } + if block.NumberU64() != check { + glog.V(logger.Debug).Infof("%v: non requested header #%d [%x], instead of #%d", p, block.NumberU64(), block.Hash().Bytes()[:4], check) + return 0, errBadPeer + } + start = check + + case <-d.bodyCh: + // Out of bounds block bodies received, ignore them + + case <-d.hashCh: + // Out of bounds eth/61 hashes received, ignore them + + case <-d.blockCh: + // Out of bounds eth/61 blocks received, ignore them + + case <-timeout: + glog.V(logger.Debug).Infof("%v: search hash timeout", p) + return 0, errTimeout + } + } + } + return start, nil +} + // process takes blocks from the queue and tries to import them into the chain. // // The algorithmic flow is as follows: @@ -796,9 +958,31 @@ func (d *Downloader) process() { } } -// DeliverBlocks injects a new batch of blocks received from a remote node. +// DeliverHashes61 injects a new batch of hashes received from a remote node into +// the download schedule. This is usually invoked through the BlockHashesMsg by +// the protocol handler. +func (d *Downloader) DeliverHashes61(id string, hashes []common.Hash) error { + // Make sure the downloader is active + if atomic.LoadInt32(&d.synchronising) == 0 { + return errNoSyncActive + } + // Deliver or abort if the sync is canceled while queuing + d.cancelLock.RLock() + cancel := d.cancelCh + d.cancelLock.RUnlock() + + select { + case d.hashCh <- hashPack{id, hashes}: + return nil + + case <-cancel: + return errNoSyncActive + } +} + +// DeliverBlocks61 injects a new batch of blocks received from a remote node. // This is usually invoked through the BlocksMsg by the protocol handler. -func (d *Downloader) DeliverBlocks(id string, blocks []*types.Block) error { +func (d *Downloader) DeliverBlocks61(id string, blocks []*types.Block) error { // Make sure the downloader is active if atomic.LoadInt32(&d.synchronising) == 0 { return errNoSyncActive @@ -817,10 +1001,9 @@ func (d *Downloader) DeliverBlocks(id string, blocks []*types.Block) error { } } -// DeliverHashes injects a new batch of hashes received from a remote node into -// the download schedule. This is usually invoked through the BlockHashesMsg by -// the protocol handler. -func (d *Downloader) DeliverHashes(id string, hashes []common.Hash) error { +// DeliverHeaders injects a new batch of blck headers received from a remote +// node into the download schedule. +func (d *Downloader) DeliverHeaders(id string, headers []*types.Header) error { // Make sure the downloader is active if atomic.LoadInt32(&d.synchronising) == 0 { return errNoSyncActive @@ -831,7 +1014,27 @@ func (d *Downloader) DeliverHashes(id string, hashes []common.Hash) error { d.cancelLock.RUnlock() select { - case d.hashCh <- hashPack{id, hashes}: + case d.headerCh <- headerPack{id, headers}: + return nil + + case <-cancel: + return errNoSyncActive + } +} + +// DeliverBodies injects a new batch of block bodies received from a remote node. +func (d *Downloader) DeliverBodies(id string, transactions [][]*types.Transaction, uncles [][]*types.Header) error { + // Make sure the downloader is active + if atomic.LoadInt32(&d.synchronising) == 0 { + return errNoSyncActive + } + // Deliver or abort if the sync is canceled while queuing + d.cancelLock.RLock() + cancel := d.cancelCh + d.cancelLock.RUnlock() + + select { + case d.bodyCh <- bodyPack{id, transactions, uncles}: return nil case <-cancel: diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index 7e3456433d..ff9f7eec84 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -78,8 +78,6 @@ type downloadTester struct { ownBlocks map[common.Hash]*types.Block // Blocks belonging to the tester peerHashes map[string][]common.Hash // Hash chain belonging to different test peers peerBlocks map[string]map[common.Hash]*types.Block // Blocks belonging to different test peers - - maxHashFetch int // Overrides the maximum number of retrieved hashes } // newTester creates a new downloader test mocker. @@ -156,7 +154,9 @@ func (dl *downloadTester) newPeer(id string, version int, hashes []common.Hash, // specific delay time on processing the network packets sent to it, simulating // potentially slow network IO. func (dl *downloadTester) newSlowPeer(id string, version int, hashes []common.Hash, blocks map[common.Hash]*types.Block, delay time.Duration) error { - err := dl.downloader.RegisterPeer(id, version, hashes[0], dl.peerGetRelHashesFn(id, delay), dl.peerGetAbsHashesFn(id, version, delay), dl.peerGetBlocksFn(id, delay)) + err := dl.downloader.RegisterPeer(id, version, hashes[0], + dl.peerGetRelHashesFn(id, delay), dl.peerGetAbsHashesFn(id, version, delay), dl.peerGetBlocksFn(id, delay), + nil, dl.peerGetAbsHeadersFn(id, version, delay), nil) if err == nil { // Assign the owned hashes and blocks to the peer (deep copy) dl.peerHashes[id] = make([]common.Hash, len(hashes)) @@ -184,13 +184,9 @@ func (dl *downloadTester) peerGetRelHashesFn(id string, delay time.Duration) fun return func(head common.Hash) error { time.Sleep(delay) - limit := MaxHashFetch - if dl.maxHashFetch > 0 { - limit = dl.maxHashFetch - } // Gather the next batch of hashes hashes := dl.peerHashes[id] - result := make([]common.Hash, 0, limit) + result := make([]common.Hash, 0, MaxHashFetch) for i, hash := range hashes { if hash == head { i++ @@ -204,7 +200,7 @@ func (dl *downloadTester) peerGetRelHashesFn(id string, delay time.Duration) fun // Delay delivery a bit to allow attacks to unfold go func() { time.Sleep(time.Millisecond) - dl.downloader.DeliverHashes(id, result) + dl.downloader.DeliverHashes61(id, result) }() return nil } @@ -217,20 +213,16 @@ func (dl *downloadTester) peerGetAbsHashesFn(id string, version int, delay time. return func(head uint64, count int) error { time.Sleep(delay) - limit := count - if dl.maxHashFetch > 0 { - limit = dl.maxHashFetch - } // Gather the next batch of hashes hashes := dl.peerHashes[id] - result := make([]common.Hash, 0, limit) - for i := 0; i < limit && len(hashes)-int(head)-1-i >= 0; i++ { + result := make([]common.Hash, 0, count) + for i := 0; i < count && len(hashes)-int(head)-1-i >= 0; i++ { result = append(result, hashes[len(hashes)-int(head)-1-i]) } // Delay delivery a bit to allow attacks to unfold go func() { time.Sleep(time.Millisecond) - dl.downloader.DeliverHashes(id, result) + dl.downloader.DeliverHashes61(id, result) }() return nil } @@ -249,22 +241,48 @@ func (dl *downloadTester) peerGetBlocksFn(id string, delay time.Duration) func([ result = append(result, block) } } - go dl.downloader.DeliverBlocks(id, result) + go dl.downloader.DeliverBlocks61(id, result) return nil } } +// peerGetAbsHeadersFn constructs a GetBlockHeaders function based on a numbered +// origin; associated with a particular peer in the download tester. The returned +// function can be used to retrieve batches of headers from the particular peer. +func (dl *downloadTester) peerGetAbsHeadersFn(id string, version int, delay time.Duration) func(uint64, int, int, bool) error { + return func(origin uint64, amount int, skip int, reverse bool) error { + time.Sleep(delay) + + // Gather the next batch of hashes + hashes := dl.peerHashes[id] + blocks := dl.peerBlocks[id] + result := make([]*types.Header, 0, amount) + for i := 0; i < amount && len(hashes)-int(origin)-1-i >= 0; i++ { + result = append(result, blocks[hashes[len(hashes)-int(origin)-1-i]].Header()) + } + // Delay delivery a bit to allow attacks to unfold + go func() { + time.Sleep(time.Millisecond) + dl.downloader.DeliverHeaders(id, result) + }() + return nil + } +} + // Tests that simple synchronization against a canonical chain works correctly. // In this test common ancestor lookup should be short circuited and not require // binary searching. -func TestCanonicalSynchronisation61(t *testing.T) { +func TestCanonicalSynchronisation61(t *testing.T) { testCanonicalSynchronisation(t, 61) } +func TestCanonicalSynchronisation62(t *testing.T) { testCanonicalSynchronisation(t, 62) } + +func testCanonicalSynchronisation(t *testing.T, protocol int) { // Create a small enough block chain to download targetBlocks := blockCacheLimit - 15 hashes, blocks := makeChain(targetBlocks, 0, genesis) tester := newTester() - tester.newPeer("peer", eth61, hashes, blocks) + tester.newPeer("peer", protocol, hashes, blocks) // Synchronise with the peer and make sure all blocks were retrieved if err := tester.sync("peer", nil); err != nil { @@ -336,14 +354,17 @@ func testThrottling(t *testing.T, protocol int) { // Tests that simple synchronization against a forked chain works correctly. In // this test common ancestor lookup should *not* be short circuited, and a full // binary search should be executed. -func TestForkedSynchronisation61(t *testing.T) { +func TestForkedSynchronisation61(t *testing.T) { testForkedSynchronisation(t, 61) } +func TestForkedSynchronisation62(t *testing.T) { testForkedSynchronisation(t, 62) } + +func testForkedSynchronisation(t *testing.T, protocol int) { // Create a long enough forked chain common, fork := MaxHashFetch, 2*MaxHashFetch hashesA, hashesB, blocksA, blocksB := makeChainFork(common+fork, fork, genesis) tester := newTester() - tester.newPeer("fork A", eth61, hashesA, blocksA) - tester.newPeer("fork B", eth61, hashesB, blocksB) + tester.newPeer("fork A", protocol, hashesA, blocksA) + tester.newPeer("fork B", protocol, hashesB, blocksB) // Synchronise with the peer and make sure all blocks were retrieved if err := tester.sync("fork A", nil); err != nil { @@ -366,10 +387,10 @@ func TestInactiveDownloader(t *testing.T) { tester := newTester() // Check that neither hashes nor blocks are accepted - if err := tester.downloader.DeliverHashes("bad peer", []common.Hash{}); err != errNoSyncActive { + if err := tester.downloader.DeliverHashes61("bad peer", []common.Hash{}); err != errNoSyncActive { t.Errorf("error mismatch: have %v, want %v", err, errNoSyncActive) } - if err := tester.downloader.DeliverBlocks("bad peer", []*types.Block{}); err != errNoSyncActive { + if err := tester.downloader.DeliverBlocks61("bad peer", []*types.Block{}); err != errNoSyncActive { t.Errorf("error mismatch: have %v, want %v", err, errNoSyncActive) } } @@ -463,6 +484,7 @@ func TestHashAttackerDropping(t *testing.T) { {errPendingQueue, false}, // There are blocks still cached, wait to exhaust, no issue {errTimeout, true}, // No hashes received in due time, drop the peer {errEmptyHashSet, true}, // No hashes were returned as a response, drop as it's a dead end + {errEmptyHeaderSet, true}, // No headers were returned as a response, drop as it's a dead end {errPeersUnavailable, true}, // Nobody had the advertised blocks, drop the advertiser {errInvalidChain, true}, // Hash chain was detected as invalid, definitely drop {errCrossCheckFailed, true}, // Hash-origin failed to pass a block cross check, drop diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index 4273b91682..e9d3ce962f 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -31,10 +31,16 @@ import ( "gopkg.in/fatih/set.v0" ) +// Hash and block fetchers belonging to eth/61 and below type relativeHashFetcherFn func(common.Hash) error type absoluteHashFetcherFn func(uint64, int) error type blockFetcherFn func([]common.Hash) error +// Block header and body fethers belonging to eth/62 and above +type relativeHeaderFetcherFn func(common.Hash, int, int, bool) error +type absoluteHeaderFetcherFn func(uint64, int, int, bool) error +type blockBodyFetcherFn func([]common.Hash) error + var ( errAlreadyFetching = errors.New("already fetching blocks from peer") errAlreadyRegistered = errors.New("peer is already registered") @@ -54,25 +60,37 @@ type peer struct { ignored *set.Set // Set of hashes not to request (didn't have previously) - getRelHashes relativeHashFetcherFn // Method to retrieve a batch of hashes from an origin hash - getAbsHashes absoluteHashFetcherFn // Method to retrieve a batch of hashes from an absolute position - getBlocks blockFetcherFn // Method to retrieve a batch of blocks + getRelHashes relativeHashFetcherFn // [eth/61] Method to retrieve a batch of hashes from an origin hash + getAbsHashes absoluteHashFetcherFn // [eth/61] Method to retrieve a batch of hashes from an absolute position + getBlocks blockFetcherFn // [eth/61] Method to retrieve a batch of blocks + + getRelHeaders relativeHeaderFetcherFn // [eth/62] Method to retrieve a batch of headers from an origin hash + getAbsHeaders absoluteHeaderFetcherFn // [eth/62] Method to retrieve a batch of headers from an absolute position + getBlockBodies blockBodyFetcherFn // [eth/62] Method to retrieve a batch of block bodies version int // Eth protocol version number to switch strategies } // newPeer create a new downloader peer, with specific hash and block retrieval // mechanisms. -func newPeer(id string, version int, head common.Hash, getRelHashes relativeHashFetcherFn, getAbsHashes absoluteHashFetcherFn, getBlocks blockFetcherFn) *peer { +func newPeer(id string, version int, head common.Hash, + getRelHashes relativeHashFetcherFn, getAbsHashes absoluteHashFetcherFn, getBlocks blockFetcherFn, // eth/61 callbacks, remove when upgrading + getRelHeaders relativeHeaderFetcherFn, getAbsHeaders absoluteHeaderFetcherFn, getBlockBodies blockBodyFetcherFn) *peer { return &peer{ - id: id, - head: head, - capacity: 1, + id: id, + head: head, + capacity: 1, + ignored: set.New(), + getRelHashes: getRelHashes, getAbsHashes: getAbsHashes, getBlocks: getBlocks, - ignored: set.New(), - version: version, + + getRelHeaders: getRelHeaders, + getAbsHeaders: getAbsHeaders, + getBlockBodies: getBlockBodies, + + version: version, } } diff --git a/eth/handler.go b/eth/handler.go index 7a5a1b754c..353a447037 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -201,7 +201,9 @@ func (pm *ProtocolManager) handle(p *peer) error { defer pm.removePeer(p.id) // Register the peer in the downloader. If the downloader considers it banned, we disconnect - if err := pm.downloader.RegisterPeer(p.id, p.version, p.Head(), p.RequestHashes, p.RequestHashesFromNumber, p.RequestBlocks); err != nil { + if err := pm.downloader.RegisterPeer(p.id, p.version, p.Head(), + p.RequestHashes, p.RequestHashesFromNumber, p.RequestBlocks, + p.RequestHeadersByHash, p.RequestHeadersByNumber, nil); err != nil { return err } // Propagate existing transactions. new transactions appearing @@ -287,7 +289,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { break } // Deliver them all to the downloader for queuing - err := pm.downloader.DeliverHashes(p.id, hashes) + err := pm.downloader.DeliverHashes61(p.id, hashes) if err != nil { glog.V(logger.Debug).Infoln(err) } @@ -333,7 +335,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } // Filter out any explicitly requested blocks, deliver the rest to the downloader if blocks := pm.fetcher.Filter(blocks); len(blocks) > 0 { - pm.downloader.DeliverBlocks(p.id, blocks) + pm.downloader.DeliverBlocks61(p.id, blocks) } // Block header query, collect the requested headers and reply @@ -399,8 +401,21 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { query.Origin.Number += (query.Skip + 1) } } + fmt.Println("Sending headers", len(headers)) return p.SendBlockHeaders(headers) + case p.version >= eth62 && msg.Code == BlockHeadersMsg: + // A batch of headers arrived to one of our previous requests + var headers []*types.Header + if err := msg.Decode(&headers); err != nil { + return errResp(ErrDecode, "msg %v: %v", msg, err) + } + // Deliver them all to the downloader for queuing + err := pm.downloader.DeliverHeaders(p.id, headers) + if err != nil { + glog.V(logger.Debug).Infoln(err) + } + case p.version >= eth62 && msg.Code == GetBlockBodiesMsg: // Decode the retrieval message msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) diff --git a/eth/peer.go b/eth/peer.go index 78de8a9d34..9475ef2814 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -185,40 +185,47 @@ func (p *peer) SendReceipts(receipts []*types.Receipt) error { // RequestHashes fetches a batch of hashes from a peer, starting at from, going // towards the genesis block. func (p *peer) RequestHashes(from common.Hash) error { - glog.V(logger.Debug).Infof("%v fetching hashes (%d) from %x...\n", p, downloader.MaxHashFetch, from[:4]) + glog.V(logger.Debug).Infof("%v fetching hashes (%d) from %x...", p, downloader.MaxHashFetch, from[:4]) return p2p.Send(p.rw, GetBlockHashesMsg, getBlockHashesData{from, uint64(downloader.MaxHashFetch)}) } // RequestHashesFromNumber fetches a batch of hashes from a peer, starting at // the requested block number, going upwards towards the genesis block. func (p *peer) RequestHashesFromNumber(from uint64, count int) error { - glog.V(logger.Debug).Infof("%v fetching hashes (%d) from #%d...\n", p, count, from) + glog.V(logger.Debug).Infof("%v fetching hashes (%d) from #%d...", p, count, from) return p2p.Send(p.rw, GetBlockHashesFromNumberMsg, getBlockHashesFromNumberData{from, uint64(count)}) } // RequestBlocks fetches a batch of blocks corresponding to the specified hashes. func (p *peer) RequestBlocks(hashes []common.Hash) error { - glog.V(logger.Debug).Infof("%v fetching %v blocks\n", p, len(hashes)) + glog.V(logger.Debug).Infof("%v fetching %v blocks", p, len(hashes)) return p2p.Send(p.rw, GetBlocksMsg, hashes) } -// RequestHeaders fetches a batch of blocks' headers corresponding to the -// specified hashes. -func (p *peer) RequestHeaders(hashes []common.Hash) error { - glog.V(logger.Debug).Infof("%v fetching %v headers\n", p, len(hashes)) - return p2p.Send(p.rw, GetBlockHeadersMsg, hashes) +// RequestHeadersByHash fetches a batch of blocks' headers corresponding to the +// specified header query, based on the hash of an origin block. +func (p *peer) RequestHeadersByHash(origin common.Hash, amount int, skip int, reverse bool) error { + glog.V(logger.Debug).Infof("%v fetching %d headers from %x, skipping %d (reverse = )", p, amount, origin[:4], skip, reverse) + return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) +} + +// RequestHeadersByNumber fetches a batch of blocks' headers corresponding to the +// specified header query, based on the number of an origin block. +func (p *peer) RequestHeadersByNumber(origin uint64, amount int, skip int, reverse bool) error { + glog.V(logger.Debug).Infof("%v fetching %d headers from #%d, skipping %d (reverse = )", p, amount, origin, skip, reverse) + return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) } // RequestNodeData fetches a batch of arbitrary data from a node's known state // data, corresponding to the specified hashes. func (p *peer) RequestNodeData(hashes []common.Hash) error { - glog.V(logger.Debug).Infof("%v fetching %v state data\n", p, len(hashes)) + glog.V(logger.Debug).Infof("%v fetching %v state data", p, len(hashes)) return p2p.Send(p.rw, GetNodeDataMsg, hashes) } // RequestReceipts fetches a batch of transaction receipts from a remote node. func (p *peer) RequestReceipts(hashes []common.Hash) error { - glog.V(logger.Debug).Infof("%v fetching %v receipts\n", p, len(hashes)) + glog.V(logger.Debug).Infof("%v fetching %v receipts", p, len(hashes)) return p2p.Send(p.rw, GetReceiptsMsg, hashes) }