forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-23 21:26:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

eth, eth/fetcher: add missing eth62 messages, test bodies

Browse source
This commit is contained in:
Péter Szilágyi 2015-08-11 18:34:33 +03:00
parent 74591a5462
commit 2686b20750
7 changed files with 212 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
eth/fetcher/fetcher.go
View file

@ -69,8 +69,9 @@ type peerDropFn func(id string)
// announce is the hash notification of the availability of a new block in the
// network.
type announce struct {
hash common.Hash // Hash of the block being announced
time time.Time // Timestamp of the announcement
hash common.Hash // Hash of the block being announced
number uint64 // Number of the block being announced (0 = unknown | old protocol)
time time.Time // Timestamp of the announcement
origin string // Identifier of the peer originating the notification
fetch blockRequesterFn // Fetcher function to retrieve
@ -152,9 +153,10 @@ func (f *Fetcher) Stop() {
// Notify announces the fetcher of the potential availability of a new block in
// the network.
func (f *Fetcher) Notify(peer string, hash common.Hash, time time.Time, fetcher blockRequesterFn) error {
func (f *Fetcher) Notify(peer string, hash common.Hash, number uint64, time time.Time, fetcher blockRequesterFn) error {
block := &announce{
hash: hash,
number: number,
time: time,
origin: peer,
fetch: fetcher,

26
eth/fetcher/fetcher_test.go
View file

@ -194,7 +194,7 @@ func TestSequentialAnnouncements(t *testing.T) {
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
for i := len(hashes) - 2; i >= 0; i-- {
tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher)
verifyImportEvent(t, imported)
}
verifyImportDone(t, imported)
@ -221,9 +221,9 @@ func TestConcurrentAnnouncements(t *testing.T) {
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
for i := len(hashes) - 2; i >= 0; i-- {
tester.fetcher.Notify("first", hashes[i], time.Now().Add(-arriveTimeout), wrapper)
tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout+time.Millisecond), wrapper)
tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout-time.Millisecond), wrapper)
tester.fetcher.Notify("first", hashes[i], 0, time.Now().Add(-arriveTimeout), wrapper)
tester.fetcher.Notify("second", hashes[i], 0, time.Now().Add(-arriveTimeout+time.Millisecond), wrapper)
tester.fetcher.Notify("second", hashes[i], 0, time.Now().Add(-arriveTimeout-time.Millisecond), wrapper)
verifyImportEvent(t, imported)
}
@ -252,7 +252,7 @@ func TestOverlappingAnnouncements(t *testing.T) {
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
for i := len(hashes) - 2; i >= 0; i-- {
tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher)
select {
case <-fetching:
case <-time.After(time.Second):
@ -286,7 +286,7 @@ func TestPendingDeduplication(t *testing.T) {
}
// Announce the same block many times until it's fetched (wait for any pending ops)
for tester.getBlock(hashes[0]) == nil {
tester.fetcher.Notify("repeater", hashes[0], time.Now().Add(-arriveTimeout), wrapper)
tester.fetcher.Notify("repeater", hashes[0], 0, time.Now().Add(-arriveTimeout), wrapper)
time.Sleep(time.Millisecond)
}
time.Sleep(delay)
@ -317,12 +317,12 @@ func TestRandomArrivalImport(t *testing.T) {
for i := len(hashes) - 1; i >= 0; i-- {
if i != skip {
tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher)
time.Sleep(time.Millisecond)
}
}
// Finally announce the skipped entry and check full import
tester.fetcher.Notify("valid", hashes[skip], time.Now().Add(-arriveTimeout), fetcher)
tester.fetcher.Notify("valid", hashes[skip], 0, time.Now().Add(-arriveTimeout), fetcher)
verifyImportCount(t, imported, len(hashes)-1)
}
@ -343,7 +343,7 @@ func TestQueueGapFill(t *testing.T) {
for i := len(hashes) - 1; i >= 0; i-- {
if i != skip {
tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), fetcher)
time.Sleep(time.Millisecond)
}
}
@ -374,7 +374,7 @@ func TestImportDeduplication(t *testing.T) {
tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
// Announce the duplicating block, wait for retrieval, and also propagate directly
tester.fetcher.Notify("valid", hashes[0], time.Now().Add(-arriveTimeout), fetcher)
tester.fetcher.Notify("valid", hashes[0], 0, time.Now().Add(-arriveTimeout), fetcher)
<-fetching
tester.fetcher.Enqueue("valid", blocks[hashes[0]])
@ -437,9 +437,9 @@ func TestHashMemoryExhaustionAttack(t *testing.T) {
// Feed the tester a huge hashset from the attacker, and a limited from the valid peer
for i := 0; i < len(attack); i++ {
if i < maxQueueDist {
tester.fetcher.Notify("valid", hashes[len(hashes)-2-i], time.Now(), valid)
tester.fetcher.Notify("valid", hashes[len(hashes)-2-i], 0, time.Now(), valid)
}
tester.fetcher.Notify("attacker", attack[i], time.Now(), attacker)
tester.fetcher.Notify("attacker", attack[i], 0, time.Now(), attacker)
}
if len(tester.fetcher.announced) != hashLimit+maxQueueDist {
t.Fatalf("queued announce count mismatch: have %d, want %d", len(tester.fetcher.announced), hashLimit+maxQueueDist)
@ -449,7 +449,7 @@ func TestHashMemoryExhaustionAttack(t *testing.T) {
// Feed the remaining valid hashes to ensure DOS protection state remains clean
for i := len(hashes) - maxQueueDist - 2; i >= 0; i-- {
tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), valid)
tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), valid)
verifyImportEvent(t, imported)
}
verifyImportDone(t, imported)

94
eth/handler.go
View file

@ -237,7 +237,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
// Status messages should never arrive after the handshake
return errResp(ErrExtraStatusMsg, "uncontrolled status message")
case (p.version == eth60 || p.version == eth61) && msg.Code == GetBlockHashesMsg:
case p.version < eth62 && msg.Code == GetBlockHashesMsg:
// Retrieve the number of hashes to return and from which origin hash
var request getBlockHashesData
if err := msg.Decode(&request); err != nil {
@ -253,7 +253,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
return p.SendBlockHashes(hashes)
case (p.version == eth60 || p.version == eth61) && msg.Code == GetBlockHashesFromNumberMsg:
case p.version < eth62 && msg.Code == GetBlockHashesFromNumberMsg:
// Retrieve and decode the number of hashes to return and from which origin number
var request getBlockHashesFromNumberData
if err := msg.Decode(&request); err != nil {
@ -280,7 +280,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
return p.SendBlockHashes(hashes)
case (p.version == eth60 || p.version == eth61) && msg.Code == BlockHashesMsg:
case p.version < eth62 && msg.Code == BlockHashesMsg:
// A batch of hashes arrived to one of our previous requests
var hashes []common.Hash
if err := msg.Decode(&hashes); err != nil {
@ -292,7 +292,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
glog.V(logger.Debug).Infoln(err)
}
case (p.version == eth60 || p.version == eth61) && msg.Code == GetBlocksMsg:
case p.version < eth62 && msg.Code == GetBlocksMsg:
// Decode the retrieval message
msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
if _, err := msgStream.List(); err != nil {
@ -320,7 +320,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
return p.SendBlocks(blocks)
case (p.version == eth60 || p.version == eth61) && msg.Code == BlocksMsg:
case p.version < eth62 && msg.Code == BlocksMsg:
// Decode the arrived block message
var blocks []*types.Block
if err := msg.Decode(&blocks); err != nil {
@ -336,7 +336,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
pm.downloader.DeliverBlocks(p.id, blocks)
}
case p.version == eth62 && msg.Code == GetBlockHeadersMsg:
case p.version >= eth62 && msg.Code == GetBlockHeadersMsg:
// Decode the retrieval message
msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
if _, err := msgStream.List(); err != nil {
@ -363,7 +363,34 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
return p.SendBlockHeaders(headers)
case p.version == eth63 && msg.Code == GetNodeDataMsg:
case p.version >= eth62 && msg.Code == GetBlockBodiesMsg:
// Decode the retrieval message
msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
if _, err := msgStream.List(); err != nil {
return err
}
// Gather blocks until the fetch or network limits is reached
var (
hash common.Hash
bytes common.StorageSize
bodies []*blockBody
)
for bytes < softResponseLimit && len(bodies) < downloader.MaxBlockFetch {
//Retrieve the hash of the next block
if err := msgStream.Decode(&hash); err == rlp.EOL {
break
} else if err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
// Retrieve the requested block, stopping if enough was found
if block := pm.chainman.GetBlock(hash); block != nil {
bodies = append(bodies, &blockBody{Transactions: block.Transactions(), Uncles: block.Uncles()})
bytes += block.Size()
}
}
return p.SendBlockBodies(bodies)
case p.version >= eth63 && msg.Code == GetNodeDataMsg:
// Decode the retrieval message
msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
if _, err := msgStream.List(); err != nil {
@ -390,7 +417,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
return p.SendNodeData(data)
case p.version == eth63 && msg.Code == GetReceiptsMsg:
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 {
@ -419,24 +446,45 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
case msg.Code == NewBlockHashesMsg:
// Retrieve and deseralize the remote new block hashes notification
var hashes []common.Hash
if err := msg.Decode(&hashes); err != nil {
break
type announce struct {
Hash common.Hash
Number uint64
}
// Mark the hashes as present at the remote node
for _, hash := range hashes {
p.MarkBlock(hash)
p.SetHead(hash)
}
// Schedule all the unknown hashes for retrieval
unknown := make([]common.Hash, 0, len(hashes))
for _, hash := range hashes {
if !pm.chainman.HasBlock(hash) {
unknown = append(unknown, hash)
var announces = []announce{}
if p.version < eth62 {
// We're running the old protocol, make block number unknown (0)
var hashes []common.Hash
if err := msg.Decode(&hashes); err != nil {
return errResp(ErrDecode, "%v: %v", msg, err)
}
for _, hash := range hashes {
announces = append(announces, announce{hash, 0})
}
} else {
// Otherwise extract both block hash and number
var request newBlockHashesData
if err := msg.Decode(&request); err != nil {
return errResp(ErrDecode, "%v: %v", msg, err)
}
for _, block := range request {
announces = append(announces, announce{block.Hash, block.Number})
}
}
for _, hash := range unknown {
pm.fetcher.Notify(p.id, hash, time.Now(), p.RequestBlocks)
// Mark the hashes as present at the remote node
for _, block := range announces {
p.MarkBlock(block.Hash)
p.SetHead(block.Hash)
}
// Schedule all the unknown hashes for retrieval
unknown := make([]announce, 0, len(announces))
for _, block := range announces {
if !pm.chainman.HasBlock(block.Hash) {
unknown = append(unknown, block)
}
}
for _, block := range unknown {
pm.fetcher.Notify(p.id, block.Hash, block.Number, time.Now(), p.RequestBlocks)
}
case msg.Code == NewBlockMsg:

95
eth/handler_test.go
View file

@ -175,9 +175,13 @@ func testGetBlocks(t *testing.T, protocol int) {
}
// Tests that block headers can be retrieved from a remote chain based on their hashes.
func TestGetBlockHeaders62(t *testing.T) {
func TestGetBlockHeaders62(t *testing.T) { testGetBlockHeaders(t, 62) }
func TestGetBlockHeaders63(t *testing.T) { testGetBlockHeaders(t, 63) }
func TestGetBlockHeaders64(t *testing.T) { testGetBlockHeaders(t, 64) }
func testGetBlockHeaders(t *testing.T, protocol int) {
pm := newTestProtocolManager(downloader.MaxHashFetch+15, nil, nil)
peer, _ := newTestPeer("peer", 62, pm, true)
peer, _ := newTestPeer("peer", protocol, pm, true)
defer peer.close()
// Create a batch of tests for various scenarios
@ -242,8 +246,84 @@ func TestGetBlockHeaders62(t *testing.T) {
}
}
// Tests that block contents can be retrieved from a remote chain based on their hashes.
func TestGetBlockBodies62(t *testing.T) { testGetBlockBodies(t, 62) }
func TestGetBlockBodies63(t *testing.T) { testGetBlockBodies(t, 63) }
func TestGetBlockBodies64(t *testing.T) { testGetBlockBodies(t, 64) }
func testGetBlockBodies(t *testing.T, protocol int) {
pm := newTestProtocolManager(downloader.MaxBlockFetch+15, nil, nil)
peer, _ := newTestPeer("peer", protocol, pm, true)
defer peer.close()
// Create a batch of tests for various scenarios
limit := downloader.MaxBlockFetch
tests := []struct {
random int // Number of blocks to fetch randomly from the chain
explicit []common.Hash // Explicitly requested blocks
available []bool // Availability of explicitly requested blocks
expected int // Total number of existing blocks to expect
}{
{1, nil, nil, 1}, // A single random block should be retrievable
{10, nil, nil, 10}, // Multiple random blocks should be retrievable
{limit, nil, nil, limit}, // The maximum possible blocks should be retrievable
{limit + 1, nil, nil, limit}, // No more that the possible block count should be returned
{0, []common.Hash{pm.chainman.Genesis().Hash()}, []bool{true}, 1}, // The genesis block should be retrievable
{0, []common.Hash{pm.chainman.CurrentBlock().Hash()}, []bool{true}, 1}, // The chains head block should be retrievable
{0, []common.Hash{common.Hash{}}, []bool{false}, 0}, // A non existent block should not be returned
// Existing and non-existing blocks interleaved should not cause problems
{0, []common.Hash{
common.Hash{},
pm.chainman.GetBlockByNumber(1).Hash(),
common.Hash{},
pm.chainman.GetBlockByNumber(10).Hash(),
common.Hash{},
pm.chainman.GetBlockByNumber(100).Hash(),
common.Hash{},
}, []bool{false, true, false, true, false, true, false}, 3},
}
// Run each of the tests and verify the results against the chain
for i, tt := range tests {
// Collect the hashes to request, and the response to expect
hashes, seen := []common.Hash{}, make(map[int64]bool)
bodies := []*blockBody{}
for j := 0; j < tt.random; j++ {
for {
num := rand.Int63n(int64(pm.chainman.CurrentBlock().NumberU64()))
if !seen[num] {
seen[num] = true
block := pm.chainman.GetBlockByNumber(uint64(num))
hashes = append(hashes, block.Hash())
if len(bodies) < tt.expected {
bodies = append(bodies, &blockBody{Transactions: block.Transactions(), Uncles: block.Uncles()})
}
break
}
}
}
for j, hash := range tt.explicit {
hashes = append(hashes, hash)
if tt.available[j] && len(bodies) < tt.expected {
block := pm.chainman.GetBlock(hash)
bodies = append(bodies, &blockBody{Transactions: block.Transactions(), Uncles: block.Uncles()})
}
}
// Send the hash request and verify the response
p2p.Send(peer.app, 0x05, hashes)
if err := p2p.ExpectMsg(peer.app, 0x06, bodies); err != nil {
t.Errorf("test %d: bodies mismatch: %v", i, err)
}
}
}
// Tests that the node state database can be retrieved based on hashes.
func TestGetNodeData63(t *testing.T) {
func TestGetNodeData63(t *testing.T) { testGetNodeData(t, 63) }
func TestGetNodeData64(t *testing.T) { testGetNodeData(t, 64) }
func testGetNodeData(t *testing.T, protocol int) {
// Define three accounts to simulate transactions with
acc1Key, _ := crypto.HexToECDSA("8a1f9a8f95be41cd7ccb6168179afb4504aefe388d1e14474d32c45c72ce7b7a")
acc2Key, _ := crypto.HexToECDSA("49a7b37aa6f6645917e7b807e9d1c00d4fa71f18343b0d4122a4d2df64dd6fee")
@ -280,7 +360,7 @@ func TestGetNodeData63(t *testing.T) {
}
// Assemble the test environment
pm := newTestProtocolManager(4, generator, nil)
peer, _ := newTestPeer("peer", 63, pm, true)
peer, _ := newTestPeer("peer", protocol, pm, true)
defer peer.close()
// Fetch for now the entire chain db
@ -329,7 +409,10 @@ func TestGetNodeData63(t *testing.T) {
}
// Tests that the transaction receipts can be retrieved based on hashes.
func TestGetReceipts63(t *testing.T) {
func TestGetReceipt63(t *testing.T) { testGetReceipt(t, 63) }
func TestGetReceipt64(t *testing.T) { testGetReceipt(t, 64) }
func testGetReceipt(t *testing.T, protocol int) {
// Define three accounts to simulate transactions with
acc1Key, _ := crypto.HexToECDSA("8a1f9a8f95be41cd7ccb6168179afb4504aefe388d1e14474d32c45c72ce7b7a")
acc2Key, _ := crypto.HexToECDSA("49a7b37aa6f6645917e7b807e9d1c00d4fa71f18343b0d4122a4d2df64dd6fee")
@ -366,7 +449,7 @@ func TestGetReceipts63(t *testing.T) {
}
// Assemble the test environment
pm := newTestProtocolManager(4, generator, nil)
peer, _ := newTestPeer("peer", 63, pm, true)
peer, _ := newTestPeer("peer", protocol, pm, true)
defer peer.close()
// Collect the hashes to request, and the response to expect

5
eth/peer.go
View file

@ -165,6 +165,11 @@ func (p *peer) SendBlockHeaders(headers []*types.Header) error {
return p2p.Send(p.rw, BlockHeadersMsg, headers)
}
// SendBlockBodies sends a batch of block contents to the remote peer.
func (p *peer) SendBlockBodies(bodies []*blockBody) error {
return p2p.Send(p.rw, BlockBodiesMsg, blockBodiesData(bodies))
}
// SendNodeData sends a batch of arbitrary internal data, corresponding to the
// hashes requested.
func (p *peer) SendNodeData(data [][]byte) error {

28
eth/protocol.go
View file

@ -29,6 +29,7 @@ const (
eth61 = 61
eth62 = 62
eth63 = 63
eth64 = 64
)
// Supported versions of the eth protocol (first is primary).
@ -54,10 +55,13 @@ const (
BlocksMsg = 0x06
NewBlockMsg = 0x07
// Protocol messages belonging to eth/61
// Protocol messages belonging to eth/61 (extension of eth/60)
GetBlockHashesFromNumberMsg = 0x08
// Protocol messages belonging to eth/62
// Protocol messages belonging to eth/62 (new protocol from scratch)
// StatusMsg = 0x00 (uncomment after eth/61 deprecation)
// NewBlockHashesMsg = 0x01 (uncomment after eth/61 deprecation)
// TxMsg = 0x02 (uncomment after eth/61 deprecation)
GetBlockHeadersMsg = 0x03
BlockHeadersMsg = 0x04
GetBlockBodiesMsg = 0x05
@ -68,6 +72,11 @@ const (
NodeDataMsg = 0x0e
GetReceiptsMsg = 0x0f
ReceiptsMsg = 0x10
// Protocol messages belonging to eth/64
GetAcctProofMsg = 0x11
GetStorageDataProof = 0x12
Proof = 0x13
)
type errCode int
@ -125,6 +134,12 @@ type statusData struct {
GenesisBlock common.Hash
}
// newBlockHashesData is the network packet for the block announcements.
type newBlockHashesData []struct {
Hash common.Hash // Hash of one particular block being announced
Number uint64 // Number of one particular block being announced
}
// getBlockHashesData is the network packet for the hash based hash retrieval.
type getBlockHashesData struct {
Hash common.Hash
@ -144,6 +159,15 @@ type newBlockData struct {
TD *big.Int
}
// blockBody represents the data content of a single block.
type blockBody struct {
Transactions []*types.Transaction // Transactions contained within a block
Uncles []*types.Header // Uncles contained within a block
}
// blockBodiesData is the network packet for block content distribution.
type blockBodiesData []*blockBody
// nodeDataData is the network response packet for a node data retrieval.
type nodeDataData []struct {
Value []byte

3
eth/protocol_test.go
View file

@ -41,6 +41,7 @@ func TestStatusMsgErrors60(t *testing.T) { testStatusMsgErrors(t, 60) }
func TestStatusMsgErrors61(t *testing.T) { testStatusMsgErrors(t, 61) }
func TestStatusMsgErrors62(t *testing.T) { testStatusMsgErrors(t, 62) }
func TestStatusMsgErrors63(t *testing.T) { testStatusMsgErrors(t, 63) }
func TestStatusMsgErrors64(t *testing.T) { testStatusMsgErrors(t, 64) }
func testStatusMsgErrors(t *testing.T, protocol int) {
pm := newTestProtocolManager(0, nil, nil)
@ -95,6 +96,7 @@ func TestRecvTransactions60(t *testing.T) { testRecvTransactions(t, 60) }
func TestRecvTransactions61(t *testing.T) { testRecvTransactions(t, 61) }
func TestRecvTransactions62(t *testing.T) { testRecvTransactions(t, 62) }
func TestRecvTransactions63(t *testing.T) { testRecvTransactions(t, 63) }
func TestRecvTransactions64(t *testing.T) { testRecvTransactions(t, 64) }
func testRecvTransactions(t *testing.T, protocol int) {
txAdded := make(chan []*types.Transaction)
@ -124,6 +126,7 @@ func TestSendTransactions60(t *testing.T) { testSendTransactions(t, 60) }
func TestSendTransactions61(t *testing.T) { testSendTransactions(t, 61) }
func TestSendTransactions62(t *testing.T) { testSendTransactions(t, 62) }
func TestSendTransactions63(t *testing.T) { testSendTransactions(t, 63) }
func TestSendTransactions64(t *testing.T) { testSendTransactions(t, 64) }
func testSendTransactions(t *testing.T, protocol int) {
pm := newTestProtocolManager(0, nil, nil)