From 8914a38d5cd58c6681ca4013162cef206db1f18a Mon Sep 17 00:00:00 2001 From: Gary Rong Date: Tue, 23 Apr 2024 15:58:39 +0800 Subject: [PATCH] eth/downloader: remove more stuff --- eth/downloader/beaconsync.go | 2 +- eth/downloader/downloader.go | 99 +++---------------- eth/downloader/fetchers.go | 45 --------- eth/downloader/fetchers_concurrent.go | 17 +--- eth/downloader/fetchers_concurrent_bodies.go | 1 - eth/downloader/fetchers_concurrent_headers.go | 97 ------------------ 6 files changed, 15 insertions(+), 246 deletions(-) delete mode 100644 eth/downloader/fetchers_concurrent_headers.go diff --git a/eth/downloader/beaconsync.go b/eth/downloader/beaconsync.go index 6dc5351b0a..8088f16af9 100644 --- a/eth/downloader/beaconsync.go +++ b/eth/downloader/beaconsync.go @@ -106,7 +106,7 @@ func (b *beaconBackfiller) resume() { }() // If the downloader fails, report an error as in beacon chain mode there // should be no errors as long as the chain we're syncing to is valid. - if err := b.downloader.synchronise("", common.Hash{}, nil, nil, mode, b.started); err != nil { + if err := b.downloader.synchronise(mode, b.started); err != nil { log.Error("Beacon backfilling failed", "err", err) return } diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index bd0e68c8f8..5b8a4bb25c 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -88,9 +88,8 @@ type Downloader struct { mode atomic.Uint32 // Synchronisation mode defining the strategy used (per sync cycle), use d.getMode() to get the SyncMode mux *event.TypeMux // Event multiplexer to announce sync operation events - genesis uint64 // Genesis block number to limit sync to (e.g. light client CHT) - queue *queue // Scheduler for selecting the hashes to download - peers *peerSet // Set of active peers from which download can proceed + queue *queue // Scheduler for selecting the hashes to download + peers *peerSet // Set of active peers from which download can proceed stateDB ethdb.Database // Database to state sync into (and deduplicate via) @@ -107,11 +106,10 @@ type Downloader struct { badBlock badBlockFn // Reports a block as rejected by the chain // Status - synchroniseMock func(id string, hash common.Hash) error // Replacement for synchronise during testing - synchronising atomic.Bool - notified atomic.Bool - committed atomic.Bool - ancientLimit uint64 // The maximum block number which can be regarded as ancient data. + synchronising atomic.Bool + notified atomic.Bool + committed atomic.Bool + ancientLimit uint64 // The maximum block number which can be regarded as ancient data. // Channels headerProcCh chan *headerTask // Channel to feed the header processor new tasks @@ -127,7 +125,6 @@ type Downloader struct { stateSyncStart chan *stateSync // Cancellation and termination - cancelPeer string // Identifier of the peer currently being used as the master (cancel on drop) cancelCh chan struct{} // Channel to cancel mid-flight syncs cancelLock sync.RWMutex // Lock to protect the cancel channel and peer in delivers cancelWg sync.WaitGroup // Make sure all fetcher goroutines have exited. @@ -136,7 +133,6 @@ type Downloader struct { quitLock sync.Mutex // Lock to prevent double closes // Testing hooks - syncInitHook func(uint64, uint64) // Method to call upon initiating a new sync run bodyFetchHook func([]*types.Header) // Method to call upon starting a block body fetch receiptFetchHook func([]*types.Header) // Method to call upon starting a receipt fetch chainInsertHook func([]*fetchResult) // Method to call upon inserting a chain of blocks (possibly in multiple invocations) @@ -318,7 +314,7 @@ func (d *Downloader) UnregisterPeer(id string) error { // synchronise will select the peer and use it for synchronising. If an empty string is given // it will use the best peer possible and synchronize if its TD is higher than our own. If any of the // checks fail an error will be returned. This method is synchronous -func (d *Downloader) synchronise(id string, hash common.Hash, td, ttd *big.Int, mode SyncMode, beaconPing chan struct{}) error { +func (d *Downloader) synchronise(mode SyncMode, beaconPing chan struct{}) error { // The beacon header syncer is async. It will start this synchronization and // will continue doing other tasks. However, if synchronization needs to be // cancelled, the syncer needs to know if we reached the startup point (and @@ -333,10 +329,6 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td, ttd *big.Int, } }() } - // Mock out the synchronisation if testing - if d.synchroniseMock != nil { - return d.synchroniseMock(id, hash) - } // Make sure only one goroutine is ever allowed past this point at once if !d.synchronising.CompareAndSwap(false, true) { return errBusy @@ -384,7 +376,6 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td, ttd *big.Int, // Create cancel channel for aborting mid-flight and mark the master peer d.cancelLock.Lock() d.cancelCh = make(chan struct{}) - d.cancelPeer = id d.cancelLock.Unlock() defer d.Cancel() // No matter what, we can't leave the cancel channel open @@ -395,7 +386,7 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td, ttd *big.Int, if beaconPing != nil { close(beaconPing) } - return d.syncToHead(hash) + return d.syncToHead() } func (d *Downloader) getMode() SyncMode { @@ -404,7 +395,7 @@ func (d *Downloader) getMode() SyncMode { // syncToHead starts a block synchronization based on the hash chain from // the specified head hash. -func (d *Downloader) syncToHead(hash common.Hash) (err error) { +func (d *Downloader) syncToHead() (err error) { d.mux.Post(StartEvent{}) defer func() { // reset on error @@ -537,9 +528,7 @@ func (d *Downloader) syncToHead(hash common.Hash) (err error) { } // Initiate the sync using a concurrent header and content retrieval algorithm d.queue.Prepare(origin+1, mode) - if d.syncInitHook != nil { - d.syncInitHook(origin, height) - } + // In beacon mode, headers are served by the skeleton syncer fetchers := []func() error{ func() error { return d.fetchHeaders(origin + 1) }, // Headers are always retrieved @@ -633,70 +622,6 @@ func (d *Downloader) Terminate() { d.Cancel() } -// fetchHead retrieves the head header and prior pivot block (if available) from -// a remote peer. -func (d *Downloader) fetchHead(p *peerConnection) (head *types.Header, pivot *types.Header, err error) { - p.log.Debug("Retrieving remote chain head") - mode := d.getMode() - - // Request the advertised remote head block and wait for the response - latest, _ := p.peer.Head() - fetch := 1 - if mode == SnapSync { - fetch = 2 // head + pivot headers - } - headers, hashes, err := d.fetchHeadersByHash(p, latest, fetch, fsMinFullBlocks-1, true) - if err != nil { - return nil, nil, err - } - // Make sure the peer gave us at least one and at most the requested headers - if len(headers) == 0 || len(headers) > fetch { - return nil, nil, fmt.Errorf("%w: returned headers %d != requested %d", errBadPeer, len(headers), fetch) - } - // The first header needs to be the head, validate against the request. If - // only 1 header was returned, make sure there's no pivot or there was not - // one requested. - head = headers[0] - if len(headers) == 1 { - if mode == SnapSync && head.Number.Uint64() > uint64(fsMinFullBlocks) { - return nil, nil, fmt.Errorf("%w: no pivot included along head header", errBadPeer) - } - p.log.Debug("Remote head identified, no pivot", "number", head.Number, "hash", hashes[0]) - return head, nil, nil - } - // At this point we have 2 headers in total and the first is the - // validated head of the chain. Check the pivot number and return, - pivot = headers[1] - if pivot.Number.Uint64() != head.Number.Uint64()-uint64(fsMinFullBlocks) { - return nil, nil, fmt.Errorf("%w: remote pivot %d != requested %d", errInvalidChain, pivot.Number, head.Number.Uint64()-uint64(fsMinFullBlocks)) - } - return head, pivot, nil -} - -// fillHeaderSkeleton concurrently retrieves headers from all our available peers -// and maps them to the provided skeleton header chain. -// -// Any partial results from the beginning of the skeleton is (if possible) forwarded -// immediately to the header processor to keep the rest of the pipeline full even -// in the case of header stalls. -// -// The method returns the entire filled skeleton and also the number of headers -// already forwarded for processing. -func (d *Downloader) fillHeaderSkeleton(from uint64, skeleton []*types.Header) ([]*types.Header, []common.Hash, int, error) { - log.Debug("Filling up skeleton", "from", from) - d.queue.ScheduleSkeleton(from, skeleton) - - err := d.concurrentFetch((*headerQueue)(d)) - if err != nil { - log.Debug("Skeleton fill failed", "err", err) - } - filled, hashes, proced := d.queue.RetrieveHeaders() - if err == nil { - log.Debug("Skeleton fill succeeded", "filled", len(filled), "processed", proced) - } - return filled, hashes, proced, err -} - // fetchBodies iteratively downloads the scheduled block bodies, taking any // available peers, reserving a chunk of blocks for each, waiting for delivery // and also periodically checking for timeouts. @@ -724,8 +649,8 @@ func (d *Downloader) fetchReceipts(from uint64) error { // queue until the stream ends or a failure occurs. func (d *Downloader) processHeaders(origin uint64) error { var ( - mode = d.getMode() - timer = time.NewTimer(time.Second) + mode = d.getMode() + timer = time.NewTimer(time.Second) ) defer timer.Stop() diff --git a/eth/downloader/fetchers.go b/eth/downloader/fetchers.go index cc4279b0da..4ebb9bbc98 100644 --- a/eth/downloader/fetchers.go +++ b/eth/downloader/fetchers.go @@ -68,48 +68,3 @@ func (d *Downloader) fetchHeadersByHash(p *peerConnection, hash common.Hash, amo return *res.Res.(*eth.BlockHeadersRequest), res.Meta.([]common.Hash), nil } } - -// fetchHeadersByNumber is a blocking version of Peer.RequestHeadersByNumber which -// handles all the cancellation, interruption and timeout mechanisms of a data -// retrieval to allow blocking API calls. -func (d *Downloader) fetchHeadersByNumber(p *peerConnection, number uint64, amount int, skip int, reverse bool) ([]*types.Header, []common.Hash, error) { - // Create the response sink and send the network request - start := time.Now() - resCh := make(chan *eth.Response) - - req, err := p.peer.RequestHeadersByNumber(number, amount, skip, reverse, resCh) - if err != nil { - return nil, nil, err - } - defer req.Close() - - // Wait until the response arrives, the request is cancelled or times out - ttl := d.peers.rates.TargetTimeout() - - timeoutTimer := time.NewTimer(ttl) - defer timeoutTimer.Stop() - - select { - case <-d.cancelCh: - return nil, nil, errCanceled - - case <-timeoutTimer.C: - // Header retrieval timed out, update the metrics - p.log.Debug("Header request timed out", "elapsed", ttl) - headerTimeoutMeter.Mark(1) - - return nil, nil, errTimeout - - case res := <-resCh: - // Headers successfully retrieved, update the metrics - headerReqTimer.Update(time.Since(start)) - headerInMeter.Mark(int64(len(*res.Res.(*eth.BlockHeadersRequest)))) - - // Don't reject the packet even if it turns out to be bad, downloader will - // disconnect the peer on its own terms. Simply delivery the headers to - // be processed by the caller - res.Done <- nil - - return *res.Res.(*eth.BlockHeadersRequest), res.Meta.([]common.Hash), nil - } -} diff --git a/eth/downloader/fetchers_concurrent.go b/eth/downloader/fetchers_concurrent.go index 35c30063f6..9d8cd114c1 100644 --- a/eth/downloader/fetchers_concurrent.go +++ b/eth/downloader/fetchers_concurrent.go @@ -154,17 +154,14 @@ func (d *Downloader) concurrentFetch(queue typedQueue) error { } sort.Sort(&peerCapacitySort{idles, caps}) - var ( - throttled bool - queued = queue.pending() - ) + var throttled bool for _, peer := range idles { // Short circuit if throttling activated or there are no more // queued tasks to be retrieved if throttled { break } - if queued = queue.pending(); queued == 0 { + if queued := queue.pending(); queued == 0 { break } // Reserve a chunk of fetches for a peer. A nil can mean either that @@ -302,16 +299,6 @@ func (d *Downloader) concurrentFetch(queue typedQueue) error { queue.updateCapacity(peer, 0, 0) } else { d.dropPeer(peer.id) - - // If this peer was the master peer, abort sync immediately - d.cancelLock.RLock() - master := peer.id == d.cancelPeer - d.cancelLock.RUnlock() - - if master { - d.cancel() - return errTimeout - } } case res := <-responses: diff --git a/eth/downloader/fetchers_concurrent_bodies.go b/eth/downloader/fetchers_concurrent_bodies.go index 5105fda66b..56359b33c9 100644 --- a/eth/downloader/fetchers_concurrent_bodies.go +++ b/eth/downloader/fetchers_concurrent_bodies.go @@ -78,7 +78,6 @@ func (q *bodyQueue) request(peer *peerConnection, req *fetchRequest, resCh chan if q.bodyFetchHook != nil { q.bodyFetchHook(req.Headers) } - hashes := make([]common.Hash, 0, len(req.Headers)) for _, header := range req.Headers { hashes = append(hashes, header.Hash()) diff --git a/eth/downloader/fetchers_concurrent_headers.go b/eth/downloader/fetchers_concurrent_headers.go deleted file mode 100644 index 8201f4ca74..0000000000 --- a/eth/downloader/fetchers_concurrent_headers.go +++ /dev/null @@ -1,97 +0,0 @@ -// Copyright 2021 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package downloader - -import ( - "time" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/eth/protocols/eth" - "github.com/ethereum/go-ethereum/log" -) - -// headerQueue implements typedQueue and is a type adapter between the generic -// concurrent fetcher and the downloader. -type headerQueue Downloader - -// waker returns a notification channel that gets pinged in case more header -// fetches have been queued up, so the fetcher might assign it to idle peers. -func (q *headerQueue) waker() chan bool { - return q.queue.headerContCh -} - -// pending returns the number of headers that are currently queued for fetching -// by the concurrent downloader. -func (q *headerQueue) pending() int { - return q.queue.PendingHeaders() -} - -// capacity is responsible for calculating how many headers a particular peer is -// estimated to be able to retrieve within the allotted round trip time. -func (q *headerQueue) capacity(peer *peerConnection, rtt time.Duration) int { - return peer.HeaderCapacity(rtt) -} - -// updateCapacity is responsible for updating how many headers a particular peer -// is estimated to be able to retrieve in a unit time. -func (q *headerQueue) updateCapacity(peer *peerConnection, items int, span time.Duration) { - peer.UpdateHeaderRate(items, span) -} - -// reserve is responsible for allocating a requested number of pending headers -// from the download queue to the specified peer. -func (q *headerQueue) reserve(peer *peerConnection, items int) (*fetchRequest, bool, bool) { - return q.queue.ReserveHeaders(peer, items), false, false -} - -// unreserve is responsible for removing the current header retrieval allocation -// assigned to a specific peer and placing it back into the pool to allow -// reassigning to some other peer. -func (q *headerQueue) unreserve(peer string) int { - fails := q.queue.ExpireHeaders(peer) - if fails > 2 { - log.Trace("Header delivery timed out", "peer", peer) - } else { - log.Debug("Header delivery stalling", "peer", peer) - } - return fails -} - -// request is responsible for converting a generic fetch request into a header -// one and sending it to the remote peer for fulfillment. -func (q *headerQueue) request(peer *peerConnection, req *fetchRequest, resCh chan *eth.Response) (*eth.Request, error) { - peer.log.Trace("Requesting new batch of headers", "from", req.From) - return peer.peer.RequestHeadersByNumber(req.From, MaxHeaderFetch, 0, false, resCh) -} - -// deliver is responsible for taking a generic response packet from the concurrent -// fetcher, unpacking the header data and delivering it to the downloader's queue. -func (q *headerQueue) deliver(peer *peerConnection, packet *eth.Response) (int, error) { - headers := *packet.Res.(*eth.BlockHeadersRequest) - hashes := packet.Meta.([]common.Hash) - - accepted, err := q.queue.DeliverHeaders(peer.id, headers, hashes, q.headerProcCh) - switch { - case err == nil && len(headers) == 0: - peer.log.Trace("Requested headers delivered") - case err == nil: - peer.log.Trace("Delivered new batch of headers", "count", len(headers), "accepted", accepted) - default: - peer.log.Debug("Failed to deliver retrieved headers", "err", err) - } - return accepted, err -}