forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

eth/fetcher, les/fetcher: reuse eth/fetcher as light fetcher

Browse source
This commit is contained in:
rjl493456442 2019-06-04 14:55:17 +08:00
parent f4c3c13ed5
commit 7c4fb12f6f
16 changed files with 1230 additions and 979 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
consensus/clique/clique.go
View file

@ -213,14 +213,14 @@ func (c *Clique) Author(header *types.Header) (common.Address, error) {
}
// VerifyHeader checks whether a header conforms to the consensus rules.
func (c *Clique) VerifyHeader(chain consensus.ChainReader, header *types.Header, seal bool) error {
func (c *Clique) VerifyHeader(chain consensus.ChainHeaderReader, header *types.Header, seal bool) error {
return c.verifyHeader(chain, header, nil)
}
// VerifyHeaders is similar to VerifyHeader, but verifies a batch of headers. The
// method returns a quit channel to abort the operations and a results channel to
// retrieve the async verifications (the order is that of the input slice).
func (c *Clique) VerifyHeaders(chain consensus.ChainReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) {
func (c *Clique) VerifyHeaders(chain consensus.ChainHeaderReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) {
abort := make(chan struct{})
results := make(chan error, len(headers))
@ -242,7 +242,7 @@ func (c *Clique) VerifyHeaders(chain consensus.ChainReader, headers []*types.Hea
// caller may optionally pass in a batch of parents (ascending order) to avoid
// looking those up from the database. This is useful for concurrently verifying
// a batch of new headers.
func (c *Clique) verifyHeader(chain consensus.ChainReader, header *types.Header, parents []*types.Header) error {
func (c *Clique) verifyHeader(chain consensus.ChainHeaderReader, header *types.Header, parents []*types.Header) error {
if header.Number == nil {
return errUnknownBlock
}
@ -305,7 +305,7 @@ func (c *Clique) verifyHeader(chain consensus.ChainReader, header *types.Header,
// rather depend on a batch of previous headers. The caller may optionally pass
// in a batch of parents (ascending order) to avoid looking those up from the
// database. This is useful for concurrently verifying a batch of new headers.
func (c *Clique) verifyCascadingFields(chain consensus.ChainReader, header *types.Header, parents []*types.Header) error {
func (c *Clique) verifyCascadingFields(chain consensus.ChainHeaderReader, header *types.Header, parents []*types.Header) error {
// The genesis block is the always valid dead-end
number := header.Number.Uint64()
if number == 0 {
@ -345,7 +345,7 @@ func (c *Clique) verifyCascadingFields(chain consensus.ChainReader, header *type
}
// snapshot retrieves the authorization snapshot at a given point in time.
func (c *Clique) snapshot(chain consensus.ChainReader, number uint64, hash common.Hash, parents []*types.Header) (*Snapshot, error) {
func (c *Clique) snapshot(chain consensus.ChainHeaderReader, number uint64, hash common.Hash, parents []*types.Header) (*Snapshot, error) {
// Search for a snapshot in memory or on disk for checkpoints
var (
headers []*types.Header
@ -436,7 +436,7 @@ func (c *Clique) VerifyUncles(chain consensus.ChainReader, block *types.Block) e
// VerifySeal implements consensus.Engine, checking whether the signature contained
// in the header satisfies the consensus protocol requirements.
func (c *Clique) VerifySeal(chain consensus.ChainReader, header *types.Header) error {
func (c *Clique) VerifySeal(chain consensus.ChainHeaderReader, header *types.Header) error {
return c.verifySeal(chain, header, nil)
}
@ -444,7 +444,7 @@ func (c *Clique) VerifySeal(chain consensus.ChainReader, header *types.Header) e
// consensus protocol requirements. The method accepts an optional list of parent
// headers that aren't yet part of the local blockchain to generate the snapshots
// from.
func (c *Clique) verifySeal(chain consensus.ChainReader, header *types.Header, parents []*types.Header) error {
func (c *Clique) verifySeal(chain consensus.ChainHeaderReader, header *types.Header, parents []*types.Header) error {
// Verifying the genesis block is not supported
number := header.Number.Uint64()
if number == 0 {
@ -654,7 +654,7 @@ func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, results c
// CalcDifficulty is the difficulty adjustment algorithm. It returns the difficulty
// that a new block should have based on the previous blocks in the chain and the
// current signer.
func (c *Clique) CalcDifficulty(chain consensus.ChainReader, time uint64, parent *types.Header) *big.Int {
func (c *Clique) CalcDifficulty(chain consensus.ChainHeaderReader, time uint64, parent *types.Header) *big.Int {
snap, err := c.snapshot(chain, parent.Number.Uint64(), parent.Hash(), nil)
if err != nil {
return nil

20
consensus/consensus.go
View file

@ -27,9 +27,9 @@ import (
"github.com/ethereum/go-ethereum/rpc"
)
// ChainReader defines a small collection of methods needed to access the local
// blockchain during header and/or uncle verification.
type ChainReader interface {
// ChainHeaderReader defines a small collection of methods needed to access the local
// blockchain during header verification.
type ChainHeaderReader interface {
// Config retrieves the blockchain's chain configuration.
Config() *params.ChainConfig
@ -44,6 +44,12 @@ type ChainReader interface {
// GetHeaderByHash retrieves a block header from the database by its hash.
GetHeaderByHash(hash common.Hash) *types.Header
}
// ChainReader defines a small collection of methods needed to access the local
// blockchain during header and/or uncle verification.
type ChainReader interface {
ChainHeaderReader
// GetBlock retrieves a block from the database by hash and number.
GetBlock(hash common.Hash, number uint64) *types.Block
@ -59,13 +65,13 @@ type Engine interface {
// VerifyHeader checks whether a header conforms to the consensus rules of a
// given engine. Verifying the seal may be done optionally here, or explicitly
// via the VerifySeal method.
VerifyHeader(chain ChainReader, header *types.Header, seal bool) error
VerifyHeader(chain ChainHeaderReader, header *types.Header, seal bool) error
// VerifyHeaders is similar to VerifyHeader, but verifies a batch of headers
// concurrently. The method returns a quit channel to abort the operations and
// a results channel to retrieve the async verifications (the order is that of
// the input slice).
VerifyHeaders(chain ChainReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error)
VerifyHeaders(chain ChainHeaderReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error)
// VerifyUncles verifies that the given block's uncles conform to the consensus
// rules of a given engine.
@ -73,7 +79,7 @@ type Engine interface {
// VerifySeal checks whether the crypto seal on a header is valid according to
// the consensus rules of the given engine.
VerifySeal(chain ChainReader, header *types.Header) error
VerifySeal(chain ChainHeaderReader, header *types.Header) error
// Prepare initializes the consensus fields of a block header according to the
// rules of a particular engine. The changes are executed inline.
@ -107,7 +113,7 @@ type Engine interface {
// CalcDifficulty is the difficulty adjustment algorithm. It returns the difficulty
// that a new block should have.
CalcDifficulty(chain ChainReader, time uint64, parent *types.Header) *big.Int
CalcDifficulty(chain ChainHeaderReader, time uint64, parent *types.Header) *big.Int
// APIs returns the RPC APIs this consensus engine provides.
APIs(chain ChainReader) []rpc.API

14
consensus/ethash/consensus.go
View file

@ -81,7 +81,7 @@ func (ethash *Ethash) Author(header *types.Header) (common.Address, error) {
// VerifyHeader checks whether a header conforms to the consensus rules of the
// stock Ethereum ethash engine.
func (ethash *Ethash) VerifyHeader(chain consensus.ChainReader, header *types.Header, seal bool) error {
func (ethash *Ethash) VerifyHeader(chain consensus.ChainHeaderReader, header *types.Header, seal bool) error {
// If we're running a full engine faking, accept any input as valid
if ethash.config.PowMode == ModeFullFake {
return nil
@ -102,7 +102,7 @@ func (ethash *Ethash) VerifyHeader(chain consensus.ChainReader, header *types.He
// VerifyHeaders is similar to VerifyHeader, but verifies a batch of headers
// concurrently. The method returns a quit channel to abort the operations and
// a results channel to retrieve the async verifications.
func (ethash *Ethash) VerifyHeaders(chain consensus.ChainReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) {
func (ethash *Ethash) VerifyHeaders(chain consensus.ChainHeaderReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) {
// If we're running a full engine faking, accept any input as valid
if ethash.config.PowMode == ModeFullFake || len(headers) == 0 {
abort, results := make(chan struct{}), make(chan error, len(headers))
@ -164,7 +164,7 @@ func (ethash *Ethash) VerifyHeaders(chain consensus.ChainReader, headers []*type
return abort, errorsOut
}
func (ethash *Ethash) verifyHeaderWorker(chain consensus.ChainReader, headers []*types.Header, seals []bool, index int) error {
func (ethash *Ethash) verifyHeaderWorker(chain consensus.ChainHeaderReader, headers []*types.Header, seals []bool, index int) error {
var parent *types.Header
if index == 0 {
parent = chain.GetHeader(headers[0].ParentHash, headers[0].Number.Uint64()-1)
@ -238,7 +238,7 @@ func (ethash *Ethash) VerifyUncles(chain consensus.ChainReader, block *types.Blo
// verifyHeader checks whether a header conforms to the consensus rules of the
// stock Ethereum ethash engine.
// See YP section 4.3.4. "Block Header Validity"
func (ethash *Ethash) verifyHeader(chain consensus.ChainReader, header, parent *types.Header, uncle bool, seal bool) error {
func (ethash *Ethash) verifyHeader(chain consensus.ChainHeaderReader, header, parent *types.Header, uncle bool, seal bool) error {
// Ensure that the header's extra-data section is of a reasonable size
if uint64(len(header.Extra)) > params.MaximumExtraDataSize {
return fmt.Errorf("extra-data too long: %d > %d", len(header.Extra), params.MaximumExtraDataSize)
@ -301,7 +301,7 @@ func (ethash *Ethash) verifyHeader(chain consensus.ChainReader, header, parent *
// CalcDifficulty is the difficulty adjustment algorithm. It returns
// the difficulty that a new block should have when created at time
// given the parent block's time and difficulty.
func (ethash *Ethash) CalcDifficulty(chain consensus.ChainReader, time uint64, parent *types.Header) *big.Int {
func (ethash *Ethash) CalcDifficulty(chain consensus.ChainHeaderReader, time uint64, parent *types.Header) *big.Int {
return CalcDifficulty(chain.Config(), time, parent)
}
@ -479,14 +479,14 @@ func calcDifficultyFrontier(time uint64, parent *types.Header) *big.Int {
// VerifySeal implements consensus.Engine, checking whether the given block satisfies
// the PoW difficulty requirements.
func (ethash *Ethash) VerifySeal(chain consensus.ChainReader, header *types.Header) error {
func (ethash *Ethash) VerifySeal(chain consensus.ChainHeaderReader, header *types.Header) error {
return ethash.verifySeal(chain, header, false)
}
// verifySeal checks whether a block satisfies the PoW difficulty requirements,
// either using the usual ethash cache for it, or alternatively using a full DAG
// to make remote mining fast.
func (ethash *Ethash) verifySeal(chain consensus.ChainReader, header *types.Header, fulldag bool) error {
func (ethash *Ethash) verifySeal(chain consensus.ChainHeaderReader, header *types.Header, fulldag bool) error {
// If we're running a fake PoW, accept any seal as valid
if ethash.config.PowMode == ModeFake || ethash.config.PowMode == ModeFullFake {
time.Sleep(ethash.fakeDelay)

225
eth/fetcher/fetcher.go
View file

@ -30,18 +30,21 @@ import (
)
const (
lightTimeout = time.Millisecond // Time allowance before an announced header is explicitly requested
arriveTimeout = 500 * time.Millisecond // Time allowance before an announced block is explicitly requested
gatherSlack = 100 * time.Millisecond // Interval used to collate almost-expired announces with fetches
fetchTimeout = 5 * time.Second // Maximum allotted time to return an explicitly requested block
maxUncleDist = 7 // Maximum allowed backward distance from the chain head
maxQueueDist = 32 // Maximum allowed distance from the chain head to queue
hashLimit = 256 // Maximum number of unique blocks a peer may have announced
hashLimit = 256 // Maximum number of unique blocks or headers a peer may have announced
blockLimit = 64 // Maximum number of unique blocks a peer may have delivered
HeaderLimit = 64 // Maximum number of unique headers a peer may have delivered
)
var (
errTerminated = errors.New("terminated")
)
var errTerminated = errors.New("terminated")
// HeaderRetrievalFn is a callback type for retrieving a header from the local chain.
type HeaderRetrievalFn func(common.Hash) *types.Header
// blockRetrievalFn is a callback type for retrieving a block from the local chain.
type blockRetrievalFn func(common.Hash) *types.Block
@ -61,6 +64,9 @@ type blockBroadcasterFn func(block *types.Block, propagate bool)
// chainHeightFn is a callback type to retrieve the current chain height.
type chainHeightFn func() uint64
// headersInsertFn is a callback type to insert a batch of headers into the local chain.
type headersInsertFn func(headers []*types.Header) (int, error)
// chainInsertFn is a callback type to insert a batch of blocks into the local chain.
type chainInsertFn func(types.Blocks) (int, error)
@ -100,12 +106,32 @@ type bodyFilterTask struct {
// inject represents a schedules import operation.
type inject struct {
origin string
block *types.Block
header *types.Header // Used for light mode fetcher which only cares about header.
block *types.Block // Used for normal mode fetcher which imports full block.
}
// number returns the block number of the injected object.
func (inject *inject) number() uint64 {
if inject.header != nil {
return inject.header.Number.Uint64()
}
return inject.block.NumberU64()
}
// number returns the block hash of the injected object.
func (inject *inject) hash() common.Hash {
if inject.header != nil {
return inject.header.Hash()
}
return inject.block.Hash()
}
// Fetcher is responsible for accumulating block announcements from various peers
// and scheduling them for retrieval.
type Fetcher struct {
light bool // The indicator whether it's a light fetcher or not.
// Various event channels
notify chan *announce
inject chan *inject
@ -126,27 +152,30 @@ type Fetcher struct {
// Block cache
queue *prque.Prque // Queue containing the import operations (block number sorted)
queues map[string]int // Per peer block counts to prevent memory exhaustion
queued map[common.Hash]*inject // Set of already queued blocks (to dedupe imports)
queued map[common.Hash]*inject // Set of already queued blocks (to dedup imports)
// Callbacks
getHeader HeaderRetrievalFn // Retrieves a header from the local chain
getBlock blockRetrievalFn // Retrieves a block from the local chain
verifyHeader headerVerifierFn // Checks if a block's headers have a valid proof of work
broadcastBlock blockBroadcasterFn // Broadcasts a block to connected peers
chainHeight chainHeightFn // Retrieves the current chain's height
insertHeaders headersInsertFn // Injects a batch of headers into the chain
insertChain chainInsertFn // Injects a batch of blocks into the chain
dropPeer peerDropFn // Drops a peer for misbehaving
// Testing hooks
announceChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a hash from the announce list
queueChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a block from the import queue
fetchingHook func([]common.Hash) // Method to call upon starting a block (eth/61) or header (eth/62) fetch
completingHook func([]common.Hash) // Method to call upon starting a block body fetch (eth/62)
importedHook func(*types.Block) // Method to call upon successful block import (both eth/61 and eth/62)
announceChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a hash from the announce list
queueChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a block from the import queue
fetchingHook func([]common.Hash) // Method to call upon starting a block (eth/61) or header (eth/62) fetch
completingHook func([]common.Hash) // Method to call upon starting a block body fetch (eth/62)
importedHook func(*types.Header, *types.Block) // Method to call upon successful header or block import (both eth/61 and eth/62)
}
// New creates a block fetcher to retrieve blocks based on hash announcements.
func New(getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher {
func New(light bool, getHeader HeaderRetrievalFn, getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertHeaders headersInsertFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher {
return &Fetcher{
light: light,
notify: make(chan *announce),
inject: make(chan *inject),
headerFilter: make(chan chan *headerFilterTask),
@ -161,10 +190,12 @@ func New(getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBloc
queue: prque.New(nil),
queues: make(map[string]int),
queued: make(map[common.Hash]*inject),
getHeader: getHeader,
getBlock: getBlock,
verifyHeader: verifyHeader,
broadcastBlock: broadcastBlock,
chainHeight: chainHeight,
insertHeaders: insertHeaders,
insertChain: insertChain,
dropPeer: dropPeer,
}
@ -202,8 +233,8 @@ func (f *Fetcher) Notify(peer string, hash common.Hash, number uint64, time time
}
}
// Enqueue tries to fill gaps the fetcher's future import queue.
func (f *Fetcher) Enqueue(peer string, block *types.Block) error {
// EnqueueBlock tries to fill gaps the fetcher's future import queue.
func (f *Fetcher) EnqueueBlock(peer string, block *types.Block) error {
op := &inject{
origin: peer,
block: block,
@ -290,12 +321,12 @@ func (f *Fetcher) loop() {
height := f.chainHeight()
for !f.queue.Empty() {
op := f.queue.PopItem().(*inject)
hash := op.block.Hash()
hash := op.hash()
if f.queueChangeHook != nil {
f.queueChangeHook(hash, false)
}
// If too high up the chain or phase, continue later
number := op.block.NumberU64()
number := op.number()
if number > height+1 {
f.queue.Push(op, -int64(number))
if f.queueChangeHook != nil {
@ -304,11 +335,15 @@ func (f *Fetcher) loop() {
break
}
// Otherwise if fresh and still unknown, try and import
if number+maxUncleDist < height || f.getBlock(hash) != nil {
if number+maxUncleDist < height || f.light && f.getHeader(hash) != nil || !f.light && f.getBlock(hash) != nil {
f.forgetBlock(hash)
continue
}
f.insert(op.origin, op.block)
if f.light {
f.importHeaders(op.origin, op.header)
} else {
f.importBlocks(op.origin, op.block)
}
}
// Wait for an outside event to occur
select {
@ -353,7 +388,10 @@ func (f *Fetcher) loop() {
case op := <-f.inject:
// A direct block insertion was requested, try and fill any pending gaps
propBroadcastInMeter.Mark(1)
f.enqueue(op.origin, op.block)
if !f.light {
f.enqueueBlock(op.origin, op.block)
}
case hash := <-f.done:
// A pending import finished, remove all traces of the notification
@ -365,13 +403,18 @@ func (f *Fetcher) loop() {
request := make(map[string][]common.Hash)
for hash, announces := range f.announced {
if time.Since(announces[0].time) > arriveTimeout-gatherSlack {
timeout := arriveTimeout - gatherSlack
if f.light {
timeout = 0
}
if time.Since(announces[0].time) > timeout {
// Pick a random peer to retrieve from, reset all others
announce := announces[rand.Intn(len(announces))]
f.forgetHash(hash)
// If the block still didn't arrive, queue for fetching
if f.getBlock(hash) == nil {
// If the block still didn't arrive or it's a light fetcher,
// queue for fetching.
if f.light && f.getHeader(hash) == nil || !f.light && f.getBlock(hash) == nil {
request[announce.origin] = append(request[announce.origin], hash)
f.fetching[hash] = announce
}
@ -394,7 +437,9 @@ func (f *Fetcher) loop() {
}()
}
// Schedule the next fetch if blocks are still pending
f.rescheduleFetch(fetchTimer)
if len(f.announced) >= 1 {
f.rescheduleFetch(fetchTimer)
}
case <-completeTimer.C:
// At least one header's timer ran out, retrieve everything
@ -423,7 +468,9 @@ func (f *Fetcher) loop() {
go f.completing[hashes[0]].fetchBodies(hashes)
}
// Schedule the next fetch if blocks are still pending
f.rescheduleComplete(completeTimer)
if len(f.fetched) >= 1 {
f.rescheduleComplete(completeTimer)
}
case filter := <-f.headerFilter:
// Headers arrived from a remote peer. Extract those that were explicitly
@ -439,7 +486,7 @@ func (f *Fetcher) loop() {
// Split the batch of headers into unknown ones (to return to the caller),
// known incomplete ones (requiring body retrievals) and completed blocks.
unknown, incomplete, complete := []*types.Header{}, []*announce{}, []*types.Block{}
unknown, incomplete, complete, lightHeaders := []*types.Header{}, []*announce{}, []*types.Block{}, []*announce{}
for _, header := range task.headers {
hash := header.Hash()
@ -452,6 +499,16 @@ func (f *Fetcher) loop() {
f.forgetHash(hash)
continue
}
// Collect all headers only if we are running in light
// mode and the headers are not imported by other means.
if f.light {
if f.getHeader(hash) == nil {
announce.header = header
lightHeaders = append(lightHeaders, announce)
f.forgetHash(hash)
}
continue
}
// Only keep if not imported by other means
if f.getBlock(hash) == nil {
announce.header = header
@ -496,10 +553,14 @@ func (f *Fetcher) loop() {
f.rescheduleComplete(completeTimer)
}
}
// Schedule the header for light fetcher import
for _, announce := range lightHeaders {
f.enqueueHeader(announce.origin, announce.header)
}
// Schedule the header-only blocks for import
for _, block := range complete {
if announce := f.completing[block.Hash()]; announce != nil {
f.enqueue(announce.origin, block)
f.enqueueBlock(announce.origin, block)
}
}
@ -555,7 +616,7 @@ func (f *Fetcher) loop() {
// Schedule the retrieved blocks for ordered import
for _, block := range blocks {
if announce := f.completing[block.Hash()]; announce != nil {
f.enqueue(announce.origin, block)
f.enqueueBlock(announce.origin, block)
}
}
}
@ -568,6 +629,12 @@ func (f *Fetcher) rescheduleFetch(fetch *time.Timer) {
if len(f.announced) == 0 {
return
}
// Schedule announcement retrieval quickly for light mode
// since server won't send any headers to client.
if f.light {
fetch.Reset(lightTimeout)
return
}
// Otherwise find the earliest expiring announcement
earliest := time.Now()
for _, announces := range f.announced {
@ -594,9 +661,44 @@ func (f *Fetcher) rescheduleComplete(complete *time.Timer) {
complete.Reset(gatherSlack - time.Since(earliest))
}
// enqueue schedules a new future import operation, if the block to be imported
// enqueueHeader schedules a new header import operation, if the header to be imported
// has not yet been seen.
func (f *Fetcher) enqueue(peer string, block *types.Block) {
func (f *Fetcher) enqueueHeader(peer string, header *types.Header) {
hash := header.Hash()
// Ensure the peer isn't DOSing us
count := f.queues[peer] + 1
if count > HeaderLimit {
log.Debug("Discarded propagated header, exceeded allowance", "peer", peer, "number", header.Number, "hash", hash, "limit", blockLimit)
propBroadcastDOSMeter.Mark(1)
f.forgetHash(hash)
return
}
// Discard any past or too distant blocks
if dist := int64(header.Number.Uint64()) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist {
log.Debug("Discarded propagated header, too far away", "peer", peer, "number", header.Number.Uint64(), "hash", hash, "distance", dist)
propBroadcastDropMeter.Mark(1)
f.forgetHash(hash)
return
}
// Schedule the block for future importing
if _, ok := f.queued[hash]; !ok {
op := &inject{
origin: peer,
header: header,
}
f.queues[peer] = count
f.queued[hash] = op
f.queue.Push(op, -int64(header.Number.Uint64()))
if f.queueChangeHook != nil {
f.queueChangeHook(op.block.Hash(), true)
}
log.Debug("Queued propagated header", "peer", peer, "number", header.Number.Uint64(), "hash", hash, "queued", f.queue.Size())
}
}
// enqueueBlock schedules a new future import operation, if the block to be imported
// has not yet been seen.
func (f *Fetcher) enqueueBlock(peer string, block *types.Block) {
hash := block.Hash()
// Ensure the peer isn't DOSing us
@ -630,17 +732,52 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) {
}
}
// insert spawns a new goroutine to run a block insertion into the chain. If the
// block's number is at the same height as the current import phase, it updates
// the phase states accordingly.
func (f *Fetcher) insert(peer string, block *types.Block) {
hash := block.Hash()
func (f *Fetcher) importHeaders(peer string, header *types.Header) {
hash := header.Hash()
log.Debug("Importing propagated header", "peer", peer, "number", header.Number, "hash", hash)
// Run the import on a new thread
log.Debug("Importing propagated block", "peer", peer, "number", block.Number(), "hash", hash)
go func() {
defer func() { f.done <- hash }()
// If the parent's unknown, abort insertion
parent := f.getHeader(header.ParentHash)
if parent == nil {
log.Debug("Unknown parent of propagated header", "peer", peer, "number", header.Number, "hash", hash, "parent", header.ParentHash)
return
}
// Quickly validate the header and propagate the block if it passes
switch err := f.verifyHeader(header); err {
case nil:
case consensus.ErrFutureBlock:
// Weird future block, don't fail, but neither propagate
default:
// Something went very wrong, drop the peer
log.Debug("Propagated header verification failed", "peer", peer, "number", header.Number, "hash", hash, "err", err)
f.dropPeer(peer)
return
}
// Run the actual import and log any issues
if _, err := f.insertHeaders([]*types.Header{header}); err != nil {
log.Debug("Propagated header import failed", "peer", peer, "number", header.Number, "hash", hash, "err", err)
return
}
// Invoke the testing hook if needed
if f.importedHook != nil {
f.importedHook(header, nil)
}
}()
}
// importBlocks spawns a new goroutine to run a block insertion into the chain. If the
// block's number is at the same height as the current import phase, it updates
// the phase states accordingly.
func (f *Fetcher) importBlocks(peer string, block *types.Block) {
hash := block.Hash()
log.Debug("Importing propagated block", "peer", peer, "number", block.Number(), "hash", hash)
go func() {
defer func() { f.done <- hash }()
// If the parent's unknown, abort insertion
parent := f.getBlock(block.ParentHash())
if parent == nil {
@ -674,11 +811,25 @@ func (f *Fetcher) insert(peer string, block *types.Block) {
// Invoke the testing hook if needed
if f.importedHook != nil {
f.importedHook(block)
f.importedHook(nil, block)
}
}()
}
func (f *Fetcher) hasHash(hash common.Hash) {
if _, ok := f.fetching[hash]; ok {
}
if _, ok := f.completing[hash]; ok {
}
if _, ok := f.queued[hash]; ok {
}
if f.light {
} else {
}
}
// forgetHash removes all traces of a block announcement from the fetcher's
// internal state.
func (f *Fetcher) forgetHash(hash common.Hash) {

322
eth/fetcher/fetcher_test.go
View file

@ -78,26 +78,36 @@ func makeChain(n int, seed byte, parent *types.Block) ([]common.Hash, map[common
type fetcherTester struct {
fetcher *Fetcher
hashes []common.Hash // Hash chain belonging to the tester
blocks map[common.Hash]*types.Block // Blocks belonging to the tester
drops map[string]bool // Map of peers dropped by the fetcher
hashes []common.Hash // Hash chain belonging to the tester
headers map[common.Hash]*types.Header // Headers belonging to the tester
blocks map[common.Hash]*types.Block // Blocks belonging to the tester
drops map[string]bool // Map of peers dropped by the fetcher
lock sync.RWMutex
}
// newTester creates a new fetcher test mocker.
func newTester() *fetcherTester {
func newTester(light bool) *fetcherTester {
tester := &fetcherTester{
hashes: []common.Hash{genesis.Hash()},
blocks: map[common.Hash]*types.Block{genesis.Hash(): genesis},
drops: make(map[string]bool),
hashes: []common.Hash{genesis.Hash()},
headers: map[common.Hash]*types.Header{genesis.Hash(): genesis.Header()},
blocks: map[common.Hash]*types.Block{genesis.Hash(): genesis},
drops: make(map[string]bool),
}
tester.fetcher = New(tester.getBlock, tester.verifyHeader, tester.broadcastBlock, tester.chainHeight, tester.insertChain, tester.dropPeer)
tester.fetcher = New(light, tester.getHeader, tester.getBlock, tester.verifyHeader, tester.broadcastBlock, tester.chainHeight, tester.insertHeaders, tester.insertChain, tester.dropPeer)
tester.fetcher.Start()
return tester
}
// getHeader retrieves a header from the tester's block chain.
func (f *fetcherTester) getHeader(hash common.Hash) *types.Header {
f.lock.RLock()
defer f.lock.RUnlock()
return f.headers[hash]
}
// getBlock retrieves a block from the tester's block chain.
func (f *fetcherTester) getBlock(hash common.Hash) *types.Block {
f.lock.RLock()
@ -120,9 +130,33 @@ func (f *fetcherTester) chainHeight() uint64 {
f.lock.RLock()
defer f.lock.RUnlock()
if f.fetcher.light {
return f.headers[f.hashes[len(f.hashes)-1]].Number.Uint64()
}
return f.blocks[f.hashes[len(f.hashes)-1]].NumberU64()
}
// insertChain injects a new headers into the simulated chain.
func (f *fetcherTester) insertHeaders(headers []*types.Header) (int, error) {
f.lock.Lock()
defer f.lock.Unlock()
for i, header := range headers {
// Make sure the parent in known
if _, ok := f.headers[header.ParentHash]; !ok {
return i, errors.New("unknown parent")
}
// Discard any new blocks if the same height already exists
if header.Number.Uint64() <= f.headers[f.hashes[len(f.hashes)-1]].Number.Uint64() {
return i, nil
}
// Otherwise build our current chain
f.hashes = append(f.hashes, header.Hash())
f.headers[header.Hash()] = header
}
return 0, nil
}
// insertChain injects a new blocks into the simulated chain.
func (f *fetcherTester) insertChain(blocks types.Blocks) (int, error) {
f.lock.Lock()
@ -233,7 +267,7 @@ func verifyCompletingEvent(t *testing.T, completing chan []common.Hash, arrive b
}
// verifyImportEvent verifies that one single event arrive on an import channel.
func verifyImportEvent(t *testing.T, imported chan *types.Block, arrive bool) {
func verifyImportEvent(t *testing.T, imported chan interface{}, arrive bool) {
if arrive {
select {
case <-imported:
@ -251,7 +285,7 @@ func verifyImportEvent(t *testing.T, imported chan *types.Block, arrive bool) {
// verifyImportCount verifies that exactly count number of events arrive on an
// import hook channel.
func verifyImportCount(t *testing.T, imported chan *types.Block, count int) {
func verifyImportCount(t *testing.T, imported chan interface{}, count int) {
for i := 0; i < count; i++ {
select {
case <-imported:
@ -263,7 +297,7 @@ func verifyImportCount(t *testing.T, imported chan *types.Block, count int) {
}
// verifyImportDone verifies that no more events are arriving on an import channel.
func verifyImportDone(t *testing.T, imported chan *types.Block) {
func verifyImportDone(t *testing.T, imported chan interface{}) {
select {
case <-imported:
t.Fatalf("extra block imported")
@ -271,45 +305,62 @@ func verifyImportDone(t *testing.T, imported chan *types.Block) {
}
}
// verifyChainHeight verifies the chain height is as expected.
func verifyChainHeight(t *testing.T, fetcher *fetcherTester, height uint64) {
if fetcher.chainHeight() != height {
t.Fatalf("chain height mismatch, got %d, want %d", fetcher.chainHeight(), height)
}
}
// Tests that a fetcher accepts block announcements and initiates retrievals for
// them, successfully importing into the local chain.
func TestSequentialAnnouncements62(t *testing.T) { testSequentialAnnouncements(t, 62) }
func TestSequentialAnnouncements63(t *testing.T) { testSequentialAnnouncements(t, 63) }
func TestSequentialAnnouncements64(t *testing.T) { testSequentialAnnouncements(t, 64) }
func TestFullSequentialAnnouncements(t *testing.T) { testSequentialAnnouncements(t, false) }
func TestLightSequentialAnnouncements(t *testing.T) { testSequentialAnnouncements(t, true) }
func testSequentialAnnouncements(t *testing.T, protocol int) {
func testSequentialAnnouncements(t *testing.T, light bool) {
// Create a chain of blocks to import
targetBlocks := 4 * hashLimit
hashes, blocks := makeChain(targetBlocks, 0, genesis)
tester := newTester()
tester := newTester(light)
headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0)
// Iteratively announce blocks until all are imported
imported := make(chan *types.Block)
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
imported := make(chan interface{})
tester.fetcher.importedHook = func(header *types.Header, block *types.Block) {
if light {
if header == nil {
t.Fatalf("Fetcher try to import empty header")
}
imported <- header
} else {
if block == nil {
t.Fatalf("Fetcher try to import empty block")
}
imported <- block
}
}
for i := len(hashes) - 2; i >= 0; i-- {
tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher)
verifyImportEvent(t, imported, true)
}
verifyImportDone(t, imported)
verifyChainHeight(t, tester, uint64(len(hashes)-1))
}
// Tests that if blocks are announced by multiple peers (or even the same buggy
// peer), they will only get downloaded at most once.
func TestConcurrentAnnouncements62(t *testing.T) { testConcurrentAnnouncements(t, 62) }
func TestConcurrentAnnouncements63(t *testing.T) { testConcurrentAnnouncements(t, 63) }
func TestConcurrentAnnouncements64(t *testing.T) { testConcurrentAnnouncements(t, 64) }
func TestFullConcurrentAnnouncements(t *testing.T) { testConcurrentAnnouncements(t, false) }
func TestLightConcurrentAnnouncements(t *testing.T) { testConcurrentAnnouncements(t, true) }
func testConcurrentAnnouncements(t *testing.T, protocol int) {
func testConcurrentAnnouncements(t *testing.T, light bool) {
// Create a chain of blocks to import
targetBlocks := 4 * hashLimit
hashes, blocks := makeChain(targetBlocks, 0, genesis)
// Assemble a tester with a built in counter for the requests
tester := newTester()
tester := newTester(light)
firstHeaderFetcher := tester.makeHeaderFetcher("first", blocks, -gatherSlack)
firstBodyFetcher := tester.makeBodyFetcher("first", blocks, 0)
secondHeaderFetcher := tester.makeHeaderFetcher("second", blocks, -gatherSlack)
@ -325,8 +376,20 @@ func testConcurrentAnnouncements(t *testing.T, protocol int) {
return secondHeaderFetcher(hash)
}
// Iteratively announce blocks until all are imported
imported := make(chan *types.Block)
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
imported := make(chan interface{})
tester.fetcher.importedHook = func(header *types.Header, block *types.Block) {
if light {
if header == nil {
t.Fatalf("Fetcher try to import empty header")
}
imported <- header
} else {
if block == nil {
t.Fatalf("Fetcher try to import empty block")
}
imported <- block
}
}
for i := len(hashes) - 2; i >= 0; i-- {
tester.fetcher.Notify("first", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), firstHeaderWrapper, firstBodyFetcher)
@ -340,31 +403,42 @@ func testConcurrentAnnouncements(t *testing.T, protocol int) {
if int(counter) != targetBlocks {
t.Fatalf("retrieval count mismatch: have %v, want %v", counter, targetBlocks)
}
verifyChainHeight(t, tester, uint64(len(hashes)-1))
}
// Tests that announcements arriving while a previous is being fetched still
// results in a valid import.
func TestOverlappingAnnouncements62(t *testing.T) { testOverlappingAnnouncements(t, 62) }
func TestOverlappingAnnouncements63(t *testing.T) { testOverlappingAnnouncements(t, 63) }
func TestOverlappingAnnouncements64(t *testing.T) { testOverlappingAnnouncements(t, 64) }
func TestFullOverlappingAnnouncements(t *testing.T) { testOverlappingAnnouncements(t, false) }
func TestLightOverlappingAnnouncements(t *testing.T) { testOverlappingAnnouncements(t, true) }
func testOverlappingAnnouncements(t *testing.T, protocol int) {
func testOverlappingAnnouncements(t *testing.T, light bool) {
// Create a chain of blocks to import
targetBlocks := 4 * hashLimit
hashes, blocks := makeChain(targetBlocks, 0, genesis)
tester := newTester()
tester := newTester(light)
headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0)
// Iteratively announce blocks, but overlap them continuously
overlap := 16
imported := make(chan *types.Block, len(hashes)-1)
imported := make(chan interface{}, len(hashes)-1)
for i := 0; i < overlap; i++ {
imported <- nil
}
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
tester.fetcher.importedHook = func(header *types.Header, block *types.Block) {
if light {
if header == nil {
t.Fatalf("Fetcher try to import empty header")
}
imported <- header
} else {
if block == nil {
t.Fatalf("Fetcher try to import empty block")
}
imported <- block
}
}
for i := len(hashes) - 2; i >= 0; i-- {
tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher)
select {
@ -375,19 +449,19 @@ func testOverlappingAnnouncements(t *testing.T, protocol int) {
}
// Wait for all the imports to complete and check count
verifyImportCount(t, imported, overlap)
verifyChainHeight(t, tester, uint64(len(hashes)-1))
}
// Tests that announces already being retrieved will not be duplicated.
func TestPendingDeduplication62(t *testing.T) { testPendingDeduplication(t, 62) }
func TestPendingDeduplication63(t *testing.T) { testPendingDeduplication(t, 63) }
func TestPendingDeduplication64(t *testing.T) { testPendingDeduplication(t, 64) }
func TestFullPendingDeduplication(t *testing.T) { testPendingDeduplication(t, false) }
func TestLightPendingDeduplication(t *testing.T) { testPendingDeduplication(t, true) }
func testPendingDeduplication(t *testing.T, protocol int) {
func testPendingDeduplication(t *testing.T, light bool) {
// Create a hash and corresponding block
hashes, blocks := makeChain(1, 0, genesis)
// Assemble a tester with a built in counter and delayed fetcher
tester := newTester()
tester := newTester(light)
headerFetcher := tester.makeHeaderFetcher("repeater", blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher("repeater", blocks, 0)
@ -404,40 +478,57 @@ func testPendingDeduplication(t *testing.T, protocol int) {
return nil
}
// Announce the same block many times until it's fetched (wait for any pending ops)
for tester.getBlock(hashes[0]) == nil {
checkNonExist := func() bool {
return tester.getBlock(hashes[0]) == nil
}
if light {
checkNonExist = func() bool {
return tester.getHeader(hashes[0]) == nil
}
}
for checkNonExist() {
tester.fetcher.Notify("repeater", hashes[0], 1, time.Now().Add(-arriveTimeout), headerWrapper, bodyFetcher)
time.Sleep(time.Millisecond)
}
time.Sleep(delay)
// Check that all blocks were imported and none fetched twice
if imported := len(tester.blocks); imported != 2 {
t.Fatalf("synchronised block mismatch: have %v, want %v", imported, 2)
}
if int(counter) != 1 {
t.Fatalf("retrieval count mismatch: have %v, want %v", counter, 1)
}
verifyChainHeight(t, tester, 1)
}
// Tests that announcements retrieved in a random order are cached and eventually
// imported when all the gaps are filled in.
func TestRandomArrivalImport62(t *testing.T) { testRandomArrivalImport(t, 62) }
func TestRandomArrivalImport63(t *testing.T) { testRandomArrivalImport(t, 63) }
func TestRandomArrivalImport64(t *testing.T) { testRandomArrivalImport(t, 64) }
func TestFullRandomArrivalImport(t *testing.T) { testRandomArrivalImport(t, false) }
func TestLightRandomArrivalImport(t *testing.T) { testRandomArrivalImport(t, true) }
func testRandomArrivalImport(t *testing.T, protocol int) {
func testRandomArrivalImport(t *testing.T, light bool) {
// Create a chain of blocks to import, and choose one to delay
targetBlocks := maxQueueDist
hashes, blocks := makeChain(targetBlocks, 0, genesis)
skip := targetBlocks / 2
tester := newTester()
tester := newTester(light)
headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0)
// Iteratively announce blocks, skipping one entry
imported := make(chan *types.Block, len(hashes)-1)
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
imported := make(chan interface{}, len(hashes)-1)
tester.fetcher.importedHook = func(header *types.Header, block *types.Block) {
if light {
if header == nil {
t.Fatalf("Fetcher try to import empty header")
}
imported <- header
} else {
if block == nil {
t.Fatalf("Fetcher try to import empty block")
}
imported <- block
}
}
for i := len(hashes) - 1; i >= 0; i-- {
if i != skip {
@ -448,27 +539,26 @@ func testRandomArrivalImport(t *testing.T, protocol int) {
// Finally announce the skipped entry and check full import
tester.fetcher.Notify("valid", hashes[skip], uint64(len(hashes)-skip-1), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher)
verifyImportCount(t, imported, len(hashes)-1)
verifyChainHeight(t, tester, uint64(len(hashes)-1))
}
// Tests that direct block enqueues (due to block propagation vs. hash announce)
// are correctly schedule, filling and import queue gaps.
func TestQueueGapFill62(t *testing.T) { testQueueGapFill(t, 62) }
func TestQueueGapFill63(t *testing.T) { testQueueGapFill(t, 63) }
func TestQueueGapFill64(t *testing.T) { testQueueGapFill(t, 64) }
func TestFullQueueGapFill(t *testing.T) { testQueueGapFill(t, false) }
func testQueueGapFill(t *testing.T, protocol int) {
func testQueueGapFill(t *testing.T, light bool) {
// Create a chain of blocks to import, and choose one to not announce at all
targetBlocks := maxQueueDist
hashes, blocks := makeChain(targetBlocks, 0, genesis)
skip := targetBlocks / 2
tester := newTester()
tester := newTester(light)
headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0)
// Iteratively announce blocks, skipping one entry
imported := make(chan *types.Block, len(hashes)-1)
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
imported := make(chan interface{}, len(hashes)-1)
tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { imported <- block }
for i := len(hashes) - 1; i >= 0; i-- {
if i != skip {
@ -477,22 +567,21 @@ func testQueueGapFill(t *testing.T, protocol int) {
}
}
// Fill the missing block directly as if propagated
tester.fetcher.Enqueue("valid", blocks[hashes[skip]])
tester.fetcher.EnqueueBlock("valid", blocks[hashes[skip]])
verifyImportCount(t, imported, len(hashes)-1)
verifyChainHeight(t, tester, uint64(len(hashes)-1))
}
// Tests that blocks arriving from various sources (multiple propagations, hash
// announces, etc) do not get scheduled for import multiple times.
func TestImportDeduplication62(t *testing.T) { testImportDeduplication(t, 62) }
func TestImportDeduplication63(t *testing.T) { testImportDeduplication(t, 63) }
func TestImportDeduplication64(t *testing.T) { testImportDeduplication(t, 64) }
func TestFullImportDeduplication(t *testing.T) { testImportDeduplication(t, false) }
func testImportDeduplication(t *testing.T, protocol int) {
func testImportDeduplication(t *testing.T, light bool) {
// Create two blocks to import (one for duplication, the other for stalling)
hashes, blocks := makeChain(2, 0, genesis)
// Create the tester and wrap the importer with a counter
tester := newTester()
tester := newTester(light)
headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0)
@ -503,20 +592,20 @@ func testImportDeduplication(t *testing.T, protocol int) {
}
// Instrument the fetching and imported events
fetching := make(chan []common.Hash)
imported := make(chan *types.Block, len(hashes)-1)
imported := make(chan interface{}, len(hashes)-1)
tester.fetcher.fetchingHook = func(hashes []common.Hash) { fetching <- hashes }
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { imported <- block }
// Announce the duplicating block, wait for retrieval, and also propagate directly
tester.fetcher.Notify("valid", hashes[0], 1, time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher)
<-fetching
tester.fetcher.Enqueue("valid", blocks[hashes[0]])
tester.fetcher.Enqueue("valid", blocks[hashes[0]])
tester.fetcher.Enqueue("valid", blocks[hashes[0]])
tester.fetcher.EnqueueBlock("valid", blocks[hashes[0]])
tester.fetcher.EnqueueBlock("valid", blocks[hashes[0]])
tester.fetcher.EnqueueBlock("valid", blocks[hashes[0]])
// Fill the missing block directly as if propagated, and check import uniqueness
tester.fetcher.Enqueue("valid", blocks[hashes[1]])
tester.fetcher.EnqueueBlock("valid", blocks[hashes[1]])
verifyImportCount(t, imported, 2)
if counter != 2 {
@ -526,7 +615,9 @@ func testImportDeduplication(t *testing.T, protocol int) {
// Tests that blocks with numbers much lower or higher than out current head get
// discarded to prevent wasting resources on useless blocks from faulty peers.
func TestDistantPropagationDiscarding(t *testing.T) {
func TestFullDistantPropagationDiscarding(t *testing.T) { testDistantPropagationDiscarding(t, false) }
func testDistantPropagationDiscarding(t *testing.T, light bool) {
// Create a long chain to import and define the discard boundaries
hashes, blocks := makeChain(3*maxQueueDist, 0, genesis)
head := hashes[len(hashes)/2]
@ -534,7 +625,7 @@ func TestDistantPropagationDiscarding(t *testing.T) {
low, high := len(hashes)/2+maxUncleDist+1, len(hashes)/2-maxQueueDist-1
// Create a tester and simulate a head block being the middle of the above chain
tester := newTester()
tester := newTester(light)
tester.lock.Lock()
tester.hashes = []common.Hash{head}
@ -542,13 +633,13 @@ func TestDistantPropagationDiscarding(t *testing.T) {
tester.lock.Unlock()
// Ensure that a block with a lower number than the threshold is discarded
tester.fetcher.Enqueue("lower", blocks[hashes[low]])
tester.fetcher.EnqueueBlock("lower", blocks[hashes[low]])
time.Sleep(10 * time.Millisecond)
if !tester.fetcher.queue.Empty() {
t.Fatalf("fetcher queued stale block")
}
// Ensure that a block with a higher number than the threshold is discarded
tester.fetcher.Enqueue("higher", blocks[hashes[high]])
tester.fetcher.EnqueueBlock("higher", blocks[hashes[high]])
time.Sleep(10 * time.Millisecond)
if !tester.fetcher.queue.Empty() {
t.Fatalf("fetcher queued future block")
@ -558,11 +649,10 @@ func TestDistantPropagationDiscarding(t *testing.T) {
// Tests that announcements with numbers much lower or higher than out current
// head get discarded to prevent wasting resources on useless blocks from faulty
// peers.
func TestDistantAnnouncementDiscarding62(t *testing.T) { testDistantAnnouncementDiscarding(t, 62) }
func TestDistantAnnouncementDiscarding63(t *testing.T) { testDistantAnnouncementDiscarding(t, 63) }
func TestDistantAnnouncementDiscarding64(t *testing.T) { testDistantAnnouncementDiscarding(t, 64) }
func TestFullDistantAnnouncementDiscarding(t *testing.T) { testDistantAnnouncementDiscarding(t, false) }
func TestLightDistantAnnouncementDiscarding(t *testing.T) { testDistantAnnouncementDiscarding(t, true) }
func testDistantAnnouncementDiscarding(t *testing.T, protocol int) {
func testDistantAnnouncementDiscarding(t *testing.T, light bool) {
// Create a long chain to import and define the discard boundaries
hashes, blocks := makeChain(3*maxQueueDist, 0, genesis)
head := hashes[len(hashes)/2]
@ -570,10 +660,11 @@ func testDistantAnnouncementDiscarding(t *testing.T, protocol int) {
low, high := len(hashes)/2+maxUncleDist+1, len(hashes)/2-maxQueueDist-1
// Create a tester and simulate a head block being the middle of the above chain
tester := newTester()
tester := newTester(light)
tester.lock.Lock()
tester.hashes = []common.Hash{head}
tester.headers = map[common.Hash]*types.Header{head: blocks[head].Header()}
tester.blocks = map[common.Hash]*types.Block{head: blocks[head]}
tester.lock.Unlock()
@ -601,20 +692,31 @@ func testDistantAnnouncementDiscarding(t *testing.T, protocol int) {
// Tests that peers announcing blocks with invalid numbers (i.e. not matching
// the headers provided afterwards) get dropped as malicious.
func TestInvalidNumberAnnouncement62(t *testing.T) { testInvalidNumberAnnouncement(t, 62) }
func TestInvalidNumberAnnouncement63(t *testing.T) { testInvalidNumberAnnouncement(t, 63) }
func TestInvalidNumberAnnouncement64(t *testing.T) { testInvalidNumberAnnouncement(t, 64) }
func TestFullInvalidNumberAnnouncement(t *testing.T) { testInvalidNumberAnnouncement(t, false) }
func TestLightInvalidNumberAnnouncement(t *testing.T) { testInvalidNumberAnnouncement(t, true) }
func testInvalidNumberAnnouncement(t *testing.T, protocol int) {
func testInvalidNumberAnnouncement(t *testing.T, light bool) {
// Create a single block to import and check numbers against
hashes, blocks := makeChain(1, 0, genesis)
tester := newTester()
tester := newTester(light)
badHeaderFetcher := tester.makeHeaderFetcher("bad", blocks, -gatherSlack)
badBodyFetcher := tester.makeBodyFetcher("bad", blocks, 0)
imported := make(chan *types.Block)
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
imported := make(chan interface{})
tester.fetcher.importedHook = func(header *types.Header, block *types.Block) {
if light {
if header == nil {
t.Fatalf("Fetcher try to import empty header")
}
imported <- header
} else {
if block == nil {
t.Fatalf("Fetcher try to import empty block")
}
imported <- block
}
}
// Announce a block with a bad number, check for immediate drop
tester.fetcher.Notify("bad", hashes[0], 2, time.Now().Add(-arriveTimeout), badHeaderFetcher, badBodyFetcher)
@ -646,15 +748,13 @@ func testInvalidNumberAnnouncement(t *testing.T, protocol int) {
// Tests that if a block is empty (i.e. header only), no body request should be
// made, and instead the header should be assembled into a whole block in itself.
func TestEmptyBlockShortCircuit62(t *testing.T) { testEmptyBlockShortCircuit(t, 62) }
func TestEmptyBlockShortCircuit63(t *testing.T) { testEmptyBlockShortCircuit(t, 63) }
func TestEmptyBlockShortCircuit64(t *testing.T) { testEmptyBlockShortCircuit(t, 64) }
func TestFullEmptyBlockShortCircuit(t *testing.T) { testEmptyBlockShortCircuit(t, false) }
func testEmptyBlockShortCircuit(t *testing.T, protocol int) {
func testEmptyBlockShortCircuit(t *testing.T, light bool) {
// Create a chain of blocks to import
hashes, blocks := makeChain(32, 0, genesis)
tester := newTester()
tester := newTester(light)
headerFetcher := tester.makeHeaderFetcher("valid", blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0)
@ -665,8 +765,20 @@ func testEmptyBlockShortCircuit(t *testing.T, protocol int) {
completing := make(chan []common.Hash)
tester.fetcher.completingHook = func(hashes []common.Hash) { completing <- hashes }
imported := make(chan *types.Block)
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
imported := make(chan interface{})
tester.fetcher.importedHook = func(header *types.Header, block *types.Block) {
if light {
if header == nil {
t.Fatalf("Fetcher try to import empty header")
}
imported <- header
} else {
if block == nil {
t.Fatalf("Fetcher try to import empty block")
}
imported <- block
}
}
// Iteratively announce blocks until all are imported
for i := len(hashes) - 2; i >= 0; i-- {
@ -687,16 +799,12 @@ func testEmptyBlockShortCircuit(t *testing.T, protocol int) {
// Tests that a peer is unable to use unbounded memory with sending infinite
// block announcements to a node, but that even in the face of such an attack,
// the fetcher remains operational.
func TestHashMemoryExhaustionAttack62(t *testing.T) { testHashMemoryExhaustionAttack(t, 62) }
func TestHashMemoryExhaustionAttack63(t *testing.T) { testHashMemoryExhaustionAttack(t, 63) }
func TestHashMemoryExhaustionAttack64(t *testing.T) { testHashMemoryExhaustionAttack(t, 64) }
func testHashMemoryExhaustionAttack(t *testing.T, protocol int) {
func TestFullHashMemoryExhaustionAttack(t *testing.T) {
// Create a tester with instrumented import hooks
tester := newTester()
tester := newTester(false)
imported, announces := make(chan *types.Block), int32(0)
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
imported, announces := make(chan interface{}), int32(0)
tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { imported <- block }
tester.fetcher.announceChangeHook = func(hash common.Hash, added bool) {
if added {
atomic.AddInt32(&announces, 1)
@ -740,10 +848,10 @@ func testHashMemoryExhaustionAttack(t *testing.T, protocol int) {
// system memory.
func TestBlockMemoryExhaustionAttack(t *testing.T) {
// Create a tester with instrumented import hooks
tester := newTester()
tester := newTester(false)
imported, enqueued := make(chan *types.Block), int32(0)
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
imported, enqueued := make(chan interface{}), int32(0)
tester.fetcher.importedHook = func(header *types.Header, block *types.Block) { imported <- block }
tester.fetcher.queueChangeHook = func(hash common.Hash, added bool) {
if added {
atomic.AddInt32(&enqueued, 1)
@ -763,7 +871,7 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) {
}
// Try to feed all the attacker blocks make sure only a limited batch is accepted
for _, block := range attack {
tester.fetcher.Enqueue("attacker", block)
tester.fetcher.EnqueueBlock("attacker", block)
}
time.Sleep(200 * time.Millisecond)
if queued := atomic.LoadInt32(&enqueued); queued != blockLimit {
@ -771,19 +879,19 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) {
}
// Queue up a batch of valid blocks, and check that a new peer is allowed to do so
for i := 0; i < maxQueueDist-1; i++ {
tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-3-i]])
tester.fetcher.EnqueueBlock("valid", blocks[hashes[len(hashes)-3-i]])
}
time.Sleep(100 * time.Millisecond)
if queued := atomic.LoadInt32(&enqueued); queued != blockLimit+maxQueueDist-1 {
t.Fatalf("queued block count mismatch: have %d, want %d", queued, blockLimit+maxQueueDist-1)
}
// Insert the missing piece (and sanity check the import)
tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-2]])
tester.fetcher.EnqueueBlock("valid", blocks[hashes[len(hashes)-2]])
verifyImportCount(t, imported, maxQueueDist)
// Insert the remaining blocks in chunks to ensure clean DOS protection
for i := maxQueueDist; i < len(hashes)-1; i++ {
tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-2-i]])
tester.fetcher.EnqueueBlock("valid", blocks[hashes[len(hashes)-2-i]])
verifyImportEvent(t, imported, true)
}
verifyImportDone(t, imported)

4
eth/handler.go
View file

@ -211,7 +211,7 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne
}
return n, err
}
manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer)
manager.fetcher = fetcher.New(false, nil, blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, nil, inserter, manager.removePeer)
return manager, nil
}
@ -683,7 +683,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
// Mark the peer as owning the block and schedule it for import
p.MarkBlock(request.Block.Hash())
pm.fetcher.Enqueue(p.id, request.Block)
pm.fetcher.EnqueueBlock(p.id, request.Block)
// Assuming the block is importable by the peer, but possibly not yet done so,
// calculate the head hash and TD that the peer truly must have.

1
les/client.go
View file

@ -226,6 +226,7 @@ func (s *LightEthereum) Start(srvr *p2p.Server) error {
// Start bloom request workers.
s.wg.Add(bloomServiceThreads)
s.startBloomHandlers(params.BloomBitsBlocksClient)
s.handler.start()
s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.config.NetworkId)

22
les/client_handler.go
View file

@ -60,10 +60,14 @@ func newClientHandler(ulcConfig *eth.ULCConfig, backend *LightEthereum) *clientH
return handler
}
func (h *clientHandler) start() {
h.fetcher.start()
}
func (h *clientHandler) stop() {
close(h.closeCh)
h.downloader.Terminate()
h.fetcher.close()
h.fetcher.stop()
h.wg.Wait()
}
@ -172,6 +176,9 @@ func (h *clientHandler) handleMsg(p *serverPeer) error {
p.Log().Trace("Valid announcement signature")
}
p.Log().Trace("Announce message content", "number", req.Number, "hash", req.Hash, "td", req.Td, "reorg", req.ReorgDepth)
// Update peer head information first
p.updateHead(req.Hash, req.Number, req.Td)
h.fetcher.announce(p, &req)
}
case BlockHeadersMsg:
@ -183,11 +190,16 @@ func (h *clientHandler) handleMsg(p *serverPeer) error {
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
headers := resp.Headers
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
if h.fetcher.requestedID(resp.ReqID) {
h.fetcher.deliverHeaders(p, resp.ReqID, resp.Headers)
} else {
if err := h.downloader.DeliverHeaders(p.id, resp.Headers); err != nil {
// Filter out any explicitly requested headers, deliver the rest to the downloader
filter := len(headers) == 1
if filter {
headers = h.fetcher.deliverHeaders(p, resp.ReqID, resp.Headers)
}
if len(headers) != 0 || !filter {
if err := h.downloader.DeliverHeaders(p.id, headers); err != nil {
log.Debug("Failed to deliver headers", "err", err)
}
}

1240
les/fetcher.go
View file

File diff suppressed because it is too large Load diff

284
les/fetcher_test.go Normal file
View file

@ -0,0 +1,284 @@
// Copyright 2019 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 les
import (
"testing"
"time"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/eth"
"github.com/ethereum/go-ethereum/p2p/enode"
)
// verifyImportEvent verifies that one single event arrive on an import channel.
func verifyImportEvent(t *testing.T, imported chan interface{}, arrive bool) {
if arrive {
select {
case <-imported:
case <-time.After(time.Second):
t.Fatalf("import timeout")
}
} else {
select {
case <-imported:
t.Fatalf("import invoked")
case <-time.After(20 * time.Millisecond):
}
}
}
// verifyImportDone verifies that no more events are arriving on an import channel.
func verifyImportDone(t *testing.T, imported chan interface{}) {
select {
case <-imported:
t.Fatalf("extra block imported")
case <-time.After(50 * time.Millisecond):
}
}
// verifyChainHeight verifies the chain height is as expected.
func verifyChainHeight(t *testing.T, fetcher *lightFetcher, height uint64) {
local := fetcher.chain.CurrentHeader().Number.Uint64()
if local != height {
t.Fatalf("chain height mismatch, got %d, want %d", local, height)
}
}
func TestSequentialAnnouncementsLes2(t *testing.T) { testSequentialAnnouncements(t, 2) }
func TestSequentialAnnouncementsLes3(t *testing.T) { testSequentialAnnouncements(t, 3) }
func testSequentialAnnouncements(t *testing.T, protocol int) {
s, c, teardown := newClientServerEnv(t, 4, protocol, nil, nil, false, false)
defer teardown()
// Create connected peer pair.
c.handler.fetcher.ignoreAnnounce = true // Ignore the first announce from peer which can trigger a resync.
p1, err1, _, err2 := newTestPeerPair("peer", protocol, s.handler, c.handler)
select {
case <-time.After(time.Millisecond * 100):
case err := <-err1:
t.Fatalf("peer 1 handshake error: %v", err)
case err := <-err2:
t.Fatalf("peer 2 handshake error: %v", err)
}
c.handler.fetcher.ignoreAnnounce = false
importCh := make(chan interface{})
c.handler.fetcher.newHeadHook = func(header *types.Header) {
importCh <- header
}
for i := uint64(1); i <= s.backend.Blockchain().CurrentHeader().Number.Uint64(); i++ {
header := s.backend.Blockchain().GetHeaderByNumber(i)
hash, number := header.Hash(), header.Number.Uint64()
td := rawdb.ReadTd(s.db, hash, number)
announce := announceData{hash, number, td, 0, nil}
if p1.cpeer.announceType == announceTypeSigned {
announce.sign(s.handler.server.privateKey)
}
p1.cpeer.sendAnnounce(announce)
verifyImportEvent(t, importCh, true)
}
verifyImportDone(t, importCh)
verifyChainHeight(t, c.handler.fetcher, 4)
}
func TestGappedAnnouncementsLes2(t *testing.T) { testGappedAnnouncements(t, 2) }
func TestGappedAnnouncementsLes3(t *testing.T) { testGappedAnnouncements(t, 3) }
func testGappedAnnouncements(t *testing.T, protocol int) {
s, c, teardown := newClientServerEnv(t, 4, protocol, nil, nil, false, false)
defer teardown()
// Create connected peer pair.
c.handler.fetcher.ignoreAnnounce = true // Ignore the first announce from peer which can trigger a resync.
p1, err1, _, err2 := newTestPeerPair("peer", protocol, s.handler, c.handler)
select {
case <-time.After(time.Millisecond * 100):
case err := <-err1:
t.Fatalf("peer 1 handshake error: %v", err)
case err := <-err2:
t.Fatalf("peer 2 handshake error: %v", err)
}
c.handler.fetcher.ignoreAnnounce = false
// Prepare announcement by latest header.
latest := s.backend.Blockchain().CurrentHeader()
hash, number := latest.Hash(), latest.Number.Uint64()
td := rawdb.ReadTd(s.db, hash, number)
// Sign the announcement if necessary.
announce := announceData{hash, number, td, 0, nil}
if p1.cpeer.announceType == announceTypeSigned {
announce.sign(s.handler.server.privateKey)
}
p1.cpeer.sendAnnounce(announce)
time.Sleep(10 * time.Millisecond)
verifyChainHeight(t, c.handler.fetcher, 4)
}
func TestTrustedAnnouncementsLes2(t *testing.T) { testTrustedAnnouncement(t, 2) }
func TestTrustedAnnouncementsLes3(t *testing.T) { testTrustedAnnouncement(t, 3) }
func testTrustedAnnouncement(t *testing.T, protocol int) {
var (
servers []*testServer
teardowns []func()
nodes []*enode.Node
ids []string
cpeers []*clientPeer
speers []*serverPeer
)
for i := 0; i < 10; i++ {
s, n, teardown := newTestServerPeer(t, 10, protocol)
servers = append(servers, s)
nodes = append(nodes, n)
teardowns = append(teardowns, teardown)
// A half of them are trusted servers.
if i < 5 {
ids = append(ids, n.String())
}
}
config := &eth.ULCConfig{
MinTrustedFraction: 60, // At least 3 approvals
TrustedServers: ids,
}
_, c, teardown := newClientServerEnv(t, 0, protocol, nil, config, false, false)
defer teardown()
defer func() {
for i := 0; i < len(teardowns); i++ {
teardowns[i]()
}
}()
c.handler.fetcher.ignoreAnnounce = true // Ignore the first announce from peer which can trigger a resync.
// Connect all server instances.
for i := 0; i < len(servers); i++ {
sp, cp, err := connect(servers[i].handler, nodes[i].ID(), c.handler, protocol)
if err != nil {
t.Fatalf("connect server and client failed, err %s", err)
}
cpeers = append(cpeers, cp)
speers = append(speers, sp)
}
c.handler.fetcher.ignoreAnnounce = false
check := func(height []uint64, expected uint64, callback func()) {
for i := 0; i < len(height); i++ {
for j := 0; j < len(servers); j++ {
h := servers[j].backend.Blockchain().GetHeaderByNumber(height[i])
hash, number := h.Hash(), h.Number.Uint64()
td := rawdb.ReadTd(servers[j].db, hash, number)
// Sign the announcement if necessary.
announce := announceData{hash, number, td, 0, nil}
p := cpeers[j]
if p.announceType == announceTypeSigned {
announce.sign(servers[j].handler.server.privateKey)
}
p.sendAnnounce(announce)
if j < 2 {
time.Sleep(10 * time.Millisecond) // Ensure the announcement has been processed.
if !c.handler.fetcher.queryAnnounced(speers[j], hash) {
t.Fatalf("the announcement from server peer %d should be kept", j+1)
}
} else if j == 2 {
// The block should be imported
}
}
}
if callback != nil {
callback()
}
time.Sleep(10 * time.Millisecond) // Ensure the announcement has been processed.
verifyChainHeight(t, c.handler.fetcher, expected)
}
check([]uint64{1}, 1, nil) // Sequential announcements
check([]uint64{4}, 4, nil) // ULC-style light syncing, rollback untrusted headers
done := make(chan struct{})
c.handler.fetcher.syncingHook = func() { <-done }
check([]uint64{6, 8}, 8, func() { done <- struct{}{} }) // ULC-style light syncing, keep the later trusted announces.
c.handler.fetcher.syncingHook = nil
check([]uint64{10}, 10, nil) // Sync the whole chain.
}
func TestAnnounceDelayLes2(t *testing.T) { testAnnounceDelay(t, 2) }
func TestAnnounceDelayLes3(t *testing.T) { testAnnounceDelay(t, 3) }
func testAnnounceDelay(t *testing.T, protocol int) {
var (
servers []*testServer
teardowns []func()
nodes []*enode.Node
cpeers []*clientPeer
speers []*serverPeer
)
for i := 0; i < 5; i++ {
s, n, teardown := newTestServerPeer(t, 10, protocol)
servers = append(servers, s)
nodes = append(nodes, n)
teardowns = append(teardowns, teardown)
}
_, c, teardown := newClientServerEnv(t, 0, protocol, nil, nil, false, false)
defer teardown()
defer func() {
for i := 0; i < len(teardowns); i++ {
teardowns[i]()
}
}()
c.handler.fetcher.ignoreAnnounce = true // Ignore the first announce from peer which can trigger a resync.
// Connect all server instances.
for i := 0; i < len(servers); i++ {
sp, cp, err := connect(servers[i].handler, nodes[i].ID(), c.handler, protocol)
if err != nil {
t.Fatalf("connect server and client failed, err %s", err)
}
cpeers = append(cpeers, cp)
speers = append(speers, sp)
}
c.handler.fetcher.ignoreAnnounce = false
delays := make(map[*serverPeer]time.Duration)
c.handler.fetcher.addDelayHook = func(p *serverPeer, delay time.Duration) { delays[p] = delay }
for i := 0; i < 2; i++ {
h := servers[i].backend.Blockchain().GetHeaderByNumber(1)
hash, number := h.Hash(), h.Number.Uint64()
td := rawdb.ReadTd(servers[i].db, hash, number)
announce := announceData{hash, number, td, 0, nil}
p := cpeers[i]
if p.announceType == announceTypeSigned {
announce.sign(servers[i].handler.server.privateKey)
}
p.sendAnnounce(announce)
}
time.Sleep(10 * time.Millisecond)
if d, exist := delays[speers[1]]; !exist || d > time.Millisecond {
t.Fatalf("the second announcement should be confirmed soon")
}
}

8
les/odr_requests.go
View file

@ -84,7 +84,7 @@ func (r *BlockRequest) GetCost(peer *serverPeer) uint64 {
// CanSend tells if a certain peer is suitable for serving the given request
func (r *BlockRequest) CanSend(peer *serverPeer) bool {
return peer.HasBlock(r.Hash, r.Number, false)
return peer.hasBlock(r.Hash, r.Number, false)
}
// Request sends an ODR request to the LES network (implementation of LesOdrRequest)
@ -140,7 +140,7 @@ func (r *ReceiptsRequest) GetCost(peer *serverPeer) uint64 {
// CanSend tells if a certain peer is suitable for serving the given request
func (r *ReceiptsRequest) CanSend(peer *serverPeer) bool {
return peer.HasBlock(r.Hash, r.Number, false)
return peer.hasBlock(r.Hash, r.Number, false)
}
// Request sends an ODR request to the LES network (implementation of LesOdrRequest)
@ -197,7 +197,7 @@ func (r *TrieRequest) GetCost(peer *serverPeer) uint64 {
// CanSend tells if a certain peer is suitable for serving the given request
func (r *TrieRequest) CanSend(peer *serverPeer) bool {
return peer.HasBlock(r.Id.BlockHash, r.Id.BlockNumber, true)
return peer.hasBlock(r.Id.BlockHash, r.Id.BlockNumber, true)
}
// Request sends an ODR request to the LES network (implementation of LesOdrRequest)
@ -251,7 +251,7 @@ func (r *CodeRequest) GetCost(peer *serverPeer) uint64 {
// CanSend tells if a certain peer is suitable for serving the given request
func (r *CodeRequest) CanSend(peer *serverPeer) bool {
return peer.HasBlock(r.Id.BlockHash, r.Id.BlockNumber, true)
return peer.hasBlock(r.Id.BlockHash, r.Id.BlockNumber, true)
}
// Request sends an ODR request to the LES network (implementation of LesOdrRequest)

4
les/odr_test.go
View file

@ -213,13 +213,13 @@ func testOdr(t *testing.T, protocol int, expFail uint64, checkCached bool, fn od
// expect retrievals to fail (except genesis block) without a les peer
client.handler.backend.peers.lock.Lock()
client.peer.speer.hasBlock = func(common.Hash, uint64, bool) bool { return false }
client.peer.speer.hasBlockCallback = func() bool { return false }
client.handler.backend.peers.lock.Unlock()
test(expFail)
// expect all retrievals to pass
client.handler.backend.peers.lock.Lock()
client.peer.speer.hasBlock = func(common.Hash, uint64, bool) bool { return true }
client.peer.speer.hasBlockCallback = nil
client.handler.backend.peers.lock.Unlock()
test(5)

24
les/peer.go
View file

@ -311,8 +311,8 @@ type serverPeer struct {
updateCount uint64
updateTime mclock.AbsTime
// Callbacks
hasBlock func(common.Hash, uint64, bool) bool // Used to determine whether the server has the specified block.
// Test Hooks
hasBlockCallback func() bool
}
func newServerPeer(version int, network uint64, trusted bool, p *p2p.Peer, rw p2p.MsgReadWriter) *serverPeer {
@ -475,11 +475,14 @@ func (p *serverPeer) getTxRelayCost(amount, size int) uint64 {
return cost
}
// HasBlock checks if the peer has a given block
func (p *serverPeer) HasBlock(hash common.Hash, number uint64, hasState bool) bool {
// hasBlock checks if the peer has a given block
func (p *serverPeer) hasBlock(hash common.Hash, number uint64, hasState bool) bool {
p.lock.RLock()
defer p.lock.RUnlock()
if p.hasBlockCallback != nil {
return p.hasBlockCallback()
}
head := p.headInfo.Number
var since, recent uint64
if hasState {
@ -489,9 +492,7 @@ func (p *serverPeer) HasBlock(hash common.Hash, number uint64, hasState bool) bo
since = p.chainSince
recent = p.chainRecent
}
hasBlock := p.hasBlock
return head >= number && number >= since && (recent == 0 || number+recent+4 > head) && hasBlock != nil && hasBlock(hash, number, hasState)
return head >= number && number >= since && (recent == 0 || number+recent+4 > head)
}
// updateFlowControl updates the flow control parameters belonging to the server
@ -516,6 +517,15 @@ func (p *serverPeer) updateFlowControl(update keyValueMap) {
}
}
// updateHead updates the head information based on the announcement from
// the peer.
func (p *serverPeer) updateHead(hash common.Hash, number uint64, td *big.Int) {
p.lock.Lock()
defer p.lock.Unlock()
p.headInfo = blockInfo{Hash: hash, Number: number, Td: td}
}
// Handshake executes the les protocol handshake, negotiating version number,
// network IDs, difficulties, head and genesis blocks.
func (p *serverPeer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis common.Hash, server *LesServer) error {

15
les/test_helper.go
View file

@ -23,6 +23,7 @@ import (
"context"
"crypto/rand"
"math/big"
"net"
"testing"
"time"
@ -221,6 +222,7 @@ func newTestClientHandler(backend *backends.SimulatedBackend, odr *LesOdr, index
if client.registrar != nil {
client.registrar.start(backend)
}
client.handler.start()
return client.handler
}
@ -280,6 +282,8 @@ func newTestServerHandler(blocks int, indexers []*core.ChainIndexer, db ethdb.Da
},
fcManager: flowcontrol.NewClientManager(nil, clock),
}
key, _ := crypto.GenerateKey()
server.privateKey = key
server.costTracker, server.minCapacity = newCostTracker(db, server.config, nil)
server.costTracker.costListHook = func() RequestCostList { return testCostList(0) } // Disable flow control mechanism.
server.handler = newServerHandler(server, simulation.Blockchain(), db, txpool, nil, func() bool { return true })
@ -349,14 +353,13 @@ func (p *testPeer) close() {
func newTestPeerPair(name string, version int, server *serverHandler, client *clientHandler) (*testPeer, <-chan error, *testPeer, <-chan error) {
// Create a message pipe to communicate through
app, net := p2p.MsgPipe()
app, s := p2p.MsgPipe()
// Generate a random id and create the peer
var id enode.ID
rand.Read(id[:])
en := enode.NewV4(&server.server.privateKey.PublicKey, net.ParseIP("127.0.0.1"), 35000, 35000)
peer1 := newClientPeer(version, NetworkId, p2p.NewPeer(id, name, nil), net)
peer2 := newServerPeer(version, NetworkId, false, p2p.NewPeer(id, name, nil), app)
peer1 := newClientPeer(version, NetworkId, p2p.NewPeer(en.ID(), name, nil), s)
peer2 := newServerPeer(version, NetworkId, false, p2p.NewPeer(en.ID(), name, nil), app)
// Start the peer on a new thread
errc1 := make(chan error, 1)
@ -375,7 +378,7 @@ func newTestPeerPair(name string, version int, server *serverHandler, client *cl
case errc1 <- client.handle(peer2):
}
}()
return &testPeer{cpeer: peer1, net: net, app: app}, errc1, &testPeer{speer: peer2, net: app, app: net}, errc2
return &testPeer{cpeer: peer1, net: s, app: app}, errc1, &testPeer{speer: peer2, net: app, app: s}, errc2
}
// handshake simulates a trivial handshake that expects the same state from the

8
les/txrelay.go
View file

@ -62,14 +62,18 @@ func (self *lesTxRelay) registerPeer(p *serverPeer) {
self.lock.Lock()
defer self.lock.Unlock()
self.peerList = self.ps.allServerPeers()
self.peerList = append(self.peerList, p)
}
func (self *lesTxRelay) unregisterPeer(p *serverPeer) {
self.lock.Lock()
defer self.lock.Unlock()
self.peerList = self.ps.allServerPeers()
for index, peer := range self.peerList {
if peer == p {
self.peerList = append(self.peerList[:index], self.peerList[index+1:]...)
}
}
}
// send sends a list of transactions to at most a given number of peers at

2
light/lightchain.go
View file

@ -256,7 +256,7 @@ func (lc *LightChain) GetBodyRLP(ctx context.Context, hash common.Hash) (rlp.Raw
return body, nil
}
// HasBlock checks if a block is fully present in the database or not, caching
// hasBlock checks if a block is fully present in the database or not, caching
// it if present.
func (lc *LightChain) HasBlock(hash common.Hash, number uint64) bool {
blk, _ := lc.GetBlock(NoOdr, hash, number)