aboutsummaryrefslogtreecommitdiffstats
path: root/eth
diff options
context:
space:
mode:
Diffstat (limited to 'eth')
-rw-r--r--eth/backend.go28
-rw-r--r--eth/downloader/downloader.go8
-rw-r--r--eth/downloader/downloader_test.go135
-rw-r--r--eth/handler.go18
-rw-r--r--eth/peer.go4
5 files changed, 64 insertions, 129 deletions
diff --git a/eth/backend.go b/eth/backend.go
index 391a610e3..9c661ad54 100644
--- a/eth/backend.go
+++ b/eth/backend.go
@@ -41,7 +41,6 @@ import (
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
- "github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/miner"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
@@ -267,42 +266,21 @@ func New(config *Config) (*Ethereum, error) {
return nil, fmt.Errorf("blockchain db err: %v", err)
}
if db, ok := blockDb.(*ethdb.LDBDatabase); ok {
- db.GetTimer = metrics.NewTimer("eth/db/block/user/gets")
- db.PutTimer = metrics.NewTimer("eth/db/block/user/puts")
- db.MissMeter = metrics.NewMeter("eth/db/block/user/misses")
- db.ReadMeter = metrics.NewMeter("eth/db/block/user/reads")
- db.WriteMeter = metrics.NewMeter("eth/db/block/user/writes")
- db.CompTimeMeter = metrics.NewMeter("eth/db/block/compact/time")
- db.CompReadMeter = metrics.NewMeter("eth/db/block/compact/input")
- db.CompWriteMeter = metrics.NewMeter("eth/db/block/compact/output")
+ db.Meter("eth/db/block/")
}
stateDb, err := newdb(filepath.Join(config.DataDir, "state"))
if err != nil {
return nil, fmt.Errorf("state db err: %v", err)
}
if db, ok := stateDb.(*ethdb.LDBDatabase); ok {
- db.GetTimer = metrics.NewTimer("eth/db/state/user/gets")
- db.PutTimer = metrics.NewTimer("eth/db/state/user/puts")
- db.MissMeter = metrics.NewMeter("eth/db/state/user/misses")
- db.ReadMeter = metrics.NewMeter("eth/db/state/user/reads")
- db.WriteMeter = metrics.NewMeter("eth/db/state/user/writes")
- db.CompTimeMeter = metrics.NewMeter("eth/db/state/compact/time")
- db.CompReadMeter = metrics.NewMeter("eth/db/state/compact/input")
- db.CompWriteMeter = metrics.NewMeter("eth/db/state/compact/output")
+ db.Meter("eth/db/state/")
}
extraDb, err := newdb(filepath.Join(config.DataDir, "extra"))
if err != nil {
return nil, fmt.Errorf("extra db err: %v", err)
}
if db, ok := extraDb.(*ethdb.LDBDatabase); ok {
- db.GetTimer = metrics.NewTimer("eth/db/extra/user/gets")
- db.PutTimer = metrics.NewTimer("eth/db/extra/user/puts")
- db.MissMeter = metrics.NewMeter("eth/db/extra/user/misses")
- db.ReadMeter = metrics.NewMeter("eth/db/extra/user/reads")
- db.WriteMeter = metrics.NewMeter("eth/db/extra/user/writes")
- db.CompTimeMeter = metrics.NewMeter("eth/db/extra/compact/time")
- db.CompReadMeter = metrics.NewMeter("eth/db/extra/compact/input")
- db.CompWriteMeter = metrics.NewMeter("eth/db/extra/compact/output")
+ db.Meter("eth/db/extra/")
}
nodeDb := filepath.Join(config.DataDir, "nodes")
diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 5ce98816d..7cf83ada3 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -804,6 +804,8 @@ func (d *Downloader) fetchHashes(p *peer, from uint64) error {
}
// Start pulling hashes, until all are exhausted
getHashes(from)
+ gotHashes := false
+
for {
select {
case <-d.cancelCh:
@@ -825,8 +827,14 @@ func (d *Downloader) fetchHashes(p *peer, from uint64) error {
case d.processCh <- false:
case <-d.cancelCh:
}
+ // Error out if no hashes were retrieved at all
+ if !gotHashes {
+ return errStallingPeer
+ }
return nil
}
+ gotHashes = true
+
// Otherwise insert all the new hashes, aborting in case of junk
glog.V(logger.Detail).Infof("%v: inserting %d hashes from #%d", p, len(hashPack.hashes), from)
diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go
index ff2e59d92..cf1d0c2fb 100644
--- a/eth/downloader/downloader_test.go
+++ b/eth/downloader/downloader_test.go
@@ -272,7 +272,7 @@ func TestSynchronisation60(t *testing.T) {
// Tests that simple synchronization against a canonical chain works correctly.
// In this test common ancestor lookup should be short circuited and not require
// binary searching.
-func TestCanonicalSynchronisation(t *testing.T) {
+func TestCanonicalSynchronisation61(t *testing.T) {
// Create a small enough block chain to download
targetBlocks := blockCacheLimit - 15
hashes, blocks := makeChain(targetBlocks, 0, genesis)
@@ -291,69 +291,16 @@ func TestCanonicalSynchronisation(t *testing.T) {
// Tests that if a large batch of blocks are being downloaded, it is throttled
// until the cached blocks are retrieved.
-func TestThrottling60(t *testing.T) {
- // Create a long block chain to download and the tester
- targetBlocks := 8 * blockCacheLimit
- hashes, blocks := makeChain(targetBlocks, 0, genesis)
-
- tester := newTester()
- tester.newPeer("peer", eth60, hashes, blocks)
-
- // Wrap the importer to allow stepping
- done := make(chan int)
- tester.downloader.insertChain = func(blocks types.Blocks) (int, error) {
- n, err := tester.insertChain(blocks)
- done <- n
- return n, err
- }
- // Start a synchronisation concurrently
- errc := make(chan error)
- go func() {
- errc <- tester.sync("peer")
- }()
- // Iteratively take some blocks, always checking the retrieval count
- for len(tester.ownBlocks) < targetBlocks+1 {
- // Wait a bit for sync to throttle itself
- var cached int
- for start := time.Now(); time.Since(start) < 3*time.Second; {
- time.Sleep(25 * time.Millisecond)
-
- cached = len(tester.downloader.queue.blockPool)
- if cached == blockCacheLimit || len(tester.ownBlocks)+cached == targetBlocks+1 {
- break
- }
- }
- // Make sure we filled up the cache, then exhaust it
- time.Sleep(25 * time.Millisecond) // give it a chance to screw up
- if cached != blockCacheLimit && len(tester.ownBlocks)+cached < targetBlocks+1 {
- t.Fatalf("block count mismatch: have %v, want %v", cached, blockCacheLimit)
- }
- <-done // finish previous blocking import
- for cached > maxBlockProcess {
- cached -= <-done
- }
- time.Sleep(25 * time.Millisecond) // yield to the insertion
- }
- <-done // finish the last blocking import
+func TestThrottling60(t *testing.T) { testThrottling(t, eth60) }
+func TestThrottling61(t *testing.T) { testThrottling(t, eth61) }
- // Check that we haven't pulled more blocks than available
- if len(tester.ownBlocks) > targetBlocks+1 {
- t.Fatalf("target block count mismatch: have %v, want %v", len(tester.ownBlocks), targetBlocks+1)
- }
- if err := <-errc; err != nil {
- t.Fatalf("block synchronization failed: %v", err)
- }
-}
-
-// Tests that if a large batch of blocks are being downloaded, it is throttled
-// until the cached blocks are retrieved.
-func TestThrottling(t *testing.T) {
+func testThrottling(t *testing.T, protocol int) {
// Create a long block chain to download and the tester
targetBlocks := 8 * blockCacheLimit
hashes, blocks := makeChain(targetBlocks, 0, genesis)
tester := newTester()
- tester.newPeer("peer", eth61, hashes, blocks)
+ tester.newPeer("peer", protocol, hashes, blocks)
// Wrap the importer to allow stepping
done := make(chan int)
@@ -404,7 +351,7 @@ func TestThrottling(t *testing.T) {
// Tests that simple synchronization against a forked chain works correctly. In
// this test common ancestor lookup should *not* be short circuited, and a full
// binary search should be executed.
-func TestForkedSynchronisation(t *testing.T) {
+func TestForkedSynchronisation61(t *testing.T) {
// Create a long enough forked chain
common, fork := MaxHashFetch, 2*MaxHashFetch
hashesA, hashesB, blocksA, blocksB := makeChainFork(common+fork, fork, genesis)
@@ -443,33 +390,10 @@ func TestInactiveDownloader(t *testing.T) {
}
// Tests that a canceled download wipes all previously accumulated state.
-func TestCancel60(t *testing.T) {
- // Create a small enough block chain to download and the tester
- targetBlocks := blockCacheLimit - 15
- hashes, blocks := makeChain(targetBlocks, 0, genesis)
-
- tester := newTester()
- tester.newPeer("peer", eth60, hashes, blocks)
+func TestCancel60(t *testing.T) { testCancel(t, eth60) }
+func TestCancel61(t *testing.T) { testCancel(t, eth61) }
- // Make sure canceling works with a pristine downloader
- tester.downloader.cancel()
- hashCount, blockCount := tester.downloader.queue.Size()
- if hashCount > 0 || blockCount > 0 {
- t.Errorf("block or hash count mismatch: %d hashes, %d blocks, want 0", hashCount, blockCount)
- }
- // Synchronise with the peer, but cancel afterwards
- if err := tester.sync("peer"); err != nil {
- t.Fatalf("failed to synchronise blocks: %v", err)
- }
- tester.downloader.cancel()
- hashCount, blockCount = tester.downloader.queue.Size()
- if hashCount > 0 || blockCount > 0 {
- t.Errorf("block or hash count mismatch: %d hashes, %d blocks, want 0", hashCount, blockCount)
- }
-}
-
-// Tests that a canceled download wipes all previously accumulated state.
-func TestCancel(t *testing.T) {
+func testCancel(t *testing.T, protocol int) {
// Create a small enough block chain to download and the tester
targetBlocks := blockCacheLimit - 15
if targetBlocks >= MaxHashFetch {
@@ -478,7 +402,7 @@ func TestCancel(t *testing.T) {
hashes, blocks := makeChain(targetBlocks, 0, genesis)
tester := newTester()
- tester.newPeer("peer", eth61, hashes, blocks)
+ tester.newPeer("peer", protocol, hashes, blocks)
// Make sure canceling works with a pristine downloader
tester.downloader.cancel()
@@ -498,7 +422,10 @@ func TestCancel(t *testing.T) {
}
// Tests that synchronisation from multiple peers works as intended (multi thread sanity test).
-func TestMultiSynchronisation(t *testing.T) {
+func TestMultiSynchronisation60(t *testing.T) { testMultiSynchronisation(t, eth60) }
+func TestMultiSynchronisation61(t *testing.T) { testMultiSynchronisation(t, eth61) }
+
+func testMultiSynchronisation(t *testing.T, protocol int) {
// Create various peers with various parts of the chain
targetPeers := 16
targetBlocks := targetPeers*blockCacheLimit - 15
@@ -507,7 +434,7 @@ func TestMultiSynchronisation(t *testing.T) {
tester := newTester()
for i := 0; i < targetPeers; i++ {
id := fmt.Sprintf("peer #%d", i)
- tester.newPeer(id, eth60, hashes[i*blockCacheLimit:], blocks)
+ tester.newPeer(id, protocol, hashes[i*blockCacheLimit:], blocks)
}
// Synchronise with the middle peer and make sure half of the blocks were retrieved
id := fmt.Sprintf("peer #%d", targetPeers/2)
@@ -528,7 +455,7 @@ func TestMultiSynchronisation(t *testing.T) {
// Tests that synchronising with a peer who's very slow at network IO does not
// stall the other peers in the system.
-func TestSlowSynchronisation(t *testing.T) {
+func TestSlowSynchronisation60(t *testing.T) {
tester := newTester()
// Create a batch of blocks, with a slow and a full speed peer
@@ -557,7 +484,7 @@ func TestSlowSynchronisation(t *testing.T) {
// Tests that if a peer returns an invalid chain with a block pointing to a non-
// existing parent, it is correctly detected and handled.
-func TestNonExistingParentAttack(t *testing.T) {
+func TestNonExistingParentAttack60(t *testing.T) {
tester := newTester()
// Forge a single-link chain with a forged header
@@ -587,7 +514,7 @@ func TestNonExistingParentAttack(t *testing.T) {
// Tests that if a malicious peers keeps sending us repeating hashes, we don't
// loop indefinitely.
-func TestRepeatingHashAttack(t *testing.T) { // TODO: Is this thing valid??
+func TestRepeatingHashAttack60(t *testing.T) { // TODO: Is this thing valid??
tester := newTester()
// Create a valid chain, but drop the last link
@@ -617,7 +544,7 @@ func TestRepeatingHashAttack(t *testing.T) { // TODO: Is this thing valid??
// Tests that if a malicious peers returns a non-existent block hash, it should
// eventually time out and the sync reattempted.
-func TestNonExistingBlockAttack(t *testing.T) {
+func TestNonExistingBlockAttack60(t *testing.T) {
tester := newTester()
// Create a valid chain, but forge the last link
@@ -639,7 +566,7 @@ func TestNonExistingBlockAttack(t *testing.T) {
// Tests that if a malicious peer is returning hashes in a weird order, that the
// sync throttler doesn't choke on them waiting for the valid blocks.
-func TestInvalidHashOrderAttack(t *testing.T) {
+func TestInvalidHashOrderAttack60(t *testing.T) {
tester := newTester()
// Create a valid long chain, but reverse some hashes within
@@ -667,7 +594,7 @@ func TestInvalidHashOrderAttack(t *testing.T) {
// Tests that if a malicious peer makes up a random hash chain and tries to push
// indefinitely, it actually gets caught with it.
-func TestMadeupHashChainAttack(t *testing.T) {
+func TestMadeupHashChainAttack60(t *testing.T) {
tester := newTester()
blockSoftTTL = 100 * time.Millisecond
crossCheckCycle = 25 * time.Millisecond
@@ -697,7 +624,7 @@ func TestMadeupHashChainAttack(t *testing.T) {
// indefinitely, one hash at a time, it actually gets caught with it. The reason
// this is separate from the classical made up chain attack is that sending hashes
// one by one prevents reliable block/parent verification.
-func TestMadeupHashChainDrippingAttack(t *testing.T) {
+func TestMadeupHashChainDrippingAttack60(t *testing.T) {
// Create a random chain of hashes to drip
randomHashes := make([]common.Hash, 16*blockCacheLimit)
for i := range randomHashes {
@@ -716,7 +643,7 @@ func TestMadeupHashChainDrippingAttack(t *testing.T) {
// Tests that if a malicious peer makes up a random block chain, and tried to
// push indefinitely, it actually gets caught with it.
-func TestMadeupBlockChainAttack(t *testing.T) {
+func TestMadeupBlockChainAttack60(t *testing.T) {
defaultBlockTTL := blockSoftTTL
defaultCrossCheckCycle := crossCheckCycle
@@ -748,7 +675,7 @@ func TestMadeupBlockChainAttack(t *testing.T) {
// Tests that if one/multiple malicious peers try to feed a banned blockchain to
// the downloader, it will not keep refetching the same chain indefinitely, but
// gradually block pieces of it, until its head is also blocked.
-func TestBannedChainStarvationAttack(t *testing.T) {
+func TestBannedChainStarvationAttack60(t *testing.T) {
n := 8 * blockCacheLimit
fork := n/2 - 23
hashes, forkHashes, blocks, forkBlocks := makeChainFork(n, fork, genesis)
@@ -792,7 +719,7 @@ func TestBannedChainStarvationAttack(t *testing.T) {
// Tests that if a peer sends excessively many/large invalid chains that are
// gradually banned, it will have an upper limit on the consumed memory and also
// the origin bad hashes will not be evacuated.
-func TestBannedChainMemoryExhaustionAttack(t *testing.T) {
+func TestBannedChainMemoryExhaustionAttack60(t *testing.T) {
// Construct a banned chain with more chunks than the ban limit
n := 8 * blockCacheLimit
fork := n/2 - 23
@@ -848,7 +775,7 @@ func TestBannedChainMemoryExhaustionAttack(t *testing.T) {
// internal state problems
//
// No, don't delete this test, it actually did happen!
-func TestOverlappingDeliveryAttack(t *testing.T) {
+func TestOverlappingDeliveryAttack60(t *testing.T) {
// Create an arbitrary batch of blocks ( < cache-size not to block)
targetBlocks := blockCacheLimit - 23
hashes, blocks := makeChain(targetBlocks, 0, genesis)
@@ -875,6 +802,16 @@ func TestOverlappingDeliveryAttack(t *testing.T) {
}
}
+// Tests that a peer advertising an high TD doesn't get to stall the downloader
+// afterwards by not sending any useful hashes.
+func TestHighTDStarvationAttack61(t *testing.T) {
+ tester := newTester()
+ tester.newPeer("attack", eth61, []common.Hash{genesis.Hash()}, nil)
+ if err := tester.sync("attack"); err != errStallingPeer {
+ t.Fatalf("synchronisation error mismatch: have %v, want %v", err, errStallingPeer)
+ }
+}
+
// Tests that misbehaving peers are disconnected, whilst behaving ones are not.
func TestHashAttackerDropping(t *testing.T) {
// Define the disconnection requirement for individual hash fetch errors
diff --git a/eth/handler.go b/eth/handler.go
index bbb251812..50e2ac99b 100644
--- a/eth/handler.go
+++ b/eth/handler.go
@@ -19,6 +19,7 @@ package eth
import (
"fmt"
"math"
+ "math/big"
"sync"
"time"
@@ -412,8 +413,10 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
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)
+ if request.TD.Cmp(p.Td()) > 0 {
+ p.SetTd(request.TD)
+ go pm.synchronise(p)
+ }
case TxMsg:
// Transactions arrived, parse all of them and deliver to the pool
@@ -452,9 +455,18 @@ func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) {
// If propagation is requested, send to a subset of the peer
if propagate {
+ // Calculate the TD of the block (it's not imported yet, so block.Td is not valid)
+ var td *big.Int
+ if parent := pm.chainman.GetBlock(block.ParentHash()); parent != nil {
+ td = new(big.Int).Add(parent.Td, block.Difficulty())
+ } else {
+ glog.V(logger.Error).Infof("propagating dangling block #%d [%x]", block.NumberU64(), hash[:4])
+ return
+ }
+ // Send the block to a subset of our peers
transfer := peers[:int(math.Sqrt(float64(len(peers))))]
for _, peer := range transfer {
- peer.SendNewBlock(block)
+ peer.SendNewBlock(block, td)
}
glog.V(logger.Detail).Infof("propagated block %x to %d peers in %v", hash[:4], len(transfer), time.Since(block.ReceivedAt))
}
diff --git a/eth/peer.go b/eth/peer.go
index ccd5d3c6f..d70ad4c94 100644
--- a/eth/peer.go
+++ b/eth/peer.go
@@ -167,12 +167,12 @@ func (p *peer) SendNewBlockHashes(hashes []common.Hash) error {
}
// SendNewBlock propagates an entire block to a remote peer.
-func (p *peer) SendNewBlock(block *types.Block) error {
+func (p *peer) SendNewBlock(block *types.Block, td *big.Int) error {
propBlockOutPacketsMeter.Mark(1)
propBlockOutTrafficMeter.Mark(block.Size().Int64())
p.knownBlocks.Add(block.Hash())
- return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, block.Td})
+ return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td})
}
// RequestHashes fetches a batch of hashes from a peer, starting at from, going
generated by cgit v1.2.3 (git 2.25.1) at 2025-03-05 09:32:06 +0800