diff --git a/eth/handler.go b/eth/handler.go index e864f796fe..b9b5f6c081 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -45,6 +45,9 @@ import ( const ( softResponseLimit = 2 * 1024 * 1024 // Target maximum size of returned blocks, headers or node data. estHeaderRlpSize = 500 // Approximate size of an RLP encoded block header + + // minimim number of peers to broadcast new blocks to + minBroadcastPeers = 4 ) // errIncompatibleConfig is returned if the requested protocols and configs are @@ -624,7 +627,14 @@ func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) { return } // Send the block to a subset of our peers - transfer := peers[:int(math.Sqrt(float64(len(peers))))] + transferLen := int(math.Sqrt(float64(len(peers)))) + if transferLen < minBroadcastPeers { + transferLen = minBroadcastPeers + } + if transferLen > len(peers) { + transferLen = len(peers) + } + transfer := peers[:transferLen] for _, peer := range transfer { peer.SendNewBlock(block, td) } diff --git a/eth/handler_test.go b/eth/handler_test.go index e8081eada4..8c3a19cf67 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -17,6 +17,7 @@ package eth import ( + "fmt" "math" "math/big" "math/rand" @@ -24,6 +25,7 @@ import ( "time" "github.com/ubiq/go-ubiq/common" + "github.com/ubiq/go-ubiq/consensus/ethash" "github.com/ubiq/go-ubiq/core" "github.com/ubiq/go-ubiq/core/state" "github.com/ubiq/go-ubiq/core/types" @@ -41,11 +43,11 @@ func TestProtocolCompatibility(t *testing.T) { // Define the compatibility chart tests := []struct { version uint - fastSync bool + mode downloader.SyncMode compatible bool }{ - {61, false, true}, {62, false, true}, {63, false, true}, - {61, true, false}, {62, true, false}, {63, true, true}, + {61, downloader.FullSync, true}, {62, downloader.FullSync, true}, {63, downloader.FullSync, true}, + {61, downloader.FastSync, false}, {62, downloader.FastSync, false}, {63, downloader.FastSync, true}, } // Make sure anything we screw up is restored backup := ProtocolVersions @@ -55,7 +57,7 @@ func TestProtocolCompatibility(t *testing.T) { for i, tt := range tests { ProtocolVersions = []uint{tt.version} - pm, err := newTestProtocolManager(tt.fastSync, 0, nil, nil) + pm, _, err := newTestProtocolManager(tt.mode, 0, nil, nil) if pm != nil { defer pm.Stop() } @@ -70,7 +72,7 @@ func TestGetBlockHeaders62(t *testing.T) { testGetBlockHeaders(t, 62) } func TestGetBlockHeaders63(t *testing.T) { testGetBlockHeaders(t, 63) } func testGetBlockHeaders(t *testing.T, protocol int) { - pm := newTestProtocolManagerMust(t, false, downloader.MaxHashFetch+15, nil, nil) + pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, downloader.MaxHashFetch+15, nil, nil) peer, _ := newTestPeer("peer", protocol, pm, true) defer peer.close() @@ -229,7 +231,7 @@ func TestGetBlockBodies62(t *testing.T) { testGetBlockBodies(t, 62) } func TestGetBlockBodies63(t *testing.T) { testGetBlockBodies(t, 63) } func testGetBlockBodies(t *testing.T, protocol int) { - pm := newTestProtocolManagerMust(t, false, downloader.MaxBlockFetch+15, nil, nil) + pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, downloader.MaxBlockFetch+15, nil, nil) peer, _ := newTestPeer("peer", protocol, pm, true) defer peer.close() @@ -241,10 +243,10 @@ func testGetBlockBodies(t *testing.T, protocol int) { 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 than the possible block count should be returned + {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 than the possible block count should be returned {0, []common.Hash{pm.blockchain.Genesis().Hash()}, []bool{true}, 1}, // The genesis block should be retrievable {0, []common.Hash{pm.blockchain.CurrentBlock().Hash()}, []bool{true}, 1}, // The chains head block should be retrievable {0, []common.Hash{{}}, []bool{false}, 0}, // A non existent block should not be returned @@ -312,12 +314,12 @@ func testGetNodeData(t *testing.T, protocol int) { switch i { case 0: // In block 1, the test bank sends account #1 some ether. - tx, _ := types.SignTx(types.NewTransaction(block.TxNonce(testBank.Address), acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), signer, testBankKey) + tx, _ := types.SignTx(types.NewTransaction(block.TxNonce(testBank), acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), signer, 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.SignTx(types.NewTransaction(block.TxNonce(testBank.Address), acc1Addr, big.NewInt(1000), params.TxGas, nil, nil), signer, testBankKey) + tx1, _ := types.SignTx(types.NewTransaction(block.TxNonce(testBank), acc1Addr, big.NewInt(1000), params.TxGas, nil, nil), signer, testBankKey) tx2, _ := types.SignTx(types.NewTransaction(block.TxNonce(acc1Addr), acc2Addr, big.NewInt(1000), params.TxGas, nil, nil), signer, acc1Key) block.AddTx(tx1) block.AddTx(tx2) @@ -336,13 +338,13 @@ func testGetNodeData(t *testing.T, protocol int) { } } // Assemble the test environment - pm := newTestProtocolManagerMust(t, false, 4, generator, nil) + pm, db := newTestProtocolManagerMust(t, downloader.FullSync, 4, generator, nil) peer, _ := newTestPeer("peer", protocol, pm, true) defer peer.close() // Fetch for now the entire chain db hashes := []common.Hash{} - for _, key := range pm.chaindb.(*ethdb.MemDatabase).Keys() { + for _, key := range db.Keys() { if len(key) == len(common.Hash{}) { hashes = append(hashes, common.BytesToHash(key)) } @@ -365,13 +367,13 @@ func testGetNodeData(t *testing.T, protocol int) { t.Errorf("data hash mismatch: have %x, want %x", hash, want) } } - statedb, _ := ethdb.NewMemDatabase() + statedb := ethdb.NewMemDatabase() for i := 0; i < len(data); i++ { statedb.Put(hashes[i].Bytes(), data[i]) } - accounts := []common.Address{testBank.Address, acc1Addr, acc2Addr} + accounts := []common.Address{testBank, acc1Addr, acc2Addr} for i := uint64(0); i <= pm.blockchain.CurrentBlock().NumberU64(); i++ { - trie, _ := state.New(pm.blockchain.GetBlockByNumber(i).Root(), statedb) + trie, _ := state.New(pm.blockchain.GetBlockByNumber(i).Root(), state.NewDatabase(statedb)) for j, acc := range accounts { state, _ := pm.blockchain.State() @@ -404,12 +406,12 @@ func testGetReceipt(t *testing.T, protocol int) { switch i { case 0: // In block 1, the test bank sends account #1 some ether. - tx, _ := types.SignTx(types.NewTransaction(block.TxNonce(testBank.Address), acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), signer, testBankKey) + tx, _ := types.SignTx(types.NewTransaction(block.TxNonce(testBank), acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), signer, 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.SignTx(types.NewTransaction(block.TxNonce(testBank.Address), acc1Addr, big.NewInt(1000), params.TxGas, nil, nil), signer, testBankKey) + tx1, _ := types.SignTx(types.NewTransaction(block.TxNonce(testBank), acc1Addr, big.NewInt(1000), params.TxGas, nil, nil), signer, testBankKey) tx2, _ := types.SignTx(types.NewTransaction(block.TxNonce(acc1Addr), acc2Addr, big.NewInt(1000), params.TxGas, nil, nil), signer, acc1Key) block.AddTx(tx1) block.AddTx(tx2) @@ -428,7 +430,7 @@ func testGetReceipt(t *testing.T, protocol int) { } } // Assemble the test environment - pm := newTestProtocolManagerMust(t, false, 4, generator, nil) + pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 4, generator, nil) peer, _ := newTestPeer("peer", protocol, pm, true) defer peer.close() @@ -438,7 +440,7 @@ func testGetReceipt(t *testing.T, protocol int) { block := pm.blockchain.GetBlockByNumber(i) hashes = append(hashes, block.Hash()) - receipts = append(receipts, core.GetBlockReceipts(pm.chaindb, block.Hash(), block.NumberU64())) + receipts = append(receipts, pm.blockchain.GetReceiptsByHash(block.Hash())) } // Send the hash request and verify the response p2p.Send(peer.app, 0x0f, hashes) @@ -446,3 +448,90 @@ func testGetReceipt(t *testing.T, protocol int) { t.Errorf("receipts mismatch: %v", err) } } + +func TestBroadcastBlock(t *testing.T) { + var tests = []struct { + totalPeers int + broadcastExpected int + }{ + {1, 1}, + {2, 2}, + {3, 3}, + {4, 4}, + {5, 4}, + {9, 4}, + {12, 4}, + {16, 4}, + {26, 5}, + {100, 10}, + } + for _, test := range tests { + testBroadcastBlock(t, test.totalPeers, test.broadcastExpected) + } +} + +func testBroadcastBlock(t *testing.T, totalPeers, broadcastExpected int) { + var ( + evmux = new(event.TypeMux) + pow = ethash.NewFaker() + db = ethdb.NewMemDatabase() + config = ¶ms.ChainConfig{} + gspec = &core.Genesis{Config: config} + genesis = gspec.MustCommit(db) + ) + blockchain, err := core.NewBlockChain(db, nil, config, pow, vm.Config{}, nil) + if err != nil { + t.Fatalf("failed to create new blockchain: %v", err) + } + pm, err := NewProtocolManager(config, downloader.FullSync, DefaultConfig.NetworkId, evmux, new(testTxPool), pow, blockchain, db, nil) + if err != nil { + t.Fatalf("failed to start test protocol manager: %v", err) + } + pm.Start(1000) + defer pm.Stop() + var peers []*testPeer + for i := 0; i < totalPeers; i++ { + peer, _ := newTestPeer(fmt.Sprintf("peer %d", i), eth63, pm, true) + defer peer.close() + peers = append(peers, peer) + } + chain, _ := core.GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, 1, func(i int, gen *core.BlockGen) {}) + pm.BroadcastBlock(chain[0], true /*propagate*/) + + errCh := make(chan error, totalPeers) + doneCh := make(chan struct{}, totalPeers) + for _, peer := range peers { + go func(p *testPeer) { + if err := p2p.ExpectMsg(p.app, NewBlockMsg, &newBlockData{Block: chain[0], TD: big.NewInt(131136)}); err != nil { + errCh <- err + } else { + doneCh <- struct{}{} + } + }(peer) + } + timeout := time.After(300 * time.Millisecond) + var receivedCount int +outer: + for { + select { + case err = <-errCh: + break outer + case <-doneCh: + receivedCount++ + if receivedCount == totalPeers { + break outer + } + case <-timeout: + break outer + } + } + for _, peer := range peers { + peer.app.Close() + } + if err != nil { + t.Errorf("error matching block by peer: %v", err) + } + if receivedCount != broadcastExpected { + t.Errorf("block broadcast to %d peers, expected %d", receivedCount, broadcastExpected) + } +}