test_helper.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678
  1. // Copyright 2016 The go-ethereum Authors
  2. // This file is part of the go-ethereum library.
  3. //
  4. // The go-ethereum library is free software: you can redistribute it and/or modify
  5. // it under the terms of the GNU Lesser General Public License as published by
  6. // the Free Software Foundation, either version 3 of the License, or
  7. // (at your option) any later version.
  8. //
  9. // The go-ethereum library is distributed in the hope that it will be useful,
  10. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. // GNU Lesser General Public License for more details.
  13. //
  14. // You should have received a copy of the GNU Lesser General Public License
  15. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
  16. // This file contains some shares testing functionality, common to multiple
  17. // different files and modules being tested. Client based network and Server
  18. // based network can be created easily with available APIs.
  19. package les
  20. import (
  21. "context"
  22. "crypto/rand"
  23. "fmt"
  24. "math/big"
  25. "sync/atomic"
  26. "testing"
  27. "time"
  28. "github.com/ethereum/go-ethereum/accounts/abi/bind"
  29. "github.com/ethereum/go-ethereum/accounts/abi/bind/backends"
  30. "github.com/ethereum/go-ethereum/common"
  31. "github.com/ethereum/go-ethereum/common/mclock"
  32. "github.com/ethereum/go-ethereum/consensus/ethash"
  33. "github.com/ethereum/go-ethereum/contracts/checkpointoracle/contract"
  34. "github.com/ethereum/go-ethereum/core"
  35. "github.com/ethereum/go-ethereum/core/forkid"
  36. "github.com/ethereum/go-ethereum/core/rawdb"
  37. "github.com/ethereum/go-ethereum/core/types"
  38. "github.com/ethereum/go-ethereum/crypto"
  39. "github.com/ethereum/go-ethereum/eth/ethconfig"
  40. "github.com/ethereum/go-ethereum/ethdb"
  41. "github.com/ethereum/go-ethereum/event"
  42. "github.com/ethereum/go-ethereum/les/checkpointoracle"
  43. "github.com/ethereum/go-ethereum/les/flowcontrol"
  44. vfs "github.com/ethereum/go-ethereum/les/vflux/server"
  45. "github.com/ethereum/go-ethereum/light"
  46. "github.com/ethereum/go-ethereum/p2p"
  47. "github.com/ethereum/go-ethereum/p2p/enode"
  48. "github.com/ethereum/go-ethereum/params"
  49. )
  50. var (
  51. bankKey, _ = crypto.GenerateKey()
  52. bankAddr = crypto.PubkeyToAddress(bankKey.PublicKey)
  53. bankFunds = big.NewInt(1000000000000000000)
  54. userKey1, _ = crypto.GenerateKey()
  55. userKey2, _ = crypto.GenerateKey()
  56. userAddr1 = crypto.PubkeyToAddress(userKey1.PublicKey)
  57. userAddr2 = crypto.PubkeyToAddress(userKey2.PublicKey)
  58. testContractAddr common.Address
  59. testContractCode = common.Hex2Bytes("606060405260cc8060106000396000f360606040526000357c01000000000000000000000000000000000000000000000000000000009004806360cd2685146041578063c16431b914606b57603f565b005b6055600480803590602001909190505060a9565b6040518082815260200191505060405180910390f35b60886004808035906020019091908035906020019091905050608a565b005b80600060005083606481101560025790900160005b50819055505b5050565b6000600060005082606481101560025790900160005b5054905060c7565b91905056")
  60. testContractCodeDeployed = testContractCode[16:]
  61. testContractDeployed = uint64(2)
  62. testEventEmitterCode = common.Hex2Bytes("60606040523415600e57600080fd5b7f57050ab73f6b9ebdd9f76b8d4997793f48cf956e965ee070551b9ca0bb71584e60405160405180910390a160358060476000396000f3006060604052600080fd00a165627a7a723058203f727efcad8b5811f8cb1fc2620ce5e8c63570d697aef968172de296ea3994140029")
  63. // Checkpoint oracle relative fields
  64. oracleAddr common.Address
  65. signerKey, _ = crypto.GenerateKey()
  66. signerAddr = crypto.PubkeyToAddress(signerKey.PublicKey)
  67. )
  68. var (
  69. // The block frequency for creating checkpoint(only used in test)
  70. sectionSize = big.NewInt(128)
  71. // The number of confirmations needed to generate a checkpoint(only used in test).
  72. processConfirms = big.NewInt(1)
  73. // The token bucket buffer limit for testing purpose.
  74. testBufLimit = uint64(1000000)
  75. // The buffer recharging speed for testing purpose.
  76. testBufRecharge = uint64(1000)
  77. )
  78. /*
  79. contract test {
  80. uint256[100] data;
  81. function Put(uint256 addr, uint256 value) {
  82. data[addr] = value;
  83. }
  84. function Get(uint256 addr) constant returns (uint256 value) {
  85. return data[addr];
  86. }
  87. }
  88. */
  89. // prepare pre-commits specified number customized blocks into chain.
  90. func prepare(n int, backend *backends.SimulatedBackend) {
  91. var (
  92. ctx = context.Background()
  93. signer = types.HomesteadSigner{}
  94. )
  95. for i := 0; i < n; i++ {
  96. switch i {
  97. case 0:
  98. // Builtin-block
  99. // number: 1
  100. // txs: 2
  101. // deploy checkpoint contract
  102. auth, _ := bind.NewKeyedTransactorWithChainID(bankKey, big.NewInt(1337))
  103. oracleAddr, _, _, _ = contract.DeployCheckpointOracle(auth, backend, []common.Address{signerAddr}, sectionSize, processConfirms, big.NewInt(1))
  104. // bankUser transfers some ether to user1
  105. nonce, _ := backend.PendingNonceAt(ctx, bankAddr)
  106. tx, _ := types.SignTx(types.NewTransaction(nonce, userAddr1, big.NewInt(10000), params.TxGas, nil, nil), signer, bankKey)
  107. backend.SendTransaction(ctx, tx, bind.PrivateTxArgs{})
  108. case 1:
  109. // Builtin-block
  110. // number: 2
  111. // txs: 4
  112. bankNonce, _ := backend.PendingNonceAt(ctx, bankAddr)
  113. userNonce1, _ := backend.PendingNonceAt(ctx, userAddr1)
  114. // bankUser transfers more ether to user1
  115. tx1, _ := types.SignTx(types.NewTransaction(bankNonce, userAddr1, big.NewInt(1000), params.TxGas, nil, nil), signer, bankKey)
  116. backend.SendTransaction(ctx, tx1, bind.PrivateTxArgs{})
  117. // user1 relays ether to user2
  118. tx2, _ := types.SignTx(types.NewTransaction(userNonce1, userAddr2, big.NewInt(1000), params.TxGas, nil, nil), signer, userKey1)
  119. backend.SendTransaction(ctx, tx2, bind.PrivateTxArgs{})
  120. // user1 deploys a test contract
  121. tx3, _ := types.SignTx(types.NewContractCreation(userNonce1+1, big.NewInt(0), 200000, big.NewInt(0), testContractCode), signer, userKey1)
  122. backend.SendTransaction(ctx, tx3, bind.PrivateTxArgs{})
  123. testContractAddr = crypto.CreateAddress(userAddr1, userNonce1+1)
  124. // user1 deploys a event contract
  125. tx4, _ := types.SignTx(types.NewContractCreation(userNonce1+2, big.NewInt(0), 200000, big.NewInt(0), testEventEmitterCode), signer, userKey1)
  126. backend.SendTransaction(ctx, tx4, bind.PrivateTxArgs{})
  127. case 2:
  128. // Builtin-block
  129. // number: 3
  130. // txs: 2
  131. // bankUser transfer some ether to signer
  132. bankNonce, _ := backend.PendingNonceAt(ctx, bankAddr)
  133. tx1, _ := types.SignTx(types.NewTransaction(bankNonce, signerAddr, big.NewInt(1000000000), params.TxGas, nil, nil), signer, bankKey)
  134. backend.SendTransaction(ctx, tx1, bind.PrivateTxArgs{})
  135. // invoke test contract
  136. data := common.Hex2Bytes("C16431B900000000000000000000000000000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000000000001")
  137. tx2, _ := types.SignTx(types.NewTransaction(bankNonce+1, testContractAddr, big.NewInt(0), 100000, nil, data), signer, bankKey)
  138. backend.SendTransaction(ctx, tx2, bind.PrivateTxArgs{})
  139. case 3:
  140. // Builtin-block
  141. // number: 4
  142. // txs: 1
  143. // invoke test contract
  144. bankNonce, _ := backend.PendingNonceAt(ctx, bankAddr)
  145. data := common.Hex2Bytes("C16431B900000000000000000000000000000000000000000000000000000000000000020000000000000000000000000000000000000000000000000000000000000002")
  146. tx, _ := types.SignTx(types.NewTransaction(bankNonce, testContractAddr, big.NewInt(0), 100000, nil, data), signer, bankKey)
  147. backend.SendTransaction(ctx, tx, bind.PrivateTxArgs{})
  148. }
  149. backend.Commit()
  150. }
  151. }
  152. // testIndexers creates a set of indexers with specified params for testing purpose.
  153. func testIndexers(db ethdb.Database, odr light.OdrBackend, config *light.IndexerConfig, disablePruning bool) []*core.ChainIndexer {
  154. var indexers [3]*core.ChainIndexer
  155. indexers[0] = light.NewChtIndexer(db, odr, config.ChtSize, config.ChtConfirms, disablePruning)
  156. indexers[1] = core.NewBloomIndexer(db, config.BloomSize, config.BloomConfirms)
  157. indexers[2] = light.NewBloomTrieIndexer(db, odr, config.BloomSize, config.BloomTrieSize, disablePruning)
  158. // make bloomTrieIndexer as a child indexer of bloom indexer.
  159. indexers[1].AddChildIndexer(indexers[2])
  160. return indexers[:]
  161. }
  162. func newTestClientHandler(backend *backends.SimulatedBackend, odr *LesOdr, indexers []*core.ChainIndexer, db ethdb.Database, peers *serverPeerSet, ulcServers []string, ulcFraction int) (*clientHandler, func()) {
  163. var (
  164. evmux = new(event.TypeMux)
  165. engine = ethash.NewFaker()
  166. gspec = core.Genesis{
  167. Config: params.AllEthashProtocolChanges,
  168. Alloc: core.GenesisAlloc{bankAddr: {Balance: bankFunds}},
  169. GasLimit: 100000000,
  170. }
  171. oracle *checkpointoracle.CheckpointOracle
  172. )
  173. genesis := gspec.MustCommit(db)
  174. chain, _ := light.NewLightChain(odr, gspec.Config, engine, nil)
  175. if indexers != nil {
  176. checkpointConfig := &params.CheckpointOracleConfig{
  177. Address: crypto.CreateAddress(bankAddr, 0),
  178. Signers: []common.Address{signerAddr},
  179. Threshold: 1,
  180. }
  181. getLocal := func(index uint64) params.TrustedCheckpoint {
  182. chtIndexer := indexers[0]
  183. sectionHead := chtIndexer.SectionHead(index)
  184. return params.TrustedCheckpoint{
  185. SectionIndex: index,
  186. SectionHead: sectionHead,
  187. CHTRoot: light.GetChtRoot(db, index, sectionHead),
  188. BloomRoot: light.GetBloomTrieRoot(db, index, sectionHead),
  189. }
  190. }
  191. oracle = checkpointoracle.New(checkpointConfig, getLocal)
  192. }
  193. client := &LightEthereum{
  194. lesCommons: lesCommons{
  195. genesis: genesis.Hash(),
  196. config: &ethconfig.Config{LightPeers: 100, NetworkId: NetworkId},
  197. chainConfig: params.AllEthashProtocolChanges,
  198. iConfig: light.TestClientIndexerConfig,
  199. chainDb: db,
  200. oracle: oracle,
  201. chainReader: chain,
  202. closeCh: make(chan struct{}),
  203. },
  204. peers: peers,
  205. reqDist: odr.retriever.dist,
  206. retriever: odr.retriever,
  207. odr: odr,
  208. engine: engine,
  209. blockchain: chain,
  210. eventMux: evmux,
  211. }
  212. client.handler = newClientHandler(ulcServers, ulcFraction, nil, client)
  213. if client.oracle != nil {
  214. client.oracle.Start(backend)
  215. }
  216. client.handler.start()
  217. return client.handler, func() {
  218. client.handler.stop()
  219. }
  220. }
  221. func newTestServerHandler(blocks int, indexers []*core.ChainIndexer, db ethdb.Database, clock mclock.Clock) (*serverHandler, *backends.SimulatedBackend, func()) {
  222. var (
  223. gspec = core.Genesis{
  224. Config: params.AllEthashProtocolChanges,
  225. Alloc: core.GenesisAlloc{bankAddr: {Balance: bankFunds}},
  226. GasLimit: 100000000,
  227. }
  228. oracle *checkpointoracle.CheckpointOracle
  229. )
  230. genesis := gspec.MustCommit(db)
  231. // create a simulation backend and pre-commit several customized block to the database.
  232. simulation := backends.NewSimulatedBackendWithDatabase(db, gspec.Alloc, 100000000)
  233. prepare(blocks, simulation)
  234. txpoolConfig := core.DefaultTxPoolConfig
  235. txpoolConfig.Journal = ""
  236. txpool := core.NewTxPool(txpoolConfig, gspec.Config, simulation.Blockchain())
  237. if indexers != nil {
  238. checkpointConfig := &params.CheckpointOracleConfig{
  239. Address: crypto.CreateAddress(bankAddr, 0),
  240. Signers: []common.Address{signerAddr},
  241. Threshold: 1,
  242. }
  243. getLocal := func(index uint64) params.TrustedCheckpoint {
  244. chtIndexer := indexers[0]
  245. sectionHead := chtIndexer.SectionHead(index)
  246. return params.TrustedCheckpoint{
  247. SectionIndex: index,
  248. SectionHead: sectionHead,
  249. CHTRoot: light.GetChtRoot(db, index, sectionHead),
  250. BloomRoot: light.GetBloomTrieRoot(db, index, sectionHead),
  251. }
  252. }
  253. oracle = checkpointoracle.New(checkpointConfig, getLocal)
  254. }
  255. server := &LesServer{
  256. lesCommons: lesCommons{
  257. genesis: genesis.Hash(),
  258. config: &ethconfig.Config{LightPeers: 100, NetworkId: NetworkId},
  259. chainConfig: params.AllEthashProtocolChanges,
  260. iConfig: light.TestServerIndexerConfig,
  261. chainDb: db,
  262. chainReader: simulation.Blockchain(),
  263. oracle: oracle,
  264. closeCh: make(chan struct{}),
  265. },
  266. peers: newClientPeerSet(),
  267. servingQueue: newServingQueue(int64(time.Millisecond*10), 1),
  268. defParams: flowcontrol.ServerParams{
  269. BufLimit: testBufLimit,
  270. MinRecharge: testBufRecharge,
  271. },
  272. fcManager: flowcontrol.NewClientManager(nil, clock),
  273. }
  274. server.costTracker, server.minCapacity = newCostTracker(db, server.config)
  275. server.costTracker.testCostList = testCostList(0) // Disable flow control mechanism.
  276. server.clientPool = vfs.NewClientPool(db, testBufRecharge, defaultConnectedBias, clock, alwaysTrueFn)
  277. server.clientPool.Start()
  278. server.clientPool.SetLimits(10000, 10000) // Assign enough capacity for clientpool
  279. server.handler = newServerHandler(server, simulation.Blockchain(), db, txpool, func() bool { return true })
  280. if server.oracle != nil {
  281. server.oracle.Start(simulation)
  282. }
  283. server.servingQueue.setThreads(4)
  284. server.handler.start()
  285. closer := func() { server.Stop() }
  286. return server.handler, simulation, closer
  287. }
  288. func alwaysTrueFn() bool {
  289. return true
  290. }
  291. // testPeer is a simulated peer to allow testing direct network calls.
  292. type testPeer struct {
  293. cpeer *clientPeer
  294. speer *serverPeer
  295. net p2p.MsgReadWriter // Network layer reader/writer to simulate remote messaging
  296. app *p2p.MsgPipeRW // Application layer reader/writer to simulate the local side
  297. }
  298. // handshakeWithServer executes the handshake with the remote server peer.
  299. func (p *testPeer) handshakeWithServer(t *testing.T, td *big.Int, head common.Hash, headNum uint64, genesis common.Hash, forkID forkid.ID) {
  300. // It only works for the simulated client peer
  301. if p.cpeer == nil {
  302. t.Fatal("handshake for client peer only")
  303. }
  304. var sendList keyValueList
  305. sendList = sendList.add("protocolVersion", uint64(p.cpeer.version))
  306. sendList = sendList.add("networkId", uint64(NetworkId))
  307. sendList = sendList.add("headTd", td)
  308. sendList = sendList.add("headHash", head)
  309. sendList = sendList.add("headNum", headNum)
  310. sendList = sendList.add("genesisHash", genesis)
  311. if p.cpeer.version >= lpv4 {
  312. sendList = sendList.add("forkID", &forkID)
  313. }
  314. if err := p2p.ExpectMsg(p.app, StatusMsg, nil); err != nil {
  315. t.Fatalf("status recv: %v", err)
  316. }
  317. if err := p2p.Send(p.app, StatusMsg, sendList); err != nil {
  318. t.Fatalf("status send: %v", err)
  319. }
  320. }
  321. // handshakeWithClient executes the handshake with the remote client peer.
  322. func (p *testPeer) handshakeWithClient(t *testing.T, td *big.Int, head common.Hash, headNum uint64, genesis common.Hash, forkID forkid.ID, costList RequestCostList, recentTxLookup uint64) {
  323. // It only works for the simulated client peer
  324. if p.speer == nil {
  325. t.Fatal("handshake for server peer only")
  326. }
  327. var sendList keyValueList
  328. sendList = sendList.add("protocolVersion", uint64(p.speer.version))
  329. sendList = sendList.add("networkId", uint64(NetworkId))
  330. sendList = sendList.add("headTd", td)
  331. sendList = sendList.add("headHash", head)
  332. sendList = sendList.add("headNum", headNum)
  333. sendList = sendList.add("genesisHash", genesis)
  334. sendList = sendList.add("serveHeaders", nil)
  335. sendList = sendList.add("serveChainSince", uint64(0))
  336. sendList = sendList.add("serveStateSince", uint64(0))
  337. sendList = sendList.add("serveRecentState", uint64(core.TriesInMemory-4))
  338. sendList = sendList.add("txRelay", nil)
  339. sendList = sendList.add("flowControl/BL", testBufLimit)
  340. sendList = sendList.add("flowControl/MRR", testBufRecharge)
  341. sendList = sendList.add("flowControl/MRC", costList)
  342. if p.speer.version >= lpv4 {
  343. sendList = sendList.add("forkID", &forkID)
  344. sendList = sendList.add("recentTxLookup", recentTxLookup)
  345. }
  346. if err := p2p.ExpectMsg(p.app, StatusMsg, nil); err != nil {
  347. t.Fatalf("status recv: %v", err)
  348. }
  349. if err := p2p.Send(p.app, StatusMsg, sendList); err != nil {
  350. t.Fatalf("status send: %v", err)
  351. }
  352. }
  353. // close terminates the local side of the peer, notifying the remote protocol
  354. // manager of termination.
  355. func (p *testPeer) close() {
  356. p.app.Close()
  357. }
  358. func newTestPeerPair(name string, version int, server *serverHandler, client *clientHandler) (*testPeer, *testPeer, error) {
  359. // Create a message pipe to communicate through
  360. app, net := p2p.MsgPipe()
  361. // Generate a random id and create the peer
  362. var id enode.ID
  363. rand.Read(id[:])
  364. peer1 := newClientPeer(version, NetworkId, p2p.NewPeer(id, name, nil), net)
  365. peer2 := newServerPeer(version, NetworkId, false, p2p.NewPeer(id, name, nil), app)
  366. // Start the peer on a new thread
  367. errc1 := make(chan error, 1)
  368. errc2 := make(chan error, 1)
  369. go func() {
  370. select {
  371. case <-server.closeCh:
  372. errc1 <- p2p.DiscQuitting
  373. case errc1 <- server.handle(peer1):
  374. }
  375. }()
  376. go func() {
  377. select {
  378. case <-client.closeCh:
  379. errc2 <- p2p.DiscQuitting
  380. case errc2 <- client.handle(peer2):
  381. }
  382. }()
  383. // Ensure the connection is established or exits when any error occurs
  384. for {
  385. select {
  386. case err := <-errc1:
  387. return nil, nil, fmt.Errorf("Failed to establish protocol connection %v", err)
  388. case err := <-errc2:
  389. return nil, nil, fmt.Errorf("Failed to establish protocol connection %v", err)
  390. default:
  391. }
  392. if atomic.LoadUint32(&peer1.serving) == 1 && atomic.LoadUint32(&peer2.serving) == 1 {
  393. break
  394. }
  395. time.Sleep(50 * time.Millisecond)
  396. }
  397. return &testPeer{cpeer: peer1, net: net, app: app}, &testPeer{speer: peer2, net: app, app: net}, nil
  398. }
  399. type indexerCallback func(*core.ChainIndexer, *core.ChainIndexer, *core.ChainIndexer)
  400. // testClient represents a client object for testing with necessary auxiliary fields.
  401. type testClient struct {
  402. clock mclock.Clock
  403. db ethdb.Database
  404. peer *testPeer
  405. handler *clientHandler
  406. chtIndexer *core.ChainIndexer
  407. bloomIndexer *core.ChainIndexer
  408. bloomTrieIndexer *core.ChainIndexer
  409. }
  410. // newRawPeer creates a new server peer connects to the server and do the handshake.
  411. func (client *testClient) newRawPeer(t *testing.T, name string, version int, recentTxLookup uint64) (*testPeer, func(), <-chan error) {
  412. // Create a message pipe to communicate through
  413. app, net := p2p.MsgPipe()
  414. // Generate a random id and create the peer
  415. var id enode.ID
  416. rand.Read(id[:])
  417. peer := newServerPeer(version, NetworkId, false, p2p.NewPeer(id, name, nil), net)
  418. // Start the peer on a new thread
  419. errCh := make(chan error, 1)
  420. go func() {
  421. select {
  422. case <-client.handler.closeCh:
  423. errCh <- p2p.DiscQuitting
  424. case errCh <- client.handler.handle(peer):
  425. }
  426. }()
  427. tp := &testPeer{
  428. app: app,
  429. net: net,
  430. speer: peer,
  431. }
  432. var (
  433. genesis = client.handler.backend.blockchain.Genesis()
  434. head = client.handler.backend.blockchain.CurrentHeader()
  435. td = client.handler.backend.blockchain.GetTd(head.Hash(), head.Number.Uint64())
  436. )
  437. forkID := forkid.NewID(client.handler.backend.blockchain.Config(), genesis.Hash(), head.Number.Uint64())
  438. tp.handshakeWithClient(t, td, head.Hash(), head.Number.Uint64(), genesis.Hash(), forkID, testCostList(0), recentTxLookup) // disable flow control by default
  439. // Ensure the connection is established or exits when any error occurs
  440. for {
  441. select {
  442. case <-errCh:
  443. return nil, nil, nil
  444. default:
  445. }
  446. if atomic.LoadUint32(&peer.serving) == 1 {
  447. break
  448. }
  449. time.Sleep(50 * time.Millisecond)
  450. }
  451. closePeer := func() {
  452. tp.speer.close()
  453. tp.close()
  454. }
  455. return tp, closePeer, errCh
  456. }
  457. // testServer represents a server object for testing with necessary auxiliary fields.
  458. type testServer struct {
  459. clock mclock.Clock
  460. backend *backends.SimulatedBackend
  461. db ethdb.Database
  462. peer *testPeer
  463. handler *serverHandler
  464. chtIndexer *core.ChainIndexer
  465. bloomIndexer *core.ChainIndexer
  466. bloomTrieIndexer *core.ChainIndexer
  467. }
  468. // newRawPeer creates a new client peer connects to the server and do the handshake.
  469. func (server *testServer) newRawPeer(t *testing.T, name string, version int) (*testPeer, func(), <-chan error) {
  470. // Create a message pipe to communicate through
  471. app, net := p2p.MsgPipe()
  472. // Generate a random id and create the peer
  473. var id enode.ID
  474. rand.Read(id[:])
  475. peer := newClientPeer(version, NetworkId, p2p.NewPeer(id, name, nil), net)
  476. // Start the peer on a new thread
  477. errCh := make(chan error, 1)
  478. go func() {
  479. select {
  480. case <-server.handler.closeCh:
  481. errCh <- p2p.DiscQuitting
  482. case errCh <- server.handler.handle(peer):
  483. }
  484. }()
  485. tp := &testPeer{
  486. app: app,
  487. net: net,
  488. cpeer: peer,
  489. }
  490. var (
  491. genesis = server.handler.blockchain.Genesis()
  492. head = server.handler.blockchain.CurrentHeader()
  493. td = server.handler.blockchain.GetTd(head.Hash(), head.Number.Uint64())
  494. )
  495. forkID := forkid.NewID(server.handler.blockchain.Config(), genesis.Hash(), head.Number.Uint64())
  496. tp.handshakeWithServer(t, td, head.Hash(), head.Number.Uint64(), genesis.Hash(), forkID)
  497. // Ensure the connection is established or exits when any error occurs
  498. for {
  499. select {
  500. case <-errCh:
  501. return nil, nil, nil
  502. default:
  503. }
  504. if atomic.LoadUint32(&peer.serving) == 1 {
  505. break
  506. }
  507. time.Sleep(50 * time.Millisecond)
  508. }
  509. closePeer := func() {
  510. tp.cpeer.close()
  511. tp.close()
  512. }
  513. return tp, closePeer, errCh
  514. }
  515. // testnetConfig wraps all the configurations for testing network.
  516. type testnetConfig struct {
  517. blocks int
  518. protocol int
  519. indexFn indexerCallback
  520. ulcServers []string
  521. ulcFraction int
  522. simClock bool
  523. connect bool
  524. nopruning bool
  525. }
  526. func newClientServerEnv(t *testing.T, config testnetConfig) (*testServer, *testClient, func()) {
  527. var (
  528. sdb = rawdb.NewMemoryDatabase()
  529. cdb = rawdb.NewMemoryDatabase()
  530. speers = newServerPeerSet()
  531. )
  532. var clock mclock.Clock = &mclock.System{}
  533. if config.simClock {
  534. clock = &mclock.Simulated{}
  535. }
  536. dist := newRequestDistributor(speers, clock)
  537. rm := newRetrieveManager(speers, dist, func() time.Duration { return time.Millisecond * 500 })
  538. odr := NewLesOdr(cdb, light.TestClientIndexerConfig, speers, rm)
  539. sindexers := testIndexers(sdb, nil, light.TestServerIndexerConfig, true)
  540. cIndexers := testIndexers(cdb, odr, light.TestClientIndexerConfig, config.nopruning)
  541. scIndexer, sbIndexer, sbtIndexer := sindexers[0], sindexers[1], sindexers[2]
  542. ccIndexer, cbIndexer, cbtIndexer := cIndexers[0], cIndexers[1], cIndexers[2]
  543. odr.SetIndexers(ccIndexer, cbIndexer, cbtIndexer)
  544. server, b, serverClose := newTestServerHandler(config.blocks, sindexers, sdb, clock)
  545. client, clientClose := newTestClientHandler(b, odr, cIndexers, cdb, speers, config.ulcServers, config.ulcFraction)
  546. scIndexer.Start(server.blockchain)
  547. sbIndexer.Start(server.blockchain)
  548. ccIndexer.Start(client.backend.blockchain)
  549. cbIndexer.Start(client.backend.blockchain)
  550. if config.indexFn != nil {
  551. config.indexFn(scIndexer, sbIndexer, sbtIndexer)
  552. }
  553. var (
  554. err error
  555. speer, cpeer *testPeer
  556. )
  557. if config.connect {
  558. done := make(chan struct{})
  559. client.syncEnd = func(_ *types.Header) { close(done) }
  560. cpeer, speer, err = newTestPeerPair("peer", config.protocol, server, client)
  561. if err != nil {
  562. t.Fatalf("Failed to connect testing peers %v", err)
  563. }
  564. select {
  565. case <-done:
  566. case <-time.After(10 * time.Second):
  567. t.Fatal("test peer did not connect and sync within 3s")
  568. }
  569. }
  570. s := &testServer{
  571. clock: clock,
  572. backend: b,
  573. db: sdb,
  574. peer: cpeer,
  575. handler: server,
  576. chtIndexer: scIndexer,
  577. bloomIndexer: sbIndexer,
  578. bloomTrieIndexer: sbtIndexer,
  579. }
  580. c := &testClient{
  581. clock: clock,
  582. db: cdb,
  583. peer: speer,
  584. handler: client,
  585. chtIndexer: ccIndexer,
  586. bloomIndexer: cbIndexer,
  587. bloomTrieIndexer: cbtIndexer,
  588. }
  589. teardown := func() {
  590. if config.connect {
  591. speer.close()
  592. cpeer.close()
  593. cpeer.cpeer.close()
  594. speer.speer.close()
  595. }
  596. ccIndexer.Close()
  597. cbIndexer.Close()
  598. scIndexer.Close()
  599. sbIndexer.Close()
  600. dist.close()
  601. serverClose()
  602. b.Close()
  603. clientClose()
  604. }
  605. return s, c, teardown
  606. }
  607. // NewFuzzerPeer creates a client peer for test purposes, and also returns
  608. // a function to close the peer: this is needed to avoid goroutine leaks in the
  609. // exec queue.
  610. func NewFuzzerPeer(version int) (p *clientPeer, closer func()) {
  611. p = newClientPeer(version, 0, p2p.NewPeer(enode.ID{}, "", nil), nil)
  612. return p, func() { p.peerCommons.close() }
  613. }