From 0add0c400f0dda50c443352a255af704e71bdd3c Mon Sep 17 00:00:00 2001 From: zelig Date: Fri, 5 Dec 2014 21:14:55 +0000 Subject: initial commit for eth-p2p integration --- eth/protocol.go | 127 ++++++++++++++++++++++++++++++++++++++++++++++++++- eth/protocol_test.go | 39 ++++++++++++++++ 2 files changed, 165 insertions(+), 1 deletion(-) diff --git a/eth/protocol.go b/eth/protocol.go index 3b5b49696..985b42901 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -20,7 +20,7 @@ type ethProtocol struct { peer *p2p.Peer id string rw p2p.MsgReadWriter -} + } // backend is the interface the ethereum protocol backend should implement // used as an argument to EthProtocol @@ -68,6 +68,7 @@ type newBlockMsgData struct { type getBlockHashesMsgData struct { Hash []byte +<<<<<<< HEAD Amount uint64 } @@ -76,15 +77,29 @@ type getBlockHashesMsgData struct { // the Dev p2p layer then runs the protocol instance on each peer func EthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool) p2p.Protocol { return p2p.Protocol{ +======= + Amount uint32 +} + +// main entrypoint, wrappers starting a server running the eth protocol +// use this constructor to attach the protocol (class) to server caps +func EthProtocol(eth backend) *p2p.Protocol { + return &p2p.Protocol{ +>>>>>>> initial commit for eth-p2p integration Name: "eth", Version: ProtocolVersion, Length: ProtocolLength, Run: func(peer *p2p.Peer, rw p2p.MsgReadWriter) error { +<<<<<<< HEAD return runEthProtocol(txPool, chainManager, blockPool, peer, rw) +======= + return runEthProtocol(eth, peer, rw) +>>>>>>> initial commit for eth-p2p integration }, } } +<<<<<<< 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) { @@ -95,6 +110,13 @@ func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPoo rw: rw, peer: peer, id: (string)(peer.Identity().Pubkey()), +======= +func runEthProtocol(eth backend, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) { + self := ðProtocol{ + eth: eth, + rw: rw, + peer: peer, +>>>>>>> initial commit for eth-p2p integration } err = self.handleStatus() if err == nil { @@ -102,7 +124,10 @@ func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPoo for { err = self.handle() if err != nil { +<<<<<<< HEAD self.blockPool.RemovePeer(self.id) +======= +>>>>>>> initial commit for eth-p2p integration break } } @@ -127,24 +152,46 @@ func (self *ethProtocol) handle() error { case StatusMsg: return ProtocolError(ErrExtraStatusMsg, "") +<<<<<<< HEAD case TxMsg: // TODO: rework using lazy RLP stream +======= + case GetTxMsg: + txs := self.eth.GetTransactions() + // TODO: rewrite using rlp flat + txsInterface := make([]interface{}, len(txs)) + for i, tx := range txs { + txsInterface[i] = tx.RlpData() + } + return self.rw.EncodeMsg(TxMsg, txsInterface...) + + case TxMsg: +>>>>>>> initial commit for eth-p2p integration var txs []*types.Transaction if err := msg.Decode(&txs); err != nil { return ProtocolError(ErrDecode, "%v", err) } +<<<<<<< HEAD self.txPool.AddTransactions(txs) +======= + self.eth.AddTransactions(txs) +>>>>>>> initial commit for eth-p2p integration case GetBlockHashesMsg: var request getBlockHashesMsgData if err := msg.Decode(&request); err != nil { return ProtocolError(ErrDecode, "%v", err) } +<<<<<<< HEAD hashes := self.chainManager.GetBlockHashesFromHash(request.Hash, request.Amount) +======= + hashes := self.eth.GetBlockHashes(request.Hash, request.Amount) +>>>>>>> initial commit for eth-p2p integration return self.rw.EncodeMsg(BlockHashesMsg, ethutil.ByteSliceToInterface(hashes)...) case BlockHashesMsg: // TODO: redo using lazy decode , this way very inefficient on known chains +<<<<<<< HEAD msgStream := rlp.NewListStream(msg.Payload, uint64(msg.Size)) var err error iter := func() (hash []byte, ok bool) { @@ -160,17 +207,45 @@ func (self *ethProtocol) handle() error { } case GetBlocksMsg: +======= + // s := rlp.NewListStream(msg.Payload, uint64(msg.Size)) var blockHashes [][]byte if err := msg.Decode(&blockHashes); err != nil { return ProtocolError(ErrDecode, "%v", err) } + fetchMore := true + for _, hash := range blockHashes { + fetchMore = self.eth.AddHash(hash, self.peer) + if !fetchMore { + break + } + } + if fetchMore { + return self.FetchHashes(blockHashes[len(blockHashes)-1]) + } + + case GetBlocksMsg: + // Limit to max 300 blocks +>>>>>>> initial commit for eth-p2p integration + var blockHashes [][]byte + if err := msg.Decode(&blockHashes); err != nil { + return ProtocolError(ErrDecode, "%v", err) + } +<<<<<<< HEAD max := int(math.Min(float64(len(blockHashes)), blockHashesBatchSize)) +======= + max := int(math.Min(float64(len(blockHashes)), 300.0)) +>>>>>>> initial commit for eth-p2p integration var blocks []interface{} for i, hash := range blockHashes { if i >= max { break } +<<<<<<< HEAD block := self.chainManager.GetBlock(hash) +======= + block := self.eth.GetBlock(hash) +>>>>>>> initial commit for eth-p2p integration if block != nil { blocks = append(blocks, block.Value().Raw()) } @@ -178,6 +253,7 @@ func (self *ethProtocol) handle() error { return self.rw.EncodeMsg(BlocksMsg, blocks...) case BlocksMsg: +<<<<<<< HEAD msgStream := rlp.NewListStream(msg.Payload, uint64(msg.Size)) for { var block *types.Block @@ -189,6 +265,22 @@ func (self *ethProtocol) handle() error { } } self.blockPool.AddBlock(block, self.id) +======= + var blocks []*types.Block + if err := msg.Decode(&blocks); err != nil { + return ProtocolError(ErrDecode, "%v", err) + } + for _, block := range blocks { + fetchHashes, err := self.eth.AddBlock(nil, block, self.peer) + if err != nil { + return ProtocolError(ErrInvalidBlock, "%v", err) + } + if fetchHashes { + if err := self.FetchHashes(block.Hash()); err != nil { + return err + } + } +>>>>>>> initial commit for eth-p2p integration } case NewBlockMsg: @@ -196,6 +288,7 @@ func (self *ethProtocol) handle() error { if err := msg.Decode(&request); err != nil { return ProtocolError(ErrDecode, "%v", err) } +<<<<<<< HEAD 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 @@ -212,6 +305,15 @@ func (self *ethProtocol) handle() error { } self.blockPool.AddBlockHashes(iter, self.id) self.blockPool.AddBlock(request.Block, self.id) +======= + var fetchHashes bool + // this should reset td and offer blockpool as candidate new peer? + if fetchHashes, err = self.eth.AddBlock(request.TD, request.Block, self.peer); err != nil { + return ProtocolError(ErrInvalidBlock, "%v", err) + } + if fetchHashes { + return self.FetchHashes(request.Block.Hash()) +>>>>>>> initial commit for eth-p2p integration } default: @@ -229,7 +331,11 @@ type statusMsgData struct { } func (self *ethProtocol) statusMsg() p2p.Msg { +<<<<<<< HEAD td, currentBlock, genesisBlock := self.chainManager.Status() +======= + td, currentBlock, genesisBlock := self.eth.Status() +>>>>>>> initial commit for eth-p2p integration return p2p.NewMsg(StatusMsg, uint32(ProtocolVersion), @@ -265,7 +371,11 @@ func (self *ethProtocol) handleStatus() error { return ProtocolError(ErrDecode, "%v", err) } +<<<<<<< HEAD _, _, genesisBlock := self.chainManager.Status() +======= + _, _, genesisBlock := self.eth.Status() +>>>>>>> initial commit for eth-p2p integration if bytes.Compare(status.GenesisBlock, genesisBlock) != 0 { return ProtocolError(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, genesisBlock) @@ -279,13 +389,22 @@ func (self *ethProtocol) handleStatus() error { return ProtocolError(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, ProtocolVersion) } +<<<<<<< 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) +======= + logger.Infof("Peer is [eth] capable (%d/%d). TD = %v ~ %x", status.ProtocolVersion, status.NetworkId, status.CurrentBlock) + + if self.eth.AddPeer(status.TD, status.CurrentBlock, self.peer) { + return self.FetchHashes(status.CurrentBlock) + } +>>>>>>> initial commit for eth-p2p integration return nil } +<<<<<<< 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) @@ -316,3 +435,9 @@ func (self *ethProtocol) protoErrorDisconnect(code int, format string, params .. } } +======= +func (self *ethProtocol) FetchHashes(from []byte) error { + logger.Debugf("Fetching hashes (%d) %x...\n", blockHashesBatchSize, from[0:4]) + return self.rw.EncodeMsg(GetBlockHashesMsg, from, blockHashesBatchSize) +} +>>>>>>> initial commit for eth-p2p integration diff --git a/eth/protocol_test.go b/eth/protocol_test.go index a166ea6cd..93696213a 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -56,11 +56,18 @@ 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) } @@ -84,12 +91,21 @@ func (self *TestBackend) GetBlockHashes(hash []byte, amount uint32) (hashes [][] return } +<<<<<<< HEAD func (self *TestBackend) AddBlockHashes(next func() ([]byte, bool), peerId string) { if self.addBlockHashes != nil { self.addBlockHashes(next, peerId) } } +======= +func (self *TestBackend) AddHash(hash []byte, peer *p2p.Peer) (more bool) { + if self.addHash != nil { + more = self.addHash(hash, peer) + } + return +} +>>>>>>> initial commit for eth-p2p integration func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) { if self.getBlock != nil { block = self.getBlock(hash) @@ -97,26 +113,41 @@ func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) { return } +<<<<<<< HEAD func (self *TestBackend) AddBlock(block *types.Block, peerId string) (err error) { if self.addBlock != nil { err = self.addBlock(block, peerId) +======= +func (self *TestBackend) AddBlock(td *big.Int, block *types.Block, peer *p2p.Peer) (fetchHashes bool, err error) { + if self.addBlock != nil { + fetchHashes, err = self.addBlock(td, block, peer) +>>>>>>> initial commit for eth-p2p integration } return } +<<<<<<< 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) +======= +func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peer *p2p.Peer) (fetchHashes bool) { + if self.addPeer != nil { + fetchHashes = self.addPeer(td, currentBlock, peer) +>>>>>>> initial commit for eth-p2p integration } return } +<<<<<<< HEAD func (self *TestBackend) RemovePeer(peerId string) { if self.removePeer != nil { self.removePeer(peerId) } } +======= +>>>>>>> initial commit for eth-p2p integration func (self *TestBackend) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) { if self.status != nil { td, currentBlock, genesisBlock = self.status() @@ -124,6 +155,7 @@ func (self *TestBackend) Status() (td *big.Int, currentBlock []byte, genesisBloc return } +<<<<<<< HEAD // TODO: refactor this into p2p/client_identity type peerId struct { pubkey []byte @@ -147,12 +179,19 @@ func testPeer() *p2p.Peer { } func TestErrNoStatusMsg(t *testing.T) { +======= +func TestEth(t *testing.T) { +>>>>>>> initial commit for eth-p2p integration quit := make(chan bool) rw := &testMsgReadWriter{make(chan p2p.Msg, 10), make(chan p2p.Msg, 10)} testBackend := &TestBackend{} var err error go func() { +<<<<<<< HEAD err = runEthProtocol(testBackend, testPeer(), rw) +======= + err = runEthProtocol(testBackend, nil, rw) +>>>>>>> initial commit for eth-p2p integration close(quit) }() statusMsg := p2p.NewMsg(4) -- cgit v1.2.3