cerc-io/plugeth
18
0
Fork 1
Code Issues 4 Pull Requests Packages Projects Releases Wiki Activity

protocol

Browse Source 
- new interface explicit backend components txPool, chainManager, blockPool
- added protoErrorDisconnect for blockpool callback (FIXME: handling peer disconnects)
This commit is contained in:
zelig 2014-12-14 18:04:50 +00:00
parent 02017ed0e0
commit c44e025898
1 changed files with 52 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

90
eth/protocol.go
View File

@ -14,7 +14,9 @@ import (
// ethProtocol represents the ethereum wire protocol // ethProtocol represents the ethereum wire protocol
// instance is running on each peer // instance is running on each peer
type ethProtocol struct { type ethProtocol struct {
eth backend txPool txPool
chainManager chainManager
blockPool blockPool
peer *p2p.Peer peer *p2p.Peer
id string id string
rw p2p.MsgReadWriter rw p2p.MsgReadWriter
@ -22,18 +24,23 @@ type ethProtocol struct {
// backend is the interface the ethereum protocol backend should implement // backend is the interface the ethereum protocol backend should implement
// used as an argument to EthProtocol // used as an argument to EthProtocol
type backend interface { type txPool interface {
GetTransactions() (txs []*types.Transaction)
AddTransactions([]*types.Transaction) AddTransactions([]*types.Transaction)
GetBlockHashes(hash []byte, amount uint32) (hashes [][]byte) }
AddBlockHashes(next func() ([]byte, bool), peerId string)
type chainManager interface {
GetBlockHashesFromHash(hash []byte, amount uint64) (hashes [][]byte)
GetBlock(hash []byte) (block *types.Block) GetBlock(hash []byte) (block *types.Block)
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) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte)
} }
type blockPool interface {
AddBlockHashes(next func() ([]byte, bool), peerId string)
AddBlock(block *types.Block, peerId string)
AddPeer(td *big.Int, currentBlock []byte, peerId string, requestHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(int, string, ...interface{})) (best bool)
RemovePeer(peerId string)
}
const ( const (
ProtocolVersion = 43 ProtocolVersion = 43
NetworkId = 0 NetworkId = 0
@ -61,28 +68,30 @@ type newBlockMsgData struct {
type getBlockHashesMsgData struct { type getBlockHashesMsgData struct {
Hash []byte Hash []byte
Amount uint32 Amount uint64
} }
// main entrypoint, wrappers starting a server running the eth protocol // 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 // the Dev p2p layer then runs the protocol instance on each peer
func EthProtocol(eth backend) *p2p.Protocol { func EthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool) p2p.Protocol {
return &p2p.Protocol{ return p2p.Protocol{
Name: "eth", Name: "eth",
Version: ProtocolVersion, Version: ProtocolVersion,
Length: ProtocolLength, Length: ProtocolLength,
Run: func(peer *p2p.Peer, rw p2p.MsgReadWriter) error { Run: func(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
return runEthProtocol(eth, peer, rw) return runEthProtocol(txPool, chainManager, blockPool, peer, rw)
}, },
} }
} }
// the main loop that handles incoming messages // the main loop that handles incoming messages
// note RemovePeer in the post-disconnect hook // note RemovePeer in the post-disconnect hook
func runEthProtocol(eth backend, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) { func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
self := &ethProtocol{ self := &ethProtocol{
eth: eth, txPool: txPool,
chainManager: chainManager,
blockPool: blockPool,
rw: rw, rw: rw,
peer: peer, peer: peer,
id: (string)(peer.Identity().Pubkey()), id: (string)(peer.Identity().Pubkey()),
@ -93,7 +102,7 @@ func runEthProtocol(eth backend, peer *p2p.Peer, rw p2p.MsgReadWriter) (err erro
for { for {
err = self.handle() err = self.handle()
if err != nil { if err != nil {
self.eth.RemovePeer(self.id) self.blockPool.RemovePeer(self.id)
break break
} }
} }
@ -118,29 +127,20 @@ func (self *ethProtocol) handle() error {
case StatusMsg: case StatusMsg:
return ProtocolError(ErrExtraStatusMsg, "") 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: case TxMsg:
// TODO: rework using lazy RLP stream // TODO: rework using lazy RLP stream
var txs []*types.Transaction var txs []*types.Transaction
if err := msg.Decode(&txs); err != nil { if err := msg.Decode(&txs); err != nil {
return ProtocolError(ErrDecode, "%v", err) return ProtocolError(ErrDecode, "%v", err)
} }
self.eth.AddTransactions(txs) self.txPool.AddTransactions(txs)
case GetBlockHashesMsg: case GetBlockHashesMsg:
var request getBlockHashesMsgData var request getBlockHashesMsgData
if err := msg.Decode(&request); err != nil { if err := msg.Decode(&request); err != nil {
return ProtocolError(ErrDecode, "%v", err) return ProtocolError(ErrDecode, "%v", err)
} }
hashes := self.eth.GetBlockHashes(request.Hash, request.Amount) hashes := self.chainManager.GetBlockHashesFromHash(request.Hash, request.Amount)
return self.rw.EncodeMsg(BlockHashesMsg, ethutil.ByteSliceToInterface(hashes)...) return self.rw.EncodeMsg(BlockHashesMsg, ethutil.ByteSliceToInterface(hashes)...)
case BlockHashesMsg: case BlockHashesMsg:
@ -154,7 +154,7 @@ func (self *ethProtocol) handle() error {
} }
return return
} }
self.eth.AddBlockHashes(iter, self.id) self.blockPool.AddBlockHashes(iter, self.id)
if err != nil && err != rlp.EOL { if err != nil && err != rlp.EOL {
return ProtocolError(ErrDecode, "%v", err) return ProtocolError(ErrDecode, "%v", err)
} }
@ -170,7 +170,7 @@ func (self *ethProtocol) handle() error {
if i >= max { if i >= max {
break break
} }
block := self.eth.GetBlock(hash) block := self.chainManager.GetBlock(hash)
if block != nil { if block != nil {
blocks = append(blocks, block.Value().Raw()) blocks = append(blocks, block.Value().Raw())
} }
@ -188,9 +188,7 @@ func (self *ethProtocol) handle() error {
return ProtocolError(ErrDecode, "%v", err) return ProtocolError(ErrDecode, "%v", err)
} }
} }
if err := self.eth.AddBlock(block, self.id); err != nil { self.blockPool.AddBlock(block, self.id)
return ProtocolError(ErrInvalidBlock, "%v", err)
}
} }
case NewBlockMsg: case NewBlockMsg:
@ -202,7 +200,7 @@ func (self *ethProtocol) handle() error {
// to simplify backend interface adding a new block // to simplify backend interface adding a new block
// uses AddPeer followed by AddHashes, AddBlock only if peer is the best peer // uses AddPeer followed by AddHashes, AddBlock only if peer is the best peer
// (or selected as new best peer) // (or selected as new best peer)
if self.eth.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.invalidBlock) { if self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) {
called := true called := true
iter := func() (hash []byte, ok bool) { iter := func() (hash []byte, ok bool) {
if called { if called {
@ -212,10 +210,8 @@ func (self *ethProtocol) handle() error {
return return
} }
} }
self.eth.AddBlockHashes(iter, self.id) self.blockPool.AddBlockHashes(iter, self.id)
if err := self.eth.AddBlock(request.Block, self.id); err != nil { self.blockPool.AddBlock(request.Block, self.id)
return ProtocolError(ErrInvalidBlock, "%v", err)
}
} }
default: default:
@ -233,7 +229,7 @@ type statusMsgData struct {
} }
func (self *ethProtocol) statusMsg() p2p.Msg { func (self *ethProtocol) statusMsg() p2p.Msg {
td, currentBlock, genesisBlock := self.eth.Status() td, currentBlock, genesisBlock := self.chainManager.Status()
return p2p.NewMsg(StatusMsg, return p2p.NewMsg(StatusMsg,
uint32(ProtocolVersion), uint32(ProtocolVersion),
@ -269,7 +265,7 @@ func (self *ethProtocol) handleStatus() error {
return ProtocolError(ErrDecode, "%v", err) return ProtocolError(ErrDecode, "%v", err)
} }
_, _, genesisBlock := self.eth.Status() _, _, genesisBlock := self.chainManager.Status()
if bytes.Compare(status.GenesisBlock, genesisBlock) != 0 { if bytes.Compare(status.GenesisBlock, genesisBlock) != 0 {
return ProtocolError(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, genesisBlock) return ProtocolError(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, genesisBlock)
@ -285,7 +281,7 @@ func (self *ethProtocol) handleStatus() error {
self.peer.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)
self.eth.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.invalidBlock) self.blockPool.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect)
return nil return nil
} }
@ -300,11 +296,6 @@ func (self *ethProtocol) requestBlocks(hashes [][]byte) error {
return self.rw.EncodeMsg(GetBlocksMsg, ethutil.ByteSliceToInterface(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) { func (self *ethProtocol) protoError(code int, format string, params ...interface{}) (err *protocolError) {
err = ProtocolError(code, format, params...) err = ProtocolError(code, format, params...)
if err.Fatal() { if err.Fatal() {
@ -314,3 +305,14 @@ func (self *ethProtocol) protoError(code int, format string, params ...interface
} }
return return
} }
func (self *ethProtocol) protoErrorDisconnect(code int, format string, params ...interface{}) {
err := ProtocolError(code, format, params...)
if err.Fatal() {
self.peer.Errorln(err)
// disconnect
} else {
self.peer.Debugln(err)
}
}