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

eth, whisper: adapt for RLP encoder switch in package p2p

Browse Source 
I have rewritten the protocol test to use p2p.MsgPipe because
p2p.NewMsg is gone.
release/1.0.1
Felix Lange 10 years ago
parent b9e0b11e7d
commit e80dda6051
5 changed files with 124 additions and 175 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. 4
      eth/backend.go
  2. 95
      eth/protocol.go
  3. 189
      eth/protocol_test.go
  4. 2
      whisper/envelope.go
  5. 9
      whisper/peer.go

4
eth/backend.go
Unescape View File

@ -405,7 +405,7 @@ func (self *Ethereum) txBroadcastLoop() {
// automatically stops if unsubscribe
for obj := range self.txSub.Chan() {
event := obj.(core.TxPreEvent)
self.net.Broadcast("eth", TxMsg, event.Tx.RlpData())
self.net.Broadcast("eth", TxMsg, []*types.Transaction{event.Tx})
}
}
@ -414,7 +414,7 @@ func (self *Ethereum) blockBroadcastLoop() {
for obj := range self.blockSub.Chan() {
switch ev := obj.(type) {
case core.NewMinedBlockEvent:
self.net.Broadcast("eth", NewBlockMsg, ev.Block.RlpData(), ev.Block.Td)
self.net.Broadcast("eth", NewBlockMsg, []interface{}{ev.Block, ev.Block.Td})
}
}
}

95
eth/protocol.go
Unescape View File

@ -89,7 +89,7 @@ type blockPool interface {
RemovePeer(peerId string)
}
// message structs used for rlp decoding
// message structs used for RLP serialization
type newBlockMsgData struct {
Block *types.Block
TD *big.Int
@ -100,6 +100,14 @@ type getBlockHashesMsgData struct {
Amount uint64
}
type statusMsgData struct {
ProtocolVersion uint32
NetworkId uint32
TD *big.Int
CurrentBlock common.Hash
GenesisBlock common.Hash
}
// main entrypoint, wrappers starting a server running the eth protocol
// use this constructor to attach the protocol ("class") to server caps
// the Dev p2p layer then runs the protocol instance on each peer
@ -132,18 +140,25 @@ func runEthProtocol(protocolVersion, networkId int, txPool txPool, chainManager
},
id: fmt.Sprintf("%x", id[:8]),
}
err = self.handleStatus()
if err == nil {
self.propagateTxs()
for {
err = self.handle()
if err != nil {
self.blockPool.RemovePeer(self.id)
break
}
// handshake.
if err := self.handleStatus(); err != nil {
return err
}
defer self.blockPool.RemovePeer(self.id)
// propagate existing transactions. new transactions appearing
// after this will be sent via broadcasts.
if err := p2p.Send(rw, TxMsg, txPool.GetTransactions()); err != nil {
return err
}
// main loop. handle incoming messages.
for {
if err := self.handle(); err != nil {
return err
}
}
return
}
func (self *ethProtocol) handle() error {
@ -186,7 +201,7 @@ func (self *ethProtocol) handle() error {
request.Amount = maxHashes
}
hashes := self.chainManager.GetBlockHashesFromHash(request.Hash, request.Amount)
return p2p.EncodeMsg(self.rw, BlockHashesMsg, rlp.Flat(hashes))
return p2p.Send(self.rw, BlockHashesMsg, hashes)
case BlockHashesMsg:
msgStream := rlp.NewStream(msg.Payload)
@ -216,7 +231,7 @@ func (self *ethProtocol) handle() error {
return err
}
var blocks []interface{}
var blocks []*types.Block
var i int
for {
i++
@ -224,7 +239,7 @@ func (self *ethProtocol) handle() error {
err := msgStream.Decode(&hash)
if err == rlp.EOL {
break
} else {
} else if err != nil {
return self.protoError(ErrDecode, "msg %v: %v", msg, err)
}
@ -236,7 +251,7 @@ func (self *ethProtocol) handle() error {
break
}
}
return p2p.EncodeMsg(self.rw, BlocksMsg, blocks...)
return p2p.Send(self.rw, BlocksMsg, blocks)
case BlocksMsg:
msgStream := rlp.NewStream(msg.Payload)
@ -283,29 +298,8 @@ func (self *ethProtocol) handle() error {
return nil
}
type statusMsgData struct {
ProtocolVersion uint32
NetworkId uint32
TD *big.Int
CurrentBlock common.Hash
GenesisBlock common.Hash
}
func (self *ethProtocol) statusMsg() p2p.Msg {
td, currentBlock, genesisBlock := self.chainManager.Status()
return p2p.NewMsg(StatusMsg,
uint32(self.protocolVersion),
uint32(self.networkId),
td,
currentBlock,
genesisBlock,
)
}
func (self *ethProtocol) handleStatus() error {
// send precanned status message
if err := self.rw.WriteMsg(self.statusMsg()); err != nil {
if err := self.sendStatus(); err != nil {
return err
}
@ -314,11 +308,9 @@ func (self *ethProtocol) handleStatus() error {
if err != nil {
return err
}
if msg.Code != StatusMsg {
return self.protoError(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg)
}
if msg.Size > ProtocolMaxMsgSize {
return self.protoError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
}
@ -351,12 +343,12 @@ func (self *ethProtocol) handleStatus() error {
func (self *ethProtocol) requestBlockHashes(from common.Hash) error {
self.peer.Debugf("fetching hashes (%d) %x...\n", maxHashes, from[0:4])
return p2p.EncodeMsg(self.rw, GetBlockHashesMsg, interface{}(from), uint64(maxHashes))
return p2p.Send(self.rw, GetBlockHashesMsg, getBlockHashesMsgData{from, maxHashes})
}
func (self *ethProtocol) requestBlocks(hashes []common.Hash) error {
self.peer.Debugf("fetching %v blocks", len(hashes))
return p2p.EncodeMsg(self.rw, GetBlocksMsg, rlp.Flat(hashes))
return p2p.Send(self.rw, GetBlocksMsg, hashes)
}
func (self *ethProtocol) protoError(code int, format string, params ...interface{}) (err *errs.Error) {
@ -365,19 +357,20 @@ func (self *ethProtocol) protoError(code int, format string, params ...interface
return
}
func (self *ethProtocol) sendStatus() error {
td, currentBlock, genesisBlock := self.chainManager.Status()
return p2p.Send(self.rw, StatusMsg, &statusMsgData{
ProtocolVersion: uint32(self.protocolVersion),
NetworkId: uint32(self.networkId),
TD: td,
CurrentBlock: currentBlock,
GenesisBlock: genesisBlock,
})
}
func (self *ethProtocol) protoErrorDisconnect(err *errs.Error) {
err.Log(self.peer.Logger)
if err.Fatal() {
self.peer.Disconnect(p2p.DiscSubprotocolError)
}
}
func (self *ethProtocol) propagateTxs() {
transactions := self.txPool.GetTransactions()
iface := make([]interface{}, len(transactions))
for i, transaction := range transactions {
iface[i] = transaction
}
self.rw.WriteMsg(p2p.NewMsg(TxMsg, iface...))
}

189
eth/protocol_test.go
Unescape View File

@ -1,8 +1,6 @@
package eth
import (
"bytes"
"io"
"log"
"math/big"
"os"
@ -29,52 +27,21 @@ func logInit() {
}
}
type testMsgReadWriter struct {
in chan p2p.Msg
out []p2p.Msg
}
func (self *testMsgReadWriter) In(msg p2p.Msg) {
self.in <- msg
}
func (self *testMsgReadWriter) Out() (msg p2p.Msg, ok bool) {
if len(self.out) > 0 {
msg = self.out[0]
self.out = self.out[1:]
ok = true
}
return
}
func (self *testMsgReadWriter) WriteMsg(msg p2p.Msg) error {
self.out = append(self.out, msg)
return nil
}
func (self *testMsgReadWriter) ReadMsg() (p2p.Msg, error) {
msg, ok := <-self.in
if !ok {
return msg, io.EOF
}
return msg, nil
}
type testTxPool struct {
getTransactions func() []*types.Transaction
addTransactions func(txs []*types.Transaction)
}
type testChainManager struct {
getBlockHashes func(hash []byte, amount uint64) (hashes [][]byte)
getBlock func(hash []byte) *types.Block
status func() (td *big.Int, currentBlock []byte, genesisBlock []byte)
getBlockHashes func(hash common.Hash, amount uint64) (hashes []common.Hash)
getBlock func(hash common.Hash) *types.Block
status func() (td *big.Int, currentBlock common.Hash, genesisBlock common.Hash)
}
type testBlockPool struct {
addBlockHashes func(next func() ([]byte, bool), peerId string)
addBlockHashes func(next func() (common.Hash, bool), peerId string)
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, peerError func(*errs.Error)) (best bool)
addPeer func(td *big.Int, currentBlock common.Hash, peerId string, requestHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool)
removePeer func(peerId string)
}
@ -93,28 +60,28 @@ func (self *testTxPool) AddTransactions(txs []*types.Transaction) {
func (self *testTxPool) GetTransactions() types.Transactions { return nil }
func (self *testChainManager) GetBlockHashesFromHash(hash []byte, amount uint64) (hashes [][]byte) {
func (self *testChainManager) GetBlockHashesFromHash(hash common.Hash, amount uint64) (hashes []common.Hash) {
if self.getBlockHashes != nil {
hashes = self.getBlockHashes(hash, amount)
}
return
}
func (self *testChainManager) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) {
func (self *testChainManager) Status() (td *big.Int, currentBlock common.Hash, genesisBlock common.Hash) {
if self.status != nil {
td, currentBlock, genesisBlock = self.status()
}
return
}
func (self *testChainManager) GetBlock(hash []byte) (block *types.Block) {
func (self *testChainManager) GetBlock(hash common.Hash) (block *types.Block) {
if self.getBlock != nil {
block = self.getBlock(hash)
}
return
}
func (self *testBlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string) {
func (self *testBlockPool) AddBlockHashes(next func() (common.Hash, bool), peerId string) {
if self.addBlockHashes != nil {
self.addBlockHashes(next, peerId)
}
@ -126,7 +93,7 @@ func (self *testBlockPool) AddBlock(block *types.Block, peerId string) {
}
}
func (self *testBlockPool) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(*errs.Error)) (best bool) {
func (self *testBlockPool) AddPeer(td *big.Int, currentBlock common.Hash, peerId string, requestBlockHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool) {
if self.addPeer != nil {
best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, peerError)
}
@ -147,28 +114,36 @@ func testPeer() *p2p.Peer {
}
type ethProtocolTester struct {
p2p.MsgReadWriter // writing to the tester feeds the protocol
quit chan error
rw *testMsgReadWriter // p2p.MsgReadWriter
txPool *testTxPool // txPool
chainManager *testChainManager // chainManager
blockPool *testBlockPool // blockPool
pipe *p2p.MsgPipeRW // the protocol read/writes on this end
txPool *testTxPool // txPool
chainManager *testChainManager // chainManager
blockPool *testBlockPool // blockPool
t *testing.T
}
func newEth(t *testing.T) *ethProtocolTester {
p1, p2 := p2p.MsgPipe()
return &ethProtocolTester{
quit: make(chan error),
rw: &testMsgReadWriter{in: make(chan p2p.Msg, 10)},
txPool: &testTxPool{},
chainManager: &testChainManager{},
blockPool: &testBlockPool{},
t: t,
MsgReadWriter: p1,
quit: make(chan error, 1),
pipe: p2,
txPool: &testTxPool{},
chainManager: &testChainManager{},
blockPool: &testBlockPool{},
t: t,
}
}
func (self *ethProtocolTester) reset() {
self.rw = &testMsgReadWriter{in: make(chan p2p.Msg, 10)}
self.quit = make(chan error)
self.pipe.Close()
p1, p2 := p2p.MsgPipe()
self.MsgReadWriter = p1
self.pipe = p2
self.quit = make(chan error, 1)
}
func (self *ethProtocolTester) checkError(expCode int, delay time.Duration) (err error) {
@ -190,33 +165,8 @@ func (self *ethProtocolTester) checkError(expCode int, delay time.Duration) (err
return
}
func (self *ethProtocolTester) In(msg p2p.Msg) {
self.rw.In(msg)
}
func (self *ethProtocolTester) Out() (p2p.Msg, bool) {
return self.rw.Out()
}
func (self *ethProtocolTester) checkMsg(i int, code uint64, val interface{}) (msg p2p.Msg) {
if i >= len(self.rw.out) {
self.t.Errorf("expected at least %v msgs, got %v", i, len(self.rw.out))
return
}
msg = self.rw.out[i]
if msg.Code != code {
self.t.Errorf("expected msg code %v, got %v", code, msg.Code)
}
if val != nil {
if err := msg.Decode(val); err != nil {
self.t.Errorf("rlp encoding error: %v", err)
}
}
return
}
func (self *ethProtocolTester) run() {
err := runEthProtocol(ProtocolVersion, NetworkId, self.txPool, self.chainManager, self.blockPool, testPeer(), self.rw)
err := runEthProtocol(ProtocolVersion, NetworkId, self.txPool, self.chainManager, self.blockPool, testPeer(), self.pipe)
self.quit <- err
}
@ -224,41 +174,52 @@ func TestStatusMsgErrors(t *testing.T) {
logInit()
eth := newEth(t)
td := common.Big1
currentBlock := []byte{1}
genesis := []byte{2}
eth.chainManager.status = func() (*big.Int, []byte, []byte) { return td, currentBlock, genesis }
go eth.run()
statusMsg := p2p.NewMsg(4)
eth.In(statusMsg)
delay := 1 * time.Second
eth.checkError(ErrNoStatusMsg, delay)
var status statusMsgData
eth.checkMsg(0, StatusMsg, &status) // first outgoing msg should be StatusMsg
if status.TD.Cmp(td) != 0 ||
status.ProtocolVersion != ProtocolVersion ||
status.NetworkId != NetworkId ||
status.TD.Cmp(td) != 0 ||
bytes.Compare(status.CurrentBlock, currentBlock) != 0 ||
bytes.Compare(status.GenesisBlock, genesis) != 0 {
t.Errorf("incorrect outgoing status")
}
eth.reset()
currentBlock := common.Hash{1}
genesis := common.Hash{2}
eth.chainManager.status = func() (*big.Int, common.Hash, common.Hash) { return td, currentBlock, genesis }
go eth.run()
statusMsg = p2p.NewMsg(0, uint32(48), uint32(0), td, currentBlock, genesis)
eth.In(statusMsg)
eth.checkError(ErrProtocolVersionMismatch, delay)
eth.reset()
go eth.run()
statusMsg = p2p.NewMsg(0, uint32(49), uint32(1), td, currentBlock, genesis)
eth.In(statusMsg)
eth.checkError(ErrNetworkIdMismatch, delay)
tests := []struct {
code uint64
data interface{}
wantErrorCode int
}{
{
code: TxMsg, data: []interface{}{},
wantErrorCode: ErrNoStatusMsg,
},
{
code: StatusMsg, data: statusMsgData{10, NetworkId, td, currentBlock, genesis},
wantErrorCode: ErrProtocolVersionMismatch,
},
{
code: StatusMsg, data: statusMsgData{ProtocolVersion, 999, td, currentBlock, genesis},
wantErrorCode: ErrNetworkIdMismatch,
},
{
code: StatusMsg, data: statusMsgData{ProtocolVersion, NetworkId, td, currentBlock, common.Hash{3}},
wantErrorCode: ErrGenesisBlockMismatch,
},
}
for _, test := range tests {
// first outgoing msg should be StatusMsg.
err := p2p.ExpectMsg(eth, StatusMsg, &statusMsgData{
ProtocolVersion: ProtocolVersion,
NetworkId: NetworkId,
TD: td,
CurrentBlock: currentBlock,
GenesisBlock: genesis,
})
if err != nil {
t.Fatalf("incorrect outgoing status: %v", err)
}
eth.reset()
go eth.run()
statusMsg = p2p.NewMsg(0, uint32(49), uint32(0), td, currentBlock, []byte{3})
eth.In(statusMsg)
eth.checkError(ErrGenesisBlockMismatch, delay)
// the send call might hang until reset because
// the protocol might not read the payload.
go p2p.Send(eth, test.code, test.data)
eth.checkError(test.wantErrorCode, 1*time.Second)
eth.reset()
go eth.run()
}
}

2
whisper/envelope.go
Unescape View File

@ -6,9 +6,9 @@ import (
"fmt"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/crypto/ecies"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/rlp"
)

9
whisper/peer.go
Unescape View File

@ -77,12 +77,11 @@ func (self *peer) broadcast(envelopes []*Envelope) error {
}
if i > 0 {
if err := p2p.EncodeMsg(self.ws, envelopesMsg, envs[:i]...); err != nil {
if err := p2p.Send(self.ws, envelopesMsg, envs[:i]); err != nil {
return err
}
self.peer.DebugDetailln("broadcasted", i, "message(s)")
}
return nil
}
@ -92,7 +91,7 @@ func (self *peer) addKnown(envelope *Envelope) {
func (self *peer) handleStatus() error {
ws := self.ws
if err := ws.WriteMsg(self.statusMsg()); err != nil {
if err := p2p.SendItems(ws, statusMsg, protocolVersion); err != nil {
return err
}
msg, err := ws.ReadMsg()
@ -115,7 +114,3 @@ func (self *peer) handleStatus() error {
}
return msg.Discard() // ignore anything after protocol version
}
func (self *peer) statusMsg() p2p.Msg {
return p2p.NewMsg(statusMsg, protocolVersion)
}

Write Preview
Loading…
Cancel
Save