From d235e2c6a5788ec4a6cff15a16f56b38a3876a0d Mon Sep 17 00:00:00 2001 From: Determinant Date: Sun, 28 Jun 2020 14:47:41 -0400 Subject: ... --- eth/api.go | 6 +- eth/api_backend.go | 19 +- eth/api_tracer.go | 37 +- eth/backend.go | 42 +- eth/bloombits.go | 6 +- eth/config.go | 4 +- eth/filters/api.go | 4 +- eth/filters/filter.go | 4 +- eth/filters/filter_system.go | 25 +- eth/gasprice/gasprice.go | 2 +- eth/gen_config.go | 4 +- eth/handler.go | 844 --------------------- eth/peer.go | 546 ------------- eth/protocol.go | 2 +- eth/sync.go | 216 ------ eth/tracers/internal/tracers/4byte_tracer.js | 86 +++ eth/tracers/internal/tracers/assets.go | 463 +++++++++++ eth/tracers/internal/tracers/bigram_tracer.js | 47 ++ eth/tracers/internal/tracers/call_tracer.js | 246 ++++++ eth/tracers/internal/tracers/evmdis_tracer.js | 93 +++ eth/tracers/internal/tracers/noop_tracer.js | 29 + eth/tracers/internal/tracers/opcount_tracer.js | 32 + eth/tracers/internal/tracers/prestate_tracer.js | 108 +++ eth/tracers/internal/tracers/tracers.go | 21 + eth/tracers/internal/tracers/trigram_tracer.js | 49 ++ eth/tracers/internal/tracers/unigram_tracer.js | 43 ++ eth/tracers/testdata/call_tracer_create.json | 58 ++ eth/tracers/testdata/call_tracer_deep_calls.json | 415 ++++++++++ eth/tracers/testdata/call_tracer_delegatecall.json | 97 +++ .../call_tracer_inner_create_oog_outer_throw.json | 77 ++ .../call_tracer_inner_throw_outer_revert.json | 81 ++ eth/tracers/testdata/call_tracer_oog.json | 60 ++ eth/tracers/testdata/call_tracer_revert.json | 58 ++ eth/tracers/testdata/call_tracer_simple.json | 78 ++ eth/tracers/testdata/call_tracer_throw.json | 62 ++ eth/tracers/tracer.go | 641 ++++++++++++++++ eth/tracers/tracers.go | 53 ++ 37 files changed, 2972 insertions(+), 1686 deletions(-) delete mode 100644 eth/handler.go delete mode 100644 eth/peer.go delete mode 100644 eth/sync.go create mode 100644 eth/tracers/internal/tracers/4byte_tracer.js create mode 100644 eth/tracers/internal/tracers/assets.go create mode 100644 eth/tracers/internal/tracers/bigram_tracer.js create mode 100644 eth/tracers/internal/tracers/call_tracer.js create mode 100644 eth/tracers/internal/tracers/evmdis_tracer.js create mode 100644 eth/tracers/internal/tracers/noop_tracer.js create mode 100644 eth/tracers/internal/tracers/opcount_tracer.js create mode 100644 eth/tracers/internal/tracers/prestate_tracer.js create mode 100644 eth/tracers/internal/tracers/tracers.go create mode 100644 eth/tracers/internal/tracers/trigram_tracer.js create mode 100644 eth/tracers/internal/tracers/unigram_tracer.js create mode 100644 eth/tracers/testdata/call_tracer_create.json create mode 100644 eth/tracers/testdata/call_tracer_deep_calls.json create mode 100644 eth/tracers/testdata/call_tracer_delegatecall.json create mode 100644 eth/tracers/testdata/call_tracer_inner_create_oog_outer_throw.json create mode 100644 eth/tracers/testdata/call_tracer_inner_throw_outer_revert.json create mode 100644 eth/tracers/testdata/call_tracer_oog.json create mode 100644 eth/tracers/testdata/call_tracer_revert.json create mode 100644 eth/tracers/testdata/call_tracer_simple.json create mode 100644 eth/tracers/testdata/call_tracer_throw.json create mode 100644 eth/tracers/tracer.go create mode 100644 eth/tracers/tracers.go (limited to 'eth') diff --git a/eth/api.go b/eth/api.go index 97a0758..b64090e 100644 --- a/eth/api.go +++ b/eth/api.go @@ -29,13 +29,13 @@ import ( "time" "github.com/ava-labs/coreth/core" + "github.com/ava-labs/coreth/core/rawdb" + "github.com/ava-labs/coreth/core/state" + "github.com/ava-labs/coreth/core/types" "github.com/ava-labs/coreth/internal/ethapi" "github.com/ava-labs/coreth/rpc" "github.com/ava-labs/go-ethereum/common" "github.com/ava-labs/go-ethereum/common/hexutil" - "github.com/ava-labs/go-ethereum/core/rawdb" - "github.com/ava-labs/go-ethereum/core/state" - "github.com/ava-labs/go-ethereum/core/types" "github.com/ava-labs/go-ethereum/rlp" "github.com/ava-labs/go-ethereum/trie" ) diff --git a/eth/api_backend.go b/eth/api_backend.go index c4ec8f0..1ca7b94 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -21,22 +21,21 @@ import ( "errors" "math/big" + "github.com/ava-labs/coreth/accounts" "github.com/ava-labs/coreth/core" + "github.com/ava-labs/coreth/core/bloombits" + "github.com/ava-labs/coreth/core/rawdb" + "github.com/ava-labs/coreth/core/state" + "github.com/ava-labs/coreth/core/types" + "github.com/ava-labs/coreth/core/vm" "github.com/ava-labs/coreth/eth/gasprice" + "github.com/ava-labs/coreth/params" "github.com/ava-labs/coreth/rpc" - "github.com/ava-labs/go-ethereum/accounts" "github.com/ava-labs/go-ethereum/common" "github.com/ava-labs/go-ethereum/common/math" - ethcore "github.com/ava-labs/go-ethereum/core" - "github.com/ava-labs/go-ethereum/core/bloombits" - "github.com/ava-labs/go-ethereum/core/rawdb" - "github.com/ava-labs/go-ethereum/core/state" - "github.com/ava-labs/go-ethereum/core/types" - "github.com/ava-labs/go-ethereum/core/vm" "github.com/ava-labs/go-ethereum/eth/downloader" "github.com/ava-labs/go-ethereum/ethdb" "github.com/ava-labs/go-ethereum/event" - "github.com/ava-labs/go-ethereum/params" ) // EthAPIBackend implements ethapi.Backend for full nodes @@ -60,7 +59,7 @@ func (b *EthAPIBackend) AcceptedBlock() *types.Block { } func (b *EthAPIBackend) SetHead(number uint64) { - b.eth.protocolManager.downloader.Cancel() + //b.eth.protocolManager.downloader.Cancel() b.eth.blockchain.SetHead(number) } @@ -142,7 +141,7 @@ func (b *EthAPIBackend) GetTd(blockHash common.Hash) *big.Int { return b.eth.blockchain.GetTdByHash(blockHash) } -func (b *EthAPIBackend) GetEVM(ctx context.Context, msg ethcore.Message, state *state.StateDB, header *types.Header) (*vm.EVM, func() error, error) { +func (b *EthAPIBackend) GetEVM(ctx context.Context, msg core.Message, state *state.StateDB, header *types.Header) (*vm.EVM, func() error, error) { state.SetBalance(msg.From(), math.MaxBig256) vmError := func() error { return nil } diff --git a/eth/api_tracer.go b/eth/api_tracer.go index b87a9fe..c8a5307 100644 --- a/eth/api_tracer.go +++ b/eth/api_tracer.go @@ -28,19 +28,18 @@ import ( "sync" "time" + "github.com/ava-labs/coreth/core" + "github.com/ava-labs/coreth/core/rawdb" + "github.com/ava-labs/coreth/core/state" + "github.com/ava-labs/coreth/core/types" + "github.com/ava-labs/coreth/core/vm" + "github.com/ava-labs/coreth/eth/tracers" "github.com/ava-labs/coreth/internal/ethapi" - myrpc "github.com/ava-labs/coreth/rpc" + "github.com/ava-labs/coreth/rpc" "github.com/ava-labs/go-ethereum/common" "github.com/ava-labs/go-ethereum/common/hexutil" - "github.com/ava-labs/go-ethereum/core" - "github.com/ava-labs/go-ethereum/core/rawdb" - "github.com/ava-labs/go-ethereum/core/state" - "github.com/ava-labs/go-ethereum/core/types" - "github.com/ava-labs/go-ethereum/core/vm" - "github.com/ava-labs/go-ethereum/eth/tracers" "github.com/ava-labs/go-ethereum/log" "github.com/ava-labs/go-ethereum/rlp" - "github.com/ava-labs/go-ethereum/rpc" "github.com/ava-labs/go-ethereum/trie" ) @@ -102,26 +101,26 @@ type txTraceTask struct { // TraceChain returns the structured logs created during the execution of EVM // between two blocks (excluding start) and returns them as a JSON object. -func (api *PrivateDebugAPI) TraceChain(ctx context.Context, start, end myrpc.BlockNumber, config *TraceConfig) (*rpc.Subscription, error) { +func (api *PrivateDebugAPI) TraceChain(ctx context.Context, start, end rpc.BlockNumber, config *TraceConfig) (*rpc.Subscription, error) { // Fetch the block interval that we want to trace var from, to *types.Block switch start { - case myrpc.PendingBlockNumber: + case rpc.PendingBlockNumber: from = api.eth.miner.PendingBlock() - case myrpc.LatestBlockNumber: + case rpc.LatestBlockNumber: from = api.eth.blockchain.CurrentBlock() - case myrpc.AcceptedBlockNumber: + case rpc.AcceptedBlockNumber: from = api.eth.AcceptedBlock() default: from = api.eth.blockchain.GetBlockByNumber(uint64(start)) } switch end { - case myrpc.PendingBlockNumber: + case rpc.PendingBlockNumber: to = api.eth.miner.PendingBlock() - case myrpc.LatestBlockNumber: + case rpc.LatestBlockNumber: to = api.eth.blockchain.CurrentBlock() - case myrpc.AcceptedBlockNumber: + case rpc.AcceptedBlockNumber: from = api.eth.AcceptedBlock() default: to = api.eth.blockchain.GetBlockByNumber(uint64(end)) @@ -357,16 +356,16 @@ func (api *PrivateDebugAPI) traceChain(ctx context.Context, start, end *types.Bl // TraceBlockByNumber returns the structured logs created during the execution of // EVM and returns them as a JSON object. -func (api *PrivateDebugAPI) TraceBlockByNumber(ctx context.Context, number myrpc.BlockNumber, config *TraceConfig) ([]*txTraceResult, error) { +func (api *PrivateDebugAPI) TraceBlockByNumber(ctx context.Context, number rpc.BlockNumber, config *TraceConfig) ([]*txTraceResult, error) { // Fetch the block that we want to trace var block *types.Block switch number { - case myrpc.PendingBlockNumber: + case rpc.PendingBlockNumber: block = api.eth.miner.PendingBlock() - case myrpc.LatestBlockNumber: + case rpc.LatestBlockNumber: block = api.eth.blockchain.CurrentBlock() - case myrpc.AcceptedBlockNumber: + case rpc.AcceptedBlockNumber: block = api.eth.AcceptedBlock() default: block = api.eth.blockchain.GetBlockByNumber(uint64(number)) diff --git a/eth/backend.go b/eth/backend.go index 6d052e2..983909c 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -23,36 +23,34 @@ import ( "math/big" "runtime" "sync" - "sync/atomic" + //"sync/atomic" + "github.com/ava-labs/coreth/accounts" + "github.com/ava-labs/coreth/accounts/abi/bind" + "github.com/ava-labs/coreth/consensus" + "github.com/ava-labs/coreth/consensus/clique" "github.com/ava-labs/coreth/consensus/dummy" + "github.com/ava-labs/coreth/consensus/ethash" "github.com/ava-labs/coreth/core" + "github.com/ava-labs/coreth/core/bloombits" + "github.com/ava-labs/coreth/core/rawdb" + "github.com/ava-labs/coreth/core/types" + "github.com/ava-labs/coreth/core/vm" "github.com/ava-labs/coreth/eth/filters" "github.com/ava-labs/coreth/eth/gasprice" "github.com/ava-labs/coreth/internal/ethapi" "github.com/ava-labs/coreth/miner" "github.com/ava-labs/coreth/node" - myparams "github.com/ava-labs/coreth/params" - "github.com/ava-labs/go-ethereum/accounts" - "github.com/ava-labs/go-ethereum/accounts/abi/bind" + "github.com/ava-labs/coreth/params" + "github.com/ava-labs/coreth/rpc" "github.com/ava-labs/go-ethereum/common" "github.com/ava-labs/go-ethereum/common/hexutil" - "github.com/ava-labs/go-ethereum/consensus" - "github.com/ava-labs/go-ethereum/consensus/clique" - "github.com/ava-labs/go-ethereum/consensus/ethash" - "github.com/ava-labs/go-ethereum/core/bloombits" - "github.com/ava-labs/go-ethereum/core/rawdb" - "github.com/ava-labs/go-ethereum/core/types" - "github.com/ava-labs/go-ethereum/core/vm" "github.com/ava-labs/go-ethereum/eth/downloader" "github.com/ava-labs/go-ethereum/ethdb" "github.com/ava-labs/go-ethereum/event" "github.com/ava-labs/go-ethereum/log" "github.com/ava-labs/go-ethereum/p2p" - //"github.com/ava-labs/go-ethereum/p2p/enr" - "github.com/ava-labs/go-ethereum/params" "github.com/ava-labs/go-ethereum/rlp" - "github.com/ava-labs/go-ethereum/rpc" ) type LesServer interface { @@ -78,10 +76,10 @@ type Ethereum struct { server *p2p.Server // Handlers - txPool *core.TxPool - blockchain *core.BlockChain - protocolManager *ProtocolManager - lesServer LesServer + txPool *core.TxPool + blockchain *core.BlockChain + //protocolManager *ProtocolManager + lesServer LesServer // DB interfaces chainDb ethdb.Database // Block chain database @@ -253,8 +251,8 @@ func makeExtraData(extra []byte) []byte { runtime.GOOS, }) } - if uint64(len(extra)) > myparams.MaximumExtraDataSize { - log.Warn("Miner extra data exceed limit", "extra", hexutil.Bytes(extra), "limit", myparams.MaximumExtraDataSize) + if uint64(len(extra)) > params.MaximumExtraDataSize { + log.Warn("Miner extra data exceed limit", "extra", hexutil.Bytes(extra), "limit", params.MaximumExtraDataSize) extra = nil } return extra @@ -486,8 +484,8 @@ func (s *Ethereum) ChainDb() ethdb.Database { return s.chainDb } func (s *Ethereum) IsListening() bool { return true } // Always listening func (s *Ethereum) EthVersion() int { return int(ProtocolVersions[0]) } func (s *Ethereum) NetVersion() uint64 { return s.networkID } -func (s *Ethereum) Downloader() *downloader.Downloader { return s.protocolManager.downloader } -func (s *Ethereum) Synced() bool { return atomic.LoadUint32(&s.protocolManager.acceptTxs) == 1 } +func (s *Ethereum) Downloader() *downloader.Downloader { return nil } // s.protocolManager.downloader } +func (s *Ethereum) Synced() bool { return true } // atomic.LoadUint32(&s.protocolManager.acceptTxs) == 1 } func (s *Ethereum) ArchiveMode() bool { return s.config.NoPruning } // Protocols implements node.Service, returning all the currently configured diff --git a/eth/bloombits.go b/eth/bloombits.go index e4792bb..7136c29 100644 --- a/eth/bloombits.go +++ b/eth/bloombits.go @@ -21,11 +21,11 @@ import ( "time" "github.com/ava-labs/coreth/core" + "github.com/ava-labs/coreth/core/bloombits" + "github.com/ava-labs/coreth/core/rawdb" + "github.com/ava-labs/coreth/core/types" "github.com/ava-labs/go-ethereum/common" "github.com/ava-labs/go-ethereum/common/bitutil" - "github.com/ava-labs/go-ethereum/core/bloombits" - "github.com/ava-labs/go-ethereum/core/rawdb" - "github.com/ava-labs/go-ethereum/core/types" "github.com/ava-labs/go-ethereum/ethdb" ) diff --git a/eth/config.go b/eth/config.go index ee0fdb7..18dbd8e 100644 --- a/eth/config.go +++ b/eth/config.go @@ -25,14 +25,14 @@ import ( "runtime" "time" + "github.com/ava-labs/coreth/consensus/ethash" "github.com/ava-labs/coreth/core" "github.com/ava-labs/coreth/eth/gasprice" "github.com/ava-labs/coreth/miner" + "github.com/ava-labs/coreth/params" "github.com/ava-labs/go-ethereum/common" "github.com/ava-labs/go-ethereum/common/hexutil" - "github.com/ava-labs/go-ethereum/consensus/ethash" "github.com/ava-labs/go-ethereum/eth/downloader" - "github.com/ava-labs/go-ethereum/params" ) // DefaultConfig contains default settings for use on the Ethereum main net. diff --git a/eth/filters/api.go b/eth/filters/api.go index 9d9e757..a1d0b21 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -25,13 +25,13 @@ import ( "sync" "time" + "github.com/ava-labs/coreth/core/types" + "github.com/ava-labs/coreth/rpc" ethereum "github.com/ava-labs/go-ethereum" "github.com/ava-labs/go-ethereum/common" "github.com/ava-labs/go-ethereum/common/hexutil" - "github.com/ava-labs/go-ethereum/core/types" "github.com/ava-labs/go-ethereum/ethdb" "github.com/ava-labs/go-ethereum/event" - "github.com/ava-labs/go-ethereum/rpc" ) var ( diff --git a/eth/filters/filter.go b/eth/filters/filter.go index b25792c..3e133d4 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -22,10 +22,10 @@ import ( "math/big" "github.com/ava-labs/coreth/core" + "github.com/ava-labs/coreth/core/bloombits" + "github.com/ava-labs/coreth/core/types" "github.com/ava-labs/coreth/rpc" "github.com/ava-labs/go-ethereum/common" - "github.com/ava-labs/go-ethereum/core/bloombits" - "github.com/ava-labs/go-ethereum/core/types" "github.com/ava-labs/go-ethereum/ethdb" "github.com/ava-labs/go-ethereum/event" ) diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index f8ee038..41427c7 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -26,14 +26,13 @@ import ( "time" "github.com/ava-labs/coreth/core" - myrpc "github.com/ava-labs/coreth/rpc" + "github.com/ava-labs/coreth/core/rawdb" + "github.com/ava-labs/coreth/core/types" + "github.com/ava-labs/coreth/rpc" ethereum "github.com/ava-labs/go-ethereum" "github.com/ava-labs/go-ethereum/common" - "github.com/ava-labs/go-ethereum/core/rawdb" - "github.com/ava-labs/go-ethereum/core/types" "github.com/ava-labs/go-ethereum/event" "github.com/ava-labs/go-ethereum/log" - "github.com/ava-labs/go-ethereum/rpc" ) // Type determines the kind of filter and is used to put the filter in to @@ -197,24 +196,24 @@ func (es *EventSystem) subscribe(sub *subscription) *Subscription { // given criteria to the given logs channel. Default value for the from and to // block is "latest". If the fromBlock > toBlock an error is returned. func (es *EventSystem) SubscribeLogs(crit ethereum.FilterQuery, logs chan []*types.Log) (*Subscription, error) { - var from, to myrpc.BlockNumber + var from, to rpc.BlockNumber if crit.FromBlock == nil { - from = myrpc.LatestBlockNumber + from = rpc.LatestBlockNumber } else { - from = myrpc.BlockNumber(crit.FromBlock.Int64()) + from = rpc.BlockNumber(crit.FromBlock.Int64()) } if crit.ToBlock == nil { - to = myrpc.LatestBlockNumber + to = rpc.LatestBlockNumber } else { - to = myrpc.BlockNumber(crit.ToBlock.Int64()) + to = rpc.BlockNumber(crit.ToBlock.Int64()) } // only interested in pending logs - if from == myrpc.PendingBlockNumber && to == myrpc.PendingBlockNumber { + if from == rpc.PendingBlockNumber && to == rpc.PendingBlockNumber { return es.subscribePendingLogs(crit, logs), nil } // only interested in new mined logs - if from == myrpc.LatestBlockNumber && to == myrpc.LatestBlockNumber { + if from == rpc.LatestBlockNumber && to == rpc.LatestBlockNumber { return es.subscribeLogs(crit, logs), nil } // only interested in mined logs within a specific block range @@ -222,11 +221,11 @@ func (es *EventSystem) SubscribeLogs(crit ethereum.FilterQuery, logs chan []*typ return es.subscribeLogs(crit, logs), nil } // interested in mined logs from a specific block number, new logs and pending logs - if from >= myrpc.LatestBlockNumber && to == myrpc.PendingBlockNumber { + if from >= rpc.LatestBlockNumber && to == rpc.PendingBlockNumber { return es.subscribeMinedPendingLogs(crit, logs), nil } // interested in logs from a specific block number to new mined blocks - if from >= 0 && to == myrpc.LatestBlockNumber { + if from >= 0 && to == rpc.LatestBlockNumber { return es.subscribeLogs(crit, logs), nil } return nil, fmt.Errorf("invalid from and to block combination: from > to") diff --git a/eth/gasprice/gasprice.go b/eth/gasprice/gasprice.go index 9b632c2..23e49a6 100644 --- a/eth/gasprice/gasprice.go +++ b/eth/gasprice/gasprice.go @@ -22,11 +22,11 @@ import ( "sort" "sync" + "github.com/ava-labs/coreth/core/types" "github.com/ava-labs/coreth/internal/ethapi" "github.com/ava-labs/coreth/params" "github.com/ava-labs/coreth/rpc" "github.com/ava-labs/go-ethereum/common" - "github.com/ava-labs/go-ethereum/core/types" ) var maxPrice = big.NewInt(500 * params.GWei) diff --git a/eth/gen_config.go b/eth/gen_config.go index 6b8ddf1..617a885 100644 --- a/eth/gen_config.go +++ b/eth/gen_config.go @@ -6,13 +6,13 @@ import ( "math/big" "time" + "github.com/ava-labs/coreth/consensus/ethash" "github.com/ava-labs/coreth/core" "github.com/ava-labs/coreth/eth/gasprice" "github.com/ava-labs/coreth/miner" + "github.com/ava-labs/coreth/params" "github.com/ava-labs/go-ethereum/common" - "github.com/ava-labs/go-ethereum/consensus/ethash" "github.com/ava-labs/go-ethereum/eth/downloader" - "github.com/ava-labs/go-ethereum/params" ) // MarshalTOML marshals as TOML. diff --git a/eth/handler.go b/eth/handler.go deleted file mode 100644 index f502973..0000000 --- a/eth/handler.go +++ /dev/null @@ -1,844 +0,0 @@ -// Copyright 2015 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 . - -package eth - -import ( - "encoding/json" - "errors" - "fmt" - "math" - "math/big" - "sync" - "sync/atomic" - "time" - - "github.com/ava-labs/coreth/core" - "github.com/ava-labs/go-ethereum/common" - "github.com/ava-labs/go-ethereum/consensus" - "github.com/ava-labs/go-ethereum/core/types" - "github.com/ava-labs/go-ethereum/eth/downloader" - "github.com/ava-labs/go-ethereum/eth/fetcher" - "github.com/ava-labs/go-ethereum/ethdb" - "github.com/ava-labs/go-ethereum/event" - "github.com/ava-labs/go-ethereum/log" - "github.com/ava-labs/go-ethereum/p2p" - "github.com/ava-labs/go-ethereum/p2p/enode" - "github.com/ava-labs/go-ethereum/params" - "github.com/ava-labs/go-ethereum/rlp" - "github.com/ava-labs/go-ethereum/trie" -) - -const ( - softResponseLimit = 2 * 1024 * 1024 // Target maximum size of returned blocks, headers or node data. - estHeaderRlpSize = 500 // Approximate size of an RLP encoded block header - - // txChanSize is the size of channel listening to NewTxsEvent. - // The number is referenced from the size of tx pool. - txChanSize = 4096 - - // minimim number of peers to broadcast new blocks to - minBroadcastPeers = 4 -) - -var ( - syncChallengeTimeout = 15 * time.Second // Time allowance for a node to reply to the sync progress challenge -) - -func errResp(code errCode, format string, v ...interface{}) error { - return fmt.Errorf("%v - %v", code, fmt.Sprintf(format, v...)) -} - -type ProtocolManager struct { - networkID uint64 - - fastSync uint32 // Flag whether fast sync is enabled (gets disabled if we already have blocks) - acceptTxs uint32 // Flag whether we're considered synchronised (enables transaction processing) - - checkpointNumber uint64 // Block number for the sync progress validator to cross reference - checkpointHash common.Hash // Block hash for the sync progress validator to cross reference - - txpool txPool - blockchain *core.BlockChain - maxPeers int - - downloader *downloader.Downloader - fetcher *fetcher.Fetcher - peers *peerSet - - eventMux *event.TypeMux - txsCh chan core.NewTxsEvent - txsSub event.Subscription - minedBlockSub *event.TypeMuxSubscription - - whitelist map[uint64]common.Hash - - // channels for fetcher, syncer, txsyncLoop - newPeerCh chan *peer - txsyncCh chan *txsync - quitSync chan struct{} - noMorePeers chan struct{} - - // wait group is used for graceful shutdowns during downloading - // and processing - wg sync.WaitGroup -} - -// NewProtocolManager returns a new Ethereum sub protocol manager. The Ethereum sub protocol manages peers capable -// with the Ethereum network. -func NewProtocolManager(config *params.ChainConfig, checkpoint *params.TrustedCheckpoint, mode downloader.SyncMode, networkID uint64, mux *event.TypeMux, txpool txPool, engine consensus.Engine, blockchain *core.BlockChain, chaindb ethdb.Database, cacheLimit int, whitelist map[uint64]common.Hash) (*ProtocolManager, error) { - // Create the protocol manager with the base fields - manager := &ProtocolManager{ - networkID: networkID, - eventMux: mux, - txpool: txpool, - blockchain: blockchain, - peers: newPeerSet(), - whitelist: whitelist, - newPeerCh: make(chan *peer), - noMorePeers: make(chan struct{}), - txsyncCh: make(chan *txsync), - quitSync: make(chan struct{}), - } - if mode == downloader.FullSync { - // The database seems empty as the current block is the genesis. Yet the fast - // block is ahead, so fast sync was enabled for this node at a certain point. - // The scenarios where this can happen is - // * if the user manually (or via a bad block) rolled back a fast sync node - // below the sync point. - // * the last fast sync is not finished while user specifies a full sync this - // time. But we don't have any recent state for full sync. - // In these cases however it's safe to reenable fast sync. - fullBlock, fastBlock := blockchain.CurrentBlock(), blockchain.CurrentFastBlock() - if fullBlock.NumberU64() == 0 && fastBlock.NumberU64() > 0 { - manager.fastSync = uint32(1) - log.Warn("Switch sync mode from full sync to fast sync") - } - } else { - if blockchain.CurrentBlock().NumberU64() > 0 { - // Print warning log if database is not empty to run fast sync. - log.Warn("Switch sync mode from fast sync to full sync") - } else { - // If fast sync was requested and our database is empty, grant it - manager.fastSync = uint32(1) - } - } - // If we have trusted checkpoints, enforce them on the chain - if checkpoint != nil { - manager.checkpointNumber = (checkpoint.SectionIndex+1)*params.CHTFrequency - 1 - manager.checkpointHash = checkpoint.SectionHead - } - - // Construct the downloader (long sync) and its backing state bloom if fast - // sync is requested. The downloader is responsible for deallocating the state - // bloom when it's done. - var stateBloom *trie.SyncBloom - if atomic.LoadUint32(&manager.fastSync) == 1 { - stateBloom = trie.NewSyncBloom(uint64(cacheLimit), chaindb) - } - manager.downloader = downloader.New(manager.checkpointNumber, chaindb, stateBloom, manager.eventMux, blockchain, nil, manager.removePeer) - - // Construct the fetcher (short sync) - validator := func(header *types.Header) error { - return engine.VerifyHeader(blockchain, header, true) - } - heighter := func() uint64 { - return blockchain.CurrentBlock().NumberU64() - } - inserter := func(blocks types.Blocks) (int, error) { - // If sync hasn't reached the checkpoint yet, deny importing weird blocks. - // - // Ideally we would also compare the head block's timestamp and similarly reject - // the propagated block if the head is too old. Unfortunately there is a corner - // case when starting new networks, where the genesis might be ancient (0 unix) - // which would prevent full nodes from accepting it. - if manager.blockchain.CurrentBlock().NumberU64() < manager.checkpointNumber { - log.Warn("Unsynced yet, discarded propagated block", "number", blocks[0].Number(), "hash", blocks[0].Hash()) - return 0, nil - } - // If fast sync is running, deny importing weird blocks. This is a problematic - // clause when starting up a new network, because fast-syncing miners might not - // accept each others' blocks until a restart. Unfortunately we haven't figured - // out a way yet where nodes can decide unilaterally whether the network is new - // or not. This should be fixed if we figure out a solution. - if atomic.LoadUint32(&manager.fastSync) == 1 { - log.Warn("Fast syncing, discarded propagated block", "number", blocks[0].Number(), "hash", blocks[0].Hash()) - return 0, nil - } - n, err := manager.blockchain.InsertChain(blocks) - if err == nil { - atomic.StoreUint32(&manager.acceptTxs, 1) // Mark initial sync done on any fetcher import - } - return n, err - } - manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer) - - return manager, nil -} - -func (pm *ProtocolManager) makeProtocol(version uint) p2p.Protocol { - length, ok := protocolLengths[version] - if !ok { - panic("makeProtocol for unknown version") - } - - return p2p.Protocol{ - Name: protocolName, - Version: version, - Length: length, - Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - peer := pm.newPeer(int(version), p, rw) - select { - case pm.newPeerCh <- peer: - pm.wg.Add(1) - defer pm.wg.Done() - return pm.handle(peer) - case <-pm.quitSync: - return p2p.DiscQuitting - } - }, - NodeInfo: func() interface{} { - return pm.NodeInfo() - }, - PeerInfo: func(id enode.ID) interface{} { - if p := pm.peers.Peer(fmt.Sprintf("%x", id[:8])); p != nil { - return p.Info() - } - return nil - }, - } -} - -func (pm *ProtocolManager) removePeer(id string) { - // Short circuit if the peer was already removed - peer := pm.peers.Peer(id) - if peer == nil { - return - } - log.Debug("Removing Ethereum peer", "peer", id) - - // Unregister the peer from the downloader and Ethereum peer set - pm.downloader.UnregisterPeer(id) - if err := pm.peers.Unregister(id); err != nil { - log.Error("Peer removal failed", "peer", id, "err", err) - } - // Hard disconnect at the networking layer - if peer != nil { - peer.Peer.Disconnect(p2p.DiscUselessPeer) - } -} - -func (pm *ProtocolManager) Start(maxPeers int) { - pm.maxPeers = maxPeers - - // broadcast transactions - pm.txsCh = make(chan core.NewTxsEvent, txChanSize) - pm.txsSub = pm.txpool.SubscribeNewTxsEvent(pm.txsCh) - go pm.txBroadcastLoop() - - // broadcast mined blocks - pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{}) - go pm.minedBroadcastLoop() - - // start sync handlers - go pm.syncer() - go pm.txsyncLoop() -} - -func (pm *ProtocolManager) Stop() { - log.Info("Stopping Ethereum protocol") - - pm.txsSub.Unsubscribe() // quits txBroadcastLoop - pm.minedBlockSub.Unsubscribe() // quits blockBroadcastLoop - - // Quit the sync loop. - // After this send has completed, no new peers will be accepted. - pm.noMorePeers <- struct{}{} - - // Quit fetcher, txsyncLoop. - close(pm.quitSync) - - // Disconnect existing sessions. - // This also closes the gate for any new registrations on the peer set. - // sessions which are already established but not added to pm.peers yet - // will exit when they try to register. - pm.peers.Close() - - // Wait for all peer handler goroutines and the loops to come down. - pm.wg.Wait() - - log.Info("Ethereum protocol stopped") -} - -func (pm *ProtocolManager) newPeer(pv int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { - return newPeer(pv, p, newMeteredMsgWriter(rw)) -} - -// handle is the callback invoked to manage the life cycle of an eth peer. When -// this function terminates, the peer is disconnected. -func (pm *ProtocolManager) handle(p *peer) error { - // Ignore maxPeers if this is a trusted peer - if pm.peers.Len() >= pm.maxPeers && !p.Peer.Info().Network.Trusted { - return p2p.DiscTooManyPeers - } - p.Log().Debug("Ethereum peer connected", "name", p.Name()) - - // Execute the Ethereum handshake - var ( - genesis = pm.blockchain.Genesis() - head = pm.blockchain.CurrentHeader() - hash = head.Hash() - number = head.Number.Uint64() - td = pm.blockchain.GetTd(hash, number) - ) - if err := p.Handshake(pm.networkID, td, hash, genesis.Hash()); err != nil { - p.Log().Debug("Ethereum handshake failed", "err", err) - return err - } - if rw, ok := p.rw.(*meteredMsgReadWriter); ok { - rw.Init(p.version) - } - // Register the peer locally - if err := pm.peers.Register(p); err != nil { - p.Log().Error("Ethereum peer registration failed", "err", err) - return err - } - defer pm.removePeer(p.id) - - // Register the peer in the downloader. If the downloader considers it banned, we disconnect - if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil { - return err - } - // Propagate existing transactions. new transactions appearing - // after this will be sent via broadcasts. - pm.syncTransactions(p) - - // If we have a trusted CHT, reject all peers below that (avoid fast sync eclipse) - if pm.checkpointHash != (common.Hash{}) { - // Request the peer's checkpoint header for chain height/weight validation - if err := p.RequestHeadersByNumber(pm.checkpointNumber, 1, 0, false); err != nil { - return err - } - // Start a timer to disconnect if the peer doesn't reply in time - p.syncDrop = time.AfterFunc(syncChallengeTimeout, func() { - p.Log().Warn("Checkpoint challenge timed out, dropping", "addr", p.RemoteAddr(), "type", p.Name()) - pm.removePeer(p.id) - }) - // Make sure it's cleaned up if the peer dies off - defer func() { - if p.syncDrop != nil { - p.syncDrop.Stop() - p.syncDrop = nil - } - }() - } - // If we have any explicit whitelist block hashes, request them - for number := range pm.whitelist { - if err := p.RequestHeadersByNumber(number, 1, 0, false); err != nil { - return err - } - } - // Handle incoming messages until the connection is torn down - for { - if err := pm.handleMsg(p); err != nil { - p.Log().Debug("Ethereum message handling failed", "err", err) - return err - } - } -} - -// handleMsg is invoked whenever an inbound message is received from a remote -// peer. The remote connection is torn down upon returning any error. -func (pm *ProtocolManager) handleMsg(p *peer) error { - // Read the next message from the remote peer, and ensure it's fully consumed - msg, err := p.rw.ReadMsg() - if err != nil { - return err - } - if msg.Size > protocolMaxMsgSize { - return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, protocolMaxMsgSize) - } - defer msg.Discard() - - // Handle the message depending on its contents - switch { - case msg.Code == StatusMsg: - // Status messages should never arrive after the handshake - return errResp(ErrExtraStatusMsg, "uncontrolled status message") - - // Block header query, collect the requested headers and reply - case msg.Code == GetBlockHeadersMsg: - // Decode the complex header query - var query getBlockHeadersData - if err := msg.Decode(&query); err != nil { - return errResp(ErrDecode, "%v: %v", msg, err) - } - hashMode := query.Origin.Hash != (common.Hash{}) - first := true - maxNonCanonical := uint64(100) - - // Gather headers until the fetch or network limits is reached - var ( - bytes common.StorageSize - headers []*types.Header - unknown bool - ) - for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit && len(headers) < downloader.MaxHeaderFetch { - // Retrieve the next header satisfying the query - var origin *types.Header - if hashMode { - if first { - first = false - origin = pm.blockchain.GetHeaderByHash(query.Origin.Hash) - if origin != nil { - query.Origin.Number = origin.Number.Uint64() - } - } else { - origin = pm.blockchain.GetHeader(query.Origin.Hash, query.Origin.Number) - } - } else { - origin = pm.blockchain.GetHeaderByNumber(query.Origin.Number) - } - if origin == nil { - break - } - headers = append(headers, origin) - bytes += estHeaderRlpSize - - // Advance to the next header of the query - switch { - case hashMode && query.Reverse: - // Hash based traversal towards the genesis block - ancestor := query.Skip + 1 - if ancestor == 0 { - unknown = true - } else { - query.Origin.Hash, query.Origin.Number = pm.blockchain.GetAncestor(query.Origin.Hash, query.Origin.Number, ancestor, &maxNonCanonical) - unknown = (query.Origin.Hash == common.Hash{}) - } - case hashMode && !query.Reverse: - // Hash based traversal towards the leaf block - var ( - current = origin.Number.Uint64() - next = current + query.Skip + 1 - ) - if next <= current { - infos, _ := json.MarshalIndent(p.Peer.Info(), "", " ") - p.Log().Warn("GetBlockHeaders skip overflow attack", "current", current, "skip", query.Skip, "next", next, "attacker", infos) - unknown = true - } else { - if header := pm.blockchain.GetHeaderByNumber(next); header != nil { - nextHash := header.Hash() - expOldHash, _ := pm.blockchain.GetAncestor(nextHash, next, query.Skip+1, &maxNonCanonical) - if expOldHash == query.Origin.Hash { - query.Origin.Hash, query.Origin.Number = nextHash, next - } else { - unknown = true - } - } else { - unknown = true - } - } - case query.Reverse: - // Number based traversal towards the genesis block - if query.Origin.Number >= query.Skip+1 { - query.Origin.Number -= query.Skip + 1 - } else { - unknown = true - } - - case !query.Reverse: - // Number based traversal towards the leaf block - query.Origin.Number += query.Skip + 1 - } - } - return p.SendBlockHeaders(headers) - - case msg.Code == BlockHeadersMsg: - // A batch of headers arrived to one of our previous requests - var headers []*types.Header - if err := msg.Decode(&headers); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) - } - // If no headers were received, but we're expencting a checkpoint header, consider it that - if len(headers) == 0 && p.syncDrop != nil { - // Stop the timer either way, decide later to drop or not - p.syncDrop.Stop() - p.syncDrop = nil - - // If we're doing a fast sync, we must enforce the checkpoint block to avoid - // eclipse attacks. Unsynced nodes are welcome to connect after we're done - // joining the network - if atomic.LoadUint32(&pm.fastSync) == 1 { - p.Log().Warn("Dropping unsynced node during fast sync", "addr", p.RemoteAddr(), "type", p.Name()) - return errors.New("unsynced node cannot serve fast sync") - } - } - // Filter out any explicitly requested headers, deliver the rest to the downloader - filter := len(headers) == 1 - if filter { - // If it's a potential sync progress check, validate the content and advertised chain weight - if p.syncDrop != nil && headers[0].Number.Uint64() == pm.checkpointNumber { - // Disable the sync drop timer - p.syncDrop.Stop() - p.syncDrop = nil - - // Validate the header and either drop the peer or continue - if headers[0].Hash() != pm.checkpointHash { - return errors.New("checkpoint hash mismatch") - } - return nil - } - // Otherwise if it's a whitelisted block, validate against the set - if want, ok := pm.whitelist[headers[0].Number.Uint64()]; ok { - if hash := headers[0].Hash(); want != hash { - p.Log().Info("Whitelist mismatch, dropping peer", "number", headers[0].Number.Uint64(), "hash", hash, "want", want) - return errors.New("whitelist block mismatch") - } - p.Log().Debug("Whitelist block verified", "number", headers[0].Number.Uint64(), "hash", want) - } - // Irrelevant of the fork checks, send the header to the fetcher just in case - headers = pm.fetcher.FilterHeaders(p.id, headers, time.Now()) - } - if len(headers) > 0 || !filter { - err := pm.downloader.DeliverHeaders(p.id, headers) - if err != nil { - log.Debug("Failed to deliver headers", "err", err) - } - } - - case msg.Code == GetBlockBodiesMsg: - // 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 ( - hash common.Hash - bytes int - bodies []rlp.RawValue - ) - for bytes < softResponseLimit && len(bodies) < downloader.MaxBlockFetch { - // Retrieve the hash of the next block - if err := msgStream.Decode(&hash); err == rlp.EOL { - break - } else if err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) - } - // Retrieve the requested block body, stopping if enough was found - if data := pm.blockchain.GetBodyRLP(hash); len(data) != 0 { - bodies = append(bodies, data) - bytes += len(data) - } - } - return p.SendBlockBodiesRLP(bodies) - - case msg.Code == BlockBodiesMsg: - // A batch of block bodies arrived to one of our previous requests - var request blockBodiesData - if err := msg.Decode(&request); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) - } - // Deliver them all to the downloader for queuing - transactions := make([][]*types.Transaction, len(request)) - uncles := make([][]*types.Header, len(request)) - - for i, body := range request { - transactions[i] = body.Transactions - uncles[i] = body.Uncles - } - // Filter out any explicitly requested bodies, deliver the rest to the downloader - filter := len(transactions) > 0 || len(uncles) > 0 - if filter { - transactions, uncles = pm.fetcher.FilterBodies(p.id, transactions, uncles, time.Now()) - } - if len(transactions) > 0 || len(uncles) > 0 || !filter { - err := pm.downloader.DeliverBodies(p.id, transactions, uncles) - if err != nil { - log.Debug("Failed to deliver bodies", "err", err) - } - } - - case p.version >= eth63 && msg.Code == GetNodeDataMsg: - // Decode the retrieval message - msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) - if _, err := msgStream.List(); err != nil { - return err - } - // Gather state data until the fetch or network limits is reached - var ( - hash common.Hash - bytes int - data [][]byte - ) - for bytes < softResponseLimit && len(data) < downloader.MaxStateFetch { - // Retrieve the hash of the next state entry - if err := msgStream.Decode(&hash); err == rlp.EOL { - break - } else if err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) - } - // Retrieve the requested state entry, stopping if enough was found - if entry, err := pm.blockchain.TrieNode(hash); err == nil { - data = append(data, entry) - bytes += len(entry) - } - } - return p.SendNodeData(data) - - case p.version >= eth63 && msg.Code == NodeDataMsg: - // A batch of node state data arrived to one of our previous requests - var data [][]byte - if err := msg.Decode(&data); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) - } - // Deliver all to the downloader - if err := pm.downloader.DeliverNodeData(p.id, data); err != nil { - log.Debug("Failed to deliver node state data", "err", err) - } - - case p.version >= eth63 && msg.Code == GetReceiptsMsg: - // Decode the retrieval message - msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) - if _, err := msgStream.List(); err != nil { - return err - } - // Gather state data until the fetch or network limits is reached - var ( - hash common.Hash - bytes int - receipts []rlp.RawValue - ) - for bytes < softResponseLimit && len(receipts) < downloader.MaxReceiptFetch { - // Retrieve the hash of the next block - if err := msgStream.Decode(&hash); err == rlp.EOL { - break - } else if err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) - } - // Retrieve the requested block's receipts, skipping if unknown to us - results := pm.blockchain.GetReceiptsByHash(hash) - if results == nil { - if header := pm.blockchain.GetHeaderByHash(hash); header == nil || header.ReceiptHash != types.EmptyRootHash { - continue - } - } - // If known, encode and queue for response packet - if encoded, err := rlp.EncodeToBytes(results); err != nil { - log.Error("Failed to encode receipt", "err", err) - } else { - receipts = append(receipts, encoded) - bytes += len(encoded) - } - } - return p.SendReceiptsRLP(receipts) - - case p.version >= eth63 && msg.Code == ReceiptsMsg: - // A batch of receipts arrived to one of our previous requests - var receipts [][]*types.Receipt - if err := msg.Decode(&receipts); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) - } - // Deliver all to the downloader - if err := pm.downloader.DeliverReceipts(p.id, receipts); err != nil { - log.Debug("Failed to deliver receipts", "err", err) - } - - case msg.Code == NewBlockHashesMsg: - var announces newBlockHashesData - if err := msg.Decode(&announces); err != nil { - return errResp(ErrDecode, "%v: %v", msg, err) - } - // Mark the hashes as present at the remote node - for _, block := range announces { - p.MarkBlock(block.Hash) - } - // Schedule all the unknown hashes for retrieval - unknown := make(newBlockHashesData, 0, len(announces)) - for _, block := range announces { - if !pm.blockchain.HasBlock(block.Hash, block.Number) { - unknown = append(unknown, block) - } - } - for _, block := range unknown { - pm.fetcher.Notify(p.id, block.Hash, block.Number, time.Now(), p.RequestOneHeader, p.RequestBodies) - } - - case msg.Code == NewBlockMsg: - // Retrieve and decode the propagated block - var request newBlockData - if err := msg.Decode(&request); err != nil { - return errResp(ErrDecode, "%v: %v", msg, err) - } - if err := request.sanityCheck(); err != nil { - return err - } - request.Block.ReceivedAt = msg.ReceivedAt - request.Block.ReceivedFrom = p - - // Mark the peer as owning the block and schedule it for import - p.MarkBlock(request.Block.Hash()) - pm.fetcher.Enqueue(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. - var ( - trueHead = request.Block.ParentHash() - trueTD = new(big.Int).Sub(request.TD, request.Block.Difficulty()) - ) - // Update the peer's total difficulty if better than the previous - if _, td := p.Head(); trueTD.Cmp(td) > 0 { - p.SetHead(trueHead, trueTD) - - // Schedule a sync if above ours. Note, this will not fire a sync for a gap of - // a single block (as the true TD is below the propagated block), however this - // scenario should easily be covered by the fetcher. - currentBlock := pm.blockchain.CurrentBlock() - if trueTD.Cmp(pm.blockchain.GetTd(currentBlock.Hash(), currentBlock.NumberU64())) > 0 { - go pm.synchronise(p) - } - } - - case msg.Code == TxMsg: - // Transactions arrived, make sure we have a valid and fresh chain to handle them - if atomic.LoadUint32(&pm.acceptTxs) == 0 { - break - } - // Transactions can be processed, parse all of them and deliver to the pool - var txs []*types.Transaction - if err := msg.Decode(&txs); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) - } - for i, tx := range txs { - // Validate and mark the remote transaction - if tx == nil { - return errResp(ErrDecode, "transaction %d is nil", i) - } - p.MarkTransaction(tx.Hash()) - } - pm.txpool.AddRemotes(txs) - - default: - return errResp(ErrInvalidMsgCode, "%v", msg.Code) - } - return nil -} - -// 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() - peers := pm.peers.PeersWithoutBlock(hash) - - // 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.blockchain.GetBlock(block.ParentHash(), block.NumberU64()-1); parent != nil { - td = new(big.Int).Add(block.Difficulty(), pm.blockchain.GetTd(block.ParentHash(), block.NumberU64()-1)) - } else { - log.Error("Propagating dangling block", "number", block.Number(), "hash", hash) - return - } - // Send the block to a subset of our peers - transferLen := int(math.Sqrt(float64(len(peers)))) - if transferLen < minBroadcastPeers { - transferLen = minBroadcastPeers - } - if transferLen > len(peers) { - transferLen = len(peers) - } - transfer := peers[:transferLen] - for _, peer := range transfer { - peer.AsyncSendNewBlock(block, td) - } - log.Trace("Propagated block", "hash", hash, "recipients", len(transfer), "duration", common.PrettyDuration(time.Since(block.ReceivedAt))) - return - } - // Otherwise if the block is indeed in out own chain, announce it - if pm.blockchain.HasBlock(hash, block.NumberU64()) { - for _, peer := range peers { - peer.AsyncSendNewBlockHash(block) - } - log.Trace("Announced block", "hash", hash, "recipients", len(peers), "duration", common.PrettyDuration(time.Since(block.ReceivedAt))) - } -} - -// BroadcastTxs will propagate a batch of transactions to all peers which are not known to -// already have the given transaction. -func (pm *ProtocolManager) BroadcastTxs(txs types.Transactions) { - var txset = make(map[*peer]types.Transactions) - - // Broadcast transactions to a batch of peers not knowing about it - for _, tx := range txs { - peers := pm.peers.PeersWithoutTx(tx.Hash()) - for _, peer := range peers { - txset[peer] = append(txset[peer], tx) - } - log.Trace("Broadcast transaction", "hash", tx.Hash(), "recipients", len(peers)) - } - // FIXME include this again: peers = peers[:int(math.Sqrt(float64(len(peers))))] - for peer, txs := range txset { - peer.AsyncSendTransactions(txs) - } -} - -// Mined broadcast loop -func (pm *ProtocolManager) minedBroadcastLoop() { - // automatically stops if unsubscribe - for obj := range pm.minedBlockSub.Chan() { - if ev, ok := obj.Data.(core.NewMinedBlockEvent); ok { - pm.BroadcastBlock(ev.Block, true) // First propagate block to peers - pm.BroadcastBlock(ev.Block, false) // Only then announce to the rest - } - } -} - -func (pm *ProtocolManager) txBroadcastLoop() { - for { - select { - case event := <-pm.txsCh: - pm.BroadcastTxs(event.Txs) - - // Err() channel will be closed when unsubscribing. - case <-pm.txsSub.Err(): - return - } - } -} - -// NodeInfo represents a short summary of the Ethereum sub-protocol metadata -// known about the host peer. -type NodeInfo struct { - Network uint64 `json:"network"` // Ethereum network ID (1=Frontier, 2=Morden, Ropsten=3, Rinkeby=4) - Difficulty *big.Int `json:"difficulty"` // Total difficulty of the host's blockchain - Genesis common.Hash `json:"genesis"` // SHA3 hash of the host's genesis block - Config *params.ChainConfig `json:"config"` // Chain configuration for the fork rules - Head common.Hash `json:"head"` // SHA3 hash of the host's best owned block -} - -// NodeInfo retrieves some protocol metadata about the running host node. -func (pm *ProtocolManager) NodeInfo() *NodeInfo { - currentBlock := pm.blockchain.CurrentBlock() - return &NodeInfo{ - Network: pm.networkID, - Difficulty: pm.blockchain.GetTd(currentBlock.Hash(), currentBlock.NumberU64()), - Genesis: pm.blockchain.Genesis().Hash(), - Config: pm.blockchain.Config(), - Head: currentBlock.Hash(), - } -} diff --git a/eth/peer.go b/eth/peer.go deleted file mode 100644 index 9ce569c..0000000 --- a/eth/peer.go +++ /dev/null @@ -1,546 +0,0 @@ -// Copyright 2015 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 . - -package eth - -import ( - "errors" - "fmt" - "math/big" - "sync" - "time" - - mapset "github.com/deckarep/golang-set" - "github.com/ava-labs/go-ethereum/common" - "github.com/ava-labs/go-ethereum/core/types" - "github.com/ava-labs/go-ethereum/p2p" - "github.com/ava-labs/go-ethereum/rlp" -) - -var ( - errClosed = errors.New("peer set is closed") - errAlreadyRegistered = errors.New("peer is already registered") - errNotRegistered = errors.New("peer is not registered") -) - -const ( - maxKnownTxs = 32768 // Maximum transactions hashes to keep in the known list (prevent DOS) - maxKnownBlocks = 1024 // Maximum block hashes to keep in the known list (prevent DOS) - - // maxQueuedTxs is the maximum number of transaction lists to queue up before - // dropping broadcasts. This is a sensitive number as a transaction list might - // contain a single transaction, or thousands. - maxQueuedTxs = 128 - - // maxQueuedProps is the maximum number of block propagations to queue up before - // dropping broadcasts. There's not much point in queueing stale blocks, so a few - // that might cover uncles should be enough. - maxQueuedProps = 4 - - // maxQueuedAnns is the maximum number of block announcements to queue up before - // dropping broadcasts. Similarly to block propagations, there's no point to queue - // above some healthy uncle limit, so use that. - maxQueuedAnns = 4 - - handshakeTimeout = 5 * time.Second -) - -// PeerInfo represents a short summary of the Ethereum sub-protocol metadata known -// about a connected peer. -type PeerInfo struct { - Version int `json:"version"` // Ethereum protocol version negotiated - Difficulty *big.Int `json:"difficulty"` // Total difficulty of the peer's blockchain - Head string `json:"head"` // SHA3 hash of the peer's best owned block -} - -// propEvent is a block propagation, waiting for its turn in the broadcast queue. -type propEvent struct { - block *types.Block - td *big.Int -} - -type peer struct { - id string - - *p2p.Peer - rw p2p.MsgReadWriter - - version int // Protocol version negotiated - syncDrop *time.Timer // Timed connection dropper if sync progress isn't validated in time - - head common.Hash - td *big.Int - lock sync.RWMutex - - knownTxs mapset.Set // Set of transaction hashes known to be known by this peer - knownBlocks mapset.Set // Set of block hashes known to be known by this peer - queuedTxs chan []*types.Transaction // Queue of transactions to broadcast to the peer - queuedProps chan *propEvent // Queue of blocks to broadcast to the peer - queuedAnns chan *types.Block // Queue of blocks to announce to the peer - term chan struct{} // Termination channel to stop the broadcaster -} - -func newPeer(version int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { - return &peer{ - Peer: p, - rw: rw, - version: version, - id: fmt.Sprintf("%x", p.ID().Bytes()[:8]), - knownTxs: mapset.NewSet(), - knownBlocks: mapset.NewSet(), - queuedTxs: make(chan []*types.Transaction, maxQueuedTxs), - queuedProps: make(chan *propEvent, maxQueuedProps), - queuedAnns: make(chan *types.Block, maxQueuedAnns), - term: make(chan struct{}), - } -} - -// broadcast is a write loop that multiplexes block propagations, announcements -// and transaction broadcasts into the remote peer. The goal is to have an async -// writer that does not lock up node internals. -func (p *peer) broadcast() { - for { - select { - case txs := <-p.queuedTxs: - if err := p.SendTransactions(txs); err != nil { - return - } - p.Log().Trace("Broadcast transactions", "count", len(txs)) - - case prop := <-p.queuedProps: - if err := p.SendNewBlock(prop.block, prop.td); err != nil { - return - } - p.Log().Trace("Propagated block", "number", prop.block.Number(), "hash", prop.block.Hash(), "td", prop.td) - - case block := <-p.queuedAnns: - if err := p.SendNewBlockHashes([]common.Hash{block.Hash()}, []uint64{block.NumberU64()}); err != nil { - return - } - p.Log().Trace("Announced block", "number", block.Number(), "hash", block.Hash()) - - case <-p.term: - return - } - } -} - -// close signals the broadcast goroutine to terminate. -func (p *peer) close() { - close(p.term) -} - -// Info gathers and returns a collection of metadata known about a peer. -func (p *peer) Info() *PeerInfo { - hash, td := p.Head() - - return &PeerInfo{ - Version: p.version, - Difficulty: td, - Head: hash.Hex(), - } -} - -// Head retrieves a copy of the current head hash and total difficulty of the -// peer. -func (p *peer) Head() (hash common.Hash, td *big.Int) { - p.lock.RLock() - defer p.lock.RUnlock() - - copy(hash[:], p.head[:]) - return hash, new(big.Int).Set(p.td) -} - -// SetHead updates the head hash and total difficulty of the peer. -func (p *peer) SetHead(hash common.Hash, td *big.Int) { - p.lock.Lock() - defer p.lock.Unlock() - - copy(p.head[:], hash[:]) - p.td.Set(td) -} - -// MarkBlock marks a block as known for the peer, ensuring that the block will -// never be propagated to this particular peer. -func (p *peer) MarkBlock(hash common.Hash) { - // If we reached the memory allowance, drop a previously known block hash - for p.knownBlocks.Cardinality() >= maxKnownBlocks { - p.knownBlocks.Pop() - } - p.knownBlocks.Add(hash) -} - -// MarkTransaction marks a transaction as known for the peer, ensuring that it -// will never be propagated to this particular peer. -func (p *peer) MarkTransaction(hash common.Hash) { - // If we reached the memory allowance, drop a previously known transaction hash - for p.knownTxs.Cardinality() >= maxKnownTxs { - p.knownTxs.Pop() - } - p.knownTxs.Add(hash) -} - -// SendTransactions sends transactions to the peer and includes the hashes -// in its transaction hash set for future reference. -func (p *peer) SendTransactions(txs types.Transactions) error { - // Mark all the transactions as known, but ensure we don't overflow our limits - for _, tx := range txs { - p.knownTxs.Add(tx.Hash()) - } - for p.knownTxs.Cardinality() >= maxKnownTxs { - p.knownTxs.Pop() - } - return p2p.Send(p.rw, TxMsg, txs) -} - -// AsyncSendTransactions queues list of transactions propagation to a remote -// peer. If the peer's broadcast queue is full, the event is silently dropped. -func (p *peer) AsyncSendTransactions(txs []*types.Transaction) { - select { - case p.queuedTxs <- txs: - // Mark all the transactions as known, but ensure we don't overflow our limits - for _, tx := range txs { - p.knownTxs.Add(tx.Hash()) - } - for p.knownTxs.Cardinality() >= maxKnownTxs { - p.knownTxs.Pop() - } - default: - p.Log().Debug("Dropping transaction propagation", "count", len(txs)) - } -} - -// SendNewBlockHashes announces the availability of a number of blocks through -// a hash notification. -func (p *peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error { - // Mark all the block hashes as known, but ensure we don't overflow our limits - for _, hash := range hashes { - p.knownBlocks.Add(hash) - } - for p.knownBlocks.Cardinality() >= maxKnownBlocks { - p.knownBlocks.Pop() - } - request := make(newBlockHashesData, len(hashes)) - for i := 0; i < len(hashes); i++ { - request[i].Hash = hashes[i] - request[i].Number = numbers[i] - } - return p2p.Send(p.rw, NewBlockHashesMsg, request) -} - -// AsyncSendNewBlockHash queues the availability of a block for propagation to a -// remote peer. If the peer's broadcast queue is full, the event is silently -// dropped. -func (p *peer) AsyncSendNewBlockHash(block *types.Block) { - select { - case p.queuedAnns <- block: - // Mark all the block hash as known, but ensure we don't overflow our limits - p.knownBlocks.Add(block.Hash()) - for p.knownBlocks.Cardinality() >= maxKnownBlocks { - p.knownBlocks.Pop() - } - default: - p.Log().Debug("Dropping block announcement", "number", block.NumberU64(), "hash", block.Hash()) - } -} - -// SendNewBlock propagates an entire block to a remote peer. -func (p *peer) SendNewBlock(block *types.Block, td *big.Int) error { - // Mark all the block hash as known, but ensure we don't overflow our limits - p.knownBlocks.Add(block.Hash()) - for p.knownBlocks.Cardinality() >= maxKnownBlocks { - p.knownBlocks.Pop() - } - return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td}) -} - -// AsyncSendNewBlock queues an entire block for propagation to a remote peer. If -// the peer's broadcast queue is full, the event is silently dropped. -func (p *peer) AsyncSendNewBlock(block *types.Block, td *big.Int) { - select { - case p.queuedProps <- &propEvent{block: block, td: td}: - // Mark all the block hash as known, but ensure we don't overflow our limits - p.knownBlocks.Add(block.Hash()) - for p.knownBlocks.Cardinality() >= maxKnownBlocks { - p.knownBlocks.Pop() - } - default: - p.Log().Debug("Dropping block propagation", "number", block.NumberU64(), "hash", block.Hash()) - } -} - -// SendBlockHeaders sends a batch of block headers to the remote peer. -func (p *peer) SendBlockHeaders(headers []*types.Header) error { - return p2p.Send(p.rw, BlockHeadersMsg, headers) -} - -// SendBlockBodies sends a batch of block contents to the remote peer. -func (p *peer) SendBlockBodies(bodies []*blockBody) error { - return p2p.Send(p.rw, BlockBodiesMsg, blockBodiesData(bodies)) -} - -// SendBlockBodiesRLP sends a batch of block contents to the remote peer from -// an already RLP encoded format. -func (p *peer) SendBlockBodiesRLP(bodies []rlp.RawValue) error { - return p2p.Send(p.rw, BlockBodiesMsg, bodies) -} - -// SendNodeDataRLP sends a batch of arbitrary internal data, corresponding to the -// hashes requested. -func (p *peer) SendNodeData(data [][]byte) error { - return p2p.Send(p.rw, NodeDataMsg, data) -} - -// SendReceiptsRLP sends a batch of transaction receipts, corresponding to the -// ones requested from an already RLP encoded format. -func (p *peer) SendReceiptsRLP(receipts []rlp.RawValue) error { - return p2p.Send(p.rw, ReceiptsMsg, receipts) -} - -// RequestOneHeader is a wrapper around the header query functions to fetch a -// single header. It is used solely by the fetcher. -func (p *peer) RequestOneHeader(hash common.Hash) error { - p.Log().Debug("Fetching single header", "hash", hash) - return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: hash}, Amount: uint64(1), Skip: uint64(0), Reverse: false}) -} - -// RequestHeadersByHash fetches a batch of blocks' headers corresponding to the -// specified header query, based on the hash of an origin block. -func (p *peer) RequestHeadersByHash(origin common.Hash, amount int, skip int, reverse bool) error { - p.Log().Debug("Fetching batch of headers", "count", amount, "fromhash", origin, "skip", skip, "reverse", reverse) - return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) -} - -// RequestHeadersByNumber fetches a batch of blocks' headers corresponding to the -// specified header query, based on the number of an origin block. -func (p *peer) RequestHeadersByNumber(origin uint64, amount int, skip int, reverse bool) error { - p.Log().Debug("Fetching batch of headers", "count", amount, "fromnum", origin, "skip", skip, "reverse", reverse) - return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) -} - -// RequestBodies fetches a batch of blocks' bodies corresponding to the hashes -// specified. -func (p *peer) RequestBodies(hashes []common.Hash) error { - p.Log().Debug("Fetching batch of block bodies", "count", len(hashes)) - return p2p.Send(p.rw, GetBlockBodiesMsg, hashes) -} - -// RequestNodeData fetches a batch of arbitrary data from a node's known state -// data, corresponding to the specified hashes. -func (p *peer) RequestNodeData(hashes []common.Hash) error { - p.Log().Debug("Fetching batch of state data", "count", len(hashes)) - return p2p.Send(p.rw, GetNodeDataMsg, hashes) -} - -// RequestReceipts fetches a batch of transaction receipts from a remote node. -func (p *peer) RequestReceipts(hashes []common.Hash) error { - p.Log().Debug("Fetching batch of receipts", "count", len(hashes)) - return p2p.Send(p.rw, GetReceiptsMsg, hashes) -} - -// Handshake executes the eth protocol handshake, negotiating version number, -// network IDs, difficulties, head and genesis blocks. -func (p *peer) Handshake(network uint64, td *big.Int, head common.Hash, genesis common.Hash) error { - // Send out own handshake in a new thread - errc := make(chan error, 2) - var status statusData // safe to read after two values have been received from errc - - go func() { - errc <- p2p.Send(p.rw, StatusMsg, &statusData{ - ProtocolVersion: uint32(p.version), - NetworkId: network, - TD: