From d1aa605f1e8639769cdd75bcec3064c29a62b34a Mon Sep 17 00:00:00 2001 From: Wenbiao Zheng Date: Mon, 27 Aug 2018 16:49:29 +0800 Subject: [PATCH 1/9] all: remove the duplicate 'the' in annotations (#17509) --- accounts/abi/bind/bind.go | 4 ++-- cmd/wnode/main.go | 2 +- common/bitutil/compress_test.go | 4 ++-- console/prompter.go | 4 ++-- core/block_validator.go | 2 +- core/state_processor.go | 2 +- core/vm/contract.go | 2 +- crypto/secp256k1/secp256.go | 2 +- eth/downloader/queue.go | 2 +- eth/fetcher/fetcher.go | 2 +- p2p/server.go | 2 +- swarm/api/api.go | 2 +- swarm/network/README.md | 2 +- swarm/network/stream/intervals/intervals.go | 2 +- swarm/network/stream/snapshot_sync_test.go | 2 +- whisper/whisperv5/whisper.go | 2 +- whisper/whisperv6/whisper.go | 2 +- 17 files changed, 20 insertions(+), 20 deletions(-) diff --git a/accounts/abi/bind/bind.go b/accounts/abi/bind/bind.go index 411177057b..172bed8fa0 100644 --- a/accounts/abi/bind/bind.go +++ b/accounts/abi/bind/bind.go @@ -207,7 +207,7 @@ func bindTypeGo(kind abi.Type) string { // The inner function of bindTypeGo, this finds the inner type of stringKind. // (Or just the type itself if it is not an array or slice) -// The length of the matched part is returned, with the the translated type. +// The length of the matched part is returned, with the translated type. func bindUnnestedTypeGo(stringKind string) (int, string) { switch { @@ -255,7 +255,7 @@ func bindTypeJava(kind abi.Type) string { // The inner function of bindTypeJava, this finds the inner type of stringKind. // (Or just the type itself if it is not an array or slice) -// The length of the matched part is returned, with the the translated type. +// The length of the matched part is returned, with the translated type. func bindUnnestedTypeJava(stringKind string) (int, string) { switch { diff --git a/cmd/wnode/main.go b/cmd/wnode/main.go index 31b65c1da9..ae79841f6b 100644 --- a/cmd/wnode/main.go +++ b/cmd/wnode/main.go @@ -754,7 +754,7 @@ func extractIDFromEnode(s string) []byte { return n.ID[:] } -// obfuscateBloom adds 16 random bits to the the bloom +// obfuscateBloom adds 16 random bits to the bloom // filter, in order to obfuscate the containing topics. // it does so deterministically within every session. // despite additional bits, it will match on average diff --git a/common/bitutil/compress_test.go b/common/bitutil/compress_test.go index 9bd1de103a..13a13011dc 100644 --- a/common/bitutil/compress_test.go +++ b/common/bitutil/compress_test.go @@ -117,7 +117,7 @@ func TestDecodingCycle(t *testing.T) { // TestCompression tests that compression works by returning either the bitset // encoded input, or the actual input if the bitset version is longer. func TestCompression(t *testing.T) { - // Check the the compression returns the bitset encoding is shorter + // Check the compression returns the bitset encoding is shorter in := hexutil.MustDecode("0x4912385c0e7b64000000") out := hexutil.MustDecode("0x80fe4912385c0e7b64") @@ -127,7 +127,7 @@ func TestCompression(t *testing.T) { if data, err := DecompressBytes(out, len(in)); err != nil || !bytes.Equal(data, in) { t.Errorf("decoding mismatch for sparse data: have %x, want %x, error %v", data, in, err) } - // Check the the compression returns the input if the bitset encoding is longer + // Check the compression returns the input if the bitset encoding is longer in = hexutil.MustDecode("0xdf7070533534333636313639343638373532313536346c1bc33339343837313070706336343035336336346c65fefb3930393233383838ac2f65fefb") out = hexutil.MustDecode("0xdf7070533534333636313639343638373532313536346c1bc33339343837313070706336343035336336346c65fefb3930393233383838ac2f65fefb") diff --git a/console/prompter.go b/console/prompter.go index 9b90034db9..29a53aeadd 100644 --- a/console/prompter.go +++ b/console/prompter.go @@ -43,7 +43,7 @@ type UserPrompter interface { // choice to be made, returning that choice. PromptConfirm(prompt string) (bool, error) - // SetHistory sets the the input scrollback history that the prompter will allow + // SetHistory sets the input scrollback history that the prompter will allow // the user to scroll back to. SetHistory(history []string) @@ -149,7 +149,7 @@ func (p *terminalPrompter) PromptConfirm(prompt string) (bool, error) { return false, err } -// SetHistory sets the the input scrollback history that the prompter will allow +// SetHistory sets the input scrollback history that the prompter will allow // the user to scroll back to. func (p *terminalPrompter) SetHistory(history []string) { p.State.ReadHistory(strings.NewReader(strings.Join(history, "\n"))) diff --git a/core/block_validator.go b/core/block_validator.go index 98958809b7..ecd6a89bc2 100644 --- a/core/block_validator.go +++ b/core/block_validator.go @@ -45,7 +45,7 @@ func NewBlockValidator(config *params.ChainConfig, blockchain *BlockChain, engin return validator } -// ValidateBody validates the given block's uncles and verifies the the block +// ValidateBody validates the given block's uncles and verifies the block // header's transaction and uncle roots. The headers are assumed to be already // validated at this point. func (v *BlockValidator) ValidateBody(block *types.Block) error { diff --git a/core/state_processor.go b/core/state_processor.go index 8e238ce1f0..1a91a57ab4 100644 --- a/core/state_processor.go +++ b/core/state_processor.go @@ -61,7 +61,7 @@ func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg allLogs []*types.Log gp = new(GasPool).AddGas(block.GasLimit()) ) - // Mutate the the block and state according to any hard-fork specs + // Mutate the block and state according to any hard-fork specs if p.config.DAOForkSupport && p.config.DAOForkBlock != nil && p.config.DAOForkBlock.Cmp(block.Number()) == 0 { misc.ApplyDAOHardFork(statedb) } diff --git a/core/vm/contract.go b/core/vm/contract.go index b466681dbd..26bca68951 100644 --- a/core/vm/contract.go +++ b/core/vm/contract.go @@ -40,7 +40,7 @@ type AccountRef common.Address func (ar AccountRef) Address() common.Address { return (common.Address)(ar) } // Contract represents an ethereum contract in the state database. It contains -// the the contract code, calling arguments. Contract implements ContractRef +// the contract code, calling arguments. Contract implements ContractRef type Contract struct { // CallerAddress is the result of the caller which initialised this // contract. However when the "call method" is delegated this value diff --git a/crypto/secp256k1/secp256.go b/crypto/secp256k1/secp256.go index 843fb12525..35d0eef34a 100644 --- a/crypto/secp256k1/secp256.go +++ b/crypto/secp256k1/secp256.go @@ -86,7 +86,7 @@ func Sign(msg []byte, seckey []byte) ([]byte, error) { return sig, nil } -// RecoverPubkey returns the the public key of the signer. +// RecoverPubkey returns the public key of the signer. // msg must be the 32-byte hash of the message to be signed. // sig must be a 65-byte compact ECDSA signature containing the // recovery id as the last element. diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index 984dd13d68..8529535ba7 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -662,7 +662,7 @@ func (q *queue) expire(timeout time.Duration, pendPool map[string]*fetchRequest, for _, header := range request.Headers { taskQueue.Push(header, -float32(header.Number.Uint64())) } - // Add the peer to the expiry report along the the number of failed requests + // Add the peer to the expiry report along the number of failed requests expiries[id] = len(request.Headers) } } diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 6383596dce..277f14b81a 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -204,7 +204,7 @@ func (f *Fetcher) Notify(peer string, hash common.Hash, number uint64, time time } } -// Enqueue tries to fill gaps the the fetcher's future import queue. +// Enqueue tries to fill gaps the fetcher's future import queue. func (f *Fetcher) Enqueue(peer string, block *types.Block) error { op := &inject{ origin: peer, diff --git a/p2p/server.go b/p2p/server.go index 8f3a511f31..f9a38fcc64 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -77,7 +77,7 @@ type Config struct { // Disabling is useful for protocol debugging (manual topology). NoDiscovery bool - // DiscoveryV5 specifies whether the the new topic-discovery based V5 discovery + // DiscoveryV5 specifies whether the new topic-discovery based V5 discovery // protocol should be started or not. DiscoveryV5 bool `toml:",omitempty"` diff --git a/swarm/api/api.go b/swarm/api/api.go index adf469cfaa..0d08eeea87 100644 --- a/swarm/api/api.go +++ b/swarm/api/api.go @@ -153,7 +153,7 @@ func NewMultiResolver(opts ...MultiResolverOption) (m *MultiResolver) { // Resolve resolves address by choosing a Resolver by TLD. // If there are more default Resolvers, or for a specific TLD, -// the Hash from the the first one which does not return error +// the Hash from the first one which does not return error // will be returned. func (m *MultiResolver) Resolve(addr string) (h common.Hash, err error) { rs, err := m.getResolveValidator(addr) diff --git a/swarm/network/README.md b/swarm/network/README.md index ad429b38be..e9c9d30d0b 100644 --- a/swarm/network/README.md +++ b/swarm/network/README.md @@ -133,7 +133,7 @@ As part of the deletion protocol then, hashes of insured chunks to be removed ar Downstream peer on the other hand needs to make sure that they can only be finger pointed about a chunk they did receive and store. For this the check of a state should be exhaustive. If historical syncing finishes on one state, all hashes before are covered, no surprises. In other words historical syncing this process is self verifying. With session syncing however, it is not enough to check going back covering the range from old offset to new. Continuity (i.e., that the new state is extension of the old) needs to be verified: after downstream peer reads the range into a buffer, it appends the buffer the last known state at the last known offset and verifies the resulting hash matches -the latest state. Past intervals of historical syncing are checked via the the session root. +the latest state. Past intervals of historical syncing are checked via the session root. Upstream peer signs the states, downstream peers can use as handover proofs. Downstream peers sign off on a state together with an initial offset. diff --git a/swarm/network/stream/intervals/intervals.go b/swarm/network/stream/intervals/intervals.go index 5fd820da87..562c3df9ae 100644 --- a/swarm/network/stream/intervals/intervals.go +++ b/swarm/network/stream/intervals/intervals.go @@ -101,7 +101,7 @@ func (i *Intervals) add(start, end uint64) { } } -// Merge adds all the intervals from the the m Interval to current one. +// Merge adds all the intervals from the m Interval to current one. func (i *Intervals) Merge(m *Intervals) { m.mu.RLock() defer m.mu.RUnlock() diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index 6acab50af4..4e1ab09fcf 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -182,7 +182,7 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { conf.addrToIDMap[string(a)] = n } - //get the the node at that index + //get the node at that index //this is the node selected for upload node := sim.RandomUpNode() item, ok := sim.NodeItem(node.ID, bucketKeyStore) diff --git a/whisper/whisperv5/whisper.go b/whisper/whisperv5/whisper.go index 5a2b25c9ec..8e0662327a 100644 --- a/whisper/whisperv5/whisper.go +++ b/whisper/whisperv5/whisper.go @@ -291,7 +291,7 @@ func (w *Whisper) AddKeyPair(key *ecdsa.PrivateKey) (string, error) { return id, nil } -// HasKeyPair checks if the the whisper node is configured with the private key +// HasKeyPair checks if the whisper node is configured with the private key // of the specified public pair. func (w *Whisper) HasKeyPair(id string) bool { w.keyMu.RLock() diff --git a/whisper/whisperv6/whisper.go b/whisper/whisperv6/whisper.go index 76d485808d..34c1256a6f 100644 --- a/whisper/whisperv6/whisper.go +++ b/whisper/whisperv6/whisper.go @@ -423,7 +423,7 @@ func (whisper *Whisper) AddKeyPair(key *ecdsa.PrivateKey) (string, error) { return id, nil } -// HasKeyPair checks if the the whisper node is configured with the private key +// HasKeyPair checks if the whisper node is configured with the private key // of the specified public pair. func (whisper *Whisper) HasKeyPair(id string) bool { whisper.keyMu.RLock() From f0488e80f7837094ac74327f0a4f5f6a270641f4 Mon Sep 17 00:00:00 2001 From: Caesar Chad Date: Mon, 27 Aug 2018 16:51:26 +0800 Subject: [PATCH 2/9] signer/storage: fix typo (#17504) --- signer/storage/aes_gcm_storage.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/signer/storage/aes_gcm_storage.go b/signer/storage/aes_gcm_storage.go index 225276667f..399637a443 100644 --- a/signer/storage/aes_gcm_storage.go +++ b/signer/storage/aes_gcm_storage.go @@ -36,7 +36,7 @@ type storedCredential struct { CipherText []byte `json:"c"` } -// AESEncryptedStorage is a storage type which is backed by a json-faile. The json-file contains +// AESEncryptedStorage is a storage type which is backed by a json-file. The json-file contains // key-value mappings, where the keys are _not_ encrypted, only the values are. type AESEncryptedStorage struct { // File to read/write credentials From 1acefafe2201ba061a58b51afd7a7edf4aa0f120 Mon Sep 17 00:00:00 2001 From: Geon Kim Date: Mon, 27 Aug 2018 17:51:58 +0900 Subject: [PATCH 3/9] swarm/api: fix typo (#17500) --- swarm/api/api.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/swarm/api/api.go b/swarm/api/api.go index 0d08eeea87..d733ad989c 100644 --- a/swarm/api/api.go +++ b/swarm/api/api.go @@ -111,7 +111,7 @@ func (e *NoResolverError) Error() string { } // MultiResolver is used to resolve URL addresses based on their TLDs. -// Each TLD can have multiple resolvers, and the resoluton from the +// Each TLD can have multiple resolvers, and the resolution from the // first one in the sequence will be returned. type MultiResolver struct { resolvers map[string][]ResolveValidator From f236ac710e075f2d3ef9352a0dc19b3bdf6f5bd8 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Tue, 28 Aug 2018 09:03:33 +0200 Subject: [PATCH 4/9] vendor: github.com/rjeczalik/notify update to master (#17527) --- .../rjeczalik/notify/watcher_fsevents_cgo.go | 6 +++--- .../rjeczalik/notify/watcher_fsevents_go1.10.go | 14 -------------- .../rjeczalik/notify/watcher_fsevents_go1.11.go | 9 --------- vendor/vendor.json | 6 +++--- 4 files changed, 6 insertions(+), 29 deletions(-) delete mode 100644 vendor/github.com/rjeczalik/notify/watcher_fsevents_go1.10.go delete mode 100644 vendor/github.com/rjeczalik/notify/watcher_fsevents_go1.11.go diff --git a/vendor/github.com/rjeczalik/notify/watcher_fsevents_cgo.go b/vendor/github.com/rjeczalik/notify/watcher_fsevents_cgo.go index 95ee704444..cf0416c37e 100644 --- a/vendor/github.com/rjeczalik/notify/watcher_fsevents_cgo.go +++ b/vendor/github.com/rjeczalik/notify/watcher_fsevents_cgo.go @@ -48,7 +48,7 @@ var wg sync.WaitGroup // used to wait until the runloop starts // started and is ready via the wg. It also serves purpose of a dummy source, // thanks to it the runloop does not return as it also has at least one source // registered. -var source = C.CFRunLoopSourceCreate(refZero, 0, &C.CFRunLoopSourceContext{ +var source = C.CFRunLoopSourceCreate(C.kCFAllocatorDefault, 0, &C.CFRunLoopSourceContext{ perform: (C.CFRunLoopPerformCallBack)(C.gosource), }) @@ -166,8 +166,8 @@ func (s *stream) Start() error { return nil } wg.Wait() - p := C.CFStringCreateWithCStringNoCopy(refZero, C.CString(s.path), C.kCFStringEncodingUTF8, refZero) - path := C.CFArrayCreate(refZero, (*unsafe.Pointer)(unsafe.Pointer(&p)), 1, nil) + p := C.CFStringCreateWithCStringNoCopy(C.kCFAllocatorDefault, C.CString(s.path), C.kCFStringEncodingUTF8, C.kCFAllocatorDefault) + path := C.CFArrayCreate(C.kCFAllocatorDefault, (*unsafe.Pointer)(unsafe.Pointer(&p)), 1, nil) ctx := C.FSEventStreamContext{} ref := C.EventStreamCreate(&ctx, C.uintptr_t(s.info), path, C.FSEventStreamEventId(atomic.LoadUint64(&since)), latency, flags) if ref == nilstream { diff --git a/vendor/github.com/rjeczalik/notify/watcher_fsevents_go1.10.go b/vendor/github.com/rjeczalik/notify/watcher_fsevents_go1.10.go deleted file mode 100644 index dd2433d20a..0000000000 --- a/vendor/github.com/rjeczalik/notify/watcher_fsevents_go1.10.go +++ /dev/null @@ -1,14 +0,0 @@ -// Copyright (c) 2018 The Notify Authors. All rights reserved. -// Use of this source code is governed by the MIT license that can be -// found in the LICENSE file. - -// +build darwin,!kqueue,cgo,!go1.11 - -package notify - -/* - #include -*/ -import "C" - -var refZero = (*C.struct___CFAllocator)(nil) diff --git a/vendor/github.com/rjeczalik/notify/watcher_fsevents_go1.11.go b/vendor/github.com/rjeczalik/notify/watcher_fsevents_go1.11.go deleted file mode 100644 index 92b406ce7c..0000000000 --- a/vendor/github.com/rjeczalik/notify/watcher_fsevents_go1.11.go +++ /dev/null @@ -1,9 +0,0 @@ -// Copyright (c) 2018 The Notify Authors. All rights reserved. -// Use of this source code is governed by the MIT license that can be -// found in the LICENSE file. - -// +build darwin,!kqueue,go1.11 - -package notify - -const refZero = 0 diff --git a/vendor/vendor.json b/vendor/vendor.json index 0bfc1ca0c1..fea2c804ea 100644 --- a/vendor/vendor.json +++ b/vendor/vendor.json @@ -370,10 +370,10 @@ "revisionTime": "2017-08-14T17:01:13Z" }, { - "checksumSHA1": "D8AVDI39CJ+jvw0HOotYU2gz54c=", + "checksumSHA1": "lU41NL1TEDtsrr0yUdp3SMB4Y9o=", "path": "github.com/rjeczalik/notify", - "revision": "4e54e7fd043e865c50bda93359fb78813a8d165b", - "revisionTime": "2018-08-08T20:39:25Z" + "revision": "0f065fa99b48b842c3fd3e2c8b194c6f2b69f6b8", + "revisionTime": "2018-08-27T19:31:19Z" }, { "checksumSHA1": "5uqO4ITTDMklKi3uNaE/D9LQ5nM=", From 0bec85f2e27d7b79f7700643cd9b38e74cecadfa Mon Sep 17 00:00:00 2001 From: Sheldon <11510383@mail.sustc.edu.cn> Date: Tue, 28 Aug 2018 15:04:33 +0800 Subject: [PATCH 5/9] core: fix typos in comment (#17531) --- core/tx_journal.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/core/tx_journal.go b/core/tx_journal.go index 1397e9fd34..41b5156d4a 100644 --- a/core/tx_journal.go +++ b/core/tx_journal.go @@ -34,7 +34,7 @@ var errNoActiveJournal = errors.New("no active journal") // devNull is a WriteCloser that just discards anything written into it. Its // goal is to allow the transaction journal to write into a fake journal when // loading transactions on startup without printing warnings due to no file -// being readt for write. +// being read for write. type devNull struct{} func (*devNull) Write(p []byte) (n int, err error) { return len(p), nil } @@ -57,7 +57,7 @@ func newTxJournal(path string) *txJournal { // load parses a transaction journal dump from disk, loading its contents into // the specified pool. func (journal *txJournal) load(add func([]*types.Transaction) []error) error { - // Skip the parsing if the journal file doens't exist at all + // Skip the parsing if the journal file doesn't exist at all if _, err := os.Stat(journal.path); os.IsNotExist(err) { return nil } @@ -78,7 +78,7 @@ func (journal *txJournal) load(add func([]*types.Transaction) []error) error { // Create a method to load a limited batch of transactions and bump the // appropriate progress counters. Then use this method to load all the - // journalled transactions in small-ish batches. + // journaled transactions in small-ish batches. loadBatch := func(txs types.Transactions) { for _, err := range add(txs) { if err != nil { @@ -103,7 +103,7 @@ func (journal *txJournal) load(add func([]*types.Transaction) []error) error { } break } - // New transaction parsed, queue up for later, import if threnshold is reached + // New transaction parsed, queue up for later, import if threshold is reached total++ if batch = append(batch, tx); batch.Len() > 1024 { From c64d72bea207ccaca3f6aded25d8730a4b8696cd Mon Sep 17 00:00:00 2001 From: Mymskmkt <1847234666@qq.com> Date: Tue, 28 Aug 2018 15:05:25 +0800 Subject: [PATCH 6/9] consensus/ethash: remove unnecessary type declaration (#17529) --- consensus/ethash/consensus.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/consensus/ethash/consensus.go b/consensus/ethash/consensus.go index 259cc5056b..af0e733ae7 100644 --- a/consensus/ethash/consensus.go +++ b/consensus/ethash/consensus.go @@ -38,10 +38,10 @@ import ( // Ethash proof-of-work protocol constants. var ( - FrontierBlockReward *big.Int = big.NewInt(5e+18) // Block reward in wei for successfully mining a block - ByzantiumBlockReward *big.Int = big.NewInt(3e+18) // Block reward in wei for successfully mining a block upward from Byzantium - maxUncles = 2 // Maximum number of uncles allowed in a single block - allowedFutureBlockTime = 15 * time.Second // Max time from current time allowed for blocks, before they're considered future blocks + FrontierBlockReward = big.NewInt(5e+18) // Block reward in wei for successfully mining a block + ByzantiumBlockReward = big.NewInt(3e+18) // Block reward in wei for successfully mining a block upward from Byzantium + maxUncles = 2 // Maximum number of uncles allowed in a single block + allowedFutureBlockTime = 15 * time.Second // Max time from current time allowed for blocks, before they're considered future blocks ) // Various error messages to mark blocks invalid. These should be private to From b69476b372a26679e5bdb33db3d508f2c955e7ff Mon Sep 17 00:00:00 2001 From: gary rong Date: Tue, 28 Aug 2018 15:08:16 +0800 Subject: [PATCH 7/9] all: make indexer configurable (#17188) --- core/chain_indexer.go | 1 - eth/backend.go | 4 +- eth/bloombits.go | 15 ++-- les/api_backend.go | 2 +- les/backend.go | 13 +-- les/bloombits.go | 4 +- les/commons.go | 6 +- les/handler.go | 13 +-- les/handler_test.go | 177 ++++++++++++++++++++------------------- les/helper_test.go | 159 +++++++++++++++++++++++++++++++---- les/odr.go | 15 +++- les/odr_requests.go | 20 ++++- les/odr_test.go | 45 ++++------ les/peer.go | 30 +++---- les/request_test.go | 43 +++------- les/server.go | 14 ++-- light/lightchain.go | 20 +++-- light/lightchain_test.go | 9 +- light/odr.go | 5 +- light/odr_test.go | 11 ++- light/odr_util.go | 18 ++-- light/postprocess.go | 174 ++++++++++++++++++++++---------------- light/trie_test.go | 2 +- light/txpool_test.go | 2 +- params/network_params.go | 32 ++++++- 25 files changed, 513 insertions(+), 321 deletions(-) diff --git a/core/chain_indexer.go b/core/chain_indexer.go index 11a7c96fa0..b80b517d91 100644 --- a/core/chain_indexer.go +++ b/core/chain_indexer.go @@ -322,7 +322,6 @@ func (c *ChainIndexer) updateLoop() { updating = false c.log.Info("Finished upgrading chain index") } - c.cascadedHead = c.storedSections*c.sectionSize - 1 for _, child := range c.children { c.log.Trace("Cascading chain index update", "head", c.cascadedHead) diff --git a/eth/backend.go b/eth/backend.go index c530d3c7c3..da7e0b2cd5 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -136,7 +136,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { gasPrice: config.MinerGasPrice, etherbase: config.Etherbase, bloomRequests: make(chan chan *bloombits.Retrieval), - bloomIndexer: NewBloomIndexer(chainDb, params.BloomBitsBlocks, bloomConfirms), + bloomIndexer: NewBloomIndexer(chainDb, params.BloomBitsBlocks, params.BloomConfirms), } log.Info("Initialising Ethereum protocol", "versions", ProtocolVersions, "network", config.NetworkId) @@ -426,7 +426,7 @@ func (s *Ethereum) Protocols() []p2p.Protocol { // Ethereum protocol implementation. func (s *Ethereum) Start(srvr *p2p.Server) error { // Start the bloom bits servicing goroutines - s.startBloomHandlers() + s.startBloomHandlers(params.BloomBitsBlocks) // Start the RPC service s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.NetVersion()) diff --git a/eth/bloombits.go b/eth/bloombits.go index eb18565e2c..c7bb561402 100644 --- a/eth/bloombits.go +++ b/eth/bloombits.go @@ -27,7 +27,6 @@ import ( "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" - "github.com/ethereum/go-ethereum/params" ) const ( @@ -50,7 +49,7 @@ const ( // startBloomHandlers starts a batch of goroutines to accept bloom bit database // retrievals from possibly a range of filters and serving the data to satisfy. -func (eth *Ethereum) startBloomHandlers() { +func (eth *Ethereum) startBloomHandlers(sectionSize uint64) { for i := 0; i < bloomServiceThreads; i++ { go func() { for { @@ -62,9 +61,9 @@ func (eth *Ethereum) startBloomHandlers() { task := <-request task.Bitsets = make([][]byte, len(task.Sections)) for i, section := range task.Sections { - head := rawdb.ReadCanonicalHash(eth.chainDb, (section+1)*params.BloomBitsBlocks-1) + head := rawdb.ReadCanonicalHash(eth.chainDb, (section+1)*sectionSize-1) if compVector, err := rawdb.ReadBloomBits(eth.chainDb, task.Bit, section, head); err == nil { - if blob, err := bitutil.DecompressBytes(compVector, int(params.BloomBitsBlocks)/8); err == nil { + if blob, err := bitutil.DecompressBytes(compVector, int(sectionSize/8)); err == nil { task.Bitsets[i] = blob } else { task.Error = err @@ -81,10 +80,6 @@ func (eth *Ethereum) startBloomHandlers() { } const ( - // bloomConfirms is the number of confirmation blocks before a bloom section is - // considered probably final and its rotated bits are calculated. - bloomConfirms = 256 - // bloomThrottling is the time to wait between processing two consecutive index // sections. It's useful during chain upgrades to prevent disk overload. bloomThrottling = 100 * time.Millisecond @@ -102,14 +97,14 @@ type BloomIndexer struct { // NewBloomIndexer returns a chain indexer that generates bloom bits data for the // canonical chain for fast logs filtering. -func NewBloomIndexer(db ethdb.Database, size, confReq uint64) *core.ChainIndexer { +func NewBloomIndexer(db ethdb.Database, size, confirms uint64) *core.ChainIndexer { backend := &BloomIndexer{ db: db, size: size, } table := ethdb.NewTable(db, string(rawdb.BloomBitsIndexPrefix)) - return core.NewChainIndexer(db, table, backend, size, confReq, bloomThrottling, "bloombits") + return core.NewChainIndexer(db, table, backend, size, confirms, bloomThrottling, "bloombits") } // Reset implements core.ChainIndexerBackend, starting a new bloombits index diff --git a/les/api_backend.go b/les/api_backend.go index 4232d3ae07..aa748a4ea1 100644 --- a/les/api_backend.go +++ b/les/api_backend.go @@ -192,7 +192,7 @@ func (b *LesApiBackend) BloomStatus() (uint64, uint64) { return 0, 0 } sections, _, _ := b.eth.bloomIndexer.Sections() - return light.BloomTrieFrequency, sections + return params.BloomBitsBlocksClient, sections } func (b *LesApiBackend) ServiceFilter(ctx context.Context, session *bloombits.MatcherSession) { diff --git a/les/backend.go b/les/backend.go index 00025ba634..75049da085 100644 --- a/les/backend.go +++ b/les/backend.go @@ -95,6 +95,7 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) { lesCommons: lesCommons{ chainDb: chainDb, config: config, + iConfig: light.DefaultClientIndexerConfig, }, chainConfig: chainConfig, eventMux: ctx.EventMux, @@ -105,16 +106,16 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) { shutdownChan: make(chan bool), networkId: config.NetworkId, bloomRequests: make(chan chan *bloombits.Retrieval), - bloomIndexer: eth.NewBloomIndexer(chainDb, light.BloomTrieFrequency, light.HelperTrieConfirmations), + bloomIndexer: eth.NewBloomIndexer(chainDb, params.BloomBitsBlocksClient, params.HelperTrieConfirmations), } leth.relay = NewLesTxRelay(peers, leth.reqDist) leth.serverPool = newServerPool(chainDb, quitSync, &leth.wg) leth.retriever = newRetrieveManager(peers, leth.reqDist, leth.serverPool) - leth.odr = NewLesOdr(chainDb, leth.retriever) - leth.chtIndexer = light.NewChtIndexer(chainDb, true, leth.odr) - leth.bloomTrieIndexer = light.NewBloomTrieIndexer(chainDb, true, leth.odr) + leth.odr = NewLesOdr(chainDb, light.DefaultClientIndexerConfig, leth.retriever) + leth.chtIndexer = light.NewChtIndexer(chainDb, leth.odr, params.CHTFrequencyClient, params.HelperTrieConfirmations) + leth.bloomTrieIndexer = light.NewBloomTrieIndexer(chainDb, leth.odr, params.BloomBitsBlocksClient, params.BloomTrieFrequency) leth.odr.SetIndexers(leth.chtIndexer, leth.bloomTrieIndexer, leth.bloomIndexer) // Note: NewLightChain adds the trusted checkpoint so it needs an ODR with @@ -135,7 +136,7 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) { } leth.txPool = light.NewTxPool(leth.chainConfig, leth.blockchain, leth.relay) - if leth.protocolManager, err = NewProtocolManager(leth.chainConfig, true, config.NetworkId, leth.eventMux, leth.engine, leth.peers, leth.blockchain, nil, chainDb, leth.odr, leth.relay, leth.serverPool, quitSync, &leth.wg); err != nil { + if leth.protocolManager, err = NewProtocolManager(leth.chainConfig, light.DefaultClientIndexerConfig, true, config.NetworkId, leth.eventMux, leth.engine, leth.peers, leth.blockchain, nil, chainDb, leth.odr, leth.relay, leth.serverPool, quitSync, &leth.wg); err != nil { return nil, err } leth.ApiBackend = &LesApiBackend{leth, nil} @@ -230,8 +231,8 @@ func (s *LightEthereum) Protocols() []p2p.Protocol { // Start implements node.Service, starting all internal goroutines needed by the // Ethereum protocol implementation. func (s *LightEthereum) Start(srvr *p2p.Server) error { - s.startBloomHandlers() log.Warn("Light client mode is an experimental feature") + s.startBloomHandlers(params.BloomBitsBlocksClient) s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.networkId) // clients are searching for the first advertised protocol in the list protocolVersion := AdvertiseProtocolVersions[0] diff --git a/les/bloombits.go b/les/bloombits.go index 2871a90064..aea0fcd5f4 100644 --- a/les/bloombits.go +++ b/les/bloombits.go @@ -43,7 +43,7 @@ const ( // startBloomHandlers starts a batch of goroutines to accept bloom bit database // retrievals from possibly a range of filters and serving the data to satisfy. -func (eth *LightEthereum) startBloomHandlers() { +func (eth *LightEthereum) startBloomHandlers(sectionSize uint64) { for i := 0; i < bloomServiceThreads; i++ { go func() { for { @@ -57,7 +57,7 @@ func (eth *LightEthereum) startBloomHandlers() { compVectors, err := light.GetBloomBits(task.Context, eth.odr, task.Bit, task.Sections) if err == nil { for i := range task.Sections { - if blob, err := bitutil.DecompressBytes(compVectors[i], int(light.BloomTrieFrequency/8)); err == nil { + if blob, err := bitutil.DecompressBytes(compVectors[i], int(sectionSize/8)); err == nil { task.Bitsets[i] = blob } else { task.Error = err diff --git a/les/commons.go b/les/commons.go index d8e9412951..a97687993d 100644 --- a/les/commons.go +++ b/les/commons.go @@ -33,6 +33,7 @@ import ( // lesCommons contains fields needed by both server and client. type lesCommons struct { config *eth.Config + iConfig *light.IndexerConfig chainDb ethdb.Database protocolManager *ProtocolManager chtIndexer, bloomTrieIndexer *core.ChainIndexer @@ -81,7 +82,7 @@ func (c *lesCommons) nodeInfo() interface{} { if !c.protocolManager.lightSync { // convert to client section size if running in server mode - sections /= light.CHTFrequencyClient / light.CHTFrequencyServer + sections /= c.iConfig.PairChtSize / c.iConfig.ChtSize } if sections2 < sections { @@ -94,7 +95,8 @@ func (c *lesCommons) nodeInfo() interface{} { if c.protocolManager.lightSync { chtRoot = light.GetChtRoot(c.chainDb, sectionIndex, sectionHead) } else { - chtRoot = light.GetChtV2Root(c.chainDb, sectionIndex, sectionHead) + idxV2 := (sectionIndex+1)*c.iConfig.PairChtSize/c.iConfig.ChtSize - 1 + chtRoot = light.GetChtRoot(c.chainDb, idxV2, sectionHead) } cht = light.TrustedCheckpoint{ SectionIdx: sectionIndex, diff --git a/les/handler.go b/les/handler.go index ca40eaabfa..243a6dabd4 100644 --- a/les/handler.go +++ b/les/handler.go @@ -94,6 +94,7 @@ type ProtocolManager struct { txrelay *LesTxRelay networkId uint64 chainConfig *params.ChainConfig + iConfig *light.IndexerConfig blockchain BlockChain chainDb ethdb.Database odr *LesOdr @@ -123,13 +124,14 @@ type ProtocolManager struct { // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable // with the ethereum network. -func NewProtocolManager(chainConfig *params.ChainConfig, lightSync bool, networkId uint64, mux *event.TypeMux, engine consensus.Engine, peers *peerSet, blockchain BlockChain, txpool txPool, chainDb ethdb.Database, odr *LesOdr, txrelay *LesTxRelay, serverPool *serverPool, quitSync chan struct{}, wg *sync.WaitGroup) (*ProtocolManager, error) { +func NewProtocolManager(chainConfig *params.ChainConfig, indexerConfig *light.IndexerConfig, lightSync bool, networkId uint64, mux *event.TypeMux, engine consensus.Engine, peers *peerSet, blockchain BlockChain, txpool txPool, chainDb ethdb.Database, odr *LesOdr, txrelay *LesTxRelay, serverPool *serverPool, quitSync chan struct{}, wg *sync.WaitGroup) (*ProtocolManager, error) { // Create the protocol manager with the base fields manager := &ProtocolManager{ lightSync: lightSync, eventMux: mux, blockchain: blockchain, chainConfig: chainConfig, + iConfig: indexerConfig, chainDb: chainDb, odr: odr, networkId: networkId, @@ -882,7 +884,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { trieDb := trie.NewDatabase(ethdb.NewTable(pm.chainDb, light.ChtTablePrefix)) for _, req := range req.Reqs { if header := pm.blockchain.GetHeaderByNumber(req.BlockNum); header != nil { - sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, req.ChtNum*light.CHTFrequencyServer-1) + sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, req.ChtNum*pm.iConfig.ChtSize-1) if root := light.GetChtRoot(pm.chainDb, req.ChtNum-1, sectionHead); root != (common.Hash{}) { trie, err := trie.New(root, trieDb) if err != nil { @@ -1137,10 +1139,11 @@ func (pm *ProtocolManager) getAccount(statedb *state.StateDB, root, hash common. func (pm *ProtocolManager) getHelperTrie(id uint, idx uint64) (common.Hash, string) { switch id { case htCanonical: - sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*light.CHTFrequencyClient-1) - return light.GetChtV2Root(pm.chainDb, idx, sectionHead), light.ChtTablePrefix + idxV1 := (idx+1)*(pm.iConfig.PairChtSize/pm.iConfig.ChtSize) - 1 + sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idxV1+1)*pm.iConfig.ChtSize-1) + return light.GetChtRoot(pm.chainDb, idxV1, sectionHead), light.ChtTablePrefix case htBloomBits: - sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*light.BloomTrieFrequency-1) + sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*pm.iConfig.BloomTrieSize-1) return light.GetBloomTrieRoot(pm.chainDb, idx, sectionHead), light.BloomTrieTablePrefix } return common.Hash{}, "" diff --git a/les/handler_test.go b/les/handler_test.go index 31aad3ed45..43be7f41b1 100644 --- a/les/handler_test.go +++ b/les/handler_test.go @@ -51,10 +51,9 @@ func TestGetBlockHeadersLes1(t *testing.T) { testGetBlockHeaders(t, 1) } func TestGetBlockHeadersLes2(t *testing.T) { testGetBlockHeaders(t, 2) } func testGetBlockHeaders(t *testing.T, protocol int) { - pm := newTestProtocolManagerMust(t, false, downloader.MaxHashFetch+15, nil, nil, nil, ethdb.NewMemDatabase()) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", protocol, pm, true) - defer peer.close() + server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, protocol, nil) + defer tearDown() + bc := server.pm.blockchain.(*core.BlockChain) // Create a "random" unknown hash for testing var unknown common.Hash @@ -167,9 +166,9 @@ func testGetBlockHeaders(t *testing.T, protocol int) { } // Send the hash request and verify the response reqID++ - cost := peer.GetRequestCost(GetBlockHeadersMsg, int(tt.query.Amount)) - sendRequest(peer.app, GetBlockHeadersMsg, reqID, cost, tt.query) - if err := expectResponse(peer.app, BlockHeadersMsg, reqID, testBufLimit, headers); err != nil { + cost := server.tPeer.GetRequestCost(GetBlockHeadersMsg, int(tt.query.Amount)) + sendRequest(server.tPeer.app, GetBlockHeadersMsg, reqID, cost, tt.query) + if err := expectResponse(server.tPeer.app, BlockHeadersMsg, reqID, testBufLimit, headers); err != nil { t.Errorf("test %d: headers mismatch: %v", i, err) } } @@ -180,10 +179,9 @@ func TestGetBlockBodiesLes1(t *testing.T) { testGetBlockBodies(t, 1) } func TestGetBlockBodiesLes2(t *testing.T) { testGetBlockBodies(t, 2) } func testGetBlockBodies(t *testing.T, protocol int) { - pm := newTestProtocolManagerMust(t, false, downloader.MaxBlockFetch+15, nil, nil, nil, ethdb.NewMemDatabase()) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", protocol, pm, true) - defer peer.close() + server, tearDown := newServerEnv(t, downloader.MaxBlockFetch+15, protocol, nil) + defer tearDown() + bc := server.pm.blockchain.(*core.BlockChain) // Create a batch of tests for various scenarios limit := MaxBodyFetch @@ -243,9 +241,9 @@ func testGetBlockBodies(t *testing.T, protocol int) { } reqID++ // Send the hash request and verify the response - cost := peer.GetRequestCost(GetBlockBodiesMsg, len(hashes)) - sendRequest(peer.app, GetBlockBodiesMsg, reqID, cost, hashes) - if err := expectResponse(peer.app, BlockBodiesMsg, reqID, testBufLimit, bodies); err != nil { + cost := server.tPeer.GetRequestCost(GetBlockBodiesMsg, len(hashes)) + sendRequest(server.tPeer.app, GetBlockBodiesMsg, reqID, cost, hashes) + if err := expectResponse(server.tPeer.app, BlockBodiesMsg, reqID, testBufLimit, bodies); err != nil { t.Errorf("test %d: bodies mismatch: %v", i, err) } } @@ -257,10 +255,9 @@ func TestGetCodeLes2(t *testing.T) { testGetCode(t, 2) } func testGetCode(t *testing.T, protocol int) { // Assemble the test environment - pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, ethdb.NewMemDatabase()) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", protocol, pm, true) - defer peer.close() + server, tearDown := newServerEnv(t, 4, protocol, nil) + defer tearDown() + bc := server.pm.blockchain.(*core.BlockChain) var codereqs []*CodeReq var codes [][]byte @@ -277,9 +274,9 @@ func testGetCode(t *testing.T, protocol int) { } } - cost := peer.GetRequestCost(GetCodeMsg, len(codereqs)) - sendRequest(peer.app, GetCodeMsg, 42, cost, codereqs) - if err := expectResponse(peer.app, CodeMsg, 42, testBufLimit, codes); err != nil { + cost := server.tPeer.GetRequestCost(GetCodeMsg, len(codereqs)) + sendRequest(server.tPeer.app, GetCodeMsg, 42, cost, codereqs) + if err := expectResponse(server.tPeer.app, CodeMsg, 42, testBufLimit, codes); err != nil { t.Errorf("codes mismatch: %v", err) } } @@ -290,11 +287,9 @@ func TestGetReceiptLes2(t *testing.T) { testGetReceipt(t, 2) } func testGetReceipt(t *testing.T, protocol int) { // Assemble the test environment - db := ethdb.NewMemDatabase() - pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", protocol, pm, true) - defer peer.close() + server, tearDown := newServerEnv(t, 4, protocol, nil) + defer tearDown() + bc := server.pm.blockchain.(*core.BlockChain) // Collect the hashes to request, and the response to expect hashes, receipts := []common.Hash{}, []types.Receipts{} @@ -302,12 +297,12 @@ func testGetReceipt(t *testing.T, protocol int) { block := bc.GetBlockByNumber(i) hashes = append(hashes, block.Hash()) - receipts = append(receipts, rawdb.ReadReceipts(db, block.Hash(), block.NumberU64())) + receipts = append(receipts, rawdb.ReadReceipts(server.db, block.Hash(), block.NumberU64())) } // Send the hash request and verify the response - cost := peer.GetRequestCost(GetReceiptsMsg, len(hashes)) - sendRequest(peer.app, GetReceiptsMsg, 42, cost, hashes) - if err := expectResponse(peer.app, ReceiptsMsg, 42, testBufLimit, receipts); err != nil { + cost := server.tPeer.GetRequestCost(GetReceiptsMsg, len(hashes)) + sendRequest(server.tPeer.app, GetReceiptsMsg, 42, cost, hashes) + if err := expectResponse(server.tPeer.app, ReceiptsMsg, 42, testBufLimit, receipts); err != nil { t.Errorf("receipts mismatch: %v", err) } } @@ -318,11 +313,9 @@ func TestGetProofsLes2(t *testing.T) { testGetProofs(t, 2) } func testGetProofs(t *testing.T, protocol int) { // Assemble the test environment - db := ethdb.NewMemDatabase() - pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", protocol, pm, true) - defer peer.close() + server, tearDown := newServerEnv(t, 4, protocol, nil) + defer tearDown() + bc := server.pm.blockchain.(*core.BlockChain) var ( proofreqs []ProofReq @@ -334,7 +327,7 @@ func testGetProofs(t *testing.T, protocol int) { for i := uint64(0); i <= bc.CurrentBlock().NumberU64(); i++ { header := bc.GetHeaderByNumber(i) root := header.Root - trie, _ := trie.New(root, trie.NewDatabase(db)) + trie, _ := trie.New(root, trie.NewDatabase(server.db)) for _, acc := range accounts { req := ProofReq{ @@ -356,15 +349,15 @@ func testGetProofs(t *testing.T, protocol int) { // Send the proof request and verify the response switch protocol { case 1: - cost := peer.GetRequestCost(GetProofsV1Msg, len(proofreqs)) - sendRequest(peer.app, GetProofsV1Msg, 42, cost, proofreqs) - if err := expectResponse(peer.app, ProofsV1Msg, 42, testBufLimit, proofsV1); err != nil { + cost := server.tPeer.GetRequestCost(GetProofsV1Msg, len(proofreqs)) + sendRequest(server.tPeer.app, GetProofsV1Msg, 42, cost, proofreqs) + if err := expectResponse(server.tPeer.app, ProofsV1Msg, 42, testBufLimit, proofsV1); err != nil { t.Errorf("proofs mismatch: %v", err) } case 2: - cost := peer.GetRequestCost(GetProofsV2Msg, len(proofreqs)) - sendRequest(peer.app, GetProofsV2Msg, 42, cost, proofreqs) - if err := expectResponse(peer.app, ProofsV2Msg, 42, testBufLimit, proofsV2.NodeList()); err != nil { + cost := server.tPeer.GetRequestCost(GetProofsV2Msg, len(proofreqs)) + sendRequest(server.tPeer.app, GetProofsV2Msg, 42, cost, proofreqs) + if err := expectResponse(server.tPeer.app, ProofsV2Msg, 42, testBufLimit, proofsV2.NodeList()); err != nil { t.Errorf("proofs mismatch: %v", err) } } @@ -375,28 +368,33 @@ func TestGetCHTProofsLes1(t *testing.T) { testGetCHTProofs(t, 1) } func TestGetCHTProofsLes2(t *testing.T) { testGetCHTProofs(t, 2) } func testGetCHTProofs(t *testing.T, protocol int) { - // Figure out the client's CHT frequency - frequency := uint64(light.CHTFrequencyClient) - if protocol == 1 { - frequency = uint64(light.CHTFrequencyServer) + config := light.TestServerIndexerConfig + frequency := config.ChtSize + if protocol == 2 { + frequency = config.PairChtSize } - // Assemble the test environment - db := ethdb.NewMemDatabase() - pm := newTestProtocolManagerMust(t, false, int(frequency)+light.HelperTrieProcessConfirmations, testChainGen, nil, nil, db) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", protocol, pm, true) - defer peer.close() - // Wait a while for the CHT indexer to process the new headers - time.Sleep(100 * time.Millisecond * time.Duration(frequency/light.CHTFrequencyServer)) // Chain indexer throttling - time.Sleep(250 * time.Millisecond) // CI tester slack + waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) { + expectSections := frequency / config.ChtSize + for { + cs, _, _ := cIndexer.Sections() + bs, _, _ := bIndexer.Sections() + if cs >= expectSections && bs >= expectSections { + break + } + time.Sleep(10 * time.Millisecond) + } + } + server, tearDown := newServerEnv(t, int(frequency+config.ChtConfirms), protocol, waitIndexers) + defer tearDown() + bc := server.pm.blockchain.(*core.BlockChain) // Assemble the proofs from the different protocols - header := bc.GetHeaderByNumber(frequency) + header := bc.GetHeaderByNumber(frequency - 1) rlp, _ := rlp.EncodeToBytes(header) key := make([]byte, 8) - binary.BigEndian.PutUint64(key, frequency) + binary.BigEndian.PutUint64(key, frequency-1) proofsV1 := []ChtResp{{ Header: header, @@ -406,41 +404,41 @@ func testGetCHTProofs(t *testing.T, protocol int) { } switch protocol { case 1: - root := light.GetChtRoot(db, 0, bc.GetHeaderByNumber(frequency-1).Hash()) - trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(db, light.ChtTablePrefix))) + root := light.GetChtRoot(server.db, 0, bc.GetHeaderByNumber(frequency-1).Hash()) + trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.ChtTablePrefix))) var proof light.NodeList trie.Prove(key, 0, &proof) proofsV1[0].Proof = proof case 2: - root := light.GetChtV2Root(db, 0, bc.GetHeaderByNumber(frequency-1).Hash()) - trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(db, light.ChtTablePrefix))) + root := light.GetChtRoot(server.db, (frequency/config.ChtSize)-1, bc.GetHeaderByNumber(frequency-1).Hash()) + trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.ChtTablePrefix))) trie.Prove(key, 0, &proofsV2.Proofs) } // Assemble the requests for the different protocols requestsV1 := []ChtReq{{ - ChtNum: 1, - BlockNum: frequency, + ChtNum: frequency / config.ChtSize, + BlockNum: frequency - 1, }} requestsV2 := []HelperTrieReq{{ Type: htCanonical, - TrieIdx: 0, + TrieIdx: frequency/config.PairChtSize - 1, Key: key, AuxReq: auxHeader, }} // Send the proof request and verify the response switch protocol { case 1: - cost := peer.GetRequestCost(GetHeaderProofsMsg, len(requestsV1)) - sendRequest(peer.app, GetHeaderProofsMsg, 42, cost, requestsV1) - if err := expectResponse(peer.app, HeaderProofsMsg, 42, testBufLimit, proofsV1); err != nil { + cost := server.tPeer.GetRequestCost(GetHeaderProofsMsg, len(requestsV1)) + sendRequest(server.tPeer.app, GetHeaderProofsMsg, 42, cost, requestsV1) + if err := expectResponse(server.tPeer.app, HeaderProofsMsg, 42, testBufLimit, proofsV1); err != nil { t.Errorf("proofs mismatch: %v", err) } case 2: - cost := peer.GetRequestCost(GetHelperTrieProofsMsg, len(requestsV2)) - sendRequest(peer.app, GetHelperTrieProofsMsg, 42, cost, requestsV2) - if err := expectResponse(peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofsV2); err != nil { + cost := server.tPeer.GetRequestCost(GetHelperTrieProofsMsg, len(requestsV2)) + sendRequest(server.tPeer.app, GetHelperTrieProofsMsg, 42, cost, requestsV2) + if err := expectResponse(server.tPeer.app, HelperTrieProofsMsg, 42, testBufLimit, proofsV2); err != nil { t.Errorf("proofs mismatch: %v", err) } } @@ -448,24 +446,31 @@ func testGetCHTProofs(t *testing.T, protocol int) { // Tests that bloombits proofs can be correctly retrieved. func TestGetBloombitsProofs(t *testing.T) { - // Assemble the test environment - db := ethdb.NewMemDatabase() - pm := newTestProtocolManagerMust(t, false, light.BloomTrieFrequency+256, testChainGen, nil, nil, db) - bc := pm.blockchain.(*core.BlockChain) - peer, _ := newTestPeer(t, "peer", 2, pm, true) - defer peer.close() + config := light.TestServerIndexerConfig - // Wait a while for the bloombits indexer to process the new headers - time.Sleep(100 * time.Millisecond * time.Duration(light.BloomTrieFrequency/4096)) // Chain indexer throttling - time.Sleep(250 * time.Millisecond) // CI tester slack + waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) { + for { + cs, _, _ := cIndexer.Sections() + bs, _, _ := bIndexer.Sections() + bts, _, _ := btIndexer.Sections() + if cs >= 8 && bs >= 8 && bts >= 1 { + break + } + time.Sleep(10 * time.Millisecond) + } + } + server, tearDown := newServerEnv(t, int(config.BloomTrieSize+config.BloomTrieConfirms), 2, waitIndexers) + defer tearDown() + bc := server.pm.blockchain.(*core.BlockChain) // Request and verify each bit of the bloom bits proofs for bit := 0; bit < 2048; bit++ { - // Assemble therequest and proofs for the bloombits + // Assemble the request and proofs for the bloombits key := make([]byte, 10) binary.BigEndian.PutUint16(key[:2], uint16(bit)) - binary.BigEndian.PutUint64(key[2:], uint64(light.BloomTrieFrequency)) + // Only the first bloom section has data. + binary.BigEndian.PutUint64(key[2:], 0) requests := []HelperTrieReq{{ Type: htBloomBits, @@ -474,14 +479,14 @@ func TestGetBloombitsProofs(t *testing.T) { }} var proofs HelperTrieResps - root := light.GetBloomTrieRoot(db, 0, bc.GetHeaderByNumber(light.BloomTrieFrequency-1).Hash()) - trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(db, light.BloomTrieTablePrefix))) + root := light.GetBloomTrieRoot(server.db, 0, bc.GetHeaderByNumber(config.BloomTrieSize-1).Hash()) + trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.BloomTrieTablePrefix))) trie.Prove(key, 0, &proofs.Proofs) // Send the proof request and verify the response - cost := peer.GetRequestCost(GetHelperTrieProofsMsg, len(requests)) - sendRequest(peer.app, GetHelperTrieProofsMsg, 42, cost, requests) - if err := expectResponse(peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofs); err != nil { + cost := server.tPeer.GetRequestCost(GetHelperTrieProofsMsg, len(requests)) + sendRequest(server.tPeer.app, GetHelperTrieProofsMsg, 42, cost, requests) + if err := expectResponse(server.tPeer.app, HelperTrieProofsMsg, 42, testBufLimit, proofs); err != nil { t.Errorf("bit %d: proofs mismatch: %v", bit, err) } } diff --git a/les/helper_test.go b/les/helper_test.go index 8817c20c7d..206ee2d920 100644 --- a/les/helper_test.go +++ b/les/helper_test.go @@ -24,6 +24,7 @@ import ( "math/big" "sync" "testing" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/consensus/ethash" @@ -123,6 +124,15 @@ func testChainGen(i int, block *core.BlockGen) { } } +// testIndexers creates a set of indexers with specified params for testing purpose. +func testIndexers(db ethdb.Database, odr light.OdrBackend, iConfig *light.IndexerConfig) (*core.ChainIndexer, *core.ChainIndexer, *core.ChainIndexer) { + chtIndexer := light.NewChtIndexer(db, odr, iConfig.ChtSize, iConfig.ChtConfirms) + bloomIndexer := eth.NewBloomIndexer(db, iConfig.BloomSize, iConfig.BloomConfirms) + bloomTrieIndexer := light.NewBloomTrieIndexer(db, odr, iConfig.BloomSize, iConfig.BloomTrieSize) + bloomIndexer.AddChildIndexer(bloomTrieIndexer) + return chtIndexer, bloomIndexer, bloomTrieIndexer +} + func testRCL() RequestCostList { cl := make(RequestCostList, len(reqList)) for i, code := range reqList { @@ -134,9 +144,9 @@ func testRCL() RequestCostList { } // newTestProtocolManager creates a new protocol manager for testing purposes, -// with the given number of blocks already known, and potential notification -// channels for different events. -func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *core.BlockGen), peers *peerSet, odr *LesOdr, db ethdb.Database) (*ProtocolManager, error) { +// with the given number of blocks already known, potential notification +// channels for different events and relative chain indexers array. +func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *core.BlockGen), odr *LesOdr, peers *peerSet, db ethdb.Database) (*ProtocolManager, error) { var ( evmux = new(event.TypeMux) engine = ethash.NewFaker() @@ -155,16 +165,6 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor chain, _ = light.NewLightChain(odr, gspec.Config, engine) } else { blockchain, _ := core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}) - - chtIndexer := light.NewChtIndexer(db, false, nil) - chtIndexer.Start(blockchain) - - bbtIndexer := light.NewBloomTrieIndexer(db, false, nil) - - bloomIndexer := eth.NewBloomIndexer(db, params.BloomBitsBlocks, light.HelperTrieProcessConfirmations) - bloomIndexer.AddChildIndexer(bbtIndexer) - bloomIndexer.Start(blockchain) - gchain, _ := core.GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, blocks, generator) if _, err := blockchain.InsertChain(gchain); err != nil { panic(err) @@ -172,7 +172,11 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor chain = blockchain } - pm, err := NewProtocolManager(gspec.Config, lightSync, NetworkId, evmux, engine, peers, chain, nil, db, odr, nil, nil, make(chan struct{}), new(sync.WaitGroup)) + indexConfig := light.TestServerIndexerConfig + if lightSync { + indexConfig = light.TestClientIndexerConfig + } + pm, err := NewProtocolManager(gspec.Config, indexConfig, lightSync, NetworkId, evmux, engine, peers, chain, nil, db, odr, nil, nil, make(chan struct{}), new(sync.WaitGroup)) if err != nil { return nil, err } @@ -193,11 +197,11 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor } // newTestProtocolManagerMust creates a new protocol manager for testing purposes, -// with the given number of blocks already known, and potential notification -// channels for different events. In case of an error, the constructor force- +// with the given number of blocks already known, potential notification +// channels for different events and relative chain indexers array. In case of an error, the constructor force- // fails the test. -func newTestProtocolManagerMust(t *testing.T, lightSync bool, blocks int, generator func(int, *core.BlockGen), peers *peerSet, odr *LesOdr, db ethdb.Database) *ProtocolManager { - pm, err := newTestProtocolManager(lightSync, blocks, generator, peers, odr, db) +func newTestProtocolManagerMust(t *testing.T, lightSync bool, blocks int, generator func(int, *core.BlockGen), odr *LesOdr, peers *peerSet, db ethdb.Database) *ProtocolManager { + pm, err := newTestProtocolManager(lightSync, blocks, generator, odr, peers, db) if err != nil { t.Fatalf("Failed to create protocol manager: %v", err) } @@ -320,3 +324,122 @@ func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNu func (p *testPeer) close() { p.app.Close() } + +// TestEntity represents a network entity for testing with necessary auxiliary fields. +type TestEntity struct { + db ethdb.Database + rPeer *peer + tPeer *testPeer + peers *peerSet + pm *ProtocolManager + // Indexers + chtIndexer *core.ChainIndexer + bloomIndexer *core.ChainIndexer + bloomTrieIndexer *core.ChainIndexer +} + +// newServerEnv creates a server testing environment with a connected test peer for testing purpose. +func newServerEnv(t *testing.T, blocks int, protocol int, waitIndexers func(*core.ChainIndexer, *core.ChainIndexer, *core.ChainIndexer)) (*TestEntity, func()) { + db := ethdb.NewMemDatabase() + cIndexer, bIndexer, btIndexer := testIndexers(db, nil, light.TestServerIndexerConfig) + + pm := newTestProtocolManagerMust(t, false, blocks, testChainGen, nil, nil, db) + peer, _ := newTestPeer(t, "peer", protocol, pm, true) + + cIndexer.Start(pm.blockchain.(*core.BlockChain)) + bIndexer.Start(pm.blockchain.(*core.BlockChain)) + + // Wait until indexers generate enough index data. + if waitIndexers != nil { + waitIndexers(cIndexer, bIndexer, btIndexer) + } + + return &TestEntity{ + db: db, + tPeer: peer, + pm: pm, + chtIndexer: cIndexer, + bloomIndexer: bIndexer, + bloomTrieIndexer: btIndexer, + }, func() { + peer.close() + // Note bloom trie indexer will be closed by it parent recursively. + cIndexer.Close() + bIndexer.Close() + } +} + +// newClientServerEnv creates a client/server arch environment with a connected les server and light client pair +// for testing purpose. +func newClientServerEnv(t *testing.T, blocks int, protocol int, waitIndexers func(*core.ChainIndexer, *core.ChainIndexer, *core.ChainIndexer), newPeer bool) (*TestEntity, *TestEntity, func()) { + db, ldb := ethdb.NewMemDatabase(), ethdb.NewMemDatabase() + peers, lPeers := newPeerSet(), newPeerSet() + + dist := newRequestDistributor(lPeers, make(chan struct{})) + rm := newRetrieveManager(lPeers, dist, nil) + odr := NewLesOdr(ldb, light.TestClientIndexerConfig, rm) + + cIndexer, bIndexer, btIndexer := testIndexers(db, nil, light.TestServerIndexerConfig) + lcIndexer, lbIndexer, lbtIndexer := testIndexers(ldb, odr, light.TestClientIndexerConfig) + odr.SetIndexers(lcIndexer, lbtIndexer, lbIndexer) + + pm := newTestProtocolManagerMust(t, false, blocks, testChainGen, nil, peers, db) + lpm := newTestProtocolManagerMust(t, true, 0, nil, odr, lPeers, ldb) + + startIndexers := func(clientMode bool, pm *ProtocolManager) { + if clientMode { + lcIndexer.Start(pm.blockchain.(*light.LightChain)) + lbIndexer.Start(pm.blockchain.(*light.LightChain)) + } else { + cIndexer.Start(pm.blockchain.(*core.BlockChain)) + bIndexer.Start(pm.blockchain.(*core.BlockChain)) + } + } + + startIndexers(false, pm) + startIndexers(true, lpm) + + // Execute wait until function if it is specified. + if waitIndexers != nil { + waitIndexers(cIndexer, bIndexer, btIndexer) + } + + var ( + peer, lPeer *peer + err1, err2 <-chan error + ) + if newPeer { + peer, err1, lPeer, err2 = newTestPeerPair("peer", protocol, pm, lpm) + select { + case <-time.After(time.Millisecond * 100): + case err := <-err1: + t.Fatalf("peer 1 handshake error: %v", err) + case err := <-err2: + t.Fatalf("peer 2 handshake error: %v", err) + } + } + + return &TestEntity{ + db: db, + pm: pm, + rPeer: peer, + peers: peers, + chtIndexer: cIndexer, + bloomIndexer: bIndexer, + bloomTrieIndexer: btIndexer, + }, &TestEntity{ + db: ldb, + pm: lpm, + rPeer: lPeer, + peers: lPeers, + chtIndexer: lcIndexer, + bloomIndexer: lbIndexer, + bloomTrieIndexer: lbtIndexer, + }, func() { + // Note bloom trie indexers will be closed by their parents recursively. + cIndexer.Close() + bIndexer.Close() + lcIndexer.Close() + lbIndexer.Close() + } +} diff --git a/les/odr.go b/les/odr.go index 2ad28d5d9f..9def05a676 100644 --- a/les/odr.go +++ b/les/odr.go @@ -28,16 +28,18 @@ import ( // LesOdr implements light.OdrBackend type LesOdr struct { db ethdb.Database + indexerConfig *light.IndexerConfig chtIndexer, bloomTrieIndexer, bloomIndexer *core.ChainIndexer retriever *retrieveManager stop chan struct{} } -func NewLesOdr(db ethdb.Database, retriever *retrieveManager) *LesOdr { +func NewLesOdr(db ethdb.Database, config *light.IndexerConfig, retriever *retrieveManager) *LesOdr { return &LesOdr{ - db: db, - retriever: retriever, - stop: make(chan struct{}), + db: db, + indexerConfig: config, + retriever: retriever, + stop: make(chan struct{}), } } @@ -73,6 +75,11 @@ func (odr *LesOdr) BloomIndexer() *core.ChainIndexer { return odr.bloomIndexer } +// IndexerConfig returns the indexer config. +func (odr *LesOdr) IndexerConfig() *light.IndexerConfig { + return odr.indexerConfig +} + const ( MsgBlockBodies = iota MsgCode diff --git a/les/odr_requests.go b/les/odr_requests.go index 075fcd92ca..9e9b2673f9 100644 --- a/les/odr_requests.go +++ b/les/odr_requests.go @@ -365,7 +365,7 @@ func (r *ChtRequest) CanSend(peer *peer) bool { peer.lock.RLock() defer peer.lock.RUnlock() - return peer.headInfo.Number >= light.HelperTrieConfirmations && r.ChtNum <= (peer.headInfo.Number-light.HelperTrieConfirmations)/light.CHTFrequencyClient + return peer.headInfo.Number >= r.Config.ChtConfirms && r.ChtNum <= (peer.headInfo.Number-r.Config.ChtConfirms)/r.Config.ChtSize } // Request sends an ODR request to the LES network (implementation of LesOdrRequest) @@ -379,7 +379,21 @@ func (r *ChtRequest) Request(reqID uint64, peer *peer) error { Key: encNum[:], AuxReq: auxHeader, } - return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []HelperTrieReq{req}) + switch peer.version { + case lpv1: + var reqsV1 ChtReq + if req.Type != htCanonical || req.AuxReq != auxHeader || len(req.Key) != 8 { + return fmt.Errorf("Request invalid in LES/1 mode") + } + blockNum := binary.BigEndian.Uint64(req.Key) + // convert HelperTrie request to old CHT request + reqsV1 = ChtReq{ChtNum: (req.TrieIdx + 1) * (r.Config.ChtSize / r.Config.PairChtSize), BlockNum: blockNum, FromLevel: req.FromLevel} + return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []ChtReq{reqsV1}) + case lpv2: + return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []HelperTrieReq{req}) + default: + panic(nil) + } } // Valid processes an ODR request reply message from the LES network @@ -484,7 +498,7 @@ func (r *BloomRequest) CanSend(peer *peer) bool { if peer.version < lpv2 { return false } - return peer.headInfo.Number >= light.HelperTrieConfirmations && r.BloomTrieNum <= (peer.headInfo.Number-light.HelperTrieConfirmations)/light.BloomTrieFrequency + return peer.headInfo.Number >= r.Config.BloomTrieConfirms && r.BloomTrieNum <= (peer.headInfo.Number-r.Config.BloomTrieConfirms)/r.Config.BloomTrieSize } // Request sends an ODR request to the LES network (implementation of LesOdrRequest) diff --git a/les/odr_test.go b/les/odr_test.go index c7c25cbe4b..e6458adf56 100644 --- a/les/odr_test.go +++ b/les/odr_test.go @@ -30,7 +30,6 @@ import ( "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/vm" - "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/params" @@ -160,36 +159,21 @@ func odrContractCall(ctx context.Context, db ethdb.Database, config *params.Chai return res } +// testOdr tests odr requests whose validation guaranteed by block headers. func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) { // Assemble the test environment - peers := newPeerSet() - dist := newRequestDistributor(peers, make(chan struct{})) - rm := newRetrieveManager(peers, dist, nil) - db := ethdb.NewMemDatabase() - ldb := ethdb.NewMemDatabase() - odr := NewLesOdr(ldb, rm) - odr.SetIndexers(light.NewChtIndexer(db, true, nil), light.NewBloomTrieIndexer(db, true, nil), eth.NewBloomIndexer(db, light.BloomTrieFrequency, light.HelperTrieConfirmations)) - pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db) - lpm := newTestProtocolManagerMust(t, true, 0, nil, peers, odr, ldb) - _, err1, lpeer, err2 := newTestPeerPair("peer", protocol, pm, lpm) - select { - case <-time.After(time.Millisecond * 100): - case err := <-err1: - t.Fatalf("peer 1 handshake error: %v", err) - case err := <-err2: - t.Fatalf("peer 1 handshake error: %v", err) - } - - lpm.synchronise(lpeer) + server, client, tearDown := newClientServerEnv(t, 4, protocol, nil, true) + defer tearDown() + client.pm.synchronise(client.rPeer) test := func(expFail uint64) { - for i := uint64(0); i <= pm.blockchain.CurrentHeader().Number.Uint64(); i++ { - bhash := rawdb.ReadCanonicalHash(db, i) - b1 := fn(light.NoOdr, db, pm.chainConfig, pm.blockchain.(*core.BlockChain), nil, bhash) + for i := uint64(0); i <= server.pm.blockchain.CurrentHeader().Number.Uint64(); i++ { + bhash := rawdb.ReadCanonicalHash(server.db, i) + b1 := fn(light.NoOdr, server.db, server.pm.chainConfig, server.pm.blockchain.(*core.BlockChain), nil, bhash) ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) defer cancel() - b2 := fn(ctx, ldb, lpm.chainConfig, nil, lpm.blockchain.(*light.LightChain), bhash) + b2 := fn(ctx, client.db, client.pm.chainConfig, nil, client.pm.blockchain.(*light.LightChain), bhash) eq := bytes.Equal(b1, b2) exp := i < expFail @@ -201,21 +185,20 @@ func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) { } } } - // temporarily remove peer to test odr fails // expect retrievals to fail (except genesis block) without a les peer - peers.Unregister(lpeer.id) + client.peers.Unregister(client.rPeer.id) time.Sleep(time.Millisecond * 10) // ensure that all peerSetNotify callbacks are executed test(expFail) // expect all retrievals to pass - peers.Register(lpeer) + client.peers.Register(client.rPeer) time.Sleep(time.Millisecond * 10) // ensure that all peerSetNotify callbacks are executed - lpeer.lock.Lock() - lpeer.hasBlock = func(common.Hash, uint64) bool { return true } - lpeer.lock.Unlock() + client.peers.lock.Lock() + client.rPeer.hasBlock = func(common.Hash, uint64) bool { return true } + client.peers.lock.Unlock() test(5) // still expect all retrievals to pass, now data should be cached locally - peers.Unregister(lpeer.id) + client.peers.Unregister(client.rPeer.id) time.Sleep(time.Millisecond * 10) // ensure that all peerSetNotify callbacks are executed test(5) } diff --git a/les/peer.go b/les/peer.go index eb7452e276..70c863c2ff 100644 --- a/les/peer.go +++ b/les/peer.go @@ -19,7 +19,6 @@ package les import ( "crypto/ecdsa" - "encoding/binary" "errors" "fmt" "math/big" @@ -36,9 +35,10 @@ import ( ) var ( - errClosed = errors.New("peer set is closed") - errAlreadyRegistered = errors.New("peer is already registered") - errNotRegistered = errors.New("peer is not registered") + errClosed = errors.New("peer set is closed") + errAlreadyRegistered = errors.New("peer is already registered") + errNotRegistered = errors.New("peer is not registered") + errInvalidHelpTrieReq = errors.New("invalid help trie request") ) const maxResponseErrors = 50 // number of invalid responses tolerated (makes the protocol less brittle but still avoids spam) @@ -284,21 +284,21 @@ func (p *peer) RequestProofs(reqID, cost uint64, reqs []ProofReq) error { } // RequestHelperTrieProofs fetches a batch of HelperTrie merkle proofs from a remote node. -func (p *peer) RequestHelperTrieProofs(reqID, cost uint64, reqs []HelperTrieReq) error { - p.Log().Debug("Fetching batch of HelperTrie proofs", "count", len(reqs)) +func (p *peer) RequestHelperTrieProofs(reqID, cost uint64, data interface{}) error { switch p.version { case lpv1: - reqsV1 := make([]ChtReq, len(reqs)) - for i, req := range reqs { - if req.Type != htCanonical || req.AuxReq != auxHeader || len(req.Key) != 8 { - return fmt.Errorf("Request invalid in LES/1 mode") - } - blockNum := binary.BigEndian.Uint64(req.Key) - // convert HelperTrie request to old CHT request - reqsV1[i] = ChtReq{ChtNum: (req.TrieIdx + 1) * (light.CHTFrequencyClient / light.CHTFrequencyServer), BlockNum: blockNum, FromLevel: req.FromLevel} + reqs, ok := data.([]ChtReq) + if !ok { + return errInvalidHelpTrieReq } - return sendRequest(p.rw, GetHeaderProofsMsg, reqID, cost, reqsV1) + p.Log().Debug("Fetching batch of header proofs", "count", len(reqs)) + return sendRequest(p.rw, GetHeaderProofsMsg, reqID, cost, reqs) case lpv2: + reqs, ok := data.([]HelperTrieReq) + if !ok { + return errInvalidHelpTrieReq + } + p.Log().Debug("Fetching batch of HelperTrie proofs", "count", len(reqs)) return sendRequest(p.rw, GetHelperTrieProofsMsg, reqID, cost, reqs) default: panic(nil) diff --git a/les/request_test.go b/les/request_test.go index db576798b4..f02c2a3d76 100644 --- a/les/request_test.go +++ b/les/request_test.go @@ -24,7 +24,6 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/crypto" - "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/light" ) @@ -84,35 +83,17 @@ func tfCodeAccess(db ethdb.Database, bhash common.Hash, num uint64) light.OdrReq func testAccess(t *testing.T, protocol int, fn accessTestFn) { // Assemble the test environment - peers := newPeerSet() - dist := newRequestDistributor(peers, make(chan struct{})) - rm := newRetrieveManager(peers, dist, nil) - db := ethdb.NewMemDatabase() - ldb := ethdb.NewMemDatabase() - odr := NewLesOdr(ldb, rm) - odr.SetIndexers(light.NewChtIndexer(db, true, nil), light.NewBloomTrieIndexer(db, true, nil), eth.NewBloomIndexer(db, light.BloomTrieFrequency, light.HelperTrieConfirmations)) - - pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db) - lpm := newTestProtocolManagerMust(t, true, 0, nil, peers, odr, ldb) - _, err1, lpeer, err2 := newTestPeerPair("peer", protocol, pm, lpm) - select { - case <-time.After(time.Millisecond * 100): - case err := <-err1: - t.Fatalf("peer 1 handshake error: %v", err) - case err := <-err2: - t.Fatalf("peer 1 handshake error: %v", err) - } - - lpm.synchronise(lpeer) + server, client, tearDown := newClientServerEnv(t, 4, protocol, nil, true) + defer tearDown() + client.pm.synchronise(client.rPeer) test := func(expFail uint64) { - for i := uint64(0); i <= pm.blockchain.CurrentHeader().Number.Uint64(); i++ { - bhash := rawdb.ReadCanonicalHash(db, i) - if req := fn(ldb, bhash, i); req != nil { + for i := uint64(0); i <= server.pm.blockchain.CurrentHeader().Number.Uint64(); i++ { + bhash := rawdb.ReadCanonicalHash(server.db, i) + if req := fn(client.db, bhash, i); req != nil { ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) defer cancel() - - err := odr.Retrieve(ctx, req) + err := client.pm.odr.Retrieve(ctx, req) got := err == nil exp := i < expFail if exp && !got { @@ -126,16 +107,16 @@ func testAccess(t *testing.T, protocol int, fn accessTestFn) { } // temporarily remove peer to test odr fails - peers.Unregister(lpeer.id) + client.peers.Unregister(client.rPeer.id) time.Sleep(time.Millisecond * 10) // ensure that all peerSetNotify callbacks are executed // expect retrievals to fail (except genesis block) without a les peer test(0) - peers.Register(lpeer) + client.peers.Register(client.rPeer) time.Sleep(time.Millisecond * 10) // ensure that all peerSetNotify callbacks are executed - lpeer.lock.Lock() - lpeer.hasBlock = func(common.Hash, uint64) bool { return true } - lpeer.lock.Unlock() + client.rPeer.lock.Lock() + client.rPeer.hasBlock = func(common.Hash, uint64) bool { return true } + client.rPeer.lock.Unlock() // expect all retrievals to pass test(5) } diff --git a/les/server.go b/les/server.go index df98d1e3a8..2fa0456d69 100644 --- a/les/server.go +++ b/les/server.go @@ -34,6 +34,7 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discv5" + "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rlp" ) @@ -50,7 +51,7 @@ type LesServer struct { func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { quitSync := make(chan struct{}) - pm, err := NewProtocolManager(eth.BlockChain().Config(), false, config.NetworkId, eth.EventMux(), eth.Engine(), newPeerSet(), eth.BlockChain(), eth.TxPool(), eth.ChainDb(), nil, nil, nil, quitSync, new(sync.WaitGroup)) + pm, err := NewProtocolManager(eth.BlockChain().Config(), light.DefaultServerIndexerConfig, false, config.NetworkId, eth.EventMux(), eth.Engine(), newPeerSet(), eth.BlockChain(), eth.TxPool(), eth.ChainDb(), nil, nil, nil, quitSync, new(sync.WaitGroup)) if err != nil { return nil, err } @@ -64,8 +65,9 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { lesCommons: lesCommons{ config: config, chainDb: eth.ChainDb(), - chtIndexer: light.NewChtIndexer(eth.ChainDb(), false, nil), - bloomTrieIndexer: light.NewBloomTrieIndexer(eth.ChainDb(), false, nil), + iConfig: light.DefaultServerIndexerConfig, + chtIndexer: light.NewChtIndexer(eth.ChainDb(), nil, params.CHTFrequencyServer, params.HelperTrieProcessConfirmations), + bloomTrieIndexer: light.NewBloomTrieIndexer(eth.ChainDb(), nil, params.BloomBitsBlocks, params.BloomTrieFrequency), protocolManager: pm, }, quitSync: quitSync, @@ -75,14 +77,14 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { logger := log.New() chtV1SectionCount, _, _ := srv.chtIndexer.Sections() // indexer still uses LES/1 4k section size for backwards server compatibility - chtV2SectionCount := chtV1SectionCount / (light.CHTFrequencyClient / light.CHTFrequencyServer) + chtV2SectionCount := chtV1SectionCount / (params.CHTFrequencyClient / params.CHTFrequencyServer) if chtV2SectionCount != 0 { // convert to LES/2 section chtLastSection := chtV2SectionCount - 1 // convert last LES/2 section index back to LES/1 index for chtIndexer.SectionHead - chtLastSectionV1 := (chtLastSection+1)*(light.CHTFrequencyClient/light.CHTFrequencyServer) - 1 + chtLastSectionV1 := (chtLastSection+1)*(params.CHTFrequencyClient/params.CHTFrequencyServer) - 1 chtSectionHead := srv.chtIndexer.SectionHead(chtLastSectionV1) - chtRoot := light.GetChtV2Root(pm.chainDb, chtLastSection, chtSectionHead) + chtRoot := light.GetChtRoot(pm.chainDb, chtLastSectionV1, chtSectionHead) logger.Info("Loaded CHT", "section", chtLastSection, "head", chtSectionHead, "root", chtRoot) } bloomTrieSectionCount, _, _ := srv.bloomTrieIndexer.Sections() diff --git a/light/lightchain.go b/light/lightchain.go index b5afe1f0e6..bd798eca24 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -48,6 +48,7 @@ var ( // interface. It only does header validation during chain insertion. type LightChain struct { hc *core.HeaderChain + indexerConfig *IndexerConfig chainDb ethdb.Database odr OdrBackend chainFeed event.Feed @@ -81,13 +82,14 @@ func NewLightChain(odr OdrBackend, config *params.ChainConfig, engine consensus. blockCache, _ := lru.New(blockCacheLimit) bc := &LightChain{ - chainDb: odr.Database(), - odr: odr, - quit: make(chan struct{}), - bodyCache: bodyCache, - bodyRLPCache: bodyRLPCache, - blockCache: blockCache, - engine: engine, + chainDb: odr.Database(), + indexerConfig: odr.IndexerConfig(), + odr: odr, + quit: make(chan struct{}), + bodyCache: bodyCache, + bodyRLPCache: bodyRLPCache, + blockCache: blockCache, + engine: engine, } var err error bc.hc, err = core.NewHeaderChain(odr.Database(), config, bc.engine, bc.getProcInterrupt) @@ -128,7 +130,7 @@ func (self *LightChain) addTrustedCheckpoint(cp TrustedCheckpoint) { if self.odr.BloomIndexer() != nil { self.odr.BloomIndexer().AddKnownSectionHead(cp.SectionIdx, cp.SectionHead) } - log.Info("Added trusted checkpoint", "chain", cp.name, "block", (cp.SectionIdx+1)*CHTFrequencyClient-1, "hash", cp.SectionHead) + log.Info("Added trusted checkpoint", "chain", cp.name, "block", (cp.SectionIdx+1)*self.indexerConfig.ChtSize-1, "hash", cp.SectionHead) } func (self *LightChain) getProcInterrupt() bool { @@ -472,7 +474,7 @@ func (self *LightChain) SyncCht(ctx context.Context) bool { head := self.CurrentHeader().Number.Uint64() sections, _, _ := self.odr.ChtIndexer().Sections() - latest := sections*CHTFrequencyClient - 1 + latest := sections*self.indexerConfig.ChtSize - 1 if clique := self.hc.Config().Clique; clique != nil { latest -= latest % clique.Epoch // epoch snapshot for clique } diff --git a/light/lightchain_test.go b/light/lightchain_test.go index 5f0baaf4cd..d45c0656d9 100644 --- a/light/lightchain_test.go +++ b/light/lightchain_test.go @@ -55,7 +55,7 @@ func newCanonical(n int) (ethdb.Database, *LightChain, error) { db := ethdb.NewMemDatabase() gspec := core.Genesis{Config: params.TestChainConfig} genesis := gspec.MustCommit(db) - blockchain, _ := NewLightChain(&dummyOdr{db: db}, gspec.Config, ethash.NewFaker()) + blockchain, _ := NewLightChain(&dummyOdr{db: db, indexerConfig: TestClientIndexerConfig}, gspec.Config, ethash.NewFaker()) // Create and inject the requested chain if n == 0 { @@ -265,7 +265,8 @@ func makeHeaderChainWithDiff(genesis *types.Block, d []int, seed byte) []*types. type dummyOdr struct { OdrBackend - db ethdb.Database + db ethdb.Database + indexerConfig *IndexerConfig } func (odr *dummyOdr) Database() ethdb.Database { @@ -276,6 +277,10 @@ func (odr *dummyOdr) Retrieve(ctx context.Context, req OdrRequest) error { return nil } +func (odr *dummyOdr) IndexerConfig() *IndexerConfig { + return odr.indexerConfig +} + // Tests that reorganizing a long difficult chain after a short easy one // overwrites the canonical numbers and links in the database. func TestReorgLongHeaders(t *testing.T) { diff --git a/light/odr.go b/light/odr.go index 83c64055a2..3cd8b2c040 100644 --- a/light/odr.go +++ b/light/odr.go @@ -44,6 +44,7 @@ type OdrBackend interface { BloomTrieIndexer() *core.ChainIndexer BloomIndexer() *core.ChainIndexer Retrieve(ctx context.Context, req OdrRequest) error + IndexerConfig() *IndexerConfig } // OdrRequest is an interface for retrieval requests @@ -136,6 +137,7 @@ func (req *ReceiptsRequest) StoreResult(db ethdb.Database) { // ChtRequest is the ODR request type for state/storage trie entries type ChtRequest struct { OdrRequest + Config *IndexerConfig ChtNum, BlockNum uint64 ChtRoot common.Hash Header *types.Header @@ -155,6 +157,7 @@ func (req *ChtRequest) StoreResult(db ethdb.Database) { // BloomRequest is the ODR request type for retrieving bloom filters from a CHT structure type BloomRequest struct { OdrRequest + Config *IndexerConfig BloomTrieNum uint64 BitIdx uint SectionIdxList []uint64 @@ -166,7 +169,7 @@ type BloomRequest struct { // StoreResult stores the retrieved data in local database func (req *BloomRequest) StoreResult(db ethdb.Database) { for i, sectionIdx := range req.SectionIdxList { - sectionHead := rawdb.ReadCanonicalHash(db, (sectionIdx+1)*BloomTrieFrequency-1) + sectionHead := rawdb.ReadCanonicalHash(db, (sectionIdx+1)*req.Config.BloomTrieSize-1) // if we don't have the canonical hash stored for this section head number, we'll still store it under // a key with a zero sectionHead. GetBloomBits will look there too if we still don't have the canonical // hash. In the unlikely case we've retrieved the section head hash since then, we'll just retrieve the diff --git a/light/odr_test.go b/light/odr_test.go index 3e7ac10118..eea5b1eab1 100644 --- a/light/odr_test.go +++ b/light/odr_test.go @@ -55,8 +55,9 @@ var ( type testOdr struct { OdrBackend - sdb, ldb ethdb.Database - disable bool + indexerConfig *IndexerConfig + sdb, ldb ethdb.Database + disable bool } func (odr *testOdr) Database() ethdb.Database { @@ -92,6 +93,10 @@ func (odr *testOdr) Retrieve(ctx context.Context, req OdrRequest) error { return nil } +func (odr *testOdr) IndexerConfig() *IndexerConfig { + return odr.indexerConfig +} + type odrTestFn func(ctx context.Context, db ethdb.Database, bc *core.BlockChain, lc *LightChain, bhash common.Hash) ([]byte, error) func TestOdrGetBlockLes1(t *testing.T) { testChainOdr(t, 1, odrGetBlock) } @@ -258,7 +263,7 @@ func testChainOdr(t *testing.T, protocol int, fn odrTestFn) { t.Fatal(err) } - odr := &testOdr{sdb: sdb, ldb: ldb} + odr := &testOdr{sdb: sdb, ldb: ldb, indexerConfig: TestClientIndexerConfig} lightchain, err := NewLightChain(odr, params.TestChainConfig, ethash.NewFullFaker()) if err != nil { t.Fatal(err) diff --git a/light/odr_util.go b/light/odr_util.go index 620af63835..9bc0f604b0 100644 --- a/light/odr_util.go +++ b/light/odr_util.go @@ -53,16 +53,16 @@ func GetHeaderByNumber(ctx context.Context, odr OdrBackend, number uint64) (*typ for chtCount > 0 && canonicalHash != sectionHead && canonicalHash != (common.Hash{}) { chtCount-- if chtCount > 0 { - sectionHeadNum = chtCount*CHTFrequencyClient - 1 + sectionHeadNum = chtCount*odr.IndexerConfig().ChtSize - 1 sectionHead = odr.ChtIndexer().SectionHead(chtCount - 1) canonicalHash = rawdb.ReadCanonicalHash(db, sectionHeadNum) } } } - if number >= chtCount*CHTFrequencyClient { + if number >= chtCount*odr.IndexerConfig().ChtSize { return nil, ErrNoTrustedCht } - r := &ChtRequest{ChtRoot: GetChtRoot(db, chtCount-1, sectionHead), ChtNum: chtCount - 1, BlockNum: number} + r := &ChtRequest{ChtRoot: GetChtRoot(db, chtCount-1, sectionHead), ChtNum: chtCount - 1, BlockNum: number, Config: odr.IndexerConfig()} if err := odr.Retrieve(ctx, r); err != nil { return nil, err } @@ -175,9 +175,9 @@ func GetBlockLogs(ctx context.Context, odr OdrBackend, hash common.Hash, number // GetBloomBits retrieves a batch of compressed bloomBits vectors belonging to the given bit index and section indexes func GetBloomBits(ctx context.Context, odr OdrBackend, bitIdx uint, sectionIdxList []uint64) ([][]byte, error) { - db := odr.Database() - result := make([][]byte, len(sectionIdxList)) var ( + db = odr.Database() + result = make([][]byte, len(sectionIdxList)) reqList []uint64 reqIdx []int ) @@ -193,7 +193,7 @@ func GetBloomBits(ctx context.Context, odr OdrBackend, bitIdx uint, sectionIdxLi for bloomTrieCount > 0 && canonicalHash != sectionHead && canonicalHash != (common.Hash{}) { bloomTrieCount-- if bloomTrieCount > 0 { - sectionHeadNum = bloomTrieCount*BloomTrieFrequency - 1 + sectionHeadNum = bloomTrieCount*odr.IndexerConfig().BloomTrieSize - 1 sectionHead = odr.BloomTrieIndexer().SectionHead(bloomTrieCount - 1) canonicalHash = rawdb.ReadCanonicalHash(db, sectionHeadNum) } @@ -201,7 +201,7 @@ func GetBloomBits(ctx context.Context, odr OdrBackend, bitIdx uint, sectionIdxLi } for i, sectionIdx := range sectionIdxList { - sectionHead := rawdb.ReadCanonicalHash(db, (sectionIdx+1)*BloomTrieFrequency-1) + sectionHead := rawdb.ReadCanonicalHash(db, (sectionIdx+1)*odr.IndexerConfig().BloomSize-1) // if we don't have the canonical hash stored for this section head number, we'll still look for // an entry with a zero sectionHead (we store it with zero section head too if we don't know it // at the time of the retrieval) @@ -209,6 +209,7 @@ func GetBloomBits(ctx context.Context, odr OdrBackend, bitIdx uint, sectionIdxLi if err == nil { result[i] = bloomBits } else { + // TODO(rjl493456442) Convert sectionIndex to BloomTrie relative index if sectionIdx >= bloomTrieCount { return nil, ErrNoTrustedBloomTrie } @@ -220,7 +221,8 @@ func GetBloomBits(ctx context.Context, odr OdrBackend, bitIdx uint, sectionIdxLi return result, nil } - r := &BloomRequest{BloomTrieRoot: GetBloomTrieRoot(db, bloomTrieCount-1, sectionHead), BloomTrieNum: bloomTrieCount - 1, BitIdx: bitIdx, SectionIdxList: reqList} + r := &BloomRequest{BloomTrieRoot: GetBloomTrieRoot(db, bloomTrieCount-1, sectionHead), BloomTrieNum: bloomTrieCount - 1, + BitIdx: bitIdx, SectionIdxList: reqList, Config: odr.IndexerConfig()} if err := odr.Retrieve(ctx, r); err != nil { return nil, err } else { diff --git a/light/postprocess.go b/light/postprocess.go index f105d57b55..7b23e48b5b 100644 --- a/light/postprocess.go +++ b/light/postprocess.go @@ -36,20 +36,75 @@ import ( "github.com/ethereum/go-ethereum/trie" ) -const ( - // CHTFrequencyClient is the block frequency for creating CHTs on the client side. - CHTFrequencyClient = 32768 +// IndexerConfig includes a set of configs for chain indexers. +type IndexerConfig struct { + // The block frequency for creating CHTs. + ChtSize uint64 - // CHTFrequencyServer is the block frequency for creating CHTs on the server side. - // Eventually this can be merged back with the client version, but that requires a - // full database upgrade, so that should be left for a suitable moment. - CHTFrequencyServer = 4096 + // A special auxiliary field represents client's chtsize for server config, otherwise represents server's chtsize. + PairChtSize uint64 - HelperTrieConfirmations = 2048 // number of confirmations before a server is expected to have the given HelperTrie available - HelperTrieProcessConfirmations = 256 // number of confirmations before a HelperTrie is generated + // The number of confirmations needed to generate/accept a canonical hash help trie. + ChtConfirms uint64 + + // The block frequency for creating new bloom bits. + BloomSize uint64 + + // The number of confirmation needed before a bloom section is considered probably final and its rotated bits + // are calculated. + BloomConfirms uint64 + + // The block frequency for creating BloomTrie. + BloomTrieSize uint64 + + // The number of confirmations needed to generate/accept a bloom trie. + BloomTrieConfirms uint64 +} + +var ( + // DefaultServerIndexerConfig wraps a set of configs as a default indexer config for server side. + DefaultServerIndexerConfig = &IndexerConfig{ + ChtSize: params.CHTFrequencyServer, + PairChtSize: params.CHTFrequencyClient, + ChtConfirms: params.HelperTrieProcessConfirmations, + BloomSize: params.BloomBitsBlocks, + BloomConfirms: params.BloomConfirms, + BloomTrieSize: params.BloomTrieFrequency, + BloomTrieConfirms: params.HelperTrieProcessConfirmations, + } + // DefaultClientIndexerConfig wraps a set of configs as a default indexer config for client side. + DefaultClientIndexerConfig = &IndexerConfig{ + ChtSize: params.CHTFrequencyClient, + PairChtSize: params.CHTFrequencyServer, + ChtConfirms: params.HelperTrieConfirmations, + BloomSize: params.BloomBitsBlocksClient, + BloomConfirms: params.HelperTrieConfirmations, + BloomTrieSize: params.BloomTrieFrequency, + BloomTrieConfirms: params.HelperTrieConfirmations, + } + // TestServerIndexerConfig wraps a set of configs as a test indexer config for server side. + TestServerIndexerConfig = &IndexerConfig{ + ChtSize: 256, + PairChtSize: 2048, + ChtConfirms: 16, + BloomSize: 256, + BloomConfirms: 16, + BloomTrieSize: 2048, + BloomTrieConfirms: 16, + } + // TestClientIndexerConfig wraps a set of configs as a test indexer config for client side. + TestClientIndexerConfig = &IndexerConfig{ + ChtSize: 2048, + PairChtSize: 256, + ChtConfirms: 128, + BloomSize: 2048, + BloomConfirms: 128, + BloomTrieSize: 2048, + BloomTrieConfirms: 128, + } ) -// TrustedCheckpoint represents a set of post-processed trie roots (CHT and BloomTrie) associated with +// trustedCheckpoint represents a set of post-processed trie roots (CHT and BloomTrie) associated with // the appropriate section index and head hash. It is used to start light syncing from this checkpoint // and avoid downloading the entire header chain while still being able to securely access old headers/logs. type TrustedCheckpoint struct { @@ -84,9 +139,9 @@ var trustedCheckpoints = map[common.Hash]TrustedCheckpoint{ } var ( - ErrNoTrustedCht = errors.New("No trusted canonical hash trie") - ErrNoTrustedBloomTrie = errors.New("No trusted bloom trie") - ErrNoHeader = errors.New("Header not found") + ErrNoTrustedCht = errors.New("no trusted canonical hash trie") + ErrNoTrustedBloomTrie = errors.New("no trusted bloom trie") + ErrNoHeader = errors.New("header not found") chtPrefix = []byte("chtRoot-") // chtPrefix + chtNum (uint64 big endian) -> trie root hash ChtTablePrefix = "cht-" ) @@ -97,8 +152,8 @@ type ChtNode struct { Td *big.Int } -// GetChtRoot reads the CHT root assoctiated to the given section from the database -// Note that sectionIdx is specified according to LES/1 CHT section size +// GetChtRoot reads the CHT root associated to the given section from the database +// Note that sectionIdx is specified according to LES/1 CHT section size. func GetChtRoot(db ethdb.Database, sectionIdx uint64, sectionHead common.Hash) common.Hash { var encNumber [8]byte binary.BigEndian.PutUint64(encNumber[:], sectionIdx) @@ -106,21 +161,15 @@ func GetChtRoot(db ethdb.Database, sectionIdx uint64, sectionHead common.Hash) c return common.BytesToHash(data) } -// GetChtV2Root reads the CHT root assoctiated to the given section from the database -// Note that sectionIdx is specified according to LES/2 CHT section size -func GetChtV2Root(db ethdb.Database, sectionIdx uint64, sectionHead common.Hash) common.Hash { - return GetChtRoot(db, (sectionIdx+1)*(CHTFrequencyClient/CHTFrequencyServer)-1, sectionHead) -} - -// StoreChtRoot writes the CHT root assoctiated to the given section into the database -// Note that sectionIdx is specified according to LES/1 CHT section size +// StoreChtRoot writes the CHT root associated to the given section into the database +// Note that sectionIdx is specified according to LES/1 CHT section size. func StoreChtRoot(db ethdb.Database, sectionIdx uint64, sectionHead, root common.Hash) { var encNumber [8]byte binary.BigEndian.PutUint64(encNumber[:], sectionIdx) db.Put(append(append(chtPrefix, encNumber[:]...), sectionHead.Bytes()...), root.Bytes()) } -// ChtIndexerBackend implements core.ChainIndexerBackend +// ChtIndexerBackend implements core.ChainIndexerBackend. type ChtIndexerBackend struct { diskdb, trieTable ethdb.Database odr OdrBackend @@ -130,33 +179,24 @@ type ChtIndexerBackend struct { trie *trie.Trie } -// NewBloomTrieIndexer creates a BloomTrie chain indexer -func NewChtIndexer(db ethdb.Database, clientMode bool, odr OdrBackend) *core.ChainIndexer { - var sectionSize, confirmReq uint64 - if clientMode { - sectionSize = CHTFrequencyClient - confirmReq = HelperTrieConfirmations - } else { - sectionSize = CHTFrequencyServer - confirmReq = HelperTrieProcessConfirmations - } - idb := ethdb.NewTable(db, "chtIndex-") +// NewChtIndexer creates a Cht chain indexer +func NewChtIndexer(db ethdb.Database, odr OdrBackend, size, confirms uint64) *core.ChainIndexer { trieTable := ethdb.NewTable(db, ChtTablePrefix) backend := &ChtIndexerBackend{ diskdb: db, odr: odr, trieTable: trieTable, triedb: trie.NewDatabase(trieTable), - sectionSize: sectionSize, + sectionSize: size, } - return core.NewChainIndexer(db, idb, backend, sectionSize, confirmReq, time.Millisecond*100, "cht") + return core.NewChainIndexer(db, ethdb.NewTable(db, "chtIndex-"), backend, size, confirms, time.Millisecond*100, "cht") } // fetchMissingNodes tries to retrieve the last entry of the latest trusted CHT from the // ODR backend in order to be able to add new entries and calculate subsequent root hashes func (c *ChtIndexerBackend) fetchMissingNodes(ctx context.Context, section uint64, root common.Hash) error { batch := c.trieTable.NewBatch() - r := &ChtRequest{ChtRoot: root, ChtNum: section - 1, BlockNum: section*c.sectionSize - 1} + r := &ChtRequest{ChtRoot: root, ChtNum: section - 1, BlockNum: section*c.sectionSize - 1, Config: c.odr.IndexerConfig()} for { err := c.odr.Retrieve(ctx, r) switch err { @@ -221,18 +261,13 @@ func (c *ChtIndexerBackend) Commit() error { } c.triedb.Commit(root, false) - if ((c.section+1)*c.sectionSize)%CHTFrequencyClient == 0 { - log.Info("Storing CHT", "section", c.section*c.sectionSize/CHTFrequencyClient, "head", fmt.Sprintf("%064x", c.lastHash), "root", fmt.Sprintf("%064x", root)) + if ((c.section+1)*c.sectionSize)%params.CHTFrequencyClient == 0 { + log.Info("Storing CHT", "section", c.section*c.sectionSize/params.CHTFrequencyClient, "head", fmt.Sprintf("%064x", c.lastHash), "root", fmt.Sprintf("%064x", root)) } StoreChtRoot(c.diskdb, c.section, c.lastHash, root) return nil } -const ( - BloomTrieFrequency = 32768 - ethBloomBitsSection = 4096 -) - var ( bloomTriePrefix = []byte("bltRoot-") // bloomTriePrefix + bloomTrieNum (uint64 big endian) -> trie root hash BloomTrieTablePrefix = "blt-" @@ -255,33 +290,31 @@ func StoreBloomTrieRoot(db ethdb.Database, sectionIdx uint64, sectionHead, root // BloomTrieIndexerBackend implements core.ChainIndexerBackend type BloomTrieIndexerBackend struct { - diskdb, trieTable ethdb.Database - odr OdrBackend - triedb *trie.Database - section, parentSectionSize, bloomTrieRatio uint64 - trie *trie.Trie - sectionHeads []common.Hash + diskdb, trieTable ethdb.Database + triedb *trie.Database + odr OdrBackend + section uint64 + parentSize uint64 + size uint64 + bloomTrieRatio uint64 + trie *trie.Trie + sectionHeads []common.Hash } // NewBloomTrieIndexer creates a BloomTrie chain indexer -func NewBloomTrieIndexer(db ethdb.Database, clientMode bool, odr OdrBackend) *core.ChainIndexer { +func NewBloomTrieIndexer(db ethdb.Database, odr OdrBackend, parentSize, size uint64) *core.ChainIndexer { trieTable := ethdb.NewTable(db, BloomTrieTablePrefix) backend := &BloomTrieIndexerBackend{ - diskdb: db, - odr: odr, - trieTable: trieTable, - triedb: trie.NewDatabase(trieTable), + diskdb: db, + odr: odr, + trieTable: trieTable, + triedb: trie.NewDatabase(trieTable), + parentSize: parentSize, + size: size, } - idb := ethdb.NewTable(db, "bltIndex-") - - if clientMode { - backend.parentSectionSize = BloomTrieFrequency - } else { - backend.parentSectionSize = ethBloomBitsSection - } - backend.bloomTrieRatio = BloomTrieFrequency / backend.parentSectionSize + backend.bloomTrieRatio = size / parentSize backend.sectionHeads = make([]common.Hash, backend.bloomTrieRatio) - return core.NewChainIndexer(db, idb, backend, BloomTrieFrequency, 0, time.Millisecond*100, "bloomtrie") + return core.NewChainIndexer(db, ethdb.NewTable(db, "bltIndex-"), backend, size, 0, time.Millisecond*100, "bloomtrie") } // fetchMissingNodes tries to retrieve the last entries of the latest trusted bloom trie from the @@ -296,7 +329,7 @@ func (b *BloomTrieIndexerBackend) fetchMissingNodes(ctx context.Context, section for i := 0; i < 20; i++ { go func() { for bitIndex := range indexCh { - r := &BloomRequest{BloomTrieRoot: root, BloomTrieNum: section - 1, BitIdx: bitIndex, SectionIdxList: []uint64{section - 1}} + r := &BloomRequest{BloomTrieRoot: root, BloomTrieNum: section - 1, BitIdx: bitIndex, SectionIdxList: []uint64{section - 1}, Config: b.odr.IndexerConfig()} for { if err := b.odr.Retrieve(ctx, r); err == ErrNoPeers { // if there are no peers to serve, retry later @@ -351,9 +384,9 @@ func (b *BloomTrieIndexerBackend) Reset(ctx context.Context, section uint64, las // Process implements core.ChainIndexerBackend func (b *BloomTrieIndexerBackend) Process(ctx context.Context, header *types.Header) error { - num := header.Number.Uint64() - b.section*BloomTrieFrequency - if (num+1)%b.parentSectionSize == 0 { - b.sectionHeads[num/b.parentSectionSize] = header.Hash() + num := header.Number.Uint64() - b.section*b.size + if (num+1)%b.parentSize == 0 { + b.sectionHeads[num/b.parentSize] = header.Hash() } return nil } @@ -372,7 +405,7 @@ func (b *BloomTrieIndexerBackend) Commit() error { if err != nil { return err } - decompData, err2 := bitutil.DecompressBytes(data, int(b.parentSectionSize/8)) + decompData, err2 := bitutil.DecompressBytes(data, int(b.parentSize/8)) if err2 != nil { return err2 } @@ -397,6 +430,5 @@ func (b *BloomTrieIndexerBackend) Commit() error { sectionHead := b.sectionHeads[b.bloomTrieRatio-1] log.Info("Storing bloom trie", "section", b.section, "head", fmt.Sprintf("%064x", sectionHead), "root", fmt.Sprintf("%064x", root), "compression", float64(compSize)/float64(decompSize)) StoreBloomTrieRoot(b.diskdb, b.section, sectionHead, root) - return nil } diff --git a/light/trie_test.go b/light/trie_test.go index 84c6f162fb..6bddfefe28 100644 --- a/light/trie_test.go +++ b/light/trie_test.go @@ -47,7 +47,7 @@ func TestNodeIterator(t *testing.T) { } ctx := context.Background() - odr := &testOdr{sdb: fulldb, ldb: lightdb} + odr := &testOdr{sdb: fulldb, ldb: lightdb, indexerConfig: TestClientIndexerConfig} head := blockchain.CurrentHeader() lightTrie, _ := NewStateDatabase(ctx, head, odr).OpenTrie(head.Root) fullTrie, _ := state.NewDatabase(fulldb).OpenTrie(head.Root) diff --git a/light/txpool_test.go b/light/txpool_test.go index ccbd83a943..204347a6e4 100644 --- a/light/txpool_test.go +++ b/light/txpool_test.go @@ -94,7 +94,7 @@ func TestTxPool(t *testing.T) { panic(err) } - odr := &testOdr{sdb: sdb, ldb: ldb} + odr := &testOdr{sdb: sdb, ldb: ldb, indexerConfig: TestClientIndexerConfig} relay := &testTxRelay{ send: make(chan int, 1), discard: make(chan int, 1), diff --git a/params/network_params.go b/params/network_params.go index 536a46d3df..f8731e8979 100644 --- a/params/network_params.go +++ b/params/network_params.go @@ -17,10 +17,38 @@ package params // These are network parameters that need to be constant between clients, but -// aren't necesarilly consensus related. +// aren't necessarily consensus related. const ( // BloomBitsBlocks is the number of blocks a single bloom bit section vector - // contains. + // contains on the server side. BloomBitsBlocks uint64 = 4096 + + // BloomBitsBlocksClient is the number of blocks a single bloom bit section vector + // contains on the light client side + BloomBitsBlocksClient uint64 = 32768 + + // BloomConfirms is the number of confirmation blocks before a bloom section is + // considered probably final and its rotated bits are calculated. + BloomConfirms = 256 + + // CHTFrequencyClient is the block frequency for creating CHTs on the client side. + CHTFrequencyClient = 32768 + + // CHTFrequencyServer is the block frequency for creating CHTs on the server side. + // Eventually this can be merged back with the client version, but that requires a + // full database upgrade, so that should be left for a suitable moment. + CHTFrequencyServer = 4096 + + // BloomTrieFrequency is the block frequency for creating BloomTrie on both + // server/client sides. + BloomTrieFrequency = 32768 + + // HelperTrieConfirmations is the number of confirmations before a client is expected + // to have the given HelperTrie available. + HelperTrieConfirmations = 2048 + + // HelperTrieProcessConfirmations is the number of confirmations before a HelperTrie + // is generated + HelperTrieProcessConfirmations = 256 ) From 63352bf4247f05d8ef255ff8c63290225c3bc671 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Felf=C3=B6ldi=20Zsolt?= Date: Tue, 28 Aug 2018 09:31:34 +0200 Subject: [PATCH 8/9] core: safe indexer operation when syncing starts before the checkpoint (#17511) --- core/chain_indexer.go | 56 ++++++++++++++++++++++++++++++++++--------- light/lightchain.go | 6 ++--- 2 files changed, 48 insertions(+), 14 deletions(-) diff --git a/core/chain_indexer.go b/core/chain_indexer.go index b80b517d91..89ee75eb29 100644 --- a/core/chain_indexer.go +++ b/core/chain_indexer.go @@ -85,6 +85,9 @@ type ChainIndexer struct { knownSections uint64 // Number of sections known to be complete (block wise) cascadedHead uint64 // Block number of the last completed section cascaded to subindexers + checkpointSections uint64 // Number of sections covered by the checkpoint + checkpointHead common.Hash // Section head belonging to the checkpoint + throttling time.Duration // Disk throttling to prevent a heavy upgrade from hogging resources log log.Logger @@ -115,12 +118,19 @@ func NewChainIndexer(chainDb, indexDb ethdb.Database, backend ChainIndexerBacken return c } -// AddKnownSectionHead marks a new section head as known/processed if it is newer -// than the already known best section head -func (c *ChainIndexer) AddKnownSectionHead(section uint64, shead common.Hash) { +// AddCheckpoint adds a checkpoint. Sections are never processed and the chain +// is not expected to be available before this point. The indexer assumes that +// the backend has sufficient information available to process subsequent sections. +// +// Note: knownSections == 0 and storedSections == checkpointSections until +// syncing reaches the checkpoint +func (c *ChainIndexer) AddCheckpoint(section uint64, shead common.Hash) { c.lock.Lock() defer c.lock.Unlock() + c.checkpointSections = section + 1 + c.checkpointHead = shead + if section < c.storedSections { return } @@ -233,16 +243,23 @@ func (c *ChainIndexer) newHead(head uint64, reorg bool) { // If a reorg happened, invalidate all sections until that point if reorg { // Revert the known section number to the reorg point - changed := head / c.sectionSize - if changed < c.knownSections { - c.knownSections = changed + known := head / c.sectionSize + stored := known + if known < c.checkpointSections { + known = 0 + } + if stored < c.checkpointSections { + stored = c.checkpointSections + } + if known < c.knownSections { + c.knownSections = known } // Revert the stored sections from the database to the reorg point - if changed < c.storedSections { - c.setValidSections(changed) + if stored < c.storedSections { + c.setValidSections(stored) } // Update the new head number to the finalized section end and notify children - head = changed * c.sectionSize + head = known * c.sectionSize if head < c.cascadedHead { c.cascadedHead = head @@ -256,7 +273,18 @@ func (c *ChainIndexer) newHead(head uint64, reorg bool) { var sections uint64 if head >= c.confirmsReq { sections = (head + 1 - c.confirmsReq) / c.sectionSize + if sections < c.checkpointSections { + sections = 0 + } if sections > c.knownSections { + if c.knownSections < c.checkpointSections { + // syncing reached the checkpoint, verify section head + syncedHead := rawdb.ReadCanonicalHash(c.chainDb, c.checkpointSections*c.sectionSize-1) + if syncedHead != c.checkpointHead { + c.log.Error("Synced chain does not match checkpoint", "number", c.checkpointSections*c.sectionSize-1, "expected", c.checkpointHead, "synced", syncedHead) + return + } + } c.knownSections = sections select { @@ -401,8 +429,14 @@ func (c *ChainIndexer) AddChildIndexer(indexer *ChainIndexer) { c.children = append(c.children, indexer) // Cascade any pending updates to new children too - if c.storedSections > 0 { - indexer.newHead(c.storedSections*c.sectionSize-1, false) + sections := c.storedSections + if c.knownSections < sections { + // if a section is "stored" but not "known" then it is a checkpoint without + // available chain data so we should not cascade it yet + sections = c.knownSections + } + if sections > 0 { + indexer.newHead(sections*c.sectionSize-1, false) } } diff --git a/light/lightchain.go b/light/lightchain.go index bd798eca24..d40a4ee6c6 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -121,14 +121,14 @@ func NewLightChain(odr OdrBackend, config *params.ChainConfig, engine consensus. func (self *LightChain) addTrustedCheckpoint(cp TrustedCheckpoint) { if self.odr.ChtIndexer() != nil { StoreChtRoot(self.chainDb, cp.SectionIdx, cp.SectionHead, cp.CHTRoot) - self.odr.ChtIndexer().AddKnownSectionHead(cp.SectionIdx, cp.SectionHead) + self.odr.ChtIndexer().AddCheckpoint(cp.SectionIdx, cp.SectionHead) } if self.odr.BloomTrieIndexer() != nil { StoreBloomTrieRoot(self.chainDb, cp.SectionIdx, cp.SectionHead, cp.BloomRoot) - self.odr.BloomTrieIndexer().AddKnownSectionHead(cp.SectionIdx, cp.SectionHead) + self.odr.BloomTrieIndexer().AddCheckpoint(cp.SectionIdx, cp.SectionHead) } if self.odr.BloomIndexer() != nil { - self.odr.BloomIndexer().AddKnownSectionHead(cp.SectionIdx, cp.SectionHead) + self.odr.BloomIndexer().AddCheckpoint(cp.SectionIdx, cp.SectionHead) } log.Info("Added trusted checkpoint", "chain", cp.name, "block", (cp.SectionIdx+1)*self.indexerConfig.ChtSize-1, "hash", cp.SectionHead) } From c1c003e4ff36c22d67662ca661fc78cde850d401 Mon Sep 17 00:00:00 2001 From: gary rong Date: Tue, 28 Aug 2018 21:59:05 +0800 Subject: [PATCH 9/9] consensus, miner: stale block mining support (#17506) * consensus, miner: stale block supporting * consensus, miner: refactor seal signature * cmd, consensus, eth: add miner noverify flag * cmd, consensus, miner: polish --- cmd/geth/main.go | 1 + cmd/geth/usage.go | 1 + cmd/utils/flags.go | 11 ++- consensus/clique/clique.go | 39 +++++---- consensus/consensus.go | 9 +- consensus/ethash/algorithm_test.go | 2 +- consensus/ethash/ethash.go | 33 +++---- consensus/ethash/ethash_test.go | 40 ++++----- consensus/ethash/sealer.go | 134 +++++++++++++++++++---------- consensus/ethash/sealer_test.go | 94 ++++++++++++++++++-- eth/api_tracer.go | 2 +- eth/backend.go | 8 +- eth/config.go | 1 + eth/gen_config.go | 7 +- les/backend.go | 2 +- miner/worker.go | 116 +++++++++++-------------- 16 files changed, 317 insertions(+), 183 deletions(-) diff --git a/cmd/geth/main.go b/cmd/geth/main.go index 7b8a8244ee..0ed67d0d57 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -108,6 +108,7 @@ var ( utils.MinerExtraDataFlag, utils.MinerLegacyExtraDataFlag, utils.MinerRecommitIntervalFlag, + utils.MinerNoVerfiyFlag, utils.NATFlag, utils.NoDiscoverFlag, utils.DiscoveryV5Flag, diff --git a/cmd/geth/usage.go b/cmd/geth/usage.go index 1e27d0ae8d..2f8260e0f6 100644 --- a/cmd/geth/usage.go +++ b/cmd/geth/usage.go @@ -192,6 +192,7 @@ var AppHelpFlagGroups = []flagGroup{ utils.MinerEtherbaseFlag, utils.MinerExtraDataFlag, utils.MinerRecommitIntervalFlag, + utils.MinerNoVerfiyFlag, }, }, { diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index b9a33ffe70..13430ad565 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -366,9 +366,13 @@ var ( } MinerRecommitIntervalFlag = cli.DurationFlag{ Name: "miner.recommit", - Usage: "Time interval to recreate the block being mined.", + Usage: "Time interval to recreate the block being mined", Value: eth.DefaultConfig.MinerRecommit, } + MinerNoVerfiyFlag = cli.BoolFlag{ + Name: "miner.noverify", + Usage: "Disable remote sealing verification", + } // Account settings UnlockedAccountFlag = cli.StringFlag{ Name: "unlock", @@ -1151,6 +1155,9 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *eth.Config) { if ctx.GlobalIsSet(MinerRecommitIntervalFlag.Name) { cfg.MinerRecommit = ctx.Duration(MinerRecommitIntervalFlag.Name) } + if ctx.GlobalIsSet(MinerNoVerfiyFlag.Name) { + cfg.MinerNoverify = ctx.Bool(MinerNoVerfiyFlag.Name) + } if ctx.GlobalIsSet(VMEnableDebugFlag.Name) { // TODO(fjl): force-enable this in --dev mode cfg.EnablePreimageRecording = ctx.GlobalBool(VMEnableDebugFlag.Name) @@ -1345,7 +1352,7 @@ func MakeChain(ctx *cli.Context, stack *node.Node) (chain *core.BlockChain, chai DatasetDir: stack.ResolvePath(eth.DefaultConfig.Ethash.DatasetDir), DatasetsInMem: eth.DefaultConfig.Ethash.DatasetsInMem, DatasetsOnDisk: eth.DefaultConfig.Ethash.DatasetsOnDisk, - }, nil) + }, nil, false) } } if gcmode := ctx.GlobalString(GCModeFlag.Name); gcmode != "full" && gcmode != "archive" { diff --git a/consensus/clique/clique.go b/consensus/clique/clique.go index 3730c91f62..5472909846 100644 --- a/consensus/clique/clique.go +++ b/consensus/clique/clique.go @@ -590,17 +590,17 @@ func (c *Clique) Authorize(signer common.Address, signFn SignerFn) { // Seal implements consensus.Engine, attempting to create a sealed block using // the local signing credentials. -func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, stop <-chan struct{}) (*types.Block, error) { +func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error { header := block.Header() // Sealing the genesis block is not supported number := header.Number.Uint64() if number == 0 { - return nil, errUnknownBlock + return errUnknownBlock } // For 0-period chains, refuse to seal empty blocks (no reward but would spin sealing) if c.config.Period == 0 && len(block.Transactions()) == 0 { - return nil, errWaitTransactions + return errWaitTransactions } // Don't hold the signer fields for the entire sealing procedure c.lock.RLock() @@ -610,10 +610,10 @@ func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, stop <-ch // Bail out if we're unauthorized to sign a block snap, err := c.snapshot(chain, number-1, header.ParentHash, nil) if err != nil { - return nil, err + return err } if _, authorized := snap.Signers[signer]; !authorized { - return nil, errUnauthorized + return errUnauthorized } // If we're amongst the recent signers, wait for the next block for seen, recent := range snap.Recents { @@ -621,8 +621,7 @@ func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, stop <-ch // Signer is among recents, only wait if the current block doesn't shift it out if limit := uint64(len(snap.Signers)/2 + 1); number < limit || seen > number-limit { log.Info("Signed recently, must wait for others") - <-stop - return nil, nil + return nil } } } @@ -635,21 +634,29 @@ func (c *Clique) Seal(chain consensus.ChainReader, block *types.Block, stop <-ch log.Trace("Out-of-turn signing requested", "wiggle", common.PrettyDuration(wiggle)) } - log.Trace("Waiting for slot to sign and propagate", "delay", common.PrettyDuration(delay)) - - select { - case <-stop: - return nil, nil - case <-time.After(delay): - } // Sign all the things! sighash, err := signFn(accounts.Account{Address: signer}, sigHash(header).Bytes()) if err != nil { - return nil, err + return err } copy(header.Extra[len(header.Extra)-extraSeal:], sighash) + // Wait until sealing is terminated or delay timeout. + log.Trace("Waiting for slot to sign and propagate", "delay", common.PrettyDuration(delay)) + go func() { + select { + case <-stop: + return + case <-time.After(delay): + } - return block.WithSeal(header), nil + select { + case results <- block.WithSeal(header): + default: + log.Warn("Sealing result is not read by miner", "sealhash", c.SealHash(header)) + } + }() + + return nil } // CalcDifficulty is the difficulty adjustment algorithm. It returns the difficulty diff --git a/consensus/consensus.go b/consensus/consensus.go index 27799f13c0..12ede7ff46 100644 --- a/consensus/consensus.go +++ b/consensus/consensus.go @@ -86,9 +86,12 @@ type Engine interface { Finalize(chain ChainReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header, receipts []*types.Receipt) (*types.Block, error) - // Seal generates a new block for the given input block with the local miner's - // seal place on top. - Seal(chain ChainReader, block *types.Block, stop <-chan struct{}) (*types.Block, error) + // Seal generates a new sealing request for the given input block and pushes + // the result into the given channel. + // + // Note, the method returns immediately and will send the result async. More + // than one result may also be returned depending on the consensus algorothm. + Seal(chain ChainReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error // SealHash returns the hash of a block prior to it being sealed. SealHash(header *types.Header) common.Hash diff --git a/consensus/ethash/algorithm_test.go b/consensus/ethash/algorithm_test.go index db22cccd0c..c58479e289 100644 --- a/consensus/ethash/algorithm_test.go +++ b/consensus/ethash/algorithm_test.go @@ -729,7 +729,7 @@ func TestConcurrentDiskCacheGeneration(t *testing.T) { go func(idx int) { defer pend.Done() - ethash := New(Config{cachedir, 0, 1, "", 0, 0, ModeNormal}, nil) + ethash := New(Config{cachedir, 0, 1, "", 0, 0, ModeNormal}, nil, false) defer ethash.Close() if err := ethash.VerifySeal(nil, block.Header()); err != nil { t.Errorf("proc %d: block verification failed: %v", idx, err) diff --git a/consensus/ethash/ethash.go b/consensus/ethash/ethash.go index d98c3371c5..b4819ca38b 100644 --- a/consensus/ethash/ethash.go +++ b/consensus/ethash/ethash.go @@ -50,7 +50,7 @@ var ( two256 = new(big.Int).Exp(big.NewInt(2), big.NewInt(256), big.NewInt(0)) // sharedEthash is a full instance that can be shared between multiple users. - sharedEthash = New(Config{"", 3, 0, "", 1, 0, ModeNormal}, nil) + sharedEthash = New(Config{"", 3, 0, "", 1, 0, ModeNormal}, nil, false) // algorithmRevision is the data structure version used for file naming. algorithmRevision = 23 @@ -405,6 +405,12 @@ type Config struct { PowMode Mode } +// sealTask wraps a seal block with relative result channel for remote sealer thread. +type sealTask struct { + block *types.Block + results chan<- *types.Block +} + // mineResult wraps the pow solution parameters for the specified block. type mineResult struct { nonce types.BlockNonce @@ -444,12 +450,11 @@ type Ethash struct { hashrate metrics.Meter // Meter tracking the average hashrate // Remote sealer related fields - workCh chan *types.Block // Notification channel to push new work to remote sealer - resultCh chan *types.Block // Channel used by mining threads to return result - fetchWorkCh chan *sealWork // Channel used for remote sealer to fetch mining work - submitWorkCh chan *mineResult // Channel used for remote sealer to submit their mining result - fetchRateCh chan chan uint64 // Channel used to gather submitted hash rate for local or remote sealer. - submitRateCh chan *hashrate // Channel used for remote sealer to submit their mining hashrate + workCh chan *sealTask // Notification channel to push new work and relative result channel to remote sealer + fetchWorkCh chan *sealWork // Channel used for remote sealer to fetch mining work + submitWorkCh chan *mineResult // Channel used for remote sealer to submit their mining result + fetchRateCh chan chan uint64 // Channel used to gather submitted hash rate for local or remote sealer. + submitRateCh chan *hashrate // Channel used for remote sealer to submit their mining hashrate // The fields below are hooks for testing shared *Ethash // Shared PoW verifier to avoid cache regeneration @@ -464,7 +469,7 @@ type Ethash struct { // New creates a full sized ethash PoW scheme and starts a background thread for // remote mining, also optionally notifying a batch of remote services of new work // packages. -func New(config Config, notify []string) *Ethash { +func New(config Config, notify []string, noverify bool) *Ethash { if config.CachesInMem <= 0 { log.Warn("One ethash cache must always be in memory", "requested", config.CachesInMem) config.CachesInMem = 1 @@ -481,36 +486,34 @@ func New(config Config, notify []string) *Ethash { datasets: newlru("dataset", config.DatasetsInMem, newDataset), update: make(chan struct{}), hashrate: metrics.NewMeter(), - workCh: make(chan *types.Block), - resultCh: make(chan *types.Block), + workCh: make(chan *sealTask), fetchWorkCh: make(chan *sealWork), submitWorkCh: make(chan *mineResult), fetchRateCh: make(chan chan uint64), submitRateCh: make(chan *hashrate), exitCh: make(chan chan error), } - go ethash.remote(notify) + go ethash.remote(notify, noverify) return ethash } // NewTester creates a small sized ethash PoW scheme useful only for testing // purposes. -func NewTester(notify []string) *Ethash { +func NewTester(notify []string, noverify bool) *Ethash { ethash := &Ethash{ config: Config{PowMode: ModeTest}, caches: newlru("cache", 1, newCache), datasets: newlru("dataset", 1, newDataset), update: make(chan struct{}), hashrate: metrics.NewMeter(), - workCh: make(chan *types.Block), - resultCh: make(chan *types.Block), + workCh: make(chan *sealTask), fetchWorkCh: make(chan *sealWork), submitWorkCh: make(chan *mineResult), fetchRateCh: make(chan chan uint64), submitRateCh: make(chan *hashrate), exitCh: make(chan chan error), } - go ethash.remote(notify) + go ethash.remote(notify, noverify) return ethash } diff --git a/consensus/ethash/ethash_test.go b/consensus/ethash/ethash_test.go index b190d63d6b..8eded2ca81 100644 --- a/consensus/ethash/ethash_test.go +++ b/consensus/ethash/ethash_test.go @@ -34,17 +34,23 @@ import ( func TestTestMode(t *testing.T) { header := &types.Header{Number: big.NewInt(1), Difficulty: big.NewInt(100)} - ethash := NewTester(nil) + ethash := NewTester(nil, false) defer ethash.Close() - block, err := ethash.Seal(nil, types.NewBlockWithHeader(header), nil) + results := make(chan *types.Block) + err := ethash.Seal(nil, types.NewBlockWithHeader(header), results, nil) if err != nil { t.Fatalf("failed to seal block: %v", err) } - header.Nonce = types.EncodeNonce(block.Nonce()) - header.MixDigest = block.MixDigest() - if err := ethash.VerifySeal(nil, header); err != nil { - t.Fatalf("unexpected verification error: %v", err) + select { + case block := <-results: + header.Nonce = types.EncodeNonce(block.Nonce()) + header.MixDigest = block.MixDigest() + if err := ethash.VerifySeal(nil, header); err != nil { + t.Fatalf("unexpected verification error: %v", err) + } + case <-time.NewTimer(time.Second).C: + t.Error("sealing result timeout") } } @@ -56,7 +62,7 @@ func TestCacheFileEvict(t *testing.T) { t.Fatal(err) } defer os.RemoveAll(tmpdir) - e := New(Config{CachesInMem: 3, CachesOnDisk: 10, CacheDir: tmpdir, PowMode: ModeTest}, nil) + e := New(Config{CachesInMem: 3, CachesOnDisk: 10, CacheDir: tmpdir, PowMode: ModeTest}, nil, false) defer e.Close() workers := 8 @@ -85,7 +91,7 @@ func verifyTest(wg *sync.WaitGroup, e *Ethash, workerIndex, epochs int) { } func TestRemoteSealer(t *testing.T) { - ethash := NewTester(nil) + ethash := NewTester(nil, false) defer ethash.Close() api := &API{ethash} @@ -97,7 +103,8 @@ func TestRemoteSealer(t *testing.T) { sealhash := ethash.SealHash(header) // Push new work. - ethash.Seal(nil, block, nil) + results := make(chan *types.Block) + ethash.Seal(nil, block, results, nil) var ( work [3]string @@ -114,20 +121,11 @@ func TestRemoteSealer(t *testing.T) { header = &types.Header{Number: big.NewInt(1), Difficulty: big.NewInt(1000)} block = types.NewBlockWithHeader(header) sealhash = ethash.SealHash(header) - ethash.Seal(nil, block, nil) + ethash.Seal(nil, block, results, nil) if work, err = api.GetWork(); err != nil || work[0] != sealhash.Hex() { t.Error("expect to return the latest pushed work") } - // Push block with higher block number. - newHead := &types.Header{Number: big.NewInt(2), Difficulty: big.NewInt(100)} - newBlock := types.NewBlockWithHeader(newHead) - newSealhash := ethash.SealHash(newHead) - ethash.Seal(nil, newBlock, nil) - - if res := api.SubmitWork(types.BlockNonce{}, newSealhash, common.Hash{}); res { - t.Error("expect to return false when submit a stale solution") - } } func TestHashRate(t *testing.T) { @@ -136,7 +134,7 @@ func TestHashRate(t *testing.T) { expect uint64 ids = []common.Hash{common.HexToHash("a"), common.HexToHash("b"), common.HexToHash("c")} ) - ethash := NewTester(nil) + ethash := NewTester(nil, false) defer ethash.Close() if tot := ethash.Hashrate(); tot != 0 { @@ -156,7 +154,7 @@ func TestHashRate(t *testing.T) { } func TestClosedRemoteSealer(t *testing.T) { - ethash := NewTester(nil) + ethash := NewTester(nil, false) time.Sleep(1 * time.Second) // ensure exit channel is listening ethash.Close() diff --git a/consensus/ethash/sealer.go b/consensus/ethash/sealer.go index a458c60f66..06c98a7811 100644 --- a/consensus/ethash/sealer.go +++ b/consensus/ethash/sealer.go @@ -35,6 +35,11 @@ import ( "github.com/ethereum/go-ethereum/log" ) +const ( + // staleThreshold is the maximum depth of the acceptable stale but valid ethash solution. + staleThreshold = 7 +) + var ( errNoMiningWork = errors.New("no mining work available yet") errInvalidSealResult = errors.New("invalid or stale proof-of-work solution") @@ -42,16 +47,21 @@ var ( // Seal implements consensus.Engine, attempting to find a nonce that satisfies // the block's difficulty requirements. -func (ethash *Ethash) Seal(chain consensus.ChainReader, block *types.Block, stop <-chan struct{}) (*types.Block, error) { +func (ethash *Ethash) Seal(chain consensus.ChainReader, block *types.Block, results chan<- *types.Block, stop <-chan struct{}) error { // If we're running a fake PoW, simply return a 0 nonce immediately if ethash.config.PowMode == ModeFake || ethash.config.PowMode == ModeFullFake { header := block.Header() header.Nonce, header.MixDigest = types.BlockNonce{}, common.Hash{} - return block.WithSeal(header), nil + select { + case results <- block.WithSeal(header): + default: + log.Warn("Sealing result is not read by miner", "mode", "fake", "sealhash", ethash.SealHash(block.Header())) + } + return nil } // If we're running a shared PoW, delegate sealing to it if ethash.shared != nil { - return ethash.shared.Seal(chain, block, stop) + return ethash.shared.Seal(chain, block, results, stop) } // Create a runner and the multiple search threads it directs abort := make(chan struct{}) @@ -62,7 +72,7 @@ func (ethash *Ethash) Seal(chain consensus.ChainReader, block *types.Block, stop seed, err := crand.Int(crand.Reader, big.NewInt(math.MaxInt64)) if err != nil { ethash.lock.Unlock() - return nil, err + return err } ethash.rand = rand.New(rand.NewSource(seed.Int64())) } @@ -75,34 +85,45 @@ func (ethash *Ethash) Seal(chain consensus.ChainReader, block *types.Block, stop } // Push new work to remote sealer if ethash.workCh != nil { - ethash.workCh <- block + ethash.workCh <- &sealTask{block: block, results: results} } - var pend sync.WaitGroup + var ( + pend sync.WaitGroup + locals = make(chan *types.Block) + ) for i := 0; i < threads; i++ { pend.Add(1) go func(id int, nonce uint64) { defer pend.Done() - ethash.mine(block, id, nonce, abort, ethash.resultCh) + ethash.mine(block, id, nonce, abort, locals) }(i, uint64(ethash.rand.Int63())) } // Wait until sealing is terminated or a nonce is found - var result *types.Block - select { - case <-stop: - // Outside abort, stop all miner threads - close(abort) - case result = <-ethash.resultCh: - // One of the threads found a block, abort all others - close(abort) - case <-ethash.update: - // Thread count was changed on user request, restart - close(abort) + go func() { + var result *types.Block + select { + case <-stop: + // Outside abort, stop all miner threads + close(abort) + case result = <-locals: + // One of the threads found a block, abort all others + select { + case results <- result: + default: + log.Warn("Sealing result is not read by miner", "mode", "local", "sealhash", ethash.SealHash(block.Header())) + } + close(abort) + case <-ethash.update: + // Thread count was changed on user request, restart + close(abort) + if err := ethash.Seal(chain, block, results, stop); err != nil { + log.Error("Failed to restart sealing after update", "err", err) + } + } + // Wait for all miners to terminate and return the block pend.Wait() - return ethash.Seal(chain, block, stop) - } - // Wait for all miners to terminate and return the block - pend.Wait() - return result, nil + }() + return nil } // mine is the actual proof-of-work miner that searches for a nonce starting from @@ -165,11 +186,12 @@ search: } // remote is a standalone goroutine to handle remote mining related stuff. -func (ethash *Ethash) remote(notify []string) { +func (ethash *Ethash) remote(notify []string, noverify bool) { var ( works = make(map[common.Hash]*types.Block) rates = make(map[common.Hash]hashrate) + results chan<- *types.Block currentBlock *types.Block currentWork [3]string @@ -226,11 +248,15 @@ func (ethash *Ethash) remote(notify []string) { // submitWork verifies the submitted pow solution, returning // whether the solution was accepted or not (not can be both a bad pow as well as // any other error, like no pending work or stale mining result). - submitWork := func(nonce types.BlockNonce, mixDigest common.Hash, hash common.Hash) bool { + submitWork := func(nonce types.BlockNonce, mixDigest common.Hash, sealhash common.Hash) bool { + if currentBlock == nil { + log.Error("Pending work without block", "sealhash", sealhash) + return false + } // Make sure the work submitted is present - block := works[hash] + block := works[sealhash] if block == nil { - log.Info("Work submitted but none pending", "hash", hash) + log.Warn("Work submitted but none pending", "sealhash", sealhash, "curnumber", currentBlock.NumberU64()) return false } // Verify the correctness of submitted result. @@ -239,26 +265,36 @@ func (ethash *Ethash) remote(notify []string) { header.MixDigest = mixDigest start := time.Now() - if err := ethash.verifySeal(nil, header, true); err != nil { - log.Warn("Invalid proof-of-work submitted", "hash", hash, "elapsed", time.Since(start), "err", err) - return false + if !noverify { + if err := ethash.verifySeal(nil, header, true); err != nil { + log.Warn("Invalid proof-of-work submitted", "sealhash", sealhash, "elapsed", time.Since(start), "err", err) + return false + } } - // Make sure the result channel is created. - if ethash.resultCh == nil { + // Make sure the result channel is assigned. + if results == nil { log.Warn("Ethash result channel is empty, submitted mining result is rejected") return false } - log.Trace("Verified correct proof-of-work", "hash", hash, "elapsed", time.Since(start)) + log.Trace("Verified correct proof-of-work", "sealhash", sealhash, "elapsed", time.Since(start)) // Solutions seems to be valid, return to the miner and notify acceptance. - select { - case ethash.resultCh <- block.WithSeal(header): - delete(works, hash) - return true - default: - log.Info("Work submitted is stale", "hash", hash) - return false + solution := block.WithSeal(header) + + // The submitted solution is within the scope of acceptance. + if solution.NumberU64()+staleThreshold > currentBlock.NumberU64() { + select { + case results <- solution: + log.Debug("Work submitted is acceptable", "number", solution.NumberU64(), "sealhash", sealhash, "hash", solution.Hash()) + return true + default: + log.Warn("Sealing result is not read by miner", "mode", "remote", "sealhash", sealhash) + return false + } } + // The submitted block is too old to accept, drop it. + log.Warn("Work submitted is too old", "number", solution.NumberU64(), "sealhash", sealhash, "hash", solution.Hash()) + return false } ticker := time.NewTicker(5 * time.Second) @@ -266,14 +302,12 @@ func (ethash *Ethash) remote(notify []string) { for { select { - case block := <-ethash.workCh: - if currentBlock != nil && block.ParentHash() != currentBlock.ParentHash() { - // Start new round mining, throw out all previous work. - works = make(map[common.Hash]*types.Block) - } + case work := <-ethash.workCh: // Update current work with new received block. // Note same work can be past twice, happens when changing CPU threads. - makeWork(block) + results = work.results + + makeWork(work.block) // Notify and requested URLs of the new work availability notifyWork() @@ -315,6 +349,14 @@ func (ethash *Ethash) remote(notify []string) { delete(rates, id) } } + // Clear stale pending blocks + if currentBlock != nil { + for hash, block := range works { + if block.NumberU64()+staleThreshold <= currentBlock.NumberU64() { + delete(works, hash) + } + } + } case errc := <-ethash.exitCh: // Exit remote loop if ethash is closed and return relevant error. diff --git a/consensus/ethash/sealer_test.go b/consensus/ethash/sealer_test.go index d1b66f9cf3..31d18b67c7 100644 --- a/consensus/ethash/sealer_test.go +++ b/consensus/ethash/sealer_test.go @@ -41,14 +41,14 @@ func TestRemoteNotify(t *testing.T) { go server.Serve(listener) // Create the custom ethash engine - ethash := NewTester([]string{"http://" + listener.Addr().String()}) + ethash := NewTester([]string{"http://" + listener.Addr().String()}, false) defer ethash.Close() // Stream a work task and ensure the notification bubbles out header := &types.Header{Number: big.NewInt(1), Difficulty: big.NewInt(100)} block := types.NewBlockWithHeader(header) - ethash.Seal(nil, block, nil) + ethash.Seal(nil, block, nil, nil) select { case work := <-sink: if want := ethash.SealHash(header).Hex(); work[0] != want { @@ -66,7 +66,7 @@ func TestRemoteNotify(t *testing.T) { } } -// Tests that pushing work packages fast to the miner doesn't cause any daa race +// Tests that pushing work packages fast to the miner doesn't cause any data race // issues in the notifications. func TestRemoteMultiNotify(t *testing.T) { // Start a simple webserver to capture notifications @@ -95,7 +95,7 @@ func TestRemoteMultiNotify(t *testing.T) { go server.Serve(listener) // Create the custom ethash engine - ethash := NewTester([]string{"http://" + listener.Addr().String()}) + ethash := NewTester([]string{"http://" + listener.Addr().String()}, false) defer ethash.Close() // Stream a lot of work task and ensure all the notifications bubble out @@ -103,7 +103,7 @@ func TestRemoteMultiNotify(t *testing.T) { header := &types.Header{Number: big.NewInt(int64(i)), Difficulty: big.NewInt(100)} block := types.NewBlockWithHeader(header) - ethash.Seal(nil, block, nil) + ethash.Seal(nil, block, nil, nil) } for i := 0; i < cap(sink); i++ { select { @@ -113,3 +113,87 @@ func TestRemoteMultiNotify(t *testing.T) { } } } + +// Tests whether stale solutions are correctly processed. +func TestStaleSubmission(t *testing.T) { + ethash := NewTester(nil, true) + defer ethash.Close() + api := &API{ethash} + + fakeNonce, fakeDigest := types.BlockNonce{0x01, 0x02, 0x03}, common.HexToHash("deadbeef") + + testcases := []struct { + headers []*types.Header + submitIndex int + submitRes bool + }{ + // Case1: submit solution for the latest mining package + { + []*types.Header{ + {ParentHash: common.BytesToHash([]byte{0xa}), Number: big.NewInt(1), Difficulty: big.NewInt(100000000)}, + }, + 0, + true, + }, + // Case2: submit solution for the previous package but have same parent. + { + []*types.Header{ + {ParentHash: common.BytesToHash([]byte{0xb}), Number: big.NewInt(2), Difficulty: big.NewInt(100000000)}, + {ParentHash: common.BytesToHash([]byte{0xb}), Number: big.NewInt(2), Difficulty: big.NewInt(100000001)}, + }, + 0, + true, + }, + // Case3: submit stale but acceptable solution + { + []*types.Header{ + {ParentHash: common.BytesToHash([]byte{0xc}), Number: big.NewInt(3), Difficulty: big.NewInt(100000000)}, + {ParentHash: common.BytesToHash([]byte{0xd}), Number: big.NewInt(9), Difficulty: big.NewInt(100000000)}, + }, + 0, + true, + }, + // Case4: submit very old solution + { + []*types.Header{ + {ParentHash: common.BytesToHash([]byte{0xe}), Number: big.NewInt(10), Difficulty: big.NewInt(100000000)}, + {ParentHash: common.BytesToHash([]byte{0xf}), Number: big.NewInt(17), Difficulty: big.NewInt(100000000)}, + }, + 0, + false, + }, + } + results := make(chan *types.Block, 16) + + for id, c := range testcases { + for _, h := range c.headers { + ethash.Seal(nil, types.NewBlockWithHeader(h), results, nil) + } + if res := api.SubmitWork(fakeNonce, ethash.SealHash(c.headers[c.submitIndex]), fakeDigest); res != c.submitRes { + t.Errorf("case %d submit result mismatch, want %t, get %t", id+1, c.submitRes, res) + } + if !c.submitRes { + continue + } + select { + case res := <-results: + if res.Header().Nonce != fakeNonce { + t.Errorf("case %d block nonce mismatch, want %s, get %s", id+1, fakeNonce, res.Header().Nonce) + } + if res.Header().MixDigest != fakeDigest { + t.Errorf("case %d block digest mismatch, want %s, get %s", id+1, fakeDigest, res.Header().MixDigest) + } + if res.Header().Difficulty.Uint64() != c.headers[c.submitIndex].Difficulty.Uint64() { + t.Errorf("case %d block difficulty mismatch, want %d, get %d", id+1, c.headers[c.submitIndex].Difficulty, res.Header().Difficulty) + } + if res.Header().Number.Uint64() != c.headers[c.submitIndex].Number.Uint64() { + t.Errorf("case %d block number mismatch, want %d, get %d", id+1, c.headers[c.submitIndex].Number.Uint64(), res.Header().Number.Uint64()) + } + if res.Header().ParentHash != c.headers[c.submitIndex].ParentHash { + t.Errorf("case %d block parent hash mismatch, want %s, get %s", id+1, c.headers[c.submitIndex].ParentHash.Hex(), res.Header().ParentHash.Hex()) + } + case <-time.NewTimer(time.Second).C: + t.Errorf("case %d fetch ethash result timeout", id+1) + } + } +} diff --git a/eth/api_tracer.go b/eth/api_tracer.go index 704a6cdbaa..0a8b9a9942 100644 --- a/eth/api_tracer.go +++ b/eth/api_tracer.go @@ -294,7 +294,7 @@ func (api *PrivateDebugAPI) traceChain(ctx context.Context, start, end *types.Bl failed = err break } - // Reference the trie twice, once for us, once for the trancer + // Reference the trie twice, once for us, once for the tracer database.TrieDB().Reference(root, common.Hash{}) if number >= origin { database.TrieDB().Reference(root, common.Hash{}) diff --git a/eth/backend.go b/eth/backend.go index da7e0b2cd5..3032e1a6d3 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -130,7 +130,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { chainConfig: chainConfig, eventMux: ctx.EventMux, accountManager: ctx.AccountManager, - engine: CreateConsensusEngine(ctx, chainConfig, &config.Ethash, config.MinerNotify, chainDb), + engine: CreateConsensusEngine(ctx, chainConfig, &config.Ethash, config.MinerNotify, config.MinerNoverify, chainDb), shutdownChan: make(chan bool), networkID: config.NetworkId, gasPrice: config.MinerGasPrice, @@ -216,7 +216,7 @@ func CreateDB(ctx *node.ServiceContext, config *Config, name string) (ethdb.Data } // CreateConsensusEngine creates the required type of consensus engine instance for an Ethereum service -func CreateConsensusEngine(ctx *node.ServiceContext, chainConfig *params.ChainConfig, config *ethash.Config, notify []string, db ethdb.Database) consensus.Engine { +func CreateConsensusEngine(ctx *node.ServiceContext, chainConfig *params.ChainConfig, config *ethash.Config, notify []string, noverify bool, db ethdb.Database) consensus.Engine { // If proof-of-authority is requested, set it up if chainConfig.Clique != nil { return clique.New(chainConfig.Clique, db) @@ -228,7 +228,7 @@ func CreateConsensusEngine(ctx *node.ServiceContext, chainConfig *params.ChainCo return ethash.NewFaker() case ethash.ModeTest: log.Warn("Ethash used in test mode") - return ethash.NewTester(nil) + return ethash.NewTester(nil, noverify) case ethash.ModeShared: log.Warn("Ethash used in shared mode") return ethash.NewShared() @@ -240,7 +240,7 @@ func CreateConsensusEngine(ctx *node.ServiceContext, chainConfig *params.ChainCo DatasetDir: config.DatasetDir, DatasetsInMem: config.DatasetsInMem, DatasetsOnDisk: config.DatasetsOnDisk, - }, notify) + }, notify, noverify) engine.SetThreads(-1) // Disable CPU mining return engine } diff --git a/eth/config.go b/eth/config.go index 5b2eca585c..517d7d8f3d 100644 --- a/eth/config.go +++ b/eth/config.go @@ -101,6 +101,7 @@ type Config struct { MinerExtraData []byte `toml:",omitempty"` MinerGasPrice *big.Int MinerRecommit time.Duration + MinerNoverify bool // Ethash options Ethash ethash.Config diff --git a/eth/gen_config.go b/eth/gen_config.go index a72e35bcd1..df4ffeb11f 100644 --- a/eth/gen_config.go +++ b/eth/gen_config.go @@ -35,6 +35,7 @@ func (c Config) MarshalTOML() (interface{}, error) { MinerExtraData hexutil.Bytes `toml:",omitempty"` MinerGasPrice *big.Int MinerRecommit time.Duration + MinerNoverify bool Ethash ethash.Config TxPool core.TxPoolConfig GPO gasprice.Config @@ -58,6 +59,7 @@ func (c Config) MarshalTOML() (interface{}, error) { enc.MinerExtraData = c.MinerExtraData enc.MinerGasPrice = c.MinerGasPrice enc.MinerRecommit = c.MinerRecommit + enc.MinerNoverify = c.MinerNoverify enc.Ethash = c.Ethash enc.TxPool = c.TxPool enc.GPO = c.GPO @@ -81,11 +83,11 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error { TrieCache *int TrieTimeout *time.Duration Etherbase *common.Address `toml:",omitempty"` - MinerThreads *int `toml:",omitempty"` MinerNotify []string `toml:",omitempty"` MinerExtraData *hexutil.Bytes `toml:",omitempty"` MinerGasPrice *big.Int MinerRecommit *time.Duration + MinerNoverify *bool Ethash *ethash.Config TxPool *core.TxPoolConfig GPO *gasprice.Config @@ -144,6 +146,9 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error { if dec.MinerRecommit != nil { c.MinerRecommit = *dec.MinerRecommit } + if dec.MinerNoverify != nil { + c.MinerNoverify = *dec.MinerNoverify + } if dec.Ethash != nil { c.Ethash = *dec.Ethash } diff --git a/les/backend.go b/les/backend.go index 75049da085..a3474a6830 100644 --- a/les/backend.go +++ b/les/backend.go @@ -102,7 +102,7 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) { peers: peers, reqDist: newRequestDistributor(peers, quitSync), accountManager: ctx.AccountManager, - engine: eth.CreateConsensusEngine(ctx, chainConfig, &config.Ethash, nil, chainDb), + engine: eth.CreateConsensusEngine(ctx, chainConfig, &config.Ethash, nil, false, chainDb), shutdownChan: make(chan bool), networkId: config.NetworkId, bloomRequests: make(chan chan *bloombits.Retrieval), diff --git a/miner/worker.go b/miner/worker.go index 1a881799d5..ca68da6e94 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -73,7 +73,7 @@ const ( // increasing upper limit or decreasing lower limit so that the limit can be reachable. intervalAdjustBias = 200 * 1000.0 * 1000.0 - // staleThreshold is the maximum distance of the acceptable stale block. + // staleThreshold is the maximum depth of the acceptable stale block. staleThreshold = 7 ) @@ -139,7 +139,7 @@ type worker struct { // Channels newWorkCh chan *newWorkReq taskCh chan *task - resultCh chan *task + resultCh chan *types.Block startCh chan struct{} exitCh chan struct{} resubmitIntervalCh chan time.Duration @@ -186,7 +186,7 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend, chainSideCh: make(chan core.ChainSideEvent, chainSideChanSize), newWorkCh: make(chan *newWorkReq), taskCh: make(chan *task), - resultCh: make(chan *task, resultQueueSize), + resultCh: make(chan *types.Block, resultQueueSize), exitCh: make(chan struct{}), startCh: make(chan struct{}, 1), resubmitIntervalCh: make(chan time.Duration), @@ -269,18 +269,10 @@ func (w *worker) isRunning() bool { return atomic.LoadInt32(&w.running) == 1 } -// close terminates all background threads maintained by the worker and cleans up buffered channels. +// close terminates all background threads maintained by the worker. // Note the worker does not support being closed multiple times. func (w *worker) close() { close(w.exitCh) - // Clean up buffered channels - for empty := false; !empty; { - select { - case <-w.resultCh: - default: - empty = true - } - } } // newWorkLoop is a standalone goroutine to submit new mining work upon received events. @@ -471,42 +463,6 @@ func (w *worker) mainLoop() { } } -// seal pushes a sealing task to consensus engine and submits the result. -func (w *worker) seal(t *task, stop <-chan struct{}) { - if w.skipSealHook != nil && w.skipSealHook(t) { - return - } - // The reason for caching task first is: - // A previous sealing action will be canceled by subsequent actions, - // however, remote miner may submit a result based on the cancelled task. - // So we should only submit the pending state corresponding to the seal result. - // TODO(rjl493456442) Replace the seal-wait logic structure - w.pendingMu.Lock() - w.pendingTasks[w.engine.SealHash(t.block.Header())] = t - w.pendingMu.Unlock() - - if block, err := w.engine.Seal(w.chain, t.block, stop); block != nil { - sealhash := w.engine.SealHash(block.Header()) - w.pendingMu.RLock() - task, exist := w.pendingTasks[sealhash] - w.pendingMu.RUnlock() - if !exist { - log.Error("Block found but no relative pending task", "number", block.Number(), "sealhash", sealhash, "hash", block.Hash()) - return - } - // Assemble sealing result - task.block = block - log.Info("Successfully sealed new block", "number", block.Number(), "sealhash", sealhash, "hash", block.Hash(), - "elapsed", common.PrettyDuration(time.Since(task.createdAt))) - select { - case w.resultCh <- task: - case <-w.exitCh: - } - } else if err != nil { - log.Warn("Block sealing failed", "err", err) - } -} - // taskLoop is a standalone goroutine to fetch sealing task from the generator and // push them to consensus engine. func (w *worker) taskLoop() { @@ -533,10 +489,20 @@ func (w *worker) taskLoop() { if sealHash == prev { continue } + // Interrupt previous sealing operation interrupt() - stopCh = make(chan struct{}) - prev = sealHash - go w.seal(task, stopCh) + stopCh, prev = make(chan struct{}), sealHash + + if w.skipSealHook != nil && w.skipSealHook(task) { + continue + } + w.pendingMu.Lock() + w.pendingTasks[w.engine.SealHash(task.block.Header())] = task + w.pendingMu.Unlock() + + if err := w.engine.Seal(w.chain, task.block, w.resultCh, stopCh); err != nil { + log.Warn("Block sealing failed", "err", err) + } case <-w.exitCh: interrupt() return @@ -549,38 +515,54 @@ func (w *worker) taskLoop() { func (w *worker) resultLoop() { for { select { - case result := <-w.resultCh: + case block := <-w.resultCh: // Short circuit when receiving empty result. - if result == nil { + if block == nil { continue } // Short circuit when receiving duplicate result caused by resubmitting. - block := result.block if w.chain.HasBlock(block.Hash(), block.NumberU64()) { continue } - // Update the block hash in all logs since it is now available and not when the - // receipt/log of individual transactions were created. - for _, r := range result.receipts { - for _, l := range r.Logs { - l.BlockHash = block.Hash() - } + var ( + sealhash = w.engine.SealHash(block.Header()) + hash = block.Hash() + ) + w.pendingMu.RLock() + task, exist := w.pendingTasks[sealhash] + w.pendingMu.RUnlock() + if !exist { + log.Error("Block found but no relative pending task", "number", block.Number(), "sealhash", sealhash, "hash", hash) + continue } - for _, log := range result.state.Logs() { - log.BlockHash = block.Hash() + // Different block could share same sealhash, deep copy here to prevent write-write conflict. + var ( + receipts = make([]*types.Receipt, len(task.receipts)) + logs []*types.Log + ) + for i, receipt := range task.receipts { + receipts[i] = new(types.Receipt) + *receipts[i] = *receipt + // Update the block hash in all logs since it is now available and not when the + // receipt/log of individual transactions were created. + for _, log := range receipt.Logs { + log.BlockHash = hash + } + logs = append(logs, receipt.Logs...) } // Commit block and state to database. - stat, err := w.chain.WriteBlockWithState(block, result.receipts, result.state) + stat, err := w.chain.WriteBlockWithState(block, receipts, task.state) if err != nil { log.Error("Failed writing block to chain", "err", err) continue } + log.Info("Successfully sealed new block", "number", block.Number(), "sealhash", sealhash, "hash", hash, + "elapsed", common.PrettyDuration(time.Since(task.createdAt))) + // Broadcast the block and announce chain insertion event w.mux.Post(core.NewMinedBlockEvent{Block: block}) - var ( - events []interface{} - logs = result.state.Logs() - ) + + var events []interface{} switch stat { case core.CanonStatTy: events = append(events, core.ChainEvent{Block: block, Hash: block.Hash(), Logs: logs})