diff --git a/eth/backend.go b/eth/backend.go index a3275cacab..052bfad0f9 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -47,6 +47,7 @@ import ( "github.com/ethereum/go-ethereum/miner" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/enr" "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rpc" @@ -66,7 +67,9 @@ type Ethereum struct { config *Config // Channel for shutting down the service - shutdownChan chan bool // Channel for shutting down the Ethereum + shutdownChan chan bool + + server *p2p.Server // Handlers txPool *core.TxPool @@ -496,7 +499,7 @@ func (s *Ethereum) EventMux() *event.TypeMux { return s.eventMux } func (s *Ethereum) Engine() consensus.Engine { return s.engine } 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(s.protocolManager.SubProtocols[0].Version) } +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 } @@ -505,15 +508,22 @@ func (s *Ethereum) ArchiveMode() bool { return s.config.NoPruni // Protocols implements node.Service, returning all the currently configured // network protocols to start. func (s *Ethereum) Protocols() []p2p.Protocol { - if s.lesServer == nil { - return s.protocolManager.SubProtocols + protos := make([]p2p.Protocol, len(ProtocolVersions)) + for i, vsn := range ProtocolVersions { + protos[i] = s.protocolManager.makeProtocol(vsn) + protos[i].Attributes = []enr.Entry{s.currentEthEntry()} } - return append(s.protocolManager.SubProtocols, s.lesServer.Protocols()...) + if s.lesServer != nil { + protos = append(protos, s.lesServer.Protocols()...) + } + return protos } // Start implements node.Service, starting all internal goroutines needed by the // Ethereum protocol implementation. func (s *Ethereum) Start(srvr *p2p.Server) error { + go s.enrUpdateLoop(srvr.LocalNode()) + // Start the bloom bits servicing goroutines s.startBloomHandlers(params.BloomBitsBlocks) diff --git a/eth/enr.go b/eth/enr_entry.go similarity index 50% rename from eth/enr.go rename to eth/enr_entry.go index e6d119bffc..3798440da4 100644 --- a/eth/enr.go +++ b/eth/enr_entry.go @@ -19,45 +19,41 @@ package eth import ( "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/forkid" - "github.com/ethereum/go-ethereum/p2p/enr" + "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/rlp" ) -// ENR is the "eth" Ethereum Node Record, advertising various information useful -// for maintaining the Ethereum computer network. -// -// Whilst only one thing is contained here for now, the ENR is a struct to permit -// future extensibility. -type ENR struct { +// ethEntry is the "eth" ENR entry which advertises eth protocol +// on the discovery network. +type ethEntry struct { ForkID forkid.ID // Fork identifier per EIP-2124 // Ignore additional fields (for forward compatibility). Rest []rlp.RawValue `rlp:"tail"` } -// NewENR calculates the Ethereum network ENR from various information available -// from the chain. -func NewENR(chain *core.BlockChain) ENR { - return ENR{ - ForkID: forkid.NewID(chain), - } +// ENRKey implements enr.Entry. +func (e ethEntry) ENRKey() string { + return "eth" } -// ENRKey implements enr.Entry, returning the key for the chain config. -func (e ENR) ENRKey() string { return "eth" } +func (eth *Ethereum) enrUpdateLoop(ln *enode.LocalNode) { + var newHead = make(chan core.ChainHeadEvent, 10) + sub := eth.blockchain.SubscribeChainHeadEvent(newHead) + defer sub.Unsubscribe() -// NewENRFilter creates an ENR filter that returns if a record should be rejected -// or not (may be rejected by another filter). -func NewENRFilter(chain *core.BlockChain) func(r *enr.Record) error { - filter := forkid.NewFilter(chain) - - return func(r *enr.Record) error { - // Retrieve the remote chain ENR entry, accept record if not found - var entry ENR - if err := r.Load(&entry); err != nil { - return nil + for { + select { + case <-newHead: + ln.Set(eth.currentEthEntry()) + case <-sub.Err(): + // Would be nice to sync with eth.Stop, but there is no + // good way to do that. + return } - // If found, run it across the fork ID validator - return filter(entry.ForkID) } } + +func (eth *Ethereum) currentEthEntry() *ethEntry { + return ðEntry{ForkID: forkid.NewID(eth.blockchain)} +} diff --git a/eth/handler.go b/eth/handler.go index 51d878dbaa..fb82c16fa7 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -37,7 +37,6 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/enode" - "github.com/ethereum/go-ethereum/p2p/enr" "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/trie" @@ -84,8 +83,6 @@ type ProtocolManager struct { fetcher *fetcher.Fetcher peers *peerSet - SubProtocols []p2p.Protocol - eventMux *event.TypeMux txsCh chan core.NewTxsEvent txsSub event.Subscription @@ -148,46 +145,7 @@ func NewProtocolManager(config *params.ChainConfig, checkpoint *params.TrustedCh manager.checkpointNumber = (checkpoint.SectionIndex+1)*params.CHTFrequency - 1 manager.checkpointHash = checkpoint.SectionHead } - // Initiate a sub-protocol for every implemented version we can handle - manager.SubProtocols = make([]p2p.Protocol, 0, len(ProtocolVersions)) - for i, version := range ProtocolVersions { - // Skip protocol version if incompatible with the mode of operation - // TODO(karalabe): hard-drop eth/62 from the code base - if atomic.LoadUint32(&manager.fastSync) == 1 && version < eth63 { - continue - } - // Compatible; initialise the sub-protocol - version := version // Closure for the run - manager.SubProtocols = append(manager.SubProtocols, p2p.Protocol{ - Name: ProtocolName, - Version: version, - Length: ProtocolLengths[i], - Attributes: []enr.Entry{NewENR(blockchain)}, - Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - peer := manager.newPeer(int(version), p, rw) - select { - case manager.newPeerCh <- peer: - manager.wg.Add(1) - defer manager.wg.Done() - return manager.handle(peer) - case <-manager.quitSync: - return p2p.DiscQuitting - } - }, - NodeInfo: func() interface{} { - return manager.NodeInfo() - }, - PeerInfo: func(id enode.ID) interface{} { - if p := manager.peers.Peer(fmt.Sprintf("%x", id[:8])); p != nil { - return p.Info() - } - return nil - }, - }) - } - if len(manager.SubProtocols) == 0 { - return nil, errIncompatibleConfig - } + // 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. @@ -235,6 +193,39 @@ func NewProtocolManager(config *params.ChainConfig, checkpoint *params.TrustedCh 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) @@ -381,8 +372,8 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err != nil { return err } - if msg.Size > ProtocolMaxMsgSize { - return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) + if msg.Size > protocolMaxMsgSize { + return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, protocolMaxMsgSize) } defer msg.Discard() diff --git a/eth/handler_test.go b/eth/handler_test.go index 445d52afca..0f1672fd44 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -38,35 +38,6 @@ import ( "github.com/ethereum/go-ethereum/params" ) -// Tests that protocol versions and modes of operations are matched up properly. -func TestProtocolCompatibility(t *testing.T) { - // Define the compatibility chart - tests := []struct { - version uint - mode downloader.SyncMode - compatible bool - }{ - {61, downloader.FullSync, true}, {62, downloader.FullSync, true}, {63, downloader.FullSync, true}, - {61, downloader.FastSync, false}, {62, downloader.FastSync, false}, {63, downloader.FastSync, true}, - } - // Make sure anything we screw up is restored - backup := ProtocolVersions - defer func() { ProtocolVersions = backup }() - - // Try all available compatibility configs and check for errors - for i, tt := range tests { - ProtocolVersions = []uint{tt.version} - - pm, _, err := newTestProtocolManager(tt.mode, 0, nil, nil) - if pm != nil { - defer pm.Stop() - } - if (err == nil && !tt.compatible) || (err != nil && tt.compatible) { - t.Errorf("test %d: compatibility mismatch: have error %v, want compatibility %v", i, err, tt.compatible) - } - } -} - // Tests that block headers can be retrieved from a remote chain based on user queries. func TestGetBlockHeaders62(t *testing.T) { testGetBlockHeaders(t, 62) } func TestGetBlockHeaders63(t *testing.T) { testGetBlockHeaders(t, 63) } diff --git a/eth/peer.go b/eth/peer.go index 208badc5ed..814c787b8c 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -394,8 +394,8 @@ func (p *peer) readStatus(network uint64, status *statusData, genesis common.Has if msg.Code != StatusMsg { return errResp(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg) } - if msg.Size > ProtocolMaxMsgSize { - return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) + if msg.Size > protocolMaxMsgSize { + return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, protocolMaxMsgSize) } // Decode the handshake and make sure everything matches if err := msg.Decode(&status); err != nil { diff --git a/eth/protocol.go b/eth/protocol.go index 497ba4c597..8418b22116 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -34,16 +34,16 @@ const ( eth63 = 63 ) -// ProtocolName is the official short name of the protocol used during capability negotiation. -var ProtocolName = "eth" +// protocolName is the official short name of the protocol used during capability negotiation. +const protocolName = "eth" // ProtocolVersions are the supported versions of the eth protocol (first is primary). -var ProtocolVersions = []uint{eth63, eth62} +var ProtocolVersions = []uint{eth63} -// ProtocolLengths are the number of implemented message corresponding to different protocol versions. -var ProtocolLengths = []uint64{17, 8} +// protocolLengths are the number of implemented message corresponding to different protocol versions. +var protocolLengths = map[uint]uint64{eth63: 17, eth62: 8} -const ProtocolMaxMsgSize = 10 * 1024 * 1024 // Maximum cap on the size of a protocol message +const protocolMaxMsgSize = 10 * 1024 * 1024 // Maximum cap on the size of a protocol message // eth protocol message codes const ( diff --git a/p2p/server.go b/p2p/server.go index b7340a5ea7..e04e7884f8 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -286,6 +286,11 @@ func (c *conn) set(f connFlag, val bool) { } } +// LocalNode returns the local node record. +func (srv *Server) LocalNode() *enode.LocalNode { + return srv.localnode +} + // Peers returns all connected peers. func (srv *Server) Peers() []*Peer { var ps []*Peer