From 74591a5462d3402e7092dfaeacff72ea48baed8a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 6 Jul 2015 14:00:27 +0300 Subject: [PATCH] eth, eth/downloader: support receipt retrieval (eth/63) --- eth/downloader/downloader.go | 11 ++++--- eth/handler.go | 42 ++++++++++++++++++++---- eth/handler_test.go | 63 ++++++++++++++++++++++++++++++++++-- eth/peer.go | 12 +++++++ 4 files changed, 114 insertions(+), 14 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index bd45af655c..6a6bce6445 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -42,11 +42,12 @@ const ( ) var ( - MinHashFetch = 512 // Minimum amount of hashes to not consider a peer stalling - MaxHashFetch = 512 // Amount of hashes to be fetched per retrieval request - MaxBlockFetch = 128 // Amount of blocks to be fetched per retrieval request - MaxHeaderFetch = 256 // Amount of block headers to be fetched per retrieval request - MaxStateFetch = 384 // Amount of node + MinHashFetch = 512 // Minimum amount of hashes to not consider a peer stalling + MaxHashFetch = 512 // Amount of hashes to be fetched per retrieval request + MaxBlockFetch = 128 // Amount of blocks to be fetched per retrieval request + MaxHeaderFetch = 256 // Amount of block headers to be fetched per retrieval request + MaxStateFetch = 384 // Amount of node state values to allow fetching per request + MaxReceiptsFetch = 384 // Amount of transaction receipts to allow fetching per request hashTTL = 5 * time.Second // Time it takes for a hash request to time out blockSoftTTL = 3 * time.Second // Request completion threshold for increasing or decreasing a peer's bandwidth diff --git a/eth/handler.go b/eth/handler.go index accc09f5ca..3f4021ea95 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -57,9 +57,10 @@ func (ep extProt) GetHashes(hash common.Hash) error { return ep.getHashes(has func (ep extProt) GetBlock(hashes []common.Hash) error { return ep.getBlocks(hashes) } type ProtocolManager struct { - txpool txPool - chainman *core.ChainManager - statedb common.Database + txpool txPool + chainman *core.ChainManager + chaindb common.Database + downloader *downloader.Downloader fetcher *fetcher.Fetcher peers *peerSet @@ -83,13 +84,13 @@ type ProtocolManager struct { // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable // with the ethereum network. -func NewProtocolManager(networkId int, mux *event.TypeMux, txpool txPool, pow pow.PoW, chainman *core.ChainManager, statedb common.Database) *ProtocolManager { +func NewProtocolManager(networkId int, mux *event.TypeMux, txpool txPool, pow pow.PoW, chainman *core.ChainManager, chaindb common.Database) *ProtocolManager { // Create the protocol manager with the base fields manager := &ProtocolManager{ eventMux: mux, txpool: txpool, chainman: chainman, - statedb: statedb, + chaindb: chaindb, peers: newPeerSet(), newPeerCh: make(chan *peer, 1), txsyncCh: make(chan *txsync), @@ -375,20 +376,47 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { data [][]byte ) for bytes < softResponseLimit && len(data) < downloader.MaxStateFetch { - //Retrieve the hash of the next state entry + // Retrieve the hash of the next state entry if err := msgStream.Decode(&hash); err == rlp.EOL { break } else if err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } // Retrieve the requested state entry, stopping if enough was found - if entry, err := pm.statedb.Get(hash.Bytes()); err == nil { + if entry, err := pm.chaindb.Get(hash.Bytes()); err == nil { data = append(data, entry) bytes += len(entry) } } return p.SendNodeData(data) + case p.version == eth63 && msg.Code == GetReceiptsMsg: + // Decode the retrieval message + msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) + if _, err := msgStream.List(); err != nil { + return err + } + // Gather state data until the fetch or network limits is reached + var ( + hash common.Hash + bytes int + receipts []*types.Receipt + ) + for bytes < softResponseLimit && len(receipts) < downloader.MaxReceiptsFetch { + // Retrieve the hash of the next transaction receipt + if err := msgStream.Decode(&hash); err == rlp.EOL { + break + } else if err != nil { + return errResp(ErrDecode, "msg %v: %v", msg, err) + } + // Retrieve the requested receipt, stopping if enough was found + if receipt := core.GetReceipt(pm.chaindb, hash); receipt != nil { + receipts = append(receipts, receipt) + bytes += len(receipt.RlpEncode()) + } + } + return p.SendReceipts(receipts) + case msg.Code == NewBlockHashesMsg: // Retrieve and deseralize the remote new block hashes notification var hashes []common.Hash diff --git a/eth/handler_test.go b/eth/handler_test.go index 84c977b0e3..1acbdfc7a0 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -283,9 +283,9 @@ func TestGetNodeData63(t *testing.T) { peer, _ := newTestPeer("peer", 63, pm, true) defer peer.close() - // Fetch for now the entire state db + // Fetch for now the entire chain db hashes := []common.Hash{} - for _, key := range pm.statedb.(*ethdb.MemDatabase).Keys() { + for _, key := range pm.chaindb.(*ethdb.MemDatabase).Keys() { hashes = append(hashes, common.BytesToHash(key)) } p2p.Send(peer.app, 0x0d, hashes) @@ -327,3 +327,62 @@ func TestGetNodeData63(t *testing.T) { } } } + +// Tests that the transaction receipts can be retrieved based on hashes. +func TestGetReceipts63(t *testing.T) { + // Define three accounts to simulate transactions with + acc1Key, _ := crypto.HexToECDSA("8a1f9a8f95be41cd7ccb6168179afb4504aefe388d1e14474d32c45c72ce7b7a") + acc2Key, _ := crypto.HexToECDSA("49a7b37aa6f6645917e7b807e9d1c00d4fa71f18343b0d4122a4d2df64dd6fee") + acc1Addr := crypto.PubkeyToAddress(acc1Key.PublicKey) + acc2Addr := crypto.PubkeyToAddress(acc2Key.PublicKey) + + // Create a chain generator with some simple transactions (blatantly stolen from @fjl/chain_makerts_test) + generator := func(i int, block *core.BlockGen) { + switch i { + case 0: + // In block 1, the test bank sends account #1 some ether. + tx, _ := types.NewTransaction(block.TxNonce(testBankAddress), acc1Addr, big.NewInt(10000), params.TxGas, nil, nil).SignECDSA(testBankKey) + block.AddTx(tx) + case 1: + // In block 2, the test bank sends some more ether to account #1. + // acc1Addr passes it on to account #2. + tx1, _ := types.NewTransaction(block.TxNonce(testBankAddress), acc1Addr, big.NewInt(1000), params.TxGas, nil, nil).SignECDSA(testBankKey) + tx2, _ := types.NewTransaction(block.TxNonce(acc1Addr), acc2Addr, big.NewInt(1000), params.TxGas, nil, nil).SignECDSA(acc1Key) + block.AddTx(tx1) + block.AddTx(tx2) + case 2: + // Block 3 is empty but was mined by account #2. + block.SetCoinbase(acc2Addr) + block.SetExtra([]byte("yeehaw")) + case 3: + // Block 4 includes blocks 2 and 3 as uncle headers (with modified extra data). + b2 := block.PrevBlock(1).Header() + b2.Extra = []byte("foo") + block.AddUncle(b2) + b3 := block.PrevBlock(2).Header() + b3.Extra = []byte("foo") + block.AddUncle(b3) + } + } + // Assemble the test environment + pm := newTestProtocolManager(4, generator, nil) + peer, _ := newTestPeer("peer", 63, pm, true) + defer peer.close() + + // Collect the hashes to request, and the response to expect + hashes := []common.Hash{} + for i := uint64(0); i <= pm.chainman.CurrentBlock().NumberU64(); i++ { + for _, tx := range pm.chainman.GetBlockByNumber(i).Transactions() { + hashes = append(hashes, tx.Hash()) + } + } + receipts := make([]*types.Receipt, len(hashes)) + for i, hash := range hashes { + receipts[i] = core.GetReceipt(pm.chaindb, hash) + } + // Send the hash request and verify the response + p2p.Send(peer.app, 0x0f, hashes) + if err := p2p.ExpectMsg(peer.app, 0x10, receipts); err != nil { + t.Errorf("receipts mismatch: %v", err) + } +} diff --git a/eth/peer.go b/eth/peer.go index c17cdfca78..65159a7d10 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -171,6 +171,12 @@ func (p *peer) SendNodeData(data [][]byte) error { return p2p.Send(p.rw, NodeDataMsg, data) } +// SendReceipts sends a batch of transaction receipts, corresponding to the ones +// requested. +func (p *peer) SendReceipts(receipts []*types.Receipt) error { + return p2p.Send(p.rw, ReceiptsMsg, receipts) +} + // RequestHashes fetches a batch of hashes from a peer, starting at from, going // towards the genesis block. func (p *peer) RequestHashes(from common.Hash) error { @@ -205,6 +211,12 @@ func (p *peer) RequestNodeData(hashes []common.Hash) error { return p2p.Send(p.rw, GetNodeDataMsg, hashes) } +// RequestReceipts fetches a batch of transaction receipts from a remote node. +func (p *peer) RequestReceipts(hashes []common.Hash) error { + glog.V(logger.Debug).Infof("%v fetching %v receipts\n", p, len(hashes)) + return p2p.Send(p.rw, GetReceiptsMsg, hashes) +} + // Handshake executes the eth protocol handshake, negotiating version number, // network IDs, difficulties, head and genesis blocks. func (p *peer) Handshake(td *big.Int, head common.Hash, genesis common.Hash) error {