From 47793b606c33e93d9a4295ba4db12165182d9e72 Mon Sep 17 00:00:00 2001 From: zelig Date: Fri, 5 Dec 2014 21:14:55 +0000 Subject: [PATCH 1/4] initial commit for eth-p2p integration --- eth/error.go | 73 +++++++++++ eth/protocol.go | 294 +++++++++++++++++++++++++++++++++++++++++++ eth/protocol_test.go | 133 ++++++++++++++++++++ 3 files changed, 500 insertions(+) create mode 100644 eth/error.go create mode 100644 eth/protocol.go create mode 100644 eth/protocol_test.go diff --git a/eth/error.go b/eth/error.go new file mode 100644 index 000000000..cb4459435 --- /dev/null +++ b/eth/error.go @@ -0,0 +1,73 @@ +package eth + +import ( + "fmt" + // "github.com/ethereum/go-ethereum/logger" +) + +const ( + ErrMsgTooLarge = iota + ErrDecode + ErrInvalidMsgCode + ErrProtocolVersionMismatch + ErrNetworkIdMismatch + ErrGenesisBlockMismatch + ErrNoStatusMsg + ErrExtraStatusMsg + ErrInvalidBlock +) + +var errorToString = map[int]string{ + ErrMsgTooLarge: "Message too long", + ErrDecode: "Invalid message", + ErrInvalidMsgCode: "Invalid message code", + ErrProtocolVersionMismatch: "Protocol version mismatch", + ErrNetworkIdMismatch: "NetworkId mismatch", + ErrGenesisBlockMismatch: "Genesis block mismatch", + ErrNoStatusMsg: "No status message", + ErrExtraStatusMsg: "Extra status message", + ErrInvalidBlock: "Invalid block", +} + +type protocolError struct { + Code int + fatal bool + message string + format string + params []interface{} + // size int +} + +func newProtocolError(code int, format string, params ...interface{}) *protocolError { + return &protocolError{Code: code, format: format, params: params} +} + +func ProtocolError(code int, format string, params ...interface{}) (err *protocolError) { + err = newProtocolError(code, format, params...) + // report(err) + if err.Fatal() { + logger.Errorln(err) + } else { + logger.Debugln(err) + } + return +} + +func (self protocolError) Error() (message string) { + message = self.message + if message == "" { + message, ok := errorToString[self.Code] + if !ok { + panic("invalid error code") + } + if self.format != "" { + message += ": " + fmt.Sprintf(self.format, self.params...) + } + self.message = message + } + return +} + +func (self *protocolError) Fatal() bool { + return self.fatal +} diff --git a/eth/protocol.go b/eth/protocol.go new file mode 100644 index 000000000..992fc7550 --- /dev/null +++ b/eth/protocol.go @@ -0,0 +1,294 @@ +package eth + +import ( + "bytes" + "math" + "math/big" + + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethutil" + ethlogger "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/p2p" +) + +var logger = ethlogger.NewLogger("SERV") + +// ethProtocol represents the ethereum wire protocol +// instance is running on each peer +type ethProtocol struct { + eth backend + td *big.Int + peer *p2p.Peer + rw p2p.MsgReadWriter +} + +// backend is the interface the ethereum protocol backend should implement +// used as an argument to EthProtocol +type backend interface { + GetTransactions() (txs []*types.Transaction) + AddTransactions(txs []*types.Transaction) + GetBlockHashes(hash []byte, amount uint32) (hashes [][]byte) + AddHash(hash []byte, peer *p2p.Peer) (more bool) + GetBlock(hash []byte) (block *types.Block) + AddBlock(td *big.Int, block *types.Block, peer *p2p.Peer) (fetchHashes bool, err error) + AddPeer(td *big.Int, currentBlock []byte, peer *p2p.Peer) (fetchHashes bool) + Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) +} + +const ( + ProtocolVersion = 43 + // 0x00 // PoC-1 + // 0x01 // PoC-2 + // 0x07 // PoC-3 + // 0x09 // PoC-4 + // 0x17 // PoC-5 + // 0x1c // PoC-6 + NetworkId = 0 + ProtocolLength = uint64(8) + ProtocolMaxMsgSize = 10 * 1024 * 1024 + + blockHashesBatchSize = 256 +) + +// eth protocol message codes +const ( + StatusMsg = iota + GetTxMsg // unused + TxMsg + GetBlockHashesMsg + BlockHashesMsg + GetBlocksMsg + BlocksMsg + NewBlockMsg +) + +// message structs used for rlp decoding +type newBlockMsgData struct { + Block *types.Block + TD *big.Int +} + +type getBlockHashesMsgData struct { + Hash []byte + 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{ + Name: "eth", + Version: ProtocolVersion, + Length: ProtocolLength, + Run: func(peer *p2p.Peer, rw p2p.MsgReadWriter) error { + return runEthProtocol(eth, peer, rw) + }, + } +} + +func runEthProtocol(eth backend, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) { + self := ðProtocol{ + eth: eth, + rw: rw, + peer: peer, + } + err = self.handleStatus() + if err == nil { + go func() { + for { + err = self.handle() + if err != nil { + break + } + } + }() + } + return +} + +func (self *ethProtocol) handle() error { + msg, err := self.rw.ReadMsg() + if err != nil { + return err + } + if msg.Size > ProtocolMaxMsgSize { + return ProtocolError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) + } + // make sure that the payload has been fully consumed + defer msg.Discard() + + switch msg.Code { + + case StatusMsg: + return ProtocolError(ErrExtraStatusMsg, "") + + 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: + var txs []*types.Transaction + if err := msg.Decode(&txs); err != nil { + return ProtocolError(ErrDecode, "%v", err) + } + self.eth.AddTransactions(txs) + + case GetBlockHashesMsg: + var request getBlockHashesMsgData + if err := msg.Decode(&request); err != nil { + return ProtocolError(ErrDecode, "%v", err) + } + hashes := self.eth.GetBlockHashes(request.Hash, request.Amount) + return self.rw.EncodeMsg(BlockHashesMsg, ethutil.ByteSliceToInterface(hashes)...) + + case BlockHashesMsg: + // TODO: redo using lazy decode , this way very inefficient on known chains + // 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 + var blockHashes [][]byte + if err := msg.Decode(&blockHashes); err != nil { + return ProtocolError(ErrDecode, "%v", err) + } + max := int(math.Min(float64(len(blockHashes)), 300.0)) + var blocks []interface{} + for i, hash := range blockHashes { + if i >= max { + break + } + block := self.eth.GetBlock(hash) + if block != nil { + blocks = append(blocks, block.Value().Raw()) + } + } + return self.rw.EncodeMsg(BlocksMsg, blocks...) + + case BlocksMsg: + 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 + } + } + } + + case NewBlockMsg: + var request newBlockMsgData + if err := msg.Decode(&request); err != nil { + return ProtocolError(ErrDecode, "%v", err) + } + 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()) + } + + default: + return ProtocolError(ErrInvalidMsgCode, "%v", msg.Code) + } + return nil +} + +type statusMsgData struct { + ProtocolVersion uint + NetworkId uint + TD *big.Int + CurrentBlock []byte + GenesisBlock []byte +} + +func (self *ethProtocol) statusMsg() p2p.Msg { + td, currentBlock, genesisBlock := self.eth.Status() + + return p2p.NewMsg(StatusMsg, + uint32(ProtocolVersion), + uint32(NetworkId), + td, + currentBlock, + genesisBlock, + ) +} + +func (self *ethProtocol) handleStatus() error { + // send precanned status message + if err := self.rw.WriteMsg(self.statusMsg()); err != nil { + return err + } + + // read and handle remote status + msg, err := self.rw.ReadMsg() + if err != nil { + return err + } + + if msg.Code != StatusMsg { + return ProtocolError(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg) + } + + if msg.Size > ProtocolMaxMsgSize { + return ProtocolError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) + } + + var status statusMsgData + if err := msg.Decode(&status); err != nil { + return ProtocolError(ErrDecode, "%v", err) + } + + _, _, genesisBlock := self.eth.Status() + + if bytes.Compare(status.GenesisBlock, genesisBlock) != 0 { + return ProtocolError(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, genesisBlock) + } + + if status.NetworkId != NetworkId { + return ProtocolError(ErrNetworkIdMismatch, "%d (!= %d)", status.NetworkId, NetworkId) + } + + if ProtocolVersion != status.ProtocolVersion { + return ProtocolError(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, ProtocolVersion) + } + + 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) + } + + return nil +} + +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) +} diff --git a/eth/protocol_test.go b/eth/protocol_test.go new file mode 100644 index 000000000..757e87fa4 --- /dev/null +++ b/eth/protocol_test.go @@ -0,0 +1,133 @@ +package eth + +import ( + "io" + "math/big" + "testing" + + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/p2p" +) + +type testMsgReadWriter struct { + in chan p2p.Msg + out chan p2p.Msg +} + +func (self *testMsgReadWriter) In(msg p2p.Msg) { + self.in <- msg +} + +func (self *testMsgReadWriter) Out(msg p2p.Msg) { + self.in <- msg +} + +func (self *testMsgReadWriter) WriteMsg(msg p2p.Msg) error { + self.out <- msg + return nil +} + +func (self *testMsgReadWriter) EncodeMsg(code uint64, data ...interface{}) error { + return self.WriteMsg(p2p.NewMsg(code, data)) +} + +func (self *testMsgReadWriter) ReadMsg() (p2p.Msg, error) { + msg, ok := <-self.in + if !ok { + return msg, io.EOF + } + return msg, nil +} + +func errorCheck(t *testing.T, expCode int, err error) { + perr, ok := err.(*protocolError) + if ok && perr != nil { + if code := perr.Code; code != expCode { + ok = false + } + } + if !ok { + t.Errorf("expected error code %v, got %v", ErrNoStatusMsg, err) + } +} + +type TestBackend struct { + getTransactions func() []*types.Transaction + addTransactions func(txs []*types.Transaction) + getBlockHashes func(hash []byte, amount uint32) (hashes [][]byte) + 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) + status func() (td *big.Int, currentBlock []byte, genesisBlock []byte) +} + +func (self *TestBackend) GetTransactions() (txs []*types.Transaction) { + if self.getTransactions != nil { + txs = self.getTransactions() + } + return +} + +func (self *TestBackend) AddTransactions(txs []*types.Transaction) { + if self.addTransactions != nil { + self.addTransactions(txs) + } +} + +func (self *TestBackend) GetBlockHashes(hash []byte, amount uint32) (hashes [][]byte) { + if self.getBlockHashes != nil { + hashes = self.getBlockHashes(hash, amount) + } + return +} + +func (self *TestBackend) AddHash(hash []byte, peer *p2p.Peer) (more bool) { + if self.addHash != nil { + more = self.addHash(hash, peer) + } + return +} +func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) { + if self.getBlock != nil { + block = self.getBlock(hash) + } + return +} + +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) + } + return +} + +func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peer *p2p.Peer) (fetchHashes bool) { + if self.addPeer != nil { + fetchHashes = self.addPeer(td, currentBlock, peer) + } + return +} + +func (self *TestBackend) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) { + if self.status != nil { + td, currentBlock, genesisBlock = self.status() + } + return +} + +func TestEth(t *testing.T) { + quit := make(chan bool) + rw := &testMsgReadWriter{make(chan p2p.Msg, 10), make(chan p2p.Msg, 10)} + testBackend := &TestBackend{} + var err error + go func() { + err = runEthProtocol(testBackend, nil, rw) + close(quit) + }() + statusMsg := p2p.NewMsg(4) + rw.In(statusMsg) + <-quit + errorCheck(t, ErrNoStatusMsg, err) + // read(t, remote, []byte("hello, world"), nil) +} From 31a9fdced84cdefd17f98f1c6a018cb1ab67ad8a Mon Sep 17 00:00:00 2001 From: zelig Date: Tue, 9 Dec 2014 23:54:02 +0000 Subject: [PATCH 2/4] no logging in error (to be refactored into p2p) --- eth/error.go | 6 ------ 1 file changed, 6 deletions(-) diff --git a/eth/error.go b/eth/error.go index cb4459435..9355d6457 100644 --- a/eth/error.go +++ b/eth/error.go @@ -2,7 +2,6 @@ package eth import ( "fmt" - // "github.com/ethereum/go-ethereum/logger" ) const ( @@ -45,11 +44,6 @@ func newProtocolError(code int, format string, params ...interface{}) *protocolE func ProtocolError(code int, format string, params ...interface{}) (err *protocolError) { err = newProtocolError(code, format, params...) // report(err) - if err.Fatal() { - logger.Errorln(err) - } else { - logger.Debugln(err) - } return } From e74f9f27dbd8917d06cbaf68395893358d476f68 Mon Sep 17 00:00:00 2001 From: zelig Date: Tue, 9 Dec 2014 23:55:50 +0000 Subject: [PATCH 3/4] eth protocol changes - changed backend interface - using callbacks for blockPool - use rlp stream for lazy decoding - use peer as logger - add id (peer pubkey) to ethProtocol fields - add testPeer to protocol test (temporary) --- eth/protocol.go | 136 +++++++++++++++++++++++++------------------ eth/protocol_test.go | 56 +++++++++++++----- 2 files changed, 122 insertions(+), 70 deletions(-) diff --git a/eth/protocol.go b/eth/protocol.go index 992fc7550..380bcc8d2 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -7,18 +7,16 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethutil" - ethlogger "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/rlp" ) -var logger = ethlogger.NewLogger("SERV") - // ethProtocol represents the ethereum wire protocol // instance is running on each peer type ethProtocol struct { eth backend - td *big.Int peer *p2p.Peer + id string rw p2p.MsgReadWriter } @@ -26,28 +24,21 @@ type ethProtocol struct { // used as an argument to EthProtocol type backend interface { GetTransactions() (txs []*types.Transaction) - AddTransactions(txs []*types.Transaction) + AddTransactions([]*types.Transaction) GetBlockHashes(hash []byte, amount uint32) (hashes [][]byte) - AddHash(hash []byte, peer *p2p.Peer) (more bool) + AddBlockHashes(next func() ([]byte, bool), peerId string) GetBlock(hash []byte) (block *types.Block) - AddBlock(td *big.Int, block *types.Block, peer *p2p.Peer) (fetchHashes bool, err error) - AddPeer(td *big.Int, currentBlock []byte, peer *p2p.Peer) (fetchHashes bool) + AddBlock(block *types.Block, peerId string) (err error) + AddPeer(td *big.Int, currentBlock []byte, peerId string, requestHashes func([]byte) error, requestBlocks func([][]byte) error, invalidBlock func(error)) (best bool) + RemovePeer(peerId string) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) } const ( - ProtocolVersion = 43 - // 0x00 // PoC-1 - // 0x01 // PoC-2 - // 0x07 // PoC-3 - // 0x09 // PoC-4 - // 0x17 // PoC-5 - // 0x1c // PoC-6 + ProtocolVersion = 43 NetworkId = 0 ProtocolLength = uint64(8) ProtocolMaxMsgSize = 10 * 1024 * 1024 - - blockHashesBatchSize = 256 ) // eth protocol message codes @@ -74,7 +65,8 @@ type getBlockHashesMsgData struct { } // 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{ Name: "eth", @@ -86,11 +78,14 @@ func EthProtocol(eth backend) *p2p.Protocol { } } +// the main loop that handles incoming messages +// note RemovePeer in the post-disconnect hook func runEthProtocol(eth backend, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) { self := ðProtocol{ eth: eth, rw: rw, peer: peer, + id: (string)(peer.Identity().Pubkey()), } err = self.handleStatus() if err == nil { @@ -98,6 +93,7 @@ func runEthProtocol(eth backend, peer *p2p.Peer, rw p2p.MsgReadWriter) (err erro for { err = self.handle() if err != nil { + self.eth.RemovePeer(self.id) break } } @@ -132,6 +128,7 @@ func (self *ethProtocol) handle() error { return self.rw.EncodeMsg(TxMsg, txsInterface...) case TxMsg: + // TODO: rework using lazy RLP stream var txs []*types.Transaction if err := msg.Decode(&txs); err != nil { return ProtocolError(ErrDecode, "%v", err) @@ -148,29 +145,26 @@ func (self *ethProtocol) handle() error { case BlockHashesMsg: // TODO: redo using lazy decode , this way very inefficient on known chains - // 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 + 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 } + 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: - // Limit to max 300 blocks var blockHashes [][]byte if err := msg.Decode(&blockHashes); err != nil { return ProtocolError(ErrDecode, "%v", err) } - max := int(math.Min(float64(len(blockHashes)), 300.0)) + max := int(math.Min(float64(len(blockHashes)), blockHashesBatchSize)) var blocks []interface{} for i, hash := range blockHashes { if i >= max { @@ -184,20 +178,19 @@ func (self *ethProtocol) handle() error { return self.rw.EncodeMsg(BlocksMsg, blocks...) case BlocksMsg: - 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 + msgStream := rlp.NewListStream(msg.Payload, uint64(msg.Size)) + for { + var block *types.Block + if err := msgStream.Decode(&block); err != nil { + if err == rlp.EOL { + break + } else { + return ProtocolError(ErrDecode, "%v", err) } } + if err := self.eth.AddBlock(block, self.id); err != nil { + return ProtocolError(ErrInvalidBlock, "%v", err) + } } case NewBlockMsg: @@ -205,13 +198,24 @@ func (self *ethProtocol) handle() error { if err := msg.Decode(&request); err != nil { return ProtocolError(ErrDecode, "%v", err) } - 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()) + 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) + if self.eth.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.invalidBlock) { + called := true + iter := func() (hash []byte, ok bool) { + if called { + called = false + return hash, true + } else { + return + } + } + self.eth.AddBlockHashes(iter, self.id) + if err := self.eth.AddBlock(request.Block, self.id); err != nil { + return ProtocolError(ErrInvalidBlock, "%v", err) + } } default: @@ -279,16 +283,34 @@ func (self *ethProtocol) handleStatus() error { return ProtocolError(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, ProtocolVersion) } - logger.Infof("Peer is [eth] capable (%d/%d). TD = %v ~ %x", status.ProtocolVersion, status.NetworkId, status.CurrentBlock) + self.peer.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) - } + self.eth.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.invalidBlock) return nil } -func (self *ethProtocol) FetchHashes(from []byte) error { - logger.Debugf("Fetching hashes (%d) %x...\n", blockHashesBatchSize, from[0:4]) +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 +} diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 757e87fa4..a166ea6cd 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -6,6 +6,7 @@ import ( "testing" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p" ) @@ -55,10 +56,11 @@ type TestBackend struct { getTransactions func() []*types.Transaction addTransactions func(txs []*types.Transaction) getBlockHashes func(hash []byte, amount uint32) (hashes [][]byte) - addHash func(hash []byte, peer *p2p.Peer) (more bool) + addBlockHashes func(next func() ([]byte, bool), peerId string) 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) + 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) status func() (td *big.Int, currentBlock []byte, genesisBlock []byte) } @@ -82,12 +84,12 @@ func (self *TestBackend) GetBlockHashes(hash []byte, amount uint32) (hashes [][] return } -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) } - return } + func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) { if self.getBlock != nil { block = self.getBlock(hash) @@ -95,20 +97,26 @@ func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) { return } -func (self *TestBackend) AddBlock(td *big.Int, block *types.Block, peer *p2p.Peer) (fetchHashes bool, err error) { +func (self *TestBackend) AddBlock(block *types.Block, peerId string) (err error) { if self.addBlock != nil { - fetchHashes, err = self.addBlock(td, block, peer) + err = self.addBlock(block, peerId) } return } -func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peer *p2p.Peer) (fetchHashes bool) { +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 { - fetchHashes = self.addPeer(td, currentBlock, peer) + best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, invalidBlock) } return } +func (self *TestBackend) RemovePeer(peerId string) { + if self.removePeer != nil { + self.removePeer(peerId) + } +} + func (self *TestBackend) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) { if self.status != nil { td, currentBlock, genesisBlock = self.status() @@ -116,13 +124,35 @@ func (self *TestBackend) Status() (td *big.Int, currentBlock []byte, genesisBloc return } -func TestEth(t *testing.T) { +// TODO: refactor this into p2p/client_identity +type peerId struct { + pubkey []byte +} + +func (self *peerId) String() string { + return "test peer" +} + +func (self *peerId) Pubkey() (pubkey []byte) { + pubkey = self.pubkey + if len(pubkey) == 0 { + pubkey = crypto.GenerateNewKeyPair().PublicKey + self.pubkey = pubkey + } + return +} + +func testPeer() *p2p.Peer { + return p2p.NewPeer(&peerId{}, []p2p.Cap{}) +} + +func TestErrNoStatusMsg(t *testing.T) { quit := make(chan bool) rw := &testMsgReadWriter{make(chan p2p.Msg, 10), make(chan p2p.Msg, 10)} testBackend := &TestBackend{} var err error go func() { - err = runEthProtocol(testBackend, nil, rw) + err = runEthProtocol(testBackend, testPeer(), rw) close(quit) }() statusMsg := p2p.NewMsg(4) From 3e38f8af23bfed66e392bb570449e739c10d2641 Mon Sep 17 00:00:00 2001 From: zelig Date: Wed, 10 Dec 2014 04:12:25 +0000 Subject: [PATCH 4/4] initial commit for eth blockpool + test --- eth/block_pool.go | 514 +++++++++++++++++++++++++++++++++++++++++ eth/block_pool_test.go | 198 ++++++++++++++++ 2 files changed, 712 insertions(+) create mode 100644 eth/block_pool.go create mode 100644 eth/block_pool_test.go diff --git a/eth/block_pool.go b/eth/block_pool.go new file mode 100644 index 000000000..2c5ebb1e3 --- /dev/null +++ b/eth/block_pool.go @@ -0,0 +1,514 @@ +package eth + +import ( + "bytes" + "fmt" + "math" + "math/big" + "sync" + "time" + + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethutil" + "github.com/ethereum/go-ethereum/event" + ethlogger "github.com/ethereum/go-ethereum/logger" +) + +var poolLogger = ethlogger.NewLogger("Blockpool") + +const ( + blockHashesBatchSize = 256 + blockBatchSize = 64 + blockRequestInterval = 10 // seconds + blockRequestRepetition = 1 + cacheTimeout = 3 // minutes + blockTimeout = 5 // minutes +) + +type poolNode struct { + hash []byte + block *types.Block + child *poolNode + parent *poolNode + root *nodePointer + knownParent bool + suicide chan bool + peer string + source string + blockRequestRoot bool + blockRequestControl *bool + blockRequestQuit *(chan bool) +} + +// the minimal interface for chain manager +type chainManager interface { + KnownBlock(hash []byte) bool + AddBlock(*types.Block) error + CheckPoW(*types.Block) bool +} + +type BlockPool struct { + chainManager chainManager + eventer event.TypeMux + + // pool Pool + lock sync.Mutex + pool map[string]*poolNode + + peersLock sync.Mutex + peers map[string]*peerInfo + peer *peerInfo + + quit chan bool + wg sync.WaitGroup + running bool +} + +type peerInfo struct { + td *big.Int + currentBlock []byte + id string + requestBlockHashes func([]byte) error + requestBlocks func([][]byte) error + invalidBlock func(error) +} + +type nodePointer struct { + hash []byte +} + +type peerChangeEvent struct { + *peerInfo +} + +func NewBlockPool(chMgr chainManager) *BlockPool { + return &BlockPool{ + chainManager: chMgr, + pool: make(map[string]*poolNode), + peers: make(map[string]*peerInfo), + quit: make(chan bool), + running: true, + } +} + +func (self *BlockPool) Stop() { + self.lock.Lock() + if !self.running { + self.lock.Unlock() + return + } + self.running = false + self.lock.Unlock() + + poolLogger.Infoln("Stopping") + + close(self.quit) + self.wg.Wait() + poolLogger.Infoln("Stopped") + +} + +// Entry point for eth protocol to add block hashes received via BlockHashesMsg +// only hashes from the best peer is handled +// this method is always responsible to initiate further hash requests until +// a known parent is reached unless cancelled by a peerChange event +func (self *BlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string) { + // subscribe to peerChangeEvent before we check for best peer + peerChange := self.eventer.Subscribe(peerChangeEvent{}) + defer peerChange.Unsubscribe() + // check if this peer is the best + peer, best := self.getPeer(peerId) + if !best { + return + } + root := &nodePointer{} + // peer is still the best + hashes := make(chan []byte) + var lastPoolNode *poolNode + + // using a for select loop so that peer change (new best peer) can abort the parallel thread that processes hashes of the earlier best peer + for { + hash, ok := next() + if ok { + hashes <- hash + } else { + break + } + select { + case <-self.quit: + return + case <-peerChange.Chan(): + // remember where we left off with this peer + if lastPoolNode != nil { + root.hash = lastPoolNode.hash + go self.killChain(lastPoolNode) + } + case hash := <-hashes: + self.lock.Lock() + defer self.lock.Unlock() + // check if known block connecting the downloaded chain to our blockchain + if self.chainManager.KnownBlock(hash) { + poolLogger.Infof("known block (%x...)\n", hash[0:4]) + if lastPoolNode != nil { + lastPoolNode.knownParent = true + go self.requestBlocksLoop(lastPoolNode) + } else { + // all hashes known if topmost one is in blockchain + } + return + } + // + var currentPoolNode *poolNode + // check if lastPoolNode has the correct parent node (hash matching), + // then just assign to currentPoolNode + if lastPoolNode != nil && lastPoolNode.parent != nil && bytes.Compare(lastPoolNode.parent.hash, hash) == 0 { + currentPoolNode = lastPoolNode.parent + } else { + // otherwise look up in pool + currentPoolNode = self.pool[string(hash)] + // if node does not exist, create it and index in the pool + if currentPoolNode == nil { + currentPoolNode = &poolNode{ + hash: hash, + } + self.pool[string(hash)] = currentPoolNode + } + } + // set up parent-child nodes (doubly linked list) + self.link(currentPoolNode, lastPoolNode) + // ! we trust the node iff + // (1) node marked as by the same peer or + // (2) it has a PoW valid block retrieved + if currentPoolNode.peer == peer.id || currentPoolNode.block != nil { + // the trusted checkpoint from which we request hashes down to known head + lastPoolNode = self.pool[string(currentPoolNode.root.hash)] + break + } + currentPoolNode.peer = peer.id + currentPoolNode.root = root + lastPoolNode = currentPoolNode + } + } + // lastPoolNode is nil if and only if the node with stored root hash is already cleaned up + // after valid block insertion, therefore in this case the blockpool active chain is connected to the blockchain, so no need to request further hashes or request blocks + if lastPoolNode != nil { + root.hash = lastPoolNode.hash + peer.requestBlockHashes(lastPoolNode.hash) + go self.requestBlocksLoop(lastPoolNode) + } + return +} + +func (self *BlockPool) requestBlocksLoop(node *poolNode) { + suicide := time.After(blockTimeout * time.Minute) + requestTimer := time.After(0) + var controlChan chan bool + closedChan := make(chan bool) + quit := make(chan bool) + close(closedChan) + requestBlocks := true + origNode := node + self.lock.Lock() + node.blockRequestRoot = true + b := false + control := &b + node.blockRequestControl = control + node.blockRequestQuit = &quit + self.lock.Unlock() + blocks := 0 + self.wg.Add(1) +loop: + for { + if requestBlocks { + controlChan = closedChan + } else { + self.lock.Lock() + if *node.blockRequestControl { + controlChan = closedChan + *node.blockRequestControl = false + } + self.lock.Unlock() + } + select { + case <-quit: + break loop + case <-suicide: + go self.killChain(origNode) + break loop + + case <-requestTimer: + requestBlocks = true + + case <-controlChan: + controlChan = nil + // this iteration takes care of requesting blocks only starting from the first node with a missing block (moving target), + // max up to the next checkpoint (n.blockRequestRoot true) + nodes := []*poolNode{} + n := node + next := node + self.lock.Lock() + for n != nil && (n == node || !n.blockRequestRoot) && (requestBlocks || n.block != nil) { + if n.block != nil { + if len(nodes) == 0 { + // nil control indicates that node is not needed anymore + // block can be inserted to blockchain and deleted if knownParent + n.blockRequestControl = nil + blocks++ + next = next.child + } else { + // this is needed to indicate that when a new chain forks from an existing one + // triggering a reorg will ? renew the blockTimeout period ??? + // if there is a block but control == nil should start fetching blocks, see link function + n.blockRequestControl = control + } + } else { + nodes = append(nodes, n) + n.blockRequestControl = control + } + n = n.child + } + // if node is connected to the blockchain, we can immediately start inserting + // blocks to the blockchain and delete nodes + if node.knownParent { + go self.insertChainFrom(node) + } + if next.blockRequestRoot && next != node { + // no more missing blocks till the checkpoint, quitting + poolLogger.Debugf("fetched %v blocks on active chain, batch %v-%v", blocks, origNode, n) + break loop + } + self.lock.Unlock() + + // reset starting node to the first descendant node with missing block + node = next + if !requestBlocks { + continue + } + go self.requestBlocks(nodes) + requestTimer = time.After(blockRequestInterval * time.Second) + } + } + self.wg.Done() + return +} + +func (self *BlockPool) requestBlocks(nodes []*poolNode) { + // distribute block request among known peers + self.peersLock.Lock() + peerCount := len(self.peers) + poolLogger.Debugf("requesting %v missing blocks from %v peers", len(nodes), peerCount) + blockHashes := make([][][]byte, peerCount) + repetitions := int(math.Max(float64(peerCount)/2.0, float64(blockRequestRepetition))) + for n, node := range nodes { + for i := 0; i < repetitions; i++ { + blockHashes[n%peerCount] = append(blockHashes[n%peerCount], node.hash) + n++ + } + } + i := 0 + for _, peer := range self.peers { + peer.requestBlocks(blockHashes[i]) + i++ + } + self.peersLock.Unlock() +} + +func (self *BlockPool) insertChainFrom(node *poolNode) { + self.lock.Lock() + defer self.lock.Unlock() + for node != nil && node.blockRequestControl == nil { + err := self.chainManager.AddBlock(node.block) + if err != nil { + poolLogger.Debugf("invalid block %v", node.hash) + poolLogger.Debugf("penalise peers %v (hash), %v (block)", node.peer, node.source) + // penalise peer in node.source + go self.killChain(node) + return + } + poolLogger.Debugf("insert block %v into blockchain", node.hash) + node = node.child + } + // if block insertion succeeds, mark the child as knownParent + // trigger request blocks reorg + if node != nil { + node.knownParent = true + *(node.blockRequestControl) = true + } +} + +// AddPeer is called by the eth protocol instance running on the peer after +// the status message has been received with total difficulty and current block hash +// AddPeer can only be used once, RemovePeer needs to be called when the peer disconnects +func (self *BlockPool) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, invalidBlock func(error)) bool { + self.peersLock.Lock() + defer self.peersLock.Unlock() + if self.peers[peerId] != nil { + panic("peer already added") + } + info := &peerInfo{ + td: td, + currentBlock: currentBlock, + id: peerId, //peer.Identity().Pubkey() + requestBlockHashes: requestBlockHashes, + requestBlocks: requestBlocks, + invalidBlock: invalidBlock, + } + self.peers[peerId] = info + poolLogger.Debugf("add new peer %v with td %v", peerId, td) + currentTD := ethutil.Big0 + if self.peer != nil { + currentTD = self.peer.td + } + if td.Cmp(currentTD) > 0 { + self.peer = info + self.eventer.Post(peerChangeEvent{info}) + poolLogger.Debugf("peer %v promoted to best peer", peerId) + requestBlockHashes(currentBlock) + return true + } + return false +} + +// RemovePeer is called by the eth protocol when the peer disconnects +func (self *BlockPool) RemovePeer(peerId string) { + self.peersLock.Lock() + defer self.peersLock.Unlock() + if self.peers[peerId] != nil { + panic("peer already removed") + } + self.peers[peerId] = nil + poolLogger.Debugf("remove peer %v", peerId[0:4]) + + // if current best peer is removed, need find a better one + if peerId == self.peer.id { + var newPeer *peerInfo + max := ethutil.Big0 + // peer with the highest self-acclaimed TD is chosen + for _, info := range self.peers { + if info.td.Cmp(max) > 0 { + max = info.td + newPeer = info + } + } + self.peer = newPeer + self.eventer.Post(peerChangeEvent{newPeer}) + if newPeer != nil { + poolLogger.Debugf("peer %v with td %v spromoted to best peer", newPeer.id[0:4], newPeer.td) + newPeer.requestBlockHashes(newPeer.currentBlock) + } else { + poolLogger.Warnln("no peers left") + } + } +} + +func (self *BlockPool) getPeer(peerId string) (*peerInfo, bool) { + self.peersLock.Lock() + defer self.peersLock.Unlock() + if self.peer.id == peerId { + return self.peer, true + } + info, ok := self.peers[peerId] + if !ok { + panic("unknown peer") + } + return info, false +} + +// if same peer gave different chain before, this will overwrite it +// if currentPoolNode existed as a non-leaf node the earlier fork is delinked +// if same parent hash is found, we can abort, we do not allow the same peer to change minds about parent of same hash, if errored first time round, will get penalized. +// if lastPoolNode had a different parent the earlier parent (with entire subtree) is delinked, this situation cannot normally arise though +// just in case reset lastPoolNode as non-root (unlikely) + +func (self *BlockPool) link(parent, child *poolNode) { + // reactivate node scheduled for suicide + if parent.suicide != nil { + close(parent.suicide) + parent.suicide = nil + } + if parent.child != child { + orphan := parent.child + orphan.parent = nil + go self.killChain(orphan) + parent.child = child + } + if child != nil { + if child.parent != parent { + orphan := child.parent + orphan.child = nil + go func() { + // if it is a aberrant reverse fork, zip down to bottom + for orphan.parent != nil { + orphan = orphan.parent + } + self.killChain(orphan) + }() + child.parent = parent + } + child.knownParent = false + } +} + +func (self *BlockPool) killChain(node *poolNode) { + if node == nil { + return + } + poolLogger.Debugf("suicide scheduled on node %v", node) + suicide := make(chan bool) + self.lock.Lock() + node.suicide = suicide + self.lock.Unlock() + timer := time.After(cacheTimeout * time.Minute) + self.wg.Add(1) + select { + case <-self.quit: + case <-suicide: + // cancel suicide = close node.suicide to reactivate node + case <-timer: + poolLogger.Debugf("suicide on node %v", node) + self.lock.Lock() + defer self.lock.Unlock() + // proceed up via child links until another suicide root found or chain ends + // abort request blocks loops that start above + // and delete nodes from pool then quit the suicide process + okToAbort := node.blockRequestRoot + for node != nil && (node.suicide == suicide || node.suicide == nil) { + self.pool[string(node.hash)] = nil + if okToAbort && node.blockRequestQuit != nil { + quit := *(node.blockRequestQuit) + if quit != nil { // not yet closed + *(node.blockRequestQuit) = nil + close(quit) + } + } else { + okToAbort = true + } + node = node.child + } + } + self.wg.Done() +} + +// AddBlock is the entry point for the eth protocol when blockmsg is received upon requests +// It has a strict interpretation of the protocol in that if the block received has not been requested, it results in an error (which can be ignored) +// block is checked for PoW +// only the first PoW-valid block for a hash is considered legit +func (self *BlockPool) AddBlock(block *types.Block, peerId string) (err error) { + hash := block.Hash() + self.lock.Lock() + defer self.lock.Unlock() + node, ok := self.pool[string(hash)] + if !ok && !self.chainManager.KnownBlock(hash) { + return fmt.Errorf("unrequested block %x", hash) + } + if node.block != nil { + return + } + // validate block for PoW + if !self.chainManager.CheckPoW(block) { + return fmt.Errorf("invalid pow on block %x", hash) + } + node.block = block + node.source = peerId + return nil +} diff --git a/eth/block_pool_test.go b/eth/block_pool_test.go new file mode 100644 index 000000000..315cc748d --- /dev/null +++ b/eth/block_pool_test.go @@ -0,0 +1,198 @@ +package eth + +import ( + "bytes" + "fmt" + "log" + "os" + "sync" + "testing" + + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethutil" + ethlogger "github.com/ethereum/go-ethereum/logger" +) + +var sys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugDetailLevel)) + +type testChainManager struct { + knownBlock func(hash []byte) bool + addBlock func(*types.Block) error + checkPoW func(*types.Block) bool +} + +func (self *testChainManager) KnownBlock(hash []byte) bool { + if self.knownBlock != nil { + return self.knownBlock(hash) + } + return false +} + +func (self *testChainManager) AddBlock(block *types.Block) error { + if self.addBlock != nil { + return self.addBlock(block) + } + return nil +} + +func (self *testChainManager) CheckPoW(block *types.Block) bool { + if self.checkPoW != nil { + return self.checkPoW(block) + } + return false +} + +func knownBlock(hashes ...[]byte) (f func([]byte) bool) { + f = func(block []byte) bool { + for _, hash := range hashes { + if bytes.Compare(block, hash) == 0 { + return true + } + } + return false + } + return +} + +func addBlock(hashes ...[]byte) (f func(*types.Block) error) { + f = func(block *types.Block) error { + for _, hash := range hashes { + if bytes.Compare(block.Hash(), hash) == 0 { + return fmt.Errorf("invalid by test") + } + } + return nil + } + return +} + +func checkPoW(hashes ...[]byte) (f func(*types.Block) bool) { + f = func(block *types.Block) bool { + for _, hash := range hashes { + if bytes.Compare(block.Hash(), hash) == 0 { + return false + } + } + return true + } + return +} + +func newTestChainManager(knownBlocks [][]byte, invalidBlocks [][]byte, invalidPoW [][]byte) *testChainManager { + return &testChainManager{ + knownBlock: knownBlock(knownBlocks...), + addBlock: addBlock(invalidBlocks...), + checkPoW: checkPoW(invalidPoW...), + } +} + +type intToHash map[int][]byte + +type hashToInt map[string]int + +type testHashPool struct { + intToHash + hashToInt +} + +func newHash(i int) []byte { + return crypto.Sha3([]byte(string(i))) +} + +func newTestBlockPool(knownBlockIndexes []int, invalidBlockIndexes []int, invalidPoWIndexes []int) (hashPool *testHashPool, blockPool *BlockPool) { + hashPool = &testHashPool{make(intToHash), make(hashToInt)} + knownBlocks := hashPool.indexesToHashes(knownBlockIndexes) + invalidBlocks := hashPool.indexesToHashes(invalidBlockIndexes) + invalidPoW := hashPool.indexesToHashes(invalidPoWIndexes) + blockPool = NewBlockPool(newTestChainManager(knownBlocks, invalidBlocks, invalidPoW)) + return +} + +func (self *testHashPool) indexesToHashes(indexes []int) (hashes [][]byte) { + for _, i := range indexes { + hash, found := self.intToHash[i] + if !found { + hash = newHash(i) + self.intToHash[i] = hash + self.hashToInt[string(hash)] = i + } + hashes = append(hashes, hash) + } + return +} + +func (self *testHashPool) hashesToIndexes(hashes [][]byte) (indexes []int) { + for _, hash := range hashes { + i, found := self.hashToInt[string(hash)] + if !found { + i = -1 + } + indexes = append(indexes, i) + } + return +} + +type protocolChecker struct { + blockHashesRequests []int + blocksRequests [][]int + invalidBlocks []error + hashPool *testHashPool + lock sync.Mutex +} + +// -1 is special: not found (a hash never seen) +func (self *protocolChecker) requestBlockHashesCallBack() (requestBlockHashesCallBack func([]byte) error) { + requestBlockHashesCallBack = func(hash []byte) error { + indexes := self.hashPool.hashesToIndexes([][]byte{hash}) + self.lock.Lock() + defer self.lock.Unlock() + self.blockHashesRequests = append(self.blockHashesRequests, indexes[0]) + return nil + } + return +} + +func (self *protocolChecker) requestBlocksCallBack() (requestBlocksCallBack func([][]byte) error) { + requestBlocksCallBack = func(hashes [][]byte) error { + indexes := self.hashPool.hashesToIndexes(hashes) + self.lock.Lock() + defer self.lock.Unlock() + self.blocksRequests = append(self.blocksRequests, indexes) + return nil + } + return +} + +func (self *protocolChecker) invalidBlockCallBack() (invalidBlockCallBack func(error)) { + invalidBlockCallBack = func(err error) { + self.invalidBlocks = append(self.invalidBlocks, err) + } + return +} + +func TestAddPeer(t *testing.T) { + ethlogger.AddLogSystem(sys) + knownBlockIndexes := []int{0, 1} + invalidBlockIndexes := []int{2, 3} + invalidPoWIndexes := []int{4, 5} + hashPool, blockPool := newTestBlockPool(knownBlockIndexes, invalidBlockIndexes, invalidPoWIndexes) + // TODO: + // hashPool, blockPool, blockChainChecker = newTestBlockPool(knownBlockIndexes, invalidBlockIndexes, invalidPoWIndexes) + peer0 := &protocolChecker{ + // blockHashesRequests: make([]int), + // blocksRequests: make([][]int), + // invalidBlocks: make([]error), + hashPool: hashPool, + } + best := blockPool.AddPeer(ethutil.Big1, newHash(100), "0", + peer0.requestBlockHashesCallBack(), + peer0.requestBlocksCallBack(), + peer0.invalidBlockCallBack(), + ) + if !best { + t.Errorf("peer not accepted as best") + } + blockPool.Stop() + +}