headrush
/
go-ethereum
forked from mirror/go-ethereum
1
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

eth protocol changes

Browse Source 
- 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)
poc8
zelig 10 years ago
parent 0add0c400f
commit 2dd8f41147
2 changed files with 132 additions and 11 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 97
      eth/protocol.go
  2. 46
      eth/protocol_test.go

97
eth/protocol.go
Unescape View File

@ -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
@ -99,6 +100,7 @@ func EthProtocol(eth backend) *p2p.Protocol {
}
}
<<<<<<< HEAD
<<<<<<< HEAD
// the main loop that handles incoming messages
// note RemovePeer in the post-disconnect hook
@ -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 := &ethProtocol{
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 {
@ -124,10 +134,14 @@ func runEthProtocol(eth backend, peer *p2p.Peer, rw p2p.MsgReadWriter) (err erro
for {
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:
@ -389,6 +447,7 @@ func (self *ethProtocol) handleStatus() error {
return ProtocolError(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, ProtocolVersion)
}
<<<<<<< HEAD
<<<<<<< HEAD
self.peer.Infof("Peer is [eth] capable (%d/%d). TD = %v ~ %x", status.ProtocolVersion, status.NetworkId, status.CurrentBlock)
@ -400,10 +459,16 @@ 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])
@ -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

46
eth/protocol_test.go
Unescape View File

@ -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)
}
@ -91,6 +84,7 @@ func (self *TestBackend) GetBlockHashes(hash []byte, amount uint32) (hashes [][]
return
}
<<<<<<< HEAD
<<<<<<< HEAD
func (self *TestBackend) AddBlockHashes(next func() ([]byte, bool), peerId string) {
if self.addBlockHashes != nil {
@ -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)
@ -113,6 +115,7 @@ func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) {
return
}
<<<<<<< HEAD
<<<<<<< HEAD
func (self *TestBackend) AddBlock(block *types.Block, peerId string) (err error) {
if self.addBlock != nil {
@ -122,10 +125,16 @@ 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 {
@ -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)

Write Preview
Loading…
Cancel
Save