diff options
| -rw-r--r-- | eth/protocol.go | 97 | ||||
| -rw-r--r-- | eth/protocol_test.go | 46 |
2 files changed, 132 insertions, 11 deletions
diff --git a/eth/protocol.go b/eth/protocol.go index 985b42901..fbc4610ec 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -82,7 +82,8 @@ func EthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool) } // main entrypoint, wrappers starting a server running the eth protocol -// use this constructor to attach the protocol (class) to server caps +// use this constructor to attach the protocol ("class") to server caps +// the Dev p2p layer then runs the protocol instance on each peer func EthProtocol(eth backend) *p2p.Protocol { return &p2p.Protocol{ >>>>>>> initial commit for eth-p2p integration @@ -100,6 +101,7 @@ func EthProtocol(eth backend) *p2p.Protocol { } <<<<<<< HEAD +<<<<<<< HEAD // the main loop that handles incoming messages // note RemovePeer in the post-disconnect hook func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) { @@ -111,12 +113,20 @@ func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPoo peer: peer, id: (string)(peer.Identity().Pubkey()), ======= +======= +// the main loop that handles incoming messages +// note RemovePeer in the post-disconnect hook +>>>>>>> eth protocol changes func runEthProtocol(eth backend, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) { self := ðProtocol{ eth: eth, rw: rw, peer: peer, +<<<<<<< HEAD >>>>>>> initial commit for eth-p2p integration +======= + id: (string)(peer.Identity().Pubkey()), +>>>>>>> eth protocol changes } err = self.handleStatus() if err == nil { @@ -125,9 +135,13 @@ func runEthProtocol(eth backend, peer *p2p.Peer, rw p2p.MsgReadWriter) (err erro err = self.handle() if err != nil { <<<<<<< HEAD +<<<<<<< HEAD self.blockPool.RemovePeer(self.id) ======= >>>>>>> initial commit for eth-p2p integration +======= + self.eth.RemovePeer(self.id) +>>>>>>> eth protocol changes break } } @@ -166,7 +180,11 @@ func (self *ethProtocol) handle() error { return self.rw.EncodeMsg(TxMsg, txsInterface...) case TxMsg: +<<<<<<< HEAD >>>>>>> initial commit for eth-p2p integration +======= + // TODO: rework using lazy RLP stream +>>>>>>> eth protocol changes var txs []*types.Transaction if err := msg.Decode(&txs); err != nil { return ProtocolError(ErrDecode, "%v", err) @@ -192,12 +210,16 @@ func (self *ethProtocol) handle() error { case BlockHashesMsg: // TODO: redo using lazy decode , this way very inefficient on known chains <<<<<<< HEAD +<<<<<<< HEAD +======= +>>>>>>> eth protocol changes msgStream := rlp.NewListStream(msg.Payload, uint64(msg.Size)) var err error iter := func() (hash []byte, ok bool) { hash, err = msgStream.Bytes() if err == nil { ok = true +<<<<<<< HEAD } return } @@ -218,24 +240,35 @@ func (self *ethProtocol) handle() error { fetchMore = self.eth.AddHash(hash, self.peer) if !fetchMore { break +======= +>>>>>>> eth protocol changes } + return } - if fetchMore { - return self.FetchHashes(blockHashes[len(blockHashes)-1]) + self.eth.AddBlockHashes(iter, self.id) + if err != nil && err != rlp.EOL { + return ProtocolError(ErrDecode, "%v", err) } case GetBlocksMsg: +<<<<<<< HEAD // Limit to max 300 blocks >>>>>>> initial commit for eth-p2p integration +======= +>>>>>>> eth protocol changes var blockHashes [][]byte if err := msg.Decode(&blockHashes); err != nil { return ProtocolError(ErrDecode, "%v", err) } <<<<<<< HEAD +<<<<<<< HEAD max := int(math.Min(float64(len(blockHashes)), blockHashesBatchSize)) ======= max := int(math.Min(float64(len(blockHashes)), 300.0)) >>>>>>> initial commit for eth-p2p integration +======= + max := int(math.Min(float64(len(blockHashes)), blockHashesBatchSize)) +>>>>>>> eth protocol changes var blocks []interface{} for i, hash := range blockHashes { if i >= max { @@ -254,6 +287,9 @@ func (self *ethProtocol) handle() error { case BlocksMsg: <<<<<<< HEAD +<<<<<<< HEAD +======= +>>>>>>> eth protocol changes msgStream := rlp.NewListStream(msg.Payload, uint64(msg.Size)) for { var block *types.Block @@ -262,6 +298,7 @@ func (self *ethProtocol) handle() error { break } else { return ProtocolError(ErrDecode, "%v", err) +<<<<<<< HEAD } } self.blockPool.AddBlock(block, self.id) @@ -281,6 +318,13 @@ func (self *ethProtocol) handle() error { } } >>>>>>> initial commit for eth-p2p integration +======= + } + } + if err := self.eth.AddBlock(block, self.id); err != nil { + return ProtocolError(ErrInvalidBlock, "%v", err) + } +>>>>>>> eth protocol changes } case NewBlockMsg: @@ -289,11 +333,18 @@ func (self *ethProtocol) handle() error { return ProtocolError(ErrDecode, "%v", err) } <<<<<<< HEAD +<<<<<<< HEAD +======= +>>>>>>> eth protocol changes hash := request.Block.Hash() // to simplify backend interface adding a new block // uses AddPeer followed by AddHashes, AddBlock only if peer is the best peer // (or selected as new best peer) +<<<<<<< HEAD if self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) { +======= + if self.eth.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.invalidBlock) { +>>>>>>> eth protocol changes called := true iter := func() (hash []byte, ok bool) { if called { @@ -303,6 +354,7 @@ func (self *ethProtocol) handle() error { return } } +<<<<<<< HEAD self.blockPool.AddBlockHashes(iter, self.id) self.blockPool.AddBlock(request.Block, self.id) ======= @@ -314,6 +366,12 @@ func (self *ethProtocol) handle() error { if fetchHashes { return self.FetchHashes(request.Block.Hash()) >>>>>>> initial commit for eth-p2p integration +======= + self.eth.AddBlockHashes(iter, self.id) + if err := self.eth.AddBlock(request.Block, self.id); err != nil { + return ProtocolError(ErrInvalidBlock, "%v", err) + } +>>>>>>> eth protocol changes } default: @@ -390,6 +448,7 @@ func (self *ethProtocol) handleStatus() error { } <<<<<<< HEAD +<<<<<<< HEAD self.peer.Infof("Peer is [eth] capable (%d/%d). TD = %v ~ %x", status.ProtocolVersion, status.NetworkId, status.CurrentBlock) self.blockPool.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) @@ -400,11 +459,17 @@ func (self *ethProtocol) handleStatus() error { return self.FetchHashes(status.CurrentBlock) } >>>>>>> initial commit for eth-p2p integration +======= + self.peer.Infof("Peer is [eth] capable (%d/%d). TD = %v ~ %x", status.ProtocolVersion, status.NetworkId, status.CurrentBlock) + + self.eth.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.invalidBlock) +>>>>>>> eth protocol changes return nil } <<<<<<< HEAD +<<<<<<< HEAD func (self *ethProtocol) requestBlockHashes(from []byte) error { self.peer.Debugf("fetching hashes (%d) %x...\n", blockHashesBatchSize, from[0:4]) return self.rw.EncodeMsg(GetBlockHashesMsg, from, blockHashesBatchSize) @@ -441,3 +506,29 @@ func (self *ethProtocol) FetchHashes(from []byte) error { return self.rw.EncodeMsg(GetBlockHashesMsg, from, blockHashesBatchSize) } >>>>>>> initial commit for eth-p2p integration +======= +func (self *ethProtocol) requestBlockHashes(from []byte) error { + self.peer.Debugf("fetching hashes (%d) %x...\n", blockHashesBatchSize, from[0:4]) + return self.rw.EncodeMsg(GetBlockHashesMsg, from, blockHashesBatchSize) +} + +func (self *ethProtocol) requestBlocks(hashes [][]byte) error { + self.peer.Debugf("fetching %v blocks", len(hashes)) + return self.rw.EncodeMsg(GetBlocksMsg, ethutil.ByteSliceToInterface(hashes)) +} + +func (self *ethProtocol) invalidBlock(err error) { + ProtocolError(ErrInvalidBlock, "%v", err) + self.peer.Disconnect(p2p.DiscSubprotocolError) +} + +func (self *ethProtocol) protoError(code int, format string, params ...interface{}) (err *protocolError) { + err = ProtocolError(code, format, params...) + if err.Fatal() { + self.peer.Errorln(err) + } else { + self.peer.Debugln(err) + } + return +} +>>>>>>> eth protocol changes diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 93696213a..322aec7b7 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -56,18 +56,11 @@ type TestBackend struct { getTransactions func() []*types.Transaction addTransactions func(txs []*types.Transaction) getBlockHashes func(hash []byte, amount uint32) (hashes [][]byte) -<<<<<<< HEAD addBlockHashes func(next func() ([]byte, bool), peerId string) getBlock func(hash []byte) *types.Block addBlock func(block *types.Block, peerId string) (err error) addPeer func(td *big.Int, currentBlock []byte, peerId string, requestHashes func([]byte) error, requestBlocks func([][]byte) error, invalidBlock func(error)) (best bool) removePeer func(peerId string) -======= - addHash func(hash []byte, peer *p2p.Peer) (more bool) - getBlock func(hash []byte) *types.Block - addBlock func(td *big.Int, block *types.Block, peer *p2p.Peer) (fetchHashes bool, err error) - addPeer func(td *big.Int, currentBlock []byte, peer *p2p.Peer) (fetchHashes bool) ->>>>>>> initial commit for eth-p2p integration status func() (td *big.Int, currentBlock []byte, genesisBlock []byte) } @@ -92,6 +85,7 @@ func (self *TestBackend) GetBlockHashes(hash []byte, amount uint32) (hashes [][] } <<<<<<< HEAD +<<<<<<< HEAD func (self *TestBackend) AddBlockHashes(next func() ([]byte, bool), peerId string) { if self.addBlockHashes != nil { self.addBlockHashes(next, peerId) @@ -102,10 +96,18 @@ func (self *TestBackend) AddBlockHashes(next func() ([]byte, bool), peerId strin func (self *TestBackend) AddHash(hash []byte, peer *p2p.Peer) (more bool) { if self.addHash != nil { more = self.addHash(hash, peer) +======= +func (self *TestBackend) AddBlockHashes(next func() ([]byte, bool), peerId string) { + if self.addBlockHashes != nil { + self.addBlockHashes(next, peerId) +>>>>>>> eth protocol changes } - return } +<<<<<<< HEAD >>>>>>> initial commit for eth-p2p integration +======= + +>>>>>>> eth protocol changes func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) { if self.getBlock != nil { block = self.getBlock(hash) @@ -114,6 +116,7 @@ func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) { } <<<<<<< HEAD +<<<<<<< HEAD func (self *TestBackend) AddBlock(block *types.Block, peerId string) (err error) { if self.addBlock != nil { err = self.addBlock(block, peerId) @@ -122,11 +125,17 @@ func (self *TestBackend) AddBlock(td *big.Int, block *types.Block, peer *p2p.Pee if self.addBlock != nil { fetchHashes, err = self.addBlock(td, block, peer) >>>>>>> initial commit for eth-p2p integration +======= +func (self *TestBackend) AddBlock(block *types.Block, peerId string) (err error) { + if self.addBlock != nil { + err = self.addBlock(block, peerId) +>>>>>>> eth protocol changes } return } <<<<<<< HEAD +<<<<<<< HEAD func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, invalidBlock func(error)) (best bool) { if self.addPeer != nil { best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, invalidBlock) @@ -135,19 +144,30 @@ func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peer *p2p.Pee if self.addPeer != nil { fetchHashes = self.addPeer(td, currentBlock, peer) >>>>>>> initial commit for eth-p2p integration +======= +func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, invalidBlock func(error)) (best bool) { + if self.addPeer != nil { + best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, invalidBlock) +>>>>>>> eth protocol changes } return } <<<<<<< HEAD +<<<<<<< HEAD +======= +>>>>>>> eth protocol changes func (self *TestBackend) RemovePeer(peerId string) { if self.removePeer != nil { self.removePeer(peerId) } } +<<<<<<< HEAD ======= >>>>>>> initial commit for eth-p2p integration +======= +>>>>>>> eth protocol changes func (self *TestBackend) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) { if self.status != nil { td, currentBlock, genesisBlock = self.status() @@ -156,6 +176,9 @@ func (self *TestBackend) Status() (td *big.Int, currentBlock []byte, genesisBloc } <<<<<<< HEAD +<<<<<<< HEAD +======= +>>>>>>> eth protocol changes // TODO: refactor this into p2p/client_identity type peerId struct { pubkey []byte @@ -179,19 +202,26 @@ func testPeer() *p2p.Peer { } func TestErrNoStatusMsg(t *testing.T) { +<<<<<<< HEAD ======= func TestEth(t *testing.T) { >>>>>>> initial commit for eth-p2p integration +======= +>>>>>>> eth protocol changes quit := make(chan bool) rw := &testMsgReadWriter{make(chan p2p.Msg, 10), make(chan p2p.Msg, 10)} testBackend := &TestBackend{} var err error go func() { <<<<<<< HEAD +<<<<<<< HEAD err = runEthProtocol(testBackend, testPeer(), rw) ======= err = runEthProtocol(testBackend, nil, rw) >>>>>>> initial commit for eth-p2p integration +======= + err = runEthProtocol(testBackend, testPeer(), rw) +>>>>>>> eth protocol changes close(quit) }() statusMsg := p2p.NewMsg(4) |