cerc-io/plugeth
18
0
Fork 1
Code Issues 4 Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #1275 from karalabe/optimise-fetcher

Browse Source 
eth/fetcher: separate the announce based sync into its own package
This commit is contained in:
Jeffrey Wilcke 2015-06-18 11:14:26 -07:00
commit 8eaaf24b1e
8 changed files with 879 additions and 275 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
eth/backend.go
View File

@ -313,7 +313,7 @@ func New(config *Config) (*Ethereum, error) {
eth.blockProcessor = core.NewBlockProcessor(stateDb, extraDb, eth.pow, eth.chainManager, eth.EventMux())
eth.chainManager.SetProcessor(eth.blockProcessor)
eth.protocolManager = NewProtocolManager(config.ProtocolVersion, config.NetworkId, eth.eventMux, eth.txPool, eth.chainManager)
eth.protocolManager = NewProtocolManager(config.ProtocolVersion, config.NetworkId, eth.eventMux, eth.txPool, eth.pow, eth.chainManager)
eth.miner = miner.New(eth, eth.EventMux(), eth.pow)
eth.miner.SetGasPrice(config.GasPrice)

3
eth/downloader/downloader.go
View File

@ -1,3 +1,4 @@
// Package downloader contains the manual full chain synchronisation.
package downloader
import (
@ -98,7 +99,7 @@ type Downloader struct {
hasBlock hashCheckFn // Checks if a block is present in the chain
getBlock blockRetrievalFn // Retrieves a block from the chain
insertChain chainInsertFn // Injects a batch of blocks into the chain
dropPeer peerDropFn // Retrieved the TD of our own chain
dropPeer peerDropFn // Drops a peer for misbehaving
// Status
synchroniseMock func(id string, hash common.Hash) error // Replacement for synchronise during testing

8
eth/downloader/downloader_test.go
View File

@ -52,6 +52,8 @@ func copyBlock(block *types.Block) *types.Block {
return createBlock(int(block.Number().Int64()), block.ParentHeaderHash, block.HeaderHash)
}
// createBlocksFromHashes assembles a collection of blocks, each having a correct
// place in the given hash chain.
func createBlocksFromHashes(hashes []common.Hash) map[common.Hash]*types.Block {
blocks := make(map[common.Hash]*types.Block)
for i := 0; i < len(hashes); i++ {
@ -64,6 +66,7 @@ func createBlocksFromHashes(hashes []common.Hash) map[common.Hash]*types.Block {
return blocks
}
// downloadTester is a test simulator for mocking out local block chain.
type downloadTester struct {
downloader *Downloader
@ -75,6 +78,7 @@ type downloadTester struct {
maxHashFetch int // Overrides the maximum number of retrieved hashes
}
// newTester creates a new downloader test mocker.
func newTester() *downloadTester {
tester := &downloadTester{
ownHashes: []common.Hash{knownHash},
@ -82,9 +86,7 @@ func newTester() *downloadTester {
peerHashes: make(map[string][]common.Hash),
peerBlocks: make(map[string]map[common.Hash]*types.Block),
}
var mux event.TypeMux
downloader := New(&mux, tester.hasBlock, tester.getBlock, tester.insertChain, tester.dropPeer)
tester.downloader = downloader
tester.downloader = New(new(event.TypeMux), tester.hasBlock, tester.getBlock, tester.insertChain, tester.dropPeer)
return tester
}

368
eth/fetcher/fetcher.go Normal file
View File

@ -0,0 +1,368 @@
// Package fetcher contains the block announcement based synchonisation.
package fetcher
import (
"errors"
"math/rand"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
"gopkg.in/karalabe/cookiejar.v2/collections/prque"
)
const (
arriveTimeout = 500 * time.Millisecond // Time allowance before an announced block is explicitly requested
fetchTimeout = 5 * time.Second // Maximum alloted time to return an explicitly requested block
maxUncleDist = 7 // Maximum allowed backward distance from the chain head
maxQueueDist = 256 // Maximum allowed distance from the chain head to queue
)
var (
errTerminated = errors.New("terminated")
)
// blockRetrievalFn is a callback type for retrieving a block from the local chain.
type blockRetrievalFn func(common.Hash) *types.Block
// blockRequesterFn is a callback type for sending a block retrieval request.
type blockRequesterFn func([]common.Hash) error
// blockValidatorFn is a callback type to verify a block's header for fast propagation.
type blockValidatorFn func(block *types.Block, parent *types.Block) error
// blockBroadcasterFn is a callback type for broadcasting a block to connected peers.
type blockBroadcasterFn func(block *types.Block, propagate bool)
// chainHeightFn is a callback type to retrieve the current chain height.
type chainHeightFn func() uint64
// chainInsertFn is a callback type to insert a batch of blocks into the local chain.
type chainInsertFn func(types.Blocks) (int, error)
// peerDropFn is a callback type for dropping a peer detected as malicious.
type peerDropFn func(id string)
// announce is the hash notification of the availability of a new block in the
// network.
type announce struct {
hash common.Hash // Hash of the block being announced
time time.Time // Timestamp of the announcement
origin string // Identifier of the peer originating the notification
fetch blockRequesterFn // Fetcher function to retrieve
}
// inject represents a schedules import operation.
type inject struct {
origin string
block *types.Block
}
// Fetcher is responsible for accumulating block announcements from various peers
// and scheduling them for retrieval.
type Fetcher struct {
// Various event channels
notify chan *announce
inject chan *inject
filter chan chan []*types.Block
done chan common.Hash
quit chan struct{}
// Announce states
announced map[common.Hash][]*announce // Announced blocks, scheduled for fetching
fetching map[common.Hash]*announce // Announced blocks, currently fetching
// Block cache
queue *prque.Prque // Queue containing the import operations (block number sorted)
queued map[common.Hash]struct{} // Presence set of already queued blocks (to dedup imports)
// Callbacks
getBlock blockRetrievalFn // Retrieves a block from the local chain
validateBlock blockValidatorFn // 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
insertChain chainInsertFn // Injects a batch of blocks into the chain
dropPeer peerDropFn // Drops a peer for misbehaving
}
// New creates a block fetcher to retrieve blocks based on hash announcements.
func New(getBlock blockRetrievalFn, validateBlock blockValidatorFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher {
return &Fetcher{
notify: make(chan *announce),
inject: make(chan *inject),
filter: make(chan chan []*types.Block),
done: make(chan common.Hash),
quit: make(chan struct{}),
announced: make(map[common.Hash][]*announce),
fetching: make(map[common.Hash]*announce),
queue: prque.New(),
queued: make(map[common.Hash]struct{}),
getBlock: getBlock,
validateBlock: validateBlock,
broadcastBlock: broadcastBlock,
chainHeight: chainHeight,
insertChain: insertChain,
dropPeer: dropPeer,
}
}
// Start boots up the announcement based synchoniser, accepting and processing
// hash notifications and block fetches until termination requested.
func (f *Fetcher) Start() {
go f.loop()
}
// Stop terminates the announcement based synchroniser, canceling all pending
// operations.
func (f *Fetcher) Stop() {
close(f.quit)
}
// Notify announces the fetcher of the potential availability of a new block in
// the network.
func (f *Fetcher) Notify(peer string, hash common.Hash, time time.Time, fetcher blockRequesterFn) error {
block := &announce{
hash: hash,
time: time,
origin: peer,
fetch: fetcher,
}
select {
case f.notify <- block:
return nil
case <-f.quit:
return errTerminated
}
}
// Enqueue tries to fill gaps the the fetcher's future import queue.
func (f *Fetcher) Enqueue(peer string, block *types.Block) error {
op := &inject{
origin: peer,
block: block,
}
select {
case f.inject <- op:
return nil
case <-f.quit:
return errTerminated
}
}
// Filter extracts all the blocks that were explicitly requested by the fetcher,
// returning those that should be handled differently.
func (f *Fetcher) Filter(blocks types.Blocks) types.Blocks {
// Send the filter channel to the fetcher
filter := make(chan []*types.Block)
select {
case f.filter <- filter:
case <-f.quit:
return nil
}
// Request the filtering of the block list
select {
case filter <- blocks:
case <-f.quit:
return nil
}
// Retrieve the blocks remaining after filtering
select {
case blocks := <-filter:
return blocks
case <-f.quit:
return nil
}
}
// Loop is the main fetcher loop, checking and processing various notification
// events.
func (f *Fetcher) loop() {
// Iterate the block fetching until a quit is requested
fetch := time.NewTimer(0)
for {
// Clean up any expired block fetches
for hash, announce := range f.fetching {
if time.Since(announce.time) > fetchTimeout {
delete(f.announced, hash)
delete(f.fetching, hash)
}
}
// Import any queued blocks that could potentially fit
height := f.chainHeight()
for !f.queue.Empty() {
op := f.queue.PopItem().(*inject)
number := op.block.NumberU64()
// If too high up the chain or phase, continue later
if number > height+1 {
f.queue.Push(op, -float32(op.block.NumberU64()))
break
}
// Otherwise if fresh and still unknown, try and import
if number+maxUncleDist < height || f.getBlock(op.block.Hash()) != nil {
continue
}
f.insert(op.origin, op.block)
}
// Wait for an outside event to occur
select {
case <-f.quit:
// Fetcher terminating, abort all operations
return
case notification := <-f.notify:
// A block was announced, schedule if it's not yet downloading
if _, ok := f.fetching[notification.hash]; ok {
break
}
f.announced[notification.hash] = append(f.announced[notification.hash], notification)
if len(f.announced) == 1 {
f.reschedule(fetch)
}
case op := <-f.inject:
// A direct block insertion was requested, try and fill any pending gaps
f.enqueue(op.origin, op.block)
case hash := <-f.done:
// A pending import finished, remove all traces of the notification
delete(f.announced, hash)
delete(f.fetching, hash)
delete(f.queued, hash)
case <-fetch.C:
// At least one block's timer ran out, check for needing retrieval
request := make(map[string][]common.Hash)
for hash, announces := range f.announced {
if time.Since(announces[0].time) > arriveTimeout {
announce := announces[rand.Intn(len(announces))]
if f.getBlock(hash) == nil {
request[announce.origin] = append(request[announce.origin], hash)
f.fetching[hash] = announce
}
delete(f.announced, hash)
}
}
// Send out all block requests
for _, hashes := range request {
go f.fetching[hashes[0]].fetch(hashes)
}
// Schedule the next fetch if blocks are still pending
f.reschedule(fetch)
case filter := <-f.filter:
// Blocks arrived, extract any explicit fetches, return all else
var blocks types.Blocks
select {
case blocks = <-filter:
case <-f.quit:
return
}
explicit, download := []*types.Block{}, []*types.Block{}
for _, block := range blocks {
hash := block.Hash()
// Filter explicitly requested blocks from hash announcements
if _, ok := f.fetching[hash]; ok {
// Discard if already imported by other means
if f.getBlock(hash) == nil {
explicit = append(explicit, block)
} else {
delete(f.fetching, hash)
}
} else {
download = append(download, block)
}
}
select {
case filter <- download:
case <-f.quit:
return
}
// Schedule the retrieved blocks for ordered import
for _, block := range explicit {
if announce := f.fetching[block.Hash()]; announce != nil {
f.enqueue(announce.origin, block)
}
}
}
}
}
// reschedule resets the specified fetch timer to the next announce timeout.
func (f *Fetcher) reschedule(fetch *time.Timer) {
// Short circuit if no blocks are announced
if len(f.announced) == 0 {
return
}
// Otherwise find the earliest expiring announcement
earliest := time.Now()
for _, announces := range f.announced {
if earliest.After(announces[0].time) {
earliest = announces[0].time
}
}
fetch.Reset(arriveTimeout - time.Since(earliest))
}
// enqueue schedules a new future import operation, if the block to be imported
// has not yet been seen.
func (f *Fetcher) enqueue(peer string, block *types.Block) {
hash := block.Hash()
// Discard any past or too distant blocks
if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist {
glog.V(logger.Detail).Infof("Peer %s: discarded block #%d [%x], distance %d", peer, block.NumberU64(), hash.Bytes()[:4], dist)
return
}
// Schedule the block for future importing
if _, ok := f.queued[hash]; !ok {
f.queued[hash] = struct{}{}
f.queue.Push(&inject{origin: peer, block: block}, -float32(block.NumberU64()))
if glog.V(logger.Debug) {
glog.Infof("Peer %s: queued block #%d [%x], total %v", peer, block.NumberU64(), hash.Bytes()[:4], f.queue.Size())
}
}
}
// 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, if updates
// the phase states accordingly.
func (f *Fetcher) insert(peer string, block *types.Block) {
hash := block.Hash()
// Run the import on a new thread
glog.V(logger.Debug).Infof("Peer %s: importing block #%d [%x]", peer, block.NumberU64(), hash[:4])
go func() {
defer func() { f.done <- hash }()
// If the parent's unknown, abort insertion
parent := f.getBlock(block.ParentHash())
if parent == nil {
return
}
// Quickly validate the header and propagate the block if it passes
if err := f.validateBlock(block, parent); err != nil {
glog.V(logger.Debug).Infof("Peer %s: block #%d [%x] verification failed: %v", peer, block.NumberU64(), hash[:4], err)
f.dropPeer(peer)
return
}
go f.broadcastBlock(block, true)
// Run the actual import and log any issues
if _, err := f.insertChain(types.Blocks{block}); err != nil {
glog.V(logger.Warn).Infof("Peer %s: block #%d [%x] import failed: %v", peer, block.NumberU64(), hash[:4], err)
return
}
// If import succeeded, broadcast the block
go f.broadcastBlock(block, false)
}()
}

397
eth/fetcher/fetcher_test.go Normal file
View File

@ -0,0 +1,397 @@
package fetcher
import (
"encoding/binary"
"errors"
"math/big"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
)
var (
knownHash = common.Hash{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1}
unknownHash = common.Hash{2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2}
bannedHash = common.Hash{3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3}
genesis = createBlock(1, common.Hash{}, knownHash)
)
// idCounter is used by the createHashes method the generate deterministic but unique hashes
var idCounter = int64(2) // #1 is the genesis block
// createHashes generates a batch of hashes rooted at a specific point in the chain.
func createHashes(amount int, root common.Hash) (hashes []common.Hash) {
hashes = make([]common.Hash, amount+1)
hashes[len(hashes)-1] = root
for i := 0; i < len(hashes)-1; i++ {
binary.BigEndian.PutUint64(hashes[i][:8], uint64(idCounter))
idCounter++
}
return
}
// createBlock assembles a new block at the given chain height.
func createBlock(i int, parent, hash common.Hash) *types.Block {
header := &types.Header{Number: big.NewInt(int64(i))}
block := types.NewBlockWithHeader(header)
block.HeaderHash = hash
block.ParentHeaderHash = parent
return block
}
// copyBlock makes a deep copy of a block suitable for local modifications.
func copyBlock(block *types.Block) *types.Block {
return createBlock(int(block.Number().Int64()), block.ParentHeaderHash, block.HeaderHash)
}
// createBlocksFromHashes assembles a collection of blocks, each having a correct
// place in the given hash chain.
func createBlocksFromHashes(hashes []common.Hash) map[common.Hash]*types.Block {
blocks := make(map[common.Hash]*types.Block)
for i := 0; i < len(hashes); i++ {
parent := knownHash
if i < len(hashes)-1 {
parent = hashes[i+1]
}
blocks[hashes[i]] = createBlock(len(hashes)-i, parent, hashes[i])
}
return blocks
}
// fetcherTester is a test simulator for mocking out local block chain.
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
lock sync.RWMutex
}
// newTester creates a new fetcher test mocker.
func newTester() *fetcherTester {
tester := &fetcherTester{
hashes: []common.Hash{knownHash},
blocks: map[common.Hash]*types.Block{knownHash: genesis},
}
tester.fetcher = New(tester.getBlock, tester.verifyBlock, tester.broadcastBlock, tester.chainHeight, tester.insertChain, tester.dropPeer)
tester.fetcher.Start()
return tester
}
// getBlock retrieves a block from the tester's block chain.
func (f *fetcherTester) getBlock(hash common.Hash) *types.Block {
f.lock.RLock()
defer f.lock.RUnlock()
return f.blocks[hash]
}
// verifyBlock is a nop placeholder for the block header verification.
func (f *fetcherTester) verifyBlock(block *types.Block, parent *types.Block) error {
return nil
}
// broadcastBlock is a nop placeholder for the block broadcasting.
func (f *fetcherTester) broadcastBlock(block *types.Block, propagate bool) {
}
// chainHeight retrieves the current height (block number) of the chain.
func (f *fetcherTester) chainHeight() uint64 {
f.lock.RLock()
defer f.lock.RUnlock()
return f.blocks[f.hashes[len(f.hashes)-1]].NumberU64()
}
// insertChain injects a new blocks into the simulated chain.
func (f *fetcherTester) insertChain(blocks types.Blocks) (int, error) {
f.lock.Lock()
defer f.lock.Unlock()
for i, block := range blocks {
// Make sure the parent in known
if _, ok := f.blocks[block.ParentHash()]; !ok {
return i, errors.New("unknown parent")
}
// Discard any new blocks if the same height already exists
if block.NumberU64() <= f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() {
return i, nil
}
// Otherwise build our current chain
f.hashes = append(f.hashes, block.Hash())
f.blocks[block.Hash()] = block
}
return 0, nil
}
// dropPeer is a nop placeholder for the peer removal.
func (f *fetcherTester) dropPeer(peer string) {
}
// peerFetcher retrieves a fetcher associated with a simulated peer.
func (f *fetcherTester) makeFetcher(blocks map[common.Hash]*types.Block) blockRequesterFn {
// Copy all the blocks to ensure they are not tampered with
closure := make(map[common.Hash]*types.Block)
for hash, block := range blocks {
closure[hash] = copyBlock(block)
}
// Create a function that returns blocks from the closure
return func(hashes []common.Hash) error {
// Gather the blocks to return
blocks := make([]*types.Block, 0, len(hashes))
for _, hash := range hashes {
if block, ok := closure[hash]; ok {
blocks = append(blocks, block)
}
}
// Return on a new thread
go f.fetcher.Filter(blocks)
return nil
}
}
// Tests that a fetcher accepts block announcements and initiates retrievals for
// them, successfully importing into the local chain.
func TestSequentialAnnouncements(t *testing.T) {
// Create a chain of blocks to import
targetBlocks := 24
hashes := createHashes(targetBlocks, knownHash)
blocks := createBlocksFromHashes(hashes)
tester := newTester()
fetcher := tester.makeFetcher(blocks)
// Iteratively announce blocks until all are imported
for i := len(hashes) - 1; i >= 0; i-- {
tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
time.Sleep(50 * time.Millisecond)
}
if imported := len(tester.blocks); imported != targetBlocks+1 {
t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+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 TestConcurrentAnnouncements(t *testing.T) {
// Create a chain of blocks to import
targetBlocks := 24
hashes := createHashes(targetBlocks, knownHash)
blocks := createBlocksFromHashes(hashes)
// Assemble a tester with a built in counter for the requests
tester := newTester()
fetcher := tester.makeFetcher(blocks)
counter := uint32(0)
wrapper := func(hashes []common.Hash) error {
atomic.AddUint32(&counter, uint32(len(hashes)))
return fetcher(hashes)
}
// Iteratively announce blocks until all are imported
for i := len(hashes) - 1; i >= 0; i-- {
tester.fetcher.Notify("first", hashes[i], time.Now().Add(-arriveTimeout), wrapper)
tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout+time.Millisecond), wrapper)
tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout-time.Millisecond), wrapper)
time.Sleep(50 * time.Millisecond)
}
if imported := len(tester.blocks); imported != targetBlocks+1 {
t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1)
}
// Make sure no blocks were retrieved twice
if int(counter) != targetBlocks {
t.Fatalf("retrieval count mismatch: have %v, want %v", counter, targetBlocks)
}
}
// Tests that announcements arriving while a previous is being fetched still
// results in a valid import.
func TestOverlappingAnnouncements(t *testing.T) {
// Create a chain of blocks to import
targetBlocks := 24
hashes := createHashes(targetBlocks, knownHash)
blocks := createBlocksFromHashes(hashes)
tester := newTester()
fetcher := tester.makeFetcher(blocks)
// Iteratively announce blocks, but overlap them continuously
delay, overlap := 50*time.Millisecond, time.Duration(5)
for i := len(hashes) - 1; i >= 0; i-- {
tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout+overlap*delay), fetcher)
time.Sleep(delay)
}
time.Sleep(overlap * delay)
if imported := len(tester.blocks); imported != targetBlocks+1 {
t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1)
}
}
// Tests that announces already being retrieved will not be duplicated.
func TestPendingDeduplication(t *testing.T) {
// Create a hash and corresponding block
hashes := createHashes(1, knownHash)
blocks := createBlocksFromHashes(hashes)
// Assemble a tester with a built in counter and delayed fetcher
tester := newTester()
fetcher := tester.makeFetcher(blocks)
delay := 50 * time.Millisecond
counter := uint32(0)
wrapper := func(hashes []common.Hash) error {
atomic.AddUint32(&counter, uint32(len(hashes)))
// Simulate a long running fetch
go func() {
time.Sleep(delay)
fetcher(hashes)
}()
return nil
}
// Announce the same block many times until it's fetched (wait for any pending ops)
for tester.getBlock(hashes[0]) == nil {
tester.fetcher.Notify("repeater", hashes[0], time.Now().Add(-arriveTimeout), wrapper)
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)
}
}
// Tests that announcements retrieved in a random order are cached and eventually
// imported when all the gaps are filled in.
func TestRandomArrivalImport(t *testing.T) {
// Create a chain of blocks to import, and choose one to delay
targetBlocks := 24
hashes := createHashes(targetBlocks, knownHash)
blocks := createBlocksFromHashes(hashes)
skip := targetBlocks / 2
tester := newTester()
fetcher := tester.makeFetcher(blocks)
// Iteratively announce blocks, skipping one entry
for i := len(hashes) - 1; i >= 0; i-- {
if i != skip {
tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
time.Sleep(50 * time.Millisecond)
}
}
// Finally announce the skipped entry and check full import
tester.fetcher.Notify("valid", hashes[skip], time.Now().Add(-arriveTimeout), fetcher)
time.Sleep(50 * time.Millisecond)
if imported := len(tester.blocks); imported != targetBlocks+1 {
t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1)
}
}
// Tests that direct block enqueues (due to block propagation vs. hash announce)
// are correctly schedule, filling and import queue gaps.
func TestQueueGapFill(t *testing.T) {
// Create a chain of blocks to import, and choose one to not announce at all
targetBlocks := 24
hashes := createHashes(targetBlocks, knownHash)
blocks := createBlocksFromHashes(hashes)
skip := targetBlocks / 2
tester := newTester()
fetcher := tester.makeFetcher(blocks)
// Iteratively announce blocks, skipping one entry
for i := len(hashes) - 1; i >= 0; i-- {
if i != skip {
tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
time.Sleep(50 * time.Millisecond)
}
}
// Fill the missing block directly as if propagated
tester.fetcher.Enqueue("valid", blocks[hashes[skip]])
time.Sleep(50 * time.Millisecond)
if imported := len(tester.blocks); imported != targetBlocks+1 {
t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1)
}
}
// Tests that blocks arriving from various sources (multiple propagations, hash
// announces, etc) do not get scheduled for import multiple times.
func TestImportDeduplication(t *testing.T) {
// Create two blocks to import (one for duplication, the other for stalling)
hashes := createHashes(2, knownHash)
blocks := createBlocksFromHashes(hashes)
// Create the tester and wrap the importer with a counter
tester := newTester()
fetcher := tester.makeFetcher(blocks)
counter := uint32(0)
tester.fetcher.insertChain = func(blocks types.Blocks) (int, error) {
atomic.AddUint32(&counter, uint32(len(blocks)))
return tester.insertChain(blocks)
}
// Announce the duplicating block, wait for retrieval, and also propagate directly
tester.fetcher.Notify("valid", hashes[0], time.Now().Add(-arriveTimeout), fetcher)
time.Sleep(50 * time.Millisecond)
tester.fetcher.Enqueue("valid", blocks[hashes[0]])
tester.fetcher.Enqueue("valid", blocks[hashes[0]])
tester.fetcher.Enqueue("valid", blocks[hashes[0]])
// Fill the missing block directly as if propagated, and check import uniqueness
tester.fetcher.Enqueue("valid", blocks[hashes[1]])
time.Sleep(50 * time.Millisecond)
if imported := len(tester.blocks); imported != 3 {
t.Fatalf("synchronised block mismatch: have %v, want %v", imported, 3)
}
if counter != 2 {
t.Fatalf("import invocation count mismatch: have %v, want %v", counter, 2)
}
}
// Tests that blocks with numbers much lower or higher than out current head get
// discarded no prevent wasting resources on useless blocks from faulty peers.
func TestDistantDiscarding(t *testing.T) {
// Create a long chain to import
hashes := createHashes(3*maxQueueDist, knownHash)
blocks := createBlocksFromHashes(hashes)
head := hashes[len(hashes)/2]
// Create a tester and simulate a head block being the middle of the above chain
tester := newTester()
tester.hashes = []common.Hash{head}
tester.blocks = map[common.Hash]*types.Block{head: blocks[head]}
// Ensure that a block with a lower number than the threshold is discarded
tester.fetcher.Enqueue("lower", blocks[hashes[0]])
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[len(hashes)-1]])
time.Sleep(10 * time.Millisecond)
if !tester.fetcher.queue.Empty() {
t.Fatalf("fetcher queued future block")
}
}

229
eth/handler.go
View File

@ -3,14 +3,15 @@ package eth
import (
"fmt"
"math"
"math/big"
"sync"
"time"
"github.com/ethereum/go-ethereum/pow"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/eth/fetcher"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
@ -45,6 +46,7 @@ type ProtocolManager struct {
txpool txPool
chainman *core.ChainManager
downloader *downloader.Downloader
fetcher *fetcher.Fetcher
peers *peerSet
SubProtocol p2p.Protocol
@ -54,11 +56,9 @@ type ProtocolManager struct {
minedBlockSub event.Subscription
// channels for fetcher, syncer, txsyncLoop
newPeerCh chan *peer
newHashCh chan []*blockAnnounce
newBlockCh chan chan []*types.Block
txsyncCh chan *txsync
quitSync chan struct{}
newPeerCh chan *peer
txsyncCh chan *txsync
quitSync chan struct{}
// wait group is used for graceful shutdowns during downloading
// and processing
@ -68,31 +68,37 @@ type ProtocolManager struct {
// NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable
// with the ethereum network.
func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpool txPool, chainman *core.ChainManager) *ProtocolManager {
func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpool txPool, pow pow.PoW, chainman *core.ChainManager) *ProtocolManager {
// Create the protocol manager and initialize peer handlers
manager := &ProtocolManager{
eventMux: mux,
txpool: txpool,
chainman: chainman,
peers: newPeerSet(),
newPeerCh: make(chan *peer, 1),
newHashCh: make(chan []*blockAnnounce, 1),
newBlockCh: make(chan chan []*types.Block),
txsyncCh: make(chan *txsync),
quitSync: make(chan struct{}),
eventMux: mux,
txpool: txpool,
chainman: chainman,
peers: newPeerSet(),
newPeerCh: make(chan *peer, 1),
txsyncCh: make(chan *txsync),
quitSync: make(chan struct{}),
}
manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.chainman.InsertChain, manager.removePeer)
manager.SubProtocol = p2p.Protocol{
Name: "eth",
Version: uint(protocolVersion),
Length: ProtocolLength,
Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
peer := manager.newPeer(protocolVersion, networkId, p, rw)
manager.newPeerCh <- peer
return manager.handle(peer)
},
}
// Construct the different synchronisation mechanisms
manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.chainman.InsertChain, manager.removePeer)
validator := func(block *types.Block, parent *types.Block) error {
return core.ValidateHeader(pow, block.Header(), parent.Header(), true)
}
heighter := func() uint64 {
return manager.chainman.CurrentBlock().NumberU64()
}
manager.fetcher = fetcher.New(manager.chainman.GetBlock, validator, manager.BroadcastBlock, heighter, manager.chainman.InsertChain, manager.removePeer)
return manager
}
@ -126,7 +132,6 @@ func (pm *ProtocolManager) Start() {
// start sync handlers
go pm.syncer()
go pm.fetcher()
go pm.txsyncLoop()
}
@ -185,7 +190,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
return nil
}
func (self *ProtocolManager) handleMsg(p *peer) error {
func (pm *ProtocolManager) handleMsg(p *peer) error {
msg, err := p.rw.ReadMsg()
if err != nil {
return err
@ -215,7 +220,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error {
RemoteId: p.ID().String(),
})
}
self.txpool.AddTransactions(txs)
pm.txpool.AddTransactions(txs)
case GetBlockHashesMsg:
var request getBlockHashesMsgData
@ -227,7 +232,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error {
request.Amount = uint64(downloader.MaxHashFetch)
}
hashes := self.chainman.GetBlockHashesFromHash(request.Hash, request.Amount)
hashes := pm.chainman.GetBlockHashesFromHash(request.Hash, request.Amount)
if glog.V(logger.Debug) {
if len(hashes) == 0 {
@ -245,40 +250,50 @@ func (self *ProtocolManager) handleMsg(p *peer) error {
if err := msgStream.Decode(&hashes); err != nil {
break
}
err := self.downloader.DeliverHashes(p.id, hashes)
err := pm.downloader.DeliverHashes(p.id, hashes)
if err != nil {
glog.V(logger.Debug).Infoln(err)
}
case GetBlocksMsg:
var blocks []*types.Block
// Decode the retrieval message
msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
if _, err := msgStream.List(); err != nil {
return err
}
// Gather blocks until the fetch or network limits is reached
var (
i int
totalsize common.StorageSize
hash common.Hash
bytes common.StorageSize
hashes []common.Hash
blocks []*types.Block
)
for {
i++
var hash common.Hash
err := msgStream.Decode(&hash)
if err == rlp.EOL {
break
} else if err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
hashes = append(hashes, hash)
block := self.chainman.GetBlock(hash)
if block != nil {
// Retrieve the requested block, stopping if enough was found
if block := pm.chainman.GetBlock(hash); block != nil {
blocks = append(blocks, block)
totalsize += block.Size()
bytes += block.Size()
if len(blocks) >= downloader.MaxBlockFetch || bytes > maxBlockRespSize {
break
}
}
if i == downloader.MaxBlockFetch || totalsize > maxBlockRespSize {
break
}
if glog.V(logger.Detail) && len(blocks) == 0 && len(hashes) > 0 {
list := "["
for _, hash := range hashes {
list += fmt.Sprintf("%x, ", hash[:4])
}
list = list[:len(list)-2] + "]"
glog.Infof("Peer %s: no blocks found for requested hashes %s", p.id, list)
}
return p.sendBlocks(blocks)
@ -291,20 +306,13 @@ func (self *ProtocolManager) handleMsg(p *peer) error {
glog.V(logger.Detail).Infoln("Decode error", err)
blocks = nil
}
// Filter out any explicitly requested blocks (cascading select to get blocking back to peer)
filter := make(chan []*types.Block)
select {
case <-self.quitSync:
case self.newBlockCh <- filter:
select {
case <-self.quitSync:
case filter <- blocks:
select {
case <-self.quitSync:
case blocks := <-filter:
self.downloader.DeliverBlocks(p.id, blocks)
}
}
// Update the receive timestamp of each block
for i:=0; i<len(blocks); i++ {
blocks[i].ReceivedAt = msg.ReceivedAt
}
// Filter out any explicitly requested blocks, deliver the rest to the downloader
if blocks := pm.fetcher.Filter(blocks); len(blocks) > 0 {
pm.downloader.DeliverBlocks(p.id, blocks)
}
case NewBlockHashesMsg:
@ -323,26 +331,16 @@ func (self *ProtocolManager) handleMsg(p *peer) error {
// Schedule all the unknown hashes for retrieval
unknown := make([]common.Hash, 0, len(hashes))
for _, hash := range hashes {
if !self.chainman.HasBlock(hash) {
if !pm.chainman.HasBlock(hash) {
unknown = append(unknown, hash)
}
}
announces := make([]*blockAnnounce, len(unknown))
for i, hash := range unknown {
announces[i] = &blockAnnounce{
hash: hash,
peer: p,
time: time.Now(),
}
}
if len(announces) > 0 {
select {
case self.newHashCh <- announces:
case <-self.quitSync:
}
for _, hash := range unknown {
pm.fetcher.Notify(p.id, hash, time.Now(), p.requestBlocks)
}
case NewBlockMsg:
// Retrieve and decode the propagated block
var request newBlockMsgData
if err := msg.Decode(&request); err != nil {
return errResp(ErrDecode, "%v: %v", msg, err)
@ -352,9 +350,24 @@ func (self *ProtocolManager) handleMsg(p *peer) error {
}
request.Block.ReceivedAt = msg.ReceivedAt
if err := self.importBlock(p, request.Block, request.TD); err != nil {
return err
}
// Mark the block's arrival for whatever reason
_, chainHead, _ := pm.chainman.Status()
jsonlogger.LogJson(&logger.EthChainReceivedNewBlock{
BlockHash: request.Block.Hash().Hex(),
BlockNumber: request.Block.Number(),
ChainHeadHash: chainHead.Hex(),
BlockPrevHash: request.Block.ParentHash().Hex(),
RemoteId: p.ID().String(),
})
// Mark the peer as owning the block and schedule it for import
p.blockHashes.Add(request.Block.Hash())
p.SetHead(request.Block.Hash())
pm.fetcher.Enqueue(p.id, request.Block)
// TODO: Schedule a sync to cover potential gaps (this needs proto update)
p.SetTd(request.TD)
go pm.synchronise(p)
default:
return errResp(ErrInvalidMsgCode, "%v", msg.Code)
@ -362,76 +375,27 @@ func (self *ProtocolManager) handleMsg(p *peer) error {
return nil
}
// importBlocks injects a new block retrieved from the given peer into the chain
// manager.
func (pm *ProtocolManager) importBlock(p *peer, block *types.Block, td *big.Int) error {
// BroadcastBlock will either propagate a block to a subset of it's peers, or
// will only announce it's availability (depending what's requested).
func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) {
hash := block.Hash()
// Mark the block as present at the remote node (don't duplicate already held data)
p.blockHashes.Add(hash)
p.SetHead(hash)
if td != nil {
p.SetTd(td)
}
// Log the block's arrival
_, chainHead, _ := pm.chainman.Status()
jsonlogger.LogJson(&logger.EthChainReceivedNewBlock{
BlockHash: hash.Hex(),
BlockNumber: block.Number(),
ChainHeadHash: chainHead.Hex(),
BlockPrevHash: block.ParentHash().Hex(),
RemoteId: p.ID().String(),
})
// If the block's already known or its difficulty is lower than ours, drop
if pm.chainman.HasBlock(hash) {
p.SetTd(pm.chainman.GetBlock(hash).Td) // update the peer's TD to the real value
return nil
}
if td != nil && pm.chainman.Td().Cmp(td) > 0 && new(big.Int).Add(block.Number(), big.NewInt(7)).Cmp(pm.chainman.CurrentBlock().Number()) < 0 {
glog.V(logger.Debug).Infof("[%s] dropped block %v due to low TD %v\n", p.id, block.Number(), td)
return nil
}
// Attempt to insert the newly received block and propagate to our peers
if pm.chainman.HasBlock(block.ParentHash()) {
if _, err := pm.chainman.InsertChain(types.Blocks{block}); err != nil {
glog.V(logger.Error).Infoln("removed peer (", p.id, ") due to block error", err)
return err
}
if td != nil && block.Td.Cmp(td) != 0 {
err := fmt.Errorf("invalid TD on block(%v) from peer(%s): block.td=%v, request.td=%v", block.Number(), p.id, block.Td, td)
glog.V(logger.Error).Infoln(err)
return err
}
pm.BroadcastBlock(hash, block)
return nil
}
// Parent of the block is unknown, try to sync with this peer if it seems to be good
if td != nil {
go pm.synchronise(p)
}
return nil
}
// BroadcastBlock will propagate the block to a subset of its connected peers,
// only notifying the rest of the block's appearance.
func (pm *ProtocolManager) BroadcastBlock(hash common.Hash, block *types.Block) {
// Retrieve all the target peers and split between full broadcast or only notification
peers := pm.peers.PeersWithoutBlock(hash)
split := int(math.Sqrt(float64(len(peers))))
transfer := peers[:split]
notify := peers[split:]
// Send out the data transfers and the notifications
for _, peer := range notify {
peer.sendNewBlockHashes([]common.Hash{hash})
// If propagation is requested, send to a subset of the peer
if propagate {
transfer := peers[:int(math.Sqrt(float64(len(peers))))]
for _, peer := range transfer {
peer.sendNewBlock(block)
}
glog.V(logger.Detail).Infof("propagated block %x to %d peers in %v", hash[:4], len(transfer), time.Since(block.ReceivedAt))
}
glog.V(logger.Detail).Infoln("broadcast hash to", len(notify), "peers.")
for _, peer := range transfer {
peer.sendNewBlock(block)
// Otherwise if the block is indeed in out own chain, announce it
if pm.chainman.HasBlock(hash) {
for _, peer := range peers {
peer.sendNewBlockHashes([]common.Hash{hash})
}
glog.V(logger.Detail).Infof("announced block %x to %d peers in %v", hash[:4], len(peers), time.Since(block.ReceivedAt))
}
glog.V(logger.Detail).Infoln("broadcast block to", len(transfer), "peers. Total processing time:", time.Since(block.ReceivedAt))
}
// BroadcastTx will propagate the block to its connected peers. It will sort
@ -453,7 +417,8 @@ func (self *ProtocolManager) minedBroadcastLoop() {
for obj := range self.minedBlockSub.Chan() {
switch ev := obj.(type) {
case core.NewMinedBlockEvent:
self.BroadcastBlock(ev.Block.Hash(), ev.Block)
self.BroadcastBlock(ev.Block, false)
self.BroadcastBlock(ev.Block, true)
}
}
}

2
eth/protocol_test.go
View File

@ -167,7 +167,7 @@ func newProtocolManagerForTesting(txAdded chan<- []*types.Transaction) *Protocol
db, _ = ethdb.NewMemDatabase()
chain, _ = core.NewChainManager(core.GenesisBlock(0, db), db, db, core.FakePow{}, em)
txpool = &fakeTxPool{added: txAdded}
pm = NewProtocolManager(ProtocolVersion, 0, em, txpool, chain)
pm = NewProtocolManager(ProtocolVersion, 0, em, txpool, core.FakePow{}, chain)
)
pm.Start()
return pm

145
eth/sync.go
View File

@ -12,11 +12,8 @@ import (
)
const (
forceSyncCycle = 10 * time.Second // Time interval to force syncs, even if few peers are available
notifyCheckCycle = 100 * time.Millisecond // Time interval to allow hash notifies to fulfill before hard fetching
notifyArriveTimeout = 500 * time.Millisecond // Time allowance before an announced block is explicitly requested
notifyFetchTimeout = 5 * time.Second // Maximum alloted time to return an explicitly requested block
minDesiredPeerCount = 5 // Amount of peers desired to start syncing
forceSyncCycle = 10 * time.Second // Time interval to force syncs, even if few peers are available
minDesiredPeerCount = 5 // Amount of peers desired to start syncing
// This is the target size for the packs of transactions sent by txsyncLoop.
// A pack can get larger than this if a single transactions exceeds this size.
@ -119,140 +116,15 @@ func (pm *ProtocolManager) txsyncLoop() {
}
}
// fetcher is responsible for collecting hash notifications, and periodically
// checking all unknown ones and individually fetching them.
func (pm *ProtocolManager) fetcher() {
announces := make(map[common.Hash][]*blockAnnounce)
request := make(map[*peer][]common.Hash)
pending := make(map[common.Hash]*blockAnnounce)
cycle := time.Tick(notifyCheckCycle)
done := make(chan common.Hash)
// Iterate the block fetching until a quit is requested
for {
select {
case notifications := <-pm.newHashCh:
// A batch of hashes the notified, schedule them for retrieval
glog.V(logger.Debug).Infof("Scheduling %d hash announcements from %s", len(notifications), notifications[0].peer.id)
for _, announce := range notifications {
// Skip if it's already pending fetch
if _, ok := pending[announce.hash]; ok {
continue
}
// Otherwise queue up the peer as a potential source
announces[announce.hash] = append(announces[announce.hash], announce)
}
case hash := <-done:
// A pending import finished, remove all traces
delete(pending, hash)
case <-cycle:
// Clean up any expired block fetches
for hash, announce := range pending {
if time.Since(announce.time) > notifyFetchTimeout {
delete(pending, hash)
}
}
// Check if any notified blocks failed to arrive
for hash, all := range announces {
if time.Since(all[0].time) > notifyArriveTimeout {
announce := all[rand.Intn(len(all))]
if !pm.chainman.HasBlock(hash) {
request[announce.peer] = append(request[announce.peer], hash)
pending[hash] = announce
}
delete(announces, hash)
}
}
if len(request) == 0 {
break
}
// Send out all block requests
for peer, hashes := range request {
glog.V(logger.Debug).Infof("Explicitly fetching %d blocks from %s", len(hashes), peer.id)
go peer.requestBlocks(hashes)
}
request = make(map[*peer][]common.Hash)
case filter := <-pm.newBlockCh:
// Blocks arrived, extract any explicit fetches, return all else
var blocks types.Blocks
select {
case blocks = <-filter:
case <-pm.quitSync:
return
}
explicit, download := []*types.Block{}, []*types.Block{}
for _, block := range blocks {
hash := block.Hash()
// Filter explicitly requested blocks from hash announcements
if _, ok := pending[hash]; ok {
// Discard if already imported by other means
if !pm.chainman.HasBlock(hash) {
explicit = append(explicit, block)
} else {
delete(pending, hash)
}
} else {
download = append(download, block)
}
}
select {
case filter <- download:
case <-pm.quitSync:
return
}
// Create a closure with the retrieved blocks and origin peers
peers := make([]*peer, 0, len(explicit))
blocks = make([]*types.Block, 0, len(explicit))
for _, block := range explicit {
hash := block.Hash()
if announce := pending[hash]; announce != nil {
// Drop the block if it surely cannot fit
if pm.chainman.HasBlock(hash) || !pm.chainman.HasBlock(block.ParentHash()) {
// delete(pending, hash) // if we drop, it will re-fetch it, wait for timeout?
continue
}
// Otherwise accumulate for import
peers = append(peers, announce.peer)
blocks = append(blocks, block)
}
}
// If any explicit fetches were replied to, import them
if count := len(blocks); count > 0 {
glog.V(logger.Debug).Infof("Importing %d explicitly fetched blocks", len(blocks))
go func() {
// Make sure all hashes are cleaned up
for _, block := range blocks {
hash := block.Hash()
defer func() { done <- hash }()
}
// Try and actually import the blocks
for i := 0; i < len(blocks); i++ {
if err := pm.importBlock(peers[i], blocks[i], nil); err != nil {
glog.V(logger.Detail).Infof("Failed to import explicitly fetched block: %v", err)
return
}
}
}()
}
case <-pm.quitSync:
return
}
}
}
// syncer is responsible for periodically synchronising with the network, both
// downloading hashes and blocks as well as retrieving cached ones.
// downloading hashes and blocks as well as handling the announcement handler.
func (pm *ProtocolManager) syncer() {
// Abort any pending syncs if we terminate
// Start and ensure cleanup of sync mechanisms
pm.fetcher.Start()
defer pm.fetcher.Stop()
defer pm.downloader.Terminate()
// Wait for different events to fire synchronisation operations
forceSync := time.Tick(forceSyncCycle)
for {
select {
@ -273,8 +145,7 @@ func (pm *ProtocolManager) syncer() {
}
}
// synchronise tries to sync up our local block chain with a remote peer, both
// adding various sanity checks as well as wrapping it with various log entries.
// synchronise tries to sync up our local block chain with a remote peer.
func (pm *ProtocolManager) synchronise(peer *peer) {
// Short circuit if no peers are available
if peer == nil {