mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-19 18:32:23 +00:00
eth/downloader: remove more stuff
This commit is contained in:
Gary Rong
Jared Wasinger
parent
1970e4c524
commit
8914a38d5c
6 changed files with 15 additions and 246 deletions
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
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
|
||||
}
|
||||
Loading…
Reference in a new issue