From 0a1eeca41e6ba5920ba65d9b41654768299bc7e3 Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 18 Mar 2015 13:00:01 +0100 Subject: conversions. -compilable- --- eth/backend.go | 10 +++++----- eth/protocol.go | 61 +++++++++++++++++++++++++++++---------------------------- 2 files changed, 36 insertions(+), 35 deletions(-) (limited to 'eth') diff --git a/eth/backend.go b/eth/backend.go index 18093008b..c26bdceb5 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -10,11 +10,11 @@ import ( "github.com/ethereum/ethash" "github.com/ethereum/go-ethereum/accounts" "github.com/ethereum/go-ethereum/blockpool" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/ethdb" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/miner" @@ -288,7 +288,7 @@ func (s *Ethereum) StartMining() error { servlogger.Errorf("Cannot start mining without coinbase: %v\n", err) return fmt.Errorf("no coinbase: %v", err) } - s.miner.Start(cb) + s.miner.Start(common.BytesToAddress(cb)) return nil } @@ -304,9 +304,9 @@ func (s *Ethereum) TxPool() *core.TxPool { return s.txPool } func (s *Ethereum) BlockPool() *blockpool.BlockPool { return s.blockPool } func (s *Ethereum) Whisper() *whisper.Whisper { return s.whisper } func (s *Ethereum) EventMux() *event.TypeMux { return s.eventMux } -func (s *Ethereum) BlockDb() common.Database { return s.blockDb } -func (s *Ethereum) StateDb() common.Database { return s.stateDb } -func (s *Ethereum) ExtraDb() common.Database { return s.extraDb } +func (s *Ethereum) BlockDb() common.Database { return s.blockDb } +func (s *Ethereum) StateDb() common.Database { return s.stateDb } +func (s *Ethereum) ExtraDb() common.Database { return s.extraDb } func (s *Ethereum) IsListening() bool { return true } // Always listening func (s *Ethereum) PeerCount() int { return s.net.PeerCount() } func (s *Ethereum) PeerInfo() int { return s.net.PeerCount() } diff --git a/eth/protocol.go b/eth/protocol.go index e368bbec5..15a8e1831 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -1,13 +1,12 @@ package eth import ( - "bytes" "fmt" "math/big" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/errs" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rlp" @@ -76,15 +75,15 @@ type txPool interface { } type chainManager interface { - GetBlockHashesFromHash(hash []byte, amount uint64) (hashes [][]byte) - GetBlock(hash []byte) (block *types.Block) - Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) + GetBlockHashesFromHash(hash common.Hash, amount uint64) (hashes []common.Hash) + GetBlock(hash common.Hash) (block *types.Block) + Status() (td *big.Int, currentBlock common.Hash, genesisBlock common.Hash) } type blockPool interface { - AddBlockHashes(next func() ([]byte, bool), peerId string) + AddBlockHashes(next func() (common.Hash, 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(*errs.Error)) (best bool) + AddPeer(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(peerId string) } @@ -95,7 +94,7 @@ type newBlockMsgData struct { } type getBlockHashesMsgData struct { - Hash []byte + Hash common.Hash Amount uint64 } @@ -167,7 +166,7 @@ func (self *ethProtocol) handle() error { } for _, tx := range txs { jsonlogger.LogJson(&logger.EthTxReceived{ - TxHash: common.Bytes2Hex(tx.Hash()), + TxHash: tx.Hash().Hex(), RemoteId: self.peer.ID().String(), }) } @@ -183,7 +182,7 @@ func (self *ethProtocol) handle() error { request.Amount = maxHashes } hashes := self.chainManager.GetBlockHashesFromHash(request.Hash, request.Amount) - return p2p.EncodeMsg(self.rw, BlockHashesMsg, common.ByteSliceToInterface(hashes)...) + return p2p.EncodeMsg(self.rw, BlockHashesMsg, rlp.Flat(hashes)) case BlockHashesMsg: msgStream := rlp.NewStream(msg.Payload) @@ -192,14 +191,16 @@ func (self *ethProtocol) handle() error { } var i int - iter := func() (hash []byte, ok bool) { - hash, err := msgStream.Bytes() + iter := func() (hash common.Hash, ok bool) { + var h common.Hash + err := msgStream.Decode(&h) if err == rlp.EOL { - return nil, false + return common.Hash{}, false } else if err != nil { self.protoError(ErrDecode, "msg %v: after %v hashes : %v", msg, i, err) - return nil, false + return common.Hash{}, false } + i++ return hash, true } @@ -215,14 +216,14 @@ func (self *ethProtocol) handle() error { var i int for { i++ - var hash []byte - if err := msgStream.Decode(&hash); err != nil { - if err == rlp.EOL { - break - } else { - return self.protoError(ErrDecode, "msg %v: %v", msg, err) - } + var hash common.Hash + err := msgStream.Decode(&hash) + if err == rlp.EOL { + break + } else { + return self.protoError(ErrDecode, "msg %v: %v", msg, err) } + block := self.chainManager.GetBlock(hash) if block != nil { blocks = append(blocks, block) @@ -259,10 +260,10 @@ func (self *ethProtocol) handle() error { _, chainHead, _ := self.chainManager.Status() jsonlogger.LogJson(&logger.EthChainReceivedNewBlock{ - BlockHash: common.Bytes2Hex(hash), + BlockHash: hash.Hex(), BlockNumber: request.Block.Number(), // this surely must be zero - ChainHeadHash: common.Bytes2Hex(chainHead), - BlockPrevHash: common.Bytes2Hex(request.Block.ParentHash()), + ChainHeadHash: chainHead.Hex(), + BlockPrevHash: request.Block.ParentHash().Hex(), RemoteId: self.peer.ID().String(), }) // to simplify backend interface adding a new block @@ -282,8 +283,8 @@ type statusMsgData struct { ProtocolVersion uint32 NetworkId uint32 TD *big.Int - CurrentBlock []byte - GenesisBlock []byte + CurrentBlock common.Hash + GenesisBlock common.Hash } func (self *ethProtocol) statusMsg() p2p.Msg { @@ -325,7 +326,7 @@ func (self *ethProtocol) handleStatus() error { _, _, genesisBlock := self.chainManager.Status() - if !bytes.Equal(status.GenesisBlock, genesisBlock) { + if status.GenesisBlock != genesisBlock { return self.protoError(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, genesisBlock) } @@ -344,14 +345,14 @@ func (self *ethProtocol) handleStatus() error { return nil } -func (self *ethProtocol) requestBlockHashes(from []byte) 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)) } -func (self *ethProtocol) requestBlocks(hashes [][]byte) error { +func (self *ethProtocol) requestBlocks(hashes []common.Hash) error { self.peer.Debugf("fetching %v blocks", len(hashes)) - return p2p.EncodeMsg(self.rw, GetBlocksMsg, common.ByteSliceToInterface(hashes)...) + return p2p.EncodeMsg(self.rw, GetBlocksMsg, rlp.Flat(hashes)) } func (self *ethProtocol) protoError(code int, format string, params ...interface{}) (err *errs.Error) { -- cgit v1.2.3 From e80dda605130479086bed363e395a102a89a574a Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Thu, 19 Mar 2015 15:18:31 +0100 Subject: eth, whisper: adapt for RLP encoder switch in package p2p I have rewritten the protocol test to use p2p.MsgPipe because p2p.NewMsg is gone. --- eth/backend.go | 4 +- eth/protocol.go | 95 ++++++++++++-------------- eth/protocol_test.go | 189 ++++++++++++++++++++------------------------------- 3 files changed, 121 insertions(+), 167 deletions(-) (limited to 'eth') diff --git a/eth/backend.go b/eth/backend.go index 9e5324688..afe314d74 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -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}) } } } diff --git a/eth/protocol.go b/eth/protocol.go index a1c2ae688..6d610a663 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -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...)) -} diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 108fb4475..7620b3854 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -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 ðProtocolTester{ - 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() + } } -- cgit v1.2.3 From 50661f0e683b4975894a0e8fe16024724adef72d Mon Sep 17 00:00:00 2001 From: zelig Date: Thu, 19 Mar 2015 22:46:54 +0000 Subject: peer suspension to disallow reconnect after disconnect on fatal error for set period (PeerSuspensionInterval) --- eth/protocol.go | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) (limited to 'eth') diff --git a/eth/protocol.go b/eth/protocol.go index 6d610a663..1999d9807 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -42,6 +42,7 @@ const ( ErrGenesisBlockMismatch ErrNoStatusMsg ErrExtraStatusMsg + ErrSuspendedPeer ) var errorToString = map[int]string{ @@ -53,6 +54,7 @@ var errorToString = map[int]string{ ErrGenesisBlockMismatch: "Genesis block mismatch", ErrNoStatusMsg: "No status message", ErrExtraStatusMsg: "Extra status message", + ErrSuspendedPeer: "Suspended peer", } // ethProtocol represents the ethereum wire protocol @@ -85,7 +87,7 @@ type chainManager interface { type blockPool interface { AddBlockHashes(next func() (common.Hash, bool), peerId string) AddBlock(block *types.Block, peerId string) - AddPeer(td *big.Int, currentBlock common.Hash, peerId string, requestHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool) + AddPeer(td *big.Int, currentBlock common.Hash, peerId string, requestHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool, suspended bool) RemovePeer(peerId string) } @@ -288,7 +290,7 @@ func (self *ethProtocol) handle() error { // to simplify backend interface adding a new block // uses AddPeer followed by AddBlock only if peer is the best peer // (or selected as new best peer) - if self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) { + if best, _ := self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect); best { self.blockPool.AddBlock(request.Block, self.id) } @@ -334,9 +336,12 @@ func (self *ethProtocol) handleStatus() error { return self.protoError(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, self.protocolVersion) } - self.peer.Infof("Peer is [eth] capable (%d/%d). TD=%v H=%x\n", status.ProtocolVersion, status.NetworkId, status.TD, status.CurrentBlock[:4]) + _, suspended := self.blockPool.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) + if suspended { + return self.protoError(ErrSuspendedPeer, "") + } - self.blockPool.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) + self.peer.Infof("Peer is [eth] capable (%d/%d). TD=%v H=%x\n", status.ProtocolVersion, status.NetworkId, status.TD, status.CurrentBlock[:4]) return nil } -- cgit v1.2.3 From 391e89d70a43b4a2153db8acac9a6af7a4f76adf Mon Sep 17 00:00:00 2001 From: zelig Date: Thu, 19 Mar 2015 22:53:15 +0000 Subject: use own total difficulty to limit best peer - update blockpool td by subscribing to ChainHeadEvent - if ahead of best peer, demote it - addPeer now take own td as current td - removePeer now take own td as current td - add relevant tests to peers_test - eth: backend now calls blockpool with eth.eventMux and chainManager.Td --- eth/backend.go | 3 ++- eth/protocol_test.go | 13 +++---------- 2 files changed, 5 insertions(+), 11 deletions(-) (limited to 'eth') diff --git a/eth/backend.go b/eth/backend.go index afe314d74..141c6c605 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -195,7 +195,8 @@ func New(config *Config) (*Ethereum, error) { hasBlock := eth.chainManager.HasBlock insertChain := eth.chainManager.InsertChain - eth.blockPool = blockpool.New(hasBlock, insertChain, eth.pow.Verify) + td := eth.chainManager.Td() + eth.blockPool = blockpool.New(hasBlock, insertChain, eth.pow.Verify, eth.EventMux(), td) netprv, err := config.nodeKey() if err != nil { diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 7620b3854..8ca6d1be6 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -41,17 +41,10 @@ type testChainManager struct { type testBlockPool struct { addBlockHashes func(next func() (common.Hash, bool), peerId string) addBlock func(block *types.Block, peerId string) (err error) - 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) + 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, suspended bool) removePeer func(peerId string) } -// func (self *testTxPool) GetTransactions() (txs []*types.Transaction) { -// if self.getTransactions != nil { -// txs = self.getTransactions() -// } -// return -// } - func (self *testTxPool) AddTransactions(txs []*types.Transaction) { if self.addTransactions != nil { self.addTransactions(txs) @@ -93,9 +86,9 @@ func (self *testBlockPool) AddBlock(block *types.Block, peerId string) { } } -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) { +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, suspended bool) { if self.addPeer != nil { - best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, peerError) + best, suspended = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, peerError) } return } -- cgit v1.2.3 From 0578df9467bb00be967da7798cc0ea8b6e7e48d7 Mon Sep 17 00:00:00 2001 From: zelig Date: Thu, 19 Mar 2015 13:02:56 +0000 Subject: remove eth/wallet.go (only commented out content) --- eth/wallet.go | 80 ----------------------------------------------------------- 1 file changed, 80 deletions(-) delete mode 100644 eth/wallet.go (limited to 'eth') diff --git a/eth/wallet.go b/eth/wallet.go deleted file mode 100644 index 9ec834309..000000000 --- a/eth/wallet.go +++ /dev/null @@ -1,80 +0,0 @@ -package eth - -/* -import ( - "crypto/ecdsa" - "errors" - "math/big" - - "github.com/ethereum/go-ethereum/core" - "github.com/ethereum/go-ethereum/core/types" -) - -type Account struct { - w *Wallet -} - -func (self *Account) Transact(to *Account, value, gas, price *big.Int, data []byte) error { - return self.w.transact(self, to, value, gas, price, data) -} - -func (self *Account) Address() []byte { - return nil -} - -func (self *Account) PrivateKey() *ecdsa.PrivateKey { - return nil -} - -type Wallet struct{} - -func NewWallet() *Wallet { - return &Wallet{} -} - -func (self *Wallet) GetAccount(i int) *Account { -} - -func (self *Wallet) transact(from, to *Account, value, gas, price *big.Int, data []byte) error { - if from.PrivateKey() == nil { - return errors.New("accounts is not owned (no private key available)") - } - - var createsContract bool - if to == nil { - createsContract = true - } - - var msg *types.Transaction - if contractCreation { - msg = types.NewContractCreationTx(value, gas, price, data) - } else { - msg = types.NewTransactionMessage(to.Address(), value, gas, price, data) - } - - state := self.chainManager.TransState() - nonce := state.GetNonce(key.Address()) - - msg.SetNonce(nonce) - msg.SignECDSA(from.PriateKey()) - - // Do some pre processing for our "pre" events and hooks - block := self.chainManager.NewBlock(from.Address()) - coinbase := state.GetOrNewStateObject(from.Address()) - coinbase.SetGasPool(block.GasLimit()) - self.blockManager.ApplyTransactions(coinbase, state, block, types.Transactions{tx}, true) - - err := self.obj.TxPool().Add(tx) - if err != nil { - return nil, err - } - state.SetNonce(key.Address(), nonce+1) - - if contractCreation { - addr := core.AddressFromMessage(tx) - pipelogger.Infof("Contract addr %x\n", addr) - } - - return tx, nil -} -*/ -- cgit v1.2.3 From 54dac59285ccc6a3af47201479ca556da2899e93 Mon Sep 17 00:00:00 2001 From: obscuren Date: Fri, 20 Mar 2015 17:42:09 +0100 Subject: wip --- eth/backend.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'eth') diff --git a/eth/backend.go b/eth/backend.go index afe314d74..52f336c1d 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -205,7 +205,7 @@ func New(config *Config) (*Ethereum, error) { ethProto := EthProtocol(config.ProtocolVersion, config.NetworkId, eth.txPool, eth.chainManager, eth.blockPool) protocols := []p2p.Protocol{ethProto} if config.Shh { - protocols = append(protocols, eth.whisper.Protocol()) + //protocols = append(protocols, eth.whisper.Protocol()) } eth.net = &p2p.Server{ -- cgit v1.2.3 From ecd10d2cf765072cd74347b9e0ca2bb85091450f Mon Sep 17 00:00:00 2001 From: obscuren Date: Fri, 20 Mar 2015 18:00:54 +0100 Subject: iterator returned wrong value --- eth/protocol.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) (limited to 'eth') diff --git a/eth/protocol.go b/eth/protocol.go index 1999d9807..494c1c1bb 100644 --- a/eth/protocol.go +++ b/eth/protocol.go @@ -213,8 +213,7 @@ func (self *ethProtocol) handle() error { var i int iter := func() (hash common.Hash, ok bool) { - var h common.Hash - err := msgStream.Decode(&h) + err := msgStream.Decode(&hash) if err == rlp.EOL { return common.Hash{}, false } else if err != nil { -- cgit v1.2.3 From 7c4ff3abb4693bf93b2b348f572f14f2cfcf9142 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Sat, 21 Mar 2015 00:50:20 +0100 Subject: eth: enable whisper again --- eth/backend.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'eth') diff --git a/eth/backend.go b/eth/backend.go index b086d6a56..141c6c605 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -206,7 +206,7 @@ func New(config *Config) (*Ethereum, error) { ethProto := EthProtocol(config.ProtocolVersion, config.NetworkId, eth.txPool, eth.chainManager, eth.blockPool) protocols := []p2p.Protocol{ethProto} if config.Shh { - //protocols = append(protocols, eth.whisper.Protocol()) + protocols = append(protocols, eth.whisper.Protocol()) } eth.net = &p2p.Server{ -- cgit v1.2.3