Implement server side packet handling for eth/62 and eth/63

cmd/geth/main.go

@@ -283,6 +283,7 @@ JavaScript API. See https://github.com/ethereum/go-ethereum/wiki/Javascipt-Conso
 		utils.DataDirFlag,
 		utils.BlockchainVersionFlag,
 		utils.OlympicFlag,
+		utils.EthVersionFlag,
 		utils.CacheFlag,
 		utils.JSpathFlag,
 		utils.ListenPortFlag,
@@ -333,6 +334,7 @@ JavaScript API. See https://github.com/ethereum/go-ethereum/wiki/Javascipt-Conso
 	app.Before = func(ctx *cli.Context) error {
 		utils.SetupLogger(ctx)
 		utils.SetupVM(ctx)
+		utils.SetupEth(ctx)
 		if ctx.GlobalBool(utils.PProfEanbledFlag.Name) {
 			utils.StartPProf(ctx)
 		}

cmd/utils/flags.go

@@ -138,6 +138,11 @@ var (
 		Name:  "olympic",
 		Usage: "Use olympic style protocol",
 	}
+	EthVersionFlag = cli.IntFlag{
+		Name:  "eth",
+		Value: 61,
+		Usage: "Highest eth protocol to advertise (temporary, dev option)",
+	}
 
 	// miner settings
 	MinerThreadsFlag = cli.IntFlag{
@@ -459,6 +464,18 @@ func SetupVM(ctx *cli.Context) {
 	vm.SetJITCacheSize(ctx.GlobalInt(VMJitCacheFlag.Name))
 }
 
+// SetupEth configures the eth packages global settings
+func SetupEth(ctx *cli.Context) {
+	version := ctx.GlobalInt(EthVersionFlag.Name)
+	for len(eth.ProtocolVersions) > 0 && eth.ProtocolVersions[0] > uint(version) {
+		eth.ProtocolVersions = eth.ProtocolVersions[1:]
+		eth.ProtocolLengths = eth.ProtocolLengths[1:]
+	}
+	if len(eth.ProtocolVersions) == 0 {
+		Fatalf("No valid eth protocols remaining")
+	}
+}
+
 // MakeChain creates a chain manager from set command line flags.
 func MakeChain(ctx *cli.Context) (chain *core.ChainManager, chainDb common.Database) {
 	datadir := ctx.GlobalString(DataDirFlag.Name)

eth/backend.go

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

eth/downloader/downloader.go

@@ -42,9 +42,12 @@ const (
 )
 
 var (
-	MinHashFetch  = 512 // Minimum amount of hashes to not consider a peer stalling
-	MaxHashFetch  = 512 // Amount of hashes to be fetched per retrieval request
-	MaxBlockFetch = 128 // Amount of blocks to be fetched per retrieval request
+	MinHashFetch     = 512 // Minimum amount of hashes to not consider a peer stalling
+	MaxHashFetch     = 512 // Amount of hashes to be fetched per retrieval request
+	MaxBlockFetch    = 128 // Amount of blocks to be fetched per retrieval request
+	MaxHeaderFetch   = 256 // Amount of block headers to be fetched per retrieval request
+	MaxStateFetch    = 384 // Amount of node state values to allow fetching per request
+	MaxReceiptsFetch = 384 // Amount of transaction receipts to allow fetching per request
 
 	hashTTL         = 5 * time.Second  // Time it takes for a hash request to time out
 	blockSoftTTL    = 3 * time.Second  // Request completion threshold for increasing or decreasing a peer's bandwidth

eth/fetcher/fetcher.go

@@ -69,8 +69,9 @@ 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
+	hash   common.Hash // Hash of the block being announced
+	number uint64      // Number of the block being announced (0 = unknown | old protocol)
+	time   time.Time   // Timestamp of the announcement
 
 	origin string           // Identifier of the peer originating the notification
 	fetch  blockRequesterFn // Fetcher function to retrieve
@@ -152,9 +153,10 @@ func (f *Fetcher) Stop() {
 
 // 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 {
+func (f *Fetcher) Notify(peer string, hash common.Hash, number uint64, time time.Time, fetcher blockRequesterFn) error {
 	block := &announce{
 		hash:   hash,
+		number: number,
 		time:   time,
 		origin: peer,
 		fetch:  fetcher,

eth/fetcher/fetcher_test.go

@@ -194,7 +194,7 @@ func TestSequentialAnnouncements(t *testing.T) {
 	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
 
 	for i := len(hashes) - 2; i >= 0; i-- {
-		tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
+		tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher)
 		verifyImportEvent(t, imported)
 	}
 	verifyImportDone(t, imported)
@@ -221,9 +221,9 @@ func TestConcurrentAnnouncements(t *testing.T) {
 	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
 
 	for i := len(hashes) - 2; 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)
+		tester.fetcher.Notify("first", hashes[i], 0, time.Now().Add(-arriveTimeout), wrapper)
+		tester.fetcher.Notify("second", hashes[i], 0, time.Now().Add(-arriveTimeout+time.Millisecond), wrapper)
+		tester.fetcher.Notify("second", hashes[i], 0, time.Now().Add(-arriveTimeout-time.Millisecond), wrapper)
 
 		verifyImportEvent(t, imported)
 	}
@@ -252,7 +252,7 @@ func TestOverlappingAnnouncements(t *testing.T) {
 	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
 
 	for i := len(hashes) - 2; i >= 0; i-- {
-		tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
+		tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher)
 		select {
 		case <-fetching:
 		case <-time.After(time.Second):
@@ -286,7 +286,7 @@ func TestPendingDeduplication(t *testing.T) {
 	}
 	// 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)
+		tester.fetcher.Notify("repeater", hashes[0], 0, time.Now().Add(-arriveTimeout), wrapper)
 		time.Sleep(time.Millisecond)
 	}
 	time.Sleep(delay)
@@ -317,12 +317,12 @@ func TestRandomArrivalImport(t *testing.T) {
 
 	for i := len(hashes) - 1; i >= 0; i-- {
 		if i != skip {
-			tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
+			tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher)
 			time.Sleep(time.Millisecond)
 		}
 	}
 	// Finally announce the skipped entry and check full import
-	tester.fetcher.Notify("valid", hashes[skip], time.Now().Add(-arriveTimeout), fetcher)
+	tester.fetcher.Notify("valid", hashes[skip], 0, time.Now().Add(-arriveTimeout), fetcher)
 	verifyImportCount(t, imported, len(hashes)-1)
 }
 
@@ -343,7 +343,7 @@ func TestQueueGapFill(t *testing.T) {
 
 	for i := len(hashes) - 1; i >= 0; i-- {
 		if i != skip {
-			tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
+			tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher)
 			time.Sleep(time.Millisecond)
 		}
 	}
@@ -374,7 +374,7 @@ func TestImportDeduplication(t *testing.T) {
 	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
 
 	// Announce the duplicating block, wait for retrieval, and also propagate directly
-	tester.fetcher.Notify("valid", hashes[0], time.Now().Add(-arriveTimeout), fetcher)
+	tester.fetcher.Notify("valid", hashes[0], 0, time.Now().Add(-arriveTimeout), fetcher)
 	<-fetching
 
 	tester.fetcher.Enqueue("valid", blocks[hashes[0]])
@@ -437,9 +437,9 @@ func TestHashMemoryExhaustionAttack(t *testing.T) {
 	// Feed the tester a huge hashset from the attacker, and a limited from the valid peer
 	for i := 0; i < len(attack); i++ {
 		if i < maxQueueDist {
-			tester.fetcher.Notify("valid", hashes[len(hashes)-2-i], time.Now(), valid)
+			tester.fetcher.Notify("valid", hashes[len(hashes)-2-i], 0, time.Now(), valid)
 		}
-		tester.fetcher.Notify("attacker", attack[i], time.Now(), attacker)
+		tester.fetcher.Notify("attacker", attack[i], 0, time.Now(), attacker)
 	}
 	if len(tester.fetcher.announced) != hashLimit+maxQueueDist {
 		t.Fatalf("queued announce count mismatch: have %d, want %d", len(tester.fetcher.announced), hashLimit+maxQueueDist)
@@ -449,7 +449,7 @@ func TestHashMemoryExhaustionAttack(t *testing.T) {
 
 	// Feed the remaining valid hashes to ensure DOS protection state remains clean
 	for i := len(hashes) - maxQueueDist - 2; i >= 0; i-- {
-		tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), valid)
+		tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), valid)
 		verifyImportEvent(t, imported)
 	}
 	verifyImportDone(t, imported)