From 52a8f7dfd39a163c221ea51c776acd57e51459ff Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Wed, 25 Sep 2019 13:09:55 +0800 Subject: [PATCH] eth/downloader: abort sync if master peer is unregistered --- eth/downloader/downloader.go | 21 ++++++------ eth/downloader/downloader_test.go | 57 +++++++++++++++++++++++++++++-- eth/downloader/statesync.go | 13 ++----- 3 files changed, 69 insertions(+), 22 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index edd0eb4d95..124cbd04f3 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -314,6 +314,14 @@ func (d *Downloader) UnregisterPeer(id string) error { } d.queue.Revoke(id) + // If this peer was the master peer, abort sync immediately + d.cancelLock.RLock() + master := id == d.cancelPeer + d.cancelLock.RUnlock() + + if master { + d.cancel() + } return nil } @@ -1275,17 +1283,10 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack) // Timeouts can occur if e.g. compaction hits at the wrong time, and can be ignored peer.log.Warn("Downloader wants to drop peer, but peerdrop-function is not set", "peer", pid) } else { + // In dropPeer function, a callback will be called which aborts + // the sync immediately. Here return the timeout error explicitly. d.dropPeer(pid) - - // If this peer was the master peer, abort sync immediately - d.cancelLock.RLock() - master := pid == d.cancelPeer - d.cancelLock.RUnlock() - - if master { - d.cancel() - return errTimeout - } + return errTimeout } } } diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index b23043b1c0..cbe3dd4dc6 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -362,11 +362,20 @@ func (dl *downloadTester) dropPeer(id string) { dl.downloader.UnregisterPeer(id) } +// setDelay adds a response delay to test peer. +func (dl *downloadTester) setDelay(id string, delay time.Duration) { + dl.lock.Lock() + defer dl.lock.Unlock() + + dl.peers[id].delayResponse = delay +} + type downloadTesterPeer struct { dl *downloadTester id string lock sync.RWMutex chain *testChain + delayResponse time.Duration missingStates map[common.Hash]bool // State entries that fast sync should not return } @@ -384,7 +393,7 @@ func (dlp *downloadTesterPeer) RequestHeadersByHash(origin common.Hash, amount i if reverse { panic("reverse header requests not supported") } - + time.Sleep(dlp.delayResponse) result := dlp.chain.headersByHash(origin, amount, skip) go dlp.dl.downloader.DeliverHeaders(dlp.id, result) return nil @@ -397,7 +406,7 @@ func (dlp *downloadTesterPeer) RequestHeadersByNumber(origin uint64, amount int, if reverse { panic("reverse header requests not supported") } - + time.Sleep(dlp.delayResponse) result := dlp.chain.headersByNumber(origin, amount, skip) go dlp.dl.downloader.DeliverHeaders(dlp.id, result) return nil @@ -407,6 +416,7 @@ func (dlp *downloadTesterPeer) RequestHeadersByNumber(origin uint64, amount int, // peer in the download tester. The returned function can be used to retrieve // batches of block bodies from the particularly requested peer. func (dlp *downloadTesterPeer) RequestBodies(hashes []common.Hash) error { + time.Sleep(dlp.delayResponse) txs, uncles := dlp.chain.bodies(hashes) go dlp.dl.downloader.DeliverBodies(dlp.id, txs, uncles) return nil @@ -416,6 +426,7 @@ func (dlp *downloadTesterPeer) RequestBodies(hashes []common.Hash) error { // peer in the download tester. The returned function can be used to retrieve // batches of block receipts from the particularly requested peer. func (dlp *downloadTesterPeer) RequestReceipts(hashes []common.Hash) error { + time.Sleep(dlp.delayResponse) receipts := dlp.chain.receipts(hashes) go dlp.dl.downloader.DeliverReceipts(dlp.id, receipts) return nil @@ -428,6 +439,7 @@ func (dlp *downloadTesterPeer) RequestNodeData(hashes []common.Hash) error { dlp.dl.lock.RLock() defer dlp.dl.lock.RUnlock() + time.Sleep(dlp.delayResponse) results := make([][]byte, 0, len(hashes)) for _, hash := range hashes { if data, err := dlp.dl.peerDb.Get(hash.Bytes()); err == nil { @@ -1656,3 +1668,44 @@ func testCheckpointEnforcement(t *testing.T, protocol int, mode SyncMode) { assertOwnChain(t, tester, chain.len()) } } + +func TestCancelMasterPeer62(t *testing.T) { testCancelMasterPeer(t, 62, FullSync) } +func TestCancelMasterPeer63Full(t *testing.T) { testCancelMasterPeer(t, 63, FullSync) } +func TestCancelMasterPeer63Fast(t *testing.T) { testCancelMasterPeer(t, 63, FastSync) } +func TestCancelMasterPeer64Full(t *testing.T) { testCancelMasterPeer(t, 64, FullSync) } +func TestCancelMasterPeer64Fast(t *testing.T) { testCancelMasterPeer(t, 64, FastSync) } +func TestCancelMasterPeer64Light(t *testing.T) { testCancelMasterPeer(t, 64, LightSync) } + +func testCancelMasterPeer(t *testing.T, protocol int, mode SyncMode) { + t.Parallel() + + // Create a new tester + tester := newTester() + defer tester.terminate() + + // Attempt to sync with the two peers + chain1 := testChainBase.shorten(MaxHeaderFetch + 64) + tester.newPeer("peer1", protocol, chain1) // Mark peer1 as the master peer + + // Add some response delay so that we have enough time to unregister master + // peer before sync finished. + tester.setDelay("peer1", time.Duration(time.Millisecond*300)) + chain2 := testChainBase.shorten(MaxHeaderFetch) + tester.newPeer("peer2", protocol, chain2) // Mark peer2 as the auxiliary peer + + var errCh = make(chan error, 1) + go func() { + errCh <- tester.sync("peer1", nil, mode) + }() + time.Sleep(time.Millisecond * 300) // Ensure we have started the syncing + tester.downloader.UnregisterPeer("peer1") // Unregister the master peer, which should abort sync + + select { + case err := <-errCh: + if err != errCanceled { + t.Fatalf("error mismatch, want %v, got %v", errCanceled, err) + } + case <-time.NewTimer(time.Second).C: + t.Fatalf("timeout") + } +} diff --git a/eth/downloader/statesync.go b/eth/downloader/statesync.go index b422557d58..96cd8167b2 100644 --- a/eth/downloader/statesync.go +++ b/eth/downloader/statesync.go @@ -316,17 +316,10 @@ func (s *stateSync) loop() (err error) { // Timeouts can occur if e.g. compaction hits at the wrong time, and can be ignored req.peer.log.Warn("Downloader wants to drop peer, but peerdrop-function is not set", "peer", req.peer.id) } else { + // In dropPeer function, a callback will be called which aborts + // the sync immediately. Here return the timeout error explicitly. s.d.dropPeer(req.peer.id) - - // If this peer was the master peer, abort sync immediately - s.d.cancelLock.RLock() - master := req.peer.id == s.d.cancelPeer - s.d.cancelLock.RUnlock() - - if master { - s.d.cancel() - return errTimeout - } + return errTimeout } } // Process all the received blobs and check for stale delivery