forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-24 05:36:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

eth/fetcher: use announce numbers for distance checking

Browse source
This commit is contained in:
Péter Szilágyi 2015-08-19 17:45:19 +03:00
parent ee35dfdbb6
commit 56e1dc406d
3 changed files with 96 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

68
eth/fetcher/fetcher.go
View file

@ -93,10 +93,12 @@ type headerFilterTask struct {
time time.Time // Arrival time of the headers
}
// body represents the data contents (transactions and uncles) of a block.
type body struct {
transactions []*types.Transaction
uncles []*types.Header
// headerFilterTask represents a batch of block bodies (transactions and uncles)
// needing fetcher filtering.
type bodyFilterTask struct {
transactions [][]*types.Transaction // Collection of transactions per block bodies
uncles [][]*types.Header // Collection of uncles per block bodies
time time.Time // Arrival time of the blocks' contents
}
// inject represents a schedules import operation.
@ -114,7 +116,7 @@ type Fetcher struct {
blockFilter chan chan []*types.Block
headerFilter chan chan *headerFilterTask
bodyFilter chan chan []*body
bodyFilter chan chan *bodyFilterTask
done chan common.Hash
quit chan struct{}
@ -151,7 +153,7 @@ func New(getBlock blockRetrievalFn, validateBlock blockValidatorFn, broadcastBlo
inject: make(chan *inject),
blockFilter: make(chan chan []*types.Block),
headerFilter: make(chan chan *headerFilterTask),
bodyFilter: make(chan chan []*body),
bodyFilter: make(chan chan *bodyFilterTask),
done: make(chan common.Hash),
quit: make(chan struct{}),
announces: make(map[string]int),
@ -277,16 +279,11 @@ func (f *Fetcher) FilterHeaders(headers []*types.Header, time time.Time) []*type
// FilterBodies extracts all the block bodies that were explicitly requested by
// the fetcher, returning those that should be handled differently.
func (f *Fetcher) FilterBodies(transactions [][]*types.Transaction, uncles [][]*types.Header) ([][]*types.Transaction, [][]*types.Header) {
func (f *Fetcher) FilterBodies(transactions [][]*types.Transaction, uncles [][]*types.Header, time time.Time) ([][]*types.Transaction, [][]*types.Header) {
glog.V(logger.Detail).Infof("[eth/62] filtering %d:%d bodies", len(transactions), len(uncles))
// Assemble the body contents into a single data struct
bodies := make([]*body, 0, len(transactions))
for i := 0; i < len(transactions) && i < len(uncles); i++ {
bodies = append(bodies, &body{transactions[i], uncles[i]})
}
// Send the filter channel to the fetcher
filter := make(chan []*body)
filter := make(chan *bodyFilterTask)
select {
case f.bodyFilter <- filter:
@ -295,19 +292,14 @@ func (f *Fetcher) FilterBodies(transactions [][]*types.Transaction, uncles [][]*
}
// Request the filtering of the body list
select {
case filter <- bodies:
case filter <- &bodyFilterTask{transactions: transactions, uncles: uncles, time: time}:
case <-f.quit:
return nil, nil
}
// Retrieve the bodies remaining after filtering
select {
case bodies := <-filter:
transactions, uncles = transactions[:0], uncles[:0]
for _, body := range bodies {
transactions = append(transactions, body.transactions)
uncles = append(uncles, body.uncles)
}
return transactions, uncles
case task := <-filter:
return task.transactions, task.uncles
case <-f.quit:
return nil, nil
}
@ -361,6 +353,14 @@ func (f *Fetcher) loop() {
glog.V(logger.Debug).Infof("Peer %s: exceeded outstanding announces (%d)", notification.origin, hashLimit)
break
}
// If we have a valid block number, check that it's potentially useful
if notification.number > 0 {
if dist := int64(notification.number) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist {
glog.V(logger.Debug).Infof("[eth/62] Peer %s: discarded announcement #%d [%x], distance %d", notification.origin, notification.number, notification.hash[:4], dist)
discardMeter.Mark(1)
break
}
}
// All is well, schedule the announce if block's not yet downloading
if _, ok := f.fetching[notification.hash]; ok {
break
@ -552,49 +552,51 @@ func (f *Fetcher) loop() {
}
case filter := <-f.bodyFilter:
// Block bodies arrived, extract any explicit completions, return all else
var bodies []*body
// Block bodies arrived, extract any explicitly requested blocks, return the rest
var task *bodyFilterTask
select {
case bodies = <-filter:
case task = <-filter:
case <-f.quit:
return
}
explicit, download := []*types.Block{}, []*body{}
for _, body := range bodies {
blocks := []*types.Block{}
for i := 0; i < len(task.transactions) && i < len(task.uncles); i++ {
// Match up a body to any possible completion request
matched := false
for hash, announce := range f.completing {
if f.queued[hash] == nil {
txnHash := types.DeriveSha(types.Transactions(body.transactions))
uncleHash := types.CalcUncleHash(body.uncles)
txnHash := types.DeriveSha(types.Transactions(task.transactions[i]))
uncleHash := types.CalcUncleHash(task.uncles[i])
if txnHash == announce.header.TxHash && uncleHash == announce.header.UncleHash {
// Mark the body matched, reassemble if still unknown
matched = true
if f.getBlock(hash) == nil {
explicit = append(explicit, types.NewBlockWithHeader(announce.header).WithBody(body.transactions, body.uncles))
blocks = append(blocks, types.NewBlockWithHeader(announce.header).WithBody(task.transactions[i], task.uncles[i]))
} else {
f.forgetHash(hash)
}
}
}
}
if !matched {
download = append(download, body)
if matched {
task.transactions = append(task.transactions[:i], task.transactions[i+1:]...)
task.uncles = append(task.uncles[:i], task.uncles[i+1:]...)
i--
continue
}
}
select {
case filter <- download:
case filter <- task:
case <-f.quit:
return
}
// Schedule the retrieved blocks for ordered import
for _, block := range explicit {
for _, block := range blocks {
if announce := f.completing[block.Hash()]; announce != nil {
f.enqueue(announce.origin, block)
}

77
eth/fetcher/fetcher_test.go
View file

@ -169,7 +169,7 @@ func (f *fetcherTester) makeHeaderFetcher(blocks map[common.Hash]*types.Block, d
}
// makeBodyFetcher retrieves a block body fetcher associated with a simulated peer.
func (f *fetcherTester) makeBodyFetcher(blocks map[common.Hash]*types.Block) bodyRequesterFn {
func (f *fetcherTester) makeBodyFetcher(blocks map[common.Hash]*types.Block, drift time.Duration) bodyRequesterFn {
closure := make(map[common.Hash]*types.Block)
for hash, block := range blocks {
closure[hash] = block
@ -187,7 +187,7 @@ func (f *fetcherTester) makeBodyFetcher(blocks map[common.Hash]*types.Block) bod
}
}
// Return on a new thread
go f.fetcher.FilterBodies(transactions, uncles)
go f.fetcher.FilterBodies(transactions, uncles, time.Now().Add(drift))
return nil
}
@ -242,7 +242,7 @@ func testConcurrentAnnouncements(t *testing.T, protocol int) {
tester := newTester()
blockFetcher := tester.makeBlockFetcher(blocks)
headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher(blocks)
bodyFetcher := tester.makeBodyFetcher(blocks, 0)
counter := uint32(0)
blockWrapper := func(hashes []common.Hash) error {
@ -292,7 +292,7 @@ func testOverlappingAnnouncements(t *testing.T, protocol int) {
tester := newTester()
blockFetcher := tester.makeBlockFetcher(blocks)
headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher(blocks)
bodyFetcher := tester.makeBodyFetcher(blocks, 0)
// Iteratively announce blocks, but overlap them continuously
fetching := make(chan []common.Hash)
@ -330,7 +330,7 @@ func testPendingDeduplication(t *testing.T, protocol int) {
tester := newTester()
blockFetcher := tester.makeBlockFetcher(blocks)
headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher(blocks)
bodyFetcher := tester.makeBodyFetcher(blocks, 0)
delay := 50 * time.Millisecond
counter := uint32(0)
@ -390,7 +390,7 @@ func testRandomArrivalImport(t *testing.T, protocol int) {
tester := newTester()
blockFetcher := tester.makeBlockFetcher(blocks)
headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher(blocks)
bodyFetcher := tester.makeBodyFetcher(blocks, 0)
// Iteratively announce blocks, skipping one entry
imported := make(chan *types.Block, len(hashes)-1)
@ -431,7 +431,7 @@ func testQueueGapFill(t *testing.T, protocol int) {
tester := newTester()
blockFetcher := tester.makeBlockFetcher(blocks)
headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher(blocks)
bodyFetcher := tester.makeBodyFetcher(blocks, 0)
// Iteratively announce blocks, skipping one entry
imported := make(chan *types.Block, len(hashes)-1)
@ -467,7 +467,7 @@ func testImportDeduplication(t *testing.T, protocol int) {
tester := newTester()
blockFetcher := tester.makeBlockFetcher(blocks)
headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher(blocks)
bodyFetcher := tester.makeBodyFetcher(blocks, 0)
counter := uint32(0)
tester.fetcher.insertChain = func(blocks types.Blocks) (int, error) {
@ -502,31 +502,74 @@ func testImportDeduplication(t *testing.T, protocol int) {
}
// 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
// discarded to prevent wasting resources on useless blocks from faulty peers.
func TestDistantPropagationDiscarding(t *testing.T) {
// Create a long chain to import and define the discard boundaries
hashes, blocks := makeChain(3*maxQueueDist, 0, genesis)
head := hashes[len(hashes)/2]
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.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]])
tester.fetcher.Enqueue("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[len(hashes)-1]])
tester.fetcher.Enqueue("higher", blocks[hashes[high]])
time.Sleep(10 * time.Millisecond)
if !tester.fetcher.queue.Empty() {
t.Fatalf("fetcher queued future block")
}
}
// 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 testDistantAnnouncementDiscarding(t *testing.T, protocol int) {
// Create a long chain to import and define the discard boundaries
hashes, blocks := makeChain(3*maxQueueDist, 0, genesis)
head := hashes[len(hashes)/2]
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.hashes = []common.Hash{head}
tester.blocks = map[common.Hash]*types.Block{head: blocks[head]}
headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack)
bodyFetcher := tester.makeBodyFetcher(blocks, 0)
fetching := make(chan struct{}, 2)
tester.fetcher.fetchingHook = func(hashes []common.Hash) { fetching <- struct{}{} }
// Ensure that a block with a lower number than the threshold is discarded
tester.fetcher.Notify("lower", hashes[low], blocks[hashes[low]].NumberU64(), time.Now().Add(-arriveTimeout), nil, headerFetcher, bodyFetcher)
select {
case <-time.After(50 * time.Millisecond):
case <-fetching:
t.Fatalf("fetcher requested stale header")
}
// Ensure that a block with a higher number than the threshold is discarded
tester.fetcher.Notify("higher", hashes[high], blocks[hashes[high]].NumberU64(), time.Now().Add(-arriveTimeout), nil, headerFetcher, bodyFetcher)
select {
case <-time.After(50 * time.Millisecond):
case <-fetching:
t.Fatalf("fetcher requested future header")
}
}
// 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.
@ -547,12 +590,12 @@ func testHashMemoryExhaustionAttack(t *testing.T, protocol int) {
hashes, blocks := makeChain(targetBlocks, 0, genesis)
validBlockFetcher := tester.makeBlockFetcher(blocks)
validHeaderFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack)
validBodyFetcher := tester.makeBodyFetcher(blocks)
validBodyFetcher := tester.makeBodyFetcher(blocks, 0)
attack, _ := makeChain(targetBlocks, 0, unknownBlock)
attackerBlockFetcher := tester.makeBlockFetcher(nil)
attackerHeaderFetcher := tester.makeHeaderFetcher(nil, -gatherSlack)
attackerBodyFetcher := tester.makeBodyFetcher(nil)
attackerBodyFetcher := tester.makeBodyFetcher(nil, 0)
// Feed the tester a huge hashset from the attacker, and a limited from the valid peer
for i := 0; i < len(attack); i++ {
@ -566,7 +609,7 @@ func testHashMemoryExhaustionAttack(t *testing.T, protocol int) {
if protocol < 62 {
tester.fetcher.Notify("attacker", attack[i], 0, time.Now(), attackerBlockFetcher, nil, nil)
} else {
tester.fetcher.Notify("attacker", attack[i], uint64(i+1), time.Now(), nil, attackerHeaderFetcher, attackerBodyFetcher)
tester.fetcher.Notify("attacker", attack[i], 1 /* don't distance drop */, time.Now(), nil, attackerHeaderFetcher, attackerBodyFetcher)
}
}
if len(tester.fetcher.announced) != hashLimit+maxQueueDist {
@ -580,7 +623,7 @@ func testHashMemoryExhaustionAttack(t *testing.T, protocol int) {
if protocol < 62 {
tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), validBlockFetcher, nil, nil)
} else {
tester.fetcher.Notify("valid", hashes[i], uint64(i+1), time.Now().Add(-arriveTimeout), nil, validHeaderFetcher, validBodyFetcher)
tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), nil, validHeaderFetcher, validBodyFetcher)
}
verifyImportEvent(t, imported)
}

2
eth/handler.go
View file

@ -436,7 +436,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
uncles[i] = body.Uncles
}
// Filter out any explicitly requested bodies, deliver the rest to the downloader
if trasactions, uncles := pm.fetcher.FilterBodies(trasactions, uncles); len(trasactions) > 0 || len(uncles) > 0 {
if trasactions, uncles := pm.fetcher.FilterBodies(trasactions, uncles, time.Now()); len(trasactions) > 0 || len(uncles) > 0 {
err := pm.downloader.DeliverBodies(p.id, trasactions, uncles)
if err != nil {
glog.V(logger.Debug).Infoln(err)