From f1578f245c1c269334cda00410c6b3b6f0e215b8 Mon Sep 17 00:00:00 2001 From: Alexey Akhunov Date: Wed, 8 Aug 2018 09:31:46 +0100 Subject: [PATCH 1/3] eth: Expose ethapi.GetAPIs to be used to construct RPC APIs without whole Ethereum node --- eth/backend.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/eth/backend.go b/eth/backend.go index 32946a0ab3..69834775ac 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -240,6 +240,15 @@ func CreateConsensusEngine(ctx *node.ServiceContext, config *ethash.Config, chai } } +// Exposes internal/ethapi/GetAPIs function so that it can be used +// by other application to construct RPC API service by simply +// implementing ethapi.Backend (which is subject with change in +// backwards incompatible way), rather than instantiating the +// entire Ethereum node +func GetAPIs(backend ethapi.Backend) []rpc.API { + return ethapi.GetAPIs(backend) +} + // APIs return the collection of RPC services the ethereum package offers. // NOTE, some of these services probably need to be moved to somewhere else. func (s *Ethereum) APIs() []rpc.API { From 0c33200e13dfb7bd8b8e98aef6bcbfea04faa2f7 Mon Sep 17 00:00:00 2001 From: Alexey Akhunov Date: Thu, 20 Sep 2018 17:35:00 +0100 Subject: [PATCH 2/3] eth: Lower bound 4 on number of peers to broadcast new blocks --- eth/handler.go | 12 +++++- eth/handler_test.go | 92 ++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 102 insertions(+), 2 deletions(-) diff --git a/eth/handler.go b/eth/handler.go index f89f68c9d2..738c901407 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -49,6 +49,9 @@ const ( // txChanSize is the size of channel listening to NewTxsEvent. // The number is referenced from the size of tx pool. txChanSize = 4096 + + // minimim number of peers to broadcast new blocks to + minBroadcastPeers = 4 ) var ( @@ -705,7 +708,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.AsyncSendNewBlock(block, td) } diff --git a/eth/handler_test.go b/eth/handler_test.go index fee4114eb6..d72e0971c7 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -17,6 +17,7 @@ package eth import ( + "fmt" "math" "math/big" "math/rand" @@ -472,8 +473,11 @@ func testDAOChallenge(t *testing.T, localForked, remoteForked bool, timeout bool config = ¶ms.ChainConfig{DAOForkBlock: big.NewInt(1), DAOForkSupport: localForked} gspec = &core.Genesis{Config: config} genesis = gspec.MustCommit(db) - blockchain, _ = core.NewBlockChain(db, nil, config, pow, vm.Config{}) ) + blockchain, err := core.NewBlockChain(db, nil, config, pow, vm.Config{}) + 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) if err != nil { t.Fatalf("failed to start test protocol manager: %v", err) @@ -520,3 +524,89 @@ func testDAOChallenge(t *testing.T, localForked, remoteForked bool, timeout bool } } } + +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{}) + 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) + 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) + } + timeoutCh := time.NewTimer(time.Millisecond*100).C + var receivedCount int + outer: for { + select { + case err =<- errCh: + break outer + case <- doneCh: + receivedCount++ + if receivedCount == totalPeers { + break outer + } + case <- timeoutCh: + 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) + } +} From bb1c03b5ee0b6cc2ad9f5dba20e4e4923a5595b2 Mon Sep 17 00:00:00 2001 From: Alexey Akhunov Date: Thu, 20 Sep 2018 17:40:53 +0100 Subject: [PATCH 3/3] Revert "eth: Lower bound 4 on number of peers to broadcast new blocks" This reverts commit 0c33200e13dfb7bd8b8e98aef6bcbfea04faa2f7. --- eth/handler.go | 12 +----- eth/handler_test.go | 92 +-------------------------------------------- 2 files changed, 2 insertions(+), 102 deletions(-) diff --git a/eth/handler.go b/eth/handler.go index 738c901407..f89f68c9d2 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -49,9 +49,6 @@ const ( // txChanSize is the size of channel listening to NewTxsEvent. // The number is referenced from the size of tx pool. txChanSize = 4096 - - // minimim number of peers to broadcast new blocks to - minBroadcastPeers = 4 ) var ( @@ -708,14 +705,7 @@ func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) { return } // Send the block to a subset of our peers - transferLen := int(math.Sqrt(float64(len(peers)))) - if transferLen < minBroadcastPeers { - transferLen = minBroadcastPeers - } - if transferLen > len(peers) { - transferLen = len(peers) - } - transfer := peers[:transferLen] + transfer := peers[:int(math.Sqrt(float64(len(peers))))] for _, peer := range transfer { peer.AsyncSendNewBlock(block, td) } diff --git a/eth/handler_test.go b/eth/handler_test.go index d72e0971c7..fee4114eb6 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -17,7 +17,6 @@ package eth import ( - "fmt" "math" "math/big" "math/rand" @@ -473,11 +472,8 @@ func testDAOChallenge(t *testing.T, localForked, remoteForked bool, timeout bool config = ¶ms.ChainConfig{DAOForkBlock: big.NewInt(1), DAOForkSupport: localForked} gspec = &core.Genesis{Config: config} genesis = gspec.MustCommit(db) + blockchain, _ = core.NewBlockChain(db, nil, config, pow, vm.Config{}) ) - blockchain, err := core.NewBlockChain(db, nil, config, pow, vm.Config{}) - 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) if err != nil { t.Fatalf("failed to start test protocol manager: %v", err) @@ -524,89 +520,3 @@ func testDAOChallenge(t *testing.T, localForked, remoteForked bool, timeout bool } } } - -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{}) - 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) - 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) - } - timeoutCh := time.NewTimer(time.Millisecond*100).C - var receivedCount int - outer: for { - select { - case err =<- errCh: - break outer - case <- doneCh: - receivedCount++ - if receivedCount == totalPeers { - break outer - } - case <- timeoutCh: - 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) - } -}