From 8bcf1c5ff20e90d9af3c92178093d32b2c5ae73d Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Sun, 5 Mar 2017 16:52:03 +0100 Subject: [PATCH 1/4] core, light: add canonical chain post-processor interface --- core/blockchain.go | 41 +++++++++- core/chain_processor.go | 176 ++++++++++++++++++++++++++++++++++++++++ core/headerchain.go | 11 ++- light/lightchain.go | 38 ++++++++- 4 files changed, 260 insertions(+), 6 deletions(-) create mode 100644 core/chain_processor.go diff --git a/core/blockchain.go b/core/blockchain.go index 794e1915fc..ecae115d7b 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -104,6 +104,8 @@ type BlockChain struct { procInterrupt int32 // interrupt signaler for block processing wg sync.WaitGroup // chain processing wait group for shutting down + chainProcessors []ChainProcessor // NewHead function called under chain lock after a new head has been written to the db + engine consensus.Engine processor Processor // block processor interface validator Validator // block and state validator interface @@ -168,6 +170,15 @@ func NewBlockChain(chainDb ethdb.Database, config *params.ChainConfig, engine co return bc, nil } +// AddChainProcessor adds a new head callback function that will be +// called under chain lock after a new head has been written to the db +func (self *BlockChain) AddChainProcessor(cp ChainProcessor) { + self.chainmu.Lock() + self.chainProcessors = append(self.chainProcessors, cp) + cp.NewHead(self.hc.CurrentHeader().Number.Uint64(), false) + self.chainmu.Unlock() +} + func (self *BlockChain) getProcInterrupt() bool { return atomic.LoadInt32(&self.procInterrupt) == 1 } @@ -243,6 +254,7 @@ func (bc *BlockChain) SetHead(head uint64) error { bc.mu.Lock() defer bc.mu.Unlock() + oldHead := bc.hc.CurrentHeader().Number.Uint64() // Rewind the header chain, deleting all block bodies until then delFn := func(hash common.Hash, num uint64) { DeleteBody(bc.chainDb, hash, num) @@ -256,6 +268,11 @@ func (bc *BlockChain) SetHead(head uint64) error { bc.blockCache.Purge() bc.futureBlocks.Purge() + if head < oldHead { + for _, cp := range bc.chainProcessors { + cp.NewHead(head, true) + } + } // Rewind the block chain, ensuring we don't end up with a stateless head block if bc.currentBlock != nil && currentHeader.Number.Uint64() < bc.currentBlock.NumberU64() { bc.currentBlock = bc.GetBlock(currentHeader.Hash(), currentHeader.Number.Uint64()) @@ -637,12 +654,15 @@ func (self *BlockChain) Rollback(chain []common.Hash) { self.mu.Lock() defer self.mu.Unlock() + var rollbackHead *types.Header + for i := len(chain) - 1; i >= 0; i-- { hash := chain[i] currentHeader := self.hc.CurrentHeader() if currentHeader.Hash() == hash { - self.hc.SetCurrentHeader(self.GetHeader(currentHeader.ParentHash, currentHeader.Number.Uint64()-1)) + rollbackHead = self.GetHeader(currentHeader.ParentHash, currentHeader.Number.Uint64()-1) + self.hc.SetCurrentHeader(rollbackHead) } if self.currentFastBlock.Hash() == hash { self.currentFastBlock = self.GetBlock(self.currentFastBlock.ParentHash(), self.currentFastBlock.NumberU64()-1) @@ -653,6 +673,12 @@ func (self *BlockChain) Rollback(chain []common.Hash) { WriteHeadBlockHash(self.chainDb, self.currentBlock.Hash()) } } + + if rollbackHead != nil { + for _, cp := range self.chainProcessors { + cp.NewHead(rollbackHead.Number.Uint64(), true) + } + } } // SetReceiptsData computes all the non-consensus fields of the receipts @@ -1316,7 +1342,16 @@ func (self *BlockChain) InsertHeaderChain(chain []*types.Header, checkFreq int) self.mu.Lock() defer self.mu.Unlock() - _, err := self.hc.WriteHeader(header) + _, err := self.hc.WriteHeader(header, func(head *types.Header) { + for _, cp := range self.chainProcessors { + cp.NewHead(head.Number.Uint64(), true) + } + }) + + for _, cp := range self.chainProcessors { + cp.NewHead(header.Number.Uint64(), false) + } + return err } @@ -1339,7 +1374,7 @@ func (self *BlockChain) writeHeader(header *types.Header) error { self.mu.Lock() defer self.mu.Unlock() - _, err := self.hc.WriteHeader(header) + _, err := self.hc.WriteHeader(header, nil) return err } diff --git a/core/chain_processor.go b/core/chain_processor.go new file mode 100644 index 0000000000..7112c3047c --- /dev/null +++ b/core/chain_processor.go @@ -0,0 +1,176 @@ +// Copyright 2017 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Package core implements the Ethereum consensus protocol. +package core + +import ( + "sync" + "time" +) + +// ChainProcessor is an external process that creates auxiliary data structures +// using the canonical chain as an input. Its callback function NewHead is called every +// time a new head is added to the chain or it gets rolled back. The callback function +// should never block because it is called under the chain's mutex lock so that the +// structures can stay consistent with the canonical chain. +// The same interface can be used for chaining processors (one using the output of the +// other). +type ChainProcessor interface { + NewHead(headNum uint64, rollBack bool) +} + +// ChainSectionProcessor is a ChainProcessor that does a post-processing job for +// equally sized sections of the canonical chain (like BlooomBits and CHT structures). +// Further child ChainProcessors can be added which use the output of this section +// processor. These child processors receive NewHead calls only after an entire section +// has been finished or in case of rollbacks that might affect already finished sections. +type ChainSectionProcessor struct { + backend ChainSectionProcessorBackend + sectionSize, confirmReq uint64 + stop chan struct{} + updateChn chan uint64 + lock sync.Mutex + calcValid bool + procWait time.Duration + stored, calcIdx, lastForwarded uint64 + childProcessors []ChainProcessor +} + +// ChainSectionProcessorBackend does the actual post-processing job. +// GetStored and SetStored are called under the canonical chain lock and should +// not block. Process is called without a chain lock and may take longer to +// finish. If the chain gets rolled back while Process is running, the results +// are considered invalid and SetStored is not called. +type ChainSectionProcessorBackend interface { + Process(idx uint64) bool + GetStored() uint64 + SetStored(count uint64) +} + +// NewChainSectionProcessor creates a new ChainSectionProcessor +func NewChainSectionProcessor(backend ChainSectionProcessorBackend, sectionSize, confirmReq uint64, procWait time.Duration, stop chan struct{}) *ChainSectionProcessor { + csp := &ChainSectionProcessor{ + backend: backend, + sectionSize: sectionSize, + confirmReq: confirmReq, + stop: stop, + procWait: procWait, + updateChn: make(chan uint64, 100), + stored: backend.GetStored(), + } + go csp.updateLoop() + return csp +} + +func (csp *ChainSectionProcessor) updateLoop() { + tryUpdate := make(chan struct{}, 1) + updating := false + var targetCnt uint64 + + for { + select { + case <-csp.stop: + return + case targetCnt = <-csp.updateChn: + if !updating { + updating = true + tryUpdate <- struct{}{} + } + case <-tryUpdate: + csp.lock.Lock() + if targetCnt > csp.stored { + csp.calcValid = true + csp.calcIdx = csp.stored + + csp.lock.Unlock() + ok := csp.backend.Process(csp.calcIdx) + csp.lock.Lock() + + if ok && csp.calcValid { + csp.stored = csp.calcIdx + 1 + csp.backend.SetStored(csp.stored) + + csp.lastForwarded = csp.stored*csp.sectionSize - 1 + for _, cp := range csp.childProcessors { + cp.NewHead(csp.lastForwarded, false) + } + + } + csp.calcValid = false + } + stored := csp.stored + csp.lock.Unlock() + + if targetCnt > stored { + go func() { + time.Sleep(csp.procWait) + tryUpdate <- struct{}{} + }() + } else { + updating = false + } + } + } +} + +// NewHead implements the ChainProcessor interface +func (csp *ChainSectionProcessor) NewHead(headNum uint64, rollback bool) { + csp.lock.Lock() + defer csp.lock.Unlock() + + if rollback { + firstChanged := headNum / csp.sectionSize + if firstChanged <= csp.calcIdx { + csp.calcValid = false + } + if firstChanged < csp.stored { + csp.stored = firstChanged + csp.backend.SetStored(csp.stored) + select { + case <-csp.stop: + case csp.updateChn <- firstChanged: + } + } + + if headNum < csp.lastForwarded { + csp.lastForwarded = headNum + for _, cp := range csp.childProcessors { + cp.NewHead(csp.lastForwarded, true) + } + } + + } else { + var newCount uint64 + if headNum >= csp.confirmReq { + newCount = (headNum + 1 - csp.confirmReq) / csp.sectionSize + if newCount > csp.stored { + go func() { + select { + case <-csp.stop: + case csp.updateChn <- newCount: + } + }() + } + } + } +} + +// AddChildProcessor adds a child ChainProcessor that can use the output of this +// section processor. +func (csp *ChainSectionProcessor) AddChildProcessor(cp ChainProcessor) { + csp.childProcessors = append(csp.childProcessors, cp) +} diff --git a/core/headerchain.go b/core/headerchain.go index 9bb7f17937..4dce739f3c 100644 --- a/core/headerchain.go +++ b/core/headerchain.go @@ -119,6 +119,10 @@ func (hc *HeaderChain) GetBlockNumber(hash common.Hash) uint64 { return number } +// ReorgCallback is a callback function to be called in case of a chain reorg. +// The parameter is the header of the last unchanged canonical block. +type ReorgCallback func(*types.Header) + // WriteHeader writes a header into the local chain, given that its parent is // already known. If the total difficulty of the newly inserted header becomes // greater than the current known TD, the canonical chain is re-routed. @@ -128,7 +132,7 @@ func (hc *HeaderChain) GetBlockNumber(hash common.Hash) uint64 { // without the real blocks. Hence, writing headers directly should only be done // in two scenarios: pure-header mode of operation (light clients), or properly // separated header/block phases (non-archive clients). -func (hc *HeaderChain) WriteHeader(header *types.Header) (status WriteStatus, err error) { +func (hc *HeaderChain) WriteHeader(header *types.Header, reorgCallback ReorgCallback) (status WriteStatus, err error) { // Cache some values to prevent constant recalculation var ( hash = header.Hash() @@ -174,6 +178,11 @@ func (hc *HeaderChain) WriteHeader(header *types.Header) (status WriteStatus, er headNumber = headHeader.Number.Uint64() - 1 headHeader = hc.GetHeader(headHash, headNumber) } + + if reorgCallback != nil && headNumber < hc.currentHeader.Number.Uint64() { + reorgCallback(headHeader) + } + // Extend the canonical chain with the new header if err := WriteCanonicalHash(hc.chainDb, hash, number); err != nil { log.Crit("Failed to insert header number", "err", err) diff --git a/light/lightchain.go b/light/lightchain.go index 5b7e570417..26517b9b40 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -64,6 +64,8 @@ type LightChain struct { procInterrupt int32 // interrupt signaler for block processing wg sync.WaitGroup + chainProcessors []core.ChainProcessor // NewHead function called under chain lock after a new head has been written to the db + engine consensus.Engine } @@ -114,6 +116,15 @@ func NewLightChain(odr OdrBackend, config *params.ChainConfig, engine consensus. return bc, nil } +// AddChainProcessor adds a new head callback function that will be +// called under chain lock after a new head has been written to the db +func (self *LightChain) AddChainProcessor(cp core.ChainProcessor) { + self.chainmu.Lock() + self.chainProcessors = append(self.chainProcessors, cp) + cp.NewHead(self.hc.CurrentHeader().Number.Uint64(), false) + self.chainmu.Unlock() +} + func (self *LightChain) getProcInterrupt() bool { return atomic.LoadInt32(&self.procInterrupt) == 1 } @@ -149,7 +160,13 @@ func (bc *LightChain) SetHead(head uint64) { bc.mu.Lock() defer bc.mu.Unlock() + oldHead := bc.hc.CurrentHeader().Number.Uint64() bc.hc.SetHead(head, nil) + if head < oldHead { + for _, cp := range bc.chainProcessors { + cp.NewHead(head, true) + } + } bc.loadLastState() } @@ -312,11 +329,20 @@ func (self *LightChain) Rollback(chain []common.Hash) { self.mu.Lock() defer self.mu.Unlock() + var rollbackHead *types.Header + for i := len(chain) - 1; i >= 0; i-- { hash := chain[i] if head := self.hc.CurrentHeader(); head.Hash() == hash { - self.hc.SetCurrentHeader(self.GetHeader(head.ParentHash, head.Number.Uint64()-1)) + rollbackHead = self.GetHeader(head.ParentHash, head.Number.Uint64()-1) + self.hc.SetCurrentHeader(rollbackHead) + } + } + + if rollbackHead != nil { + for _, cp := range self.chainProcessors { + cp.NewHead(rollbackHead.Number.Uint64(), true) } } } @@ -367,7 +393,15 @@ func (self *LightChain) InsertHeaderChain(chain []*types.Header, checkFreq int) self.mu.Lock() defer self.mu.Unlock() - status, err := self.hc.WriteHeader(header) + status, err := self.hc.WriteHeader(header, func(head *types.Header) { + for _, cp := range self.chainProcessors { + cp.NewHead(head.Number.Uint64(), true) + } + }) + + for _, cp := range self.chainProcessors { + cp.NewHead(header.Number.Uint64(), false) + } switch status { case core.CanonStatTy: From 712b17a01909f3e99ac1e36efd86a3100a5efb31 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Fri, 5 May 2017 21:25:06 +0200 Subject: [PATCH 2/4] core: added updateMsg to ChainSectionProcessorBackend interface --- core/chain_processor.go | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/core/chain_processor.go b/core/chain_processor.go index 7112c3047c..28335001f6 100644 --- a/core/chain_processor.go +++ b/core/chain_processor.go @@ -59,6 +59,7 @@ type ChainSectionProcessorBackend interface { Process(idx uint64) bool GetStored() uint64 SetStored(count uint64) + UpdateMsg(done, all uint64) } // NewChainSectionProcessor creates a new ChainSectionProcessor @@ -80,6 +81,7 @@ func (csp *ChainSectionProcessor) updateLoop() { tryUpdate := make(chan struct{}, 1) updating := false var targetCnt uint64 + updateMsg := false for { select { @@ -93,6 +95,10 @@ func (csp *ChainSectionProcessor) updateLoop() { case <-tryUpdate: csp.lock.Lock() if targetCnt > csp.stored { + if !updateMsg && targetCnt > csp.stored+1 { + updateMsg = true + csp.backend.UpdateMsg(csp.stored, targetCnt) + } csp.calcValid = true csp.calcIdx = csp.stored @@ -103,12 +109,16 @@ func (csp *ChainSectionProcessor) updateLoop() { if ok && csp.calcValid { csp.stored = csp.calcIdx + 1 csp.backend.SetStored(csp.stored) - + if updateMsg { + csp.backend.UpdateMsg(csp.stored, targetCnt) + if csp.stored >= targetCnt { + updateMsg = false + } + } csp.lastForwarded = csp.stored*csp.sectionSize - 1 for _, cp := range csp.childProcessors { cp.NewHead(csp.lastForwarded, false) } - } csp.calcValid = false } From f421827f71a1ac470913b918e5f25f55409dcedb Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Fri, 5 May 2017 22:08:28 +0200 Subject: [PATCH 3/4] core: added comments and renamed variables in ChainSectionProcessor --- core/chain_processor.go | 40 +++++++++++++++++++++++----------------- 1 file changed, 23 insertions(+), 17 deletions(-) diff --git a/core/chain_processor.go b/core/chain_processor.go index 28335001f6..9008a7d5d3 100644 --- a/core/chain_processor.go +++ b/core/chain_processor.go @@ -26,7 +26,8 @@ import ( // using the canonical chain as an input. Its callback function NewHead is called every // time a new head is added to the chain or it gets rolled back. The callback function // should never block because it is called under the chain's mutex lock so that the -// structures can stay consistent with the canonical chain. +// structures can stay consistent with the canonical chain. It should also not call +// any chain functions that use this mutex because that would cause a deadlock. // The same interface can be used for chaining processors (one using the output of the // other). type ChainProcessor interface { @@ -42,7 +43,7 @@ type ChainSectionProcessor struct { backend ChainSectionProcessorBackend sectionSize, confirmReq uint64 stop chan struct{} - updateChn chan uint64 + updateCh chan uint64 lock sync.Mutex calcValid bool procWait time.Duration @@ -56,13 +57,18 @@ type ChainSectionProcessor struct { // finish. If the chain gets rolled back while Process is running, the results // are considered invalid and SetStored is not called. type ChainSectionProcessorBackend interface { - Process(idx uint64) bool - GetStored() uint64 - SetStored(count uint64) - UpdateMsg(done, all uint64) + Process(idx uint64) bool // do the processing of section idx but do not mark it as valid yet; return false if processing failed + GetStored() uint64 // return the number of sections processed, stored and marked as valid + SetStored(count uint64) // set the number of stored and valid processed sections (called after Process returned true if no reorg happened in the meantime) + UpdateMsg(done, all uint64) // print a progress update message if necessary (only called when multiple sections need to be processed) } // NewChainSectionProcessor creates a new ChainSectionProcessor +// backend: an implementation of ChainSectionProcessorBackend +// sectionSize: the size of processable sections +// confirmReq: required number of confirmation blocks before a new section is being processed +// procWait: waiting time between processing sections (simple way of limiting the resource usage of a db upgrade) +// stop: quit channel func NewChainSectionProcessor(backend ChainSectionProcessorBackend, sectionSize, confirmReq uint64, procWait time.Duration, stop chan struct{}) *ChainSectionProcessor { csp := &ChainSectionProcessor{ backend: backend, @@ -70,7 +76,7 @@ func NewChainSectionProcessor(backend ChainSectionProcessorBackend, sectionSize, confirmReq: confirmReq, stop: stop, procWait: procWait, - updateChn: make(chan uint64, 100), + updateCh: make(chan uint64, 100), stored: backend.GetStored(), } go csp.updateLoop() @@ -80,24 +86,24 @@ func NewChainSectionProcessor(backend ChainSectionProcessorBackend, sectionSize, func (csp *ChainSectionProcessor) updateLoop() { tryUpdate := make(chan struct{}, 1) updating := false - var targetCnt uint64 + var targetCount uint64 updateMsg := false for { select { case <-csp.stop: return - case targetCnt = <-csp.updateChn: + case targetCount = <-csp.updateCh: if !updating { updating = true tryUpdate <- struct{}{} } case <-tryUpdate: csp.lock.Lock() - if targetCnt > csp.stored { - if !updateMsg && targetCnt > csp.stored+1 { + if targetCount > csp.stored { + if !updateMsg && targetCount > csp.stored+1 { updateMsg = true - csp.backend.UpdateMsg(csp.stored, targetCnt) + csp.backend.UpdateMsg(csp.stored, targetCount) } csp.calcValid = true csp.calcIdx = csp.stored @@ -110,8 +116,8 @@ func (csp *ChainSectionProcessor) updateLoop() { csp.stored = csp.calcIdx + 1 csp.backend.SetStored(csp.stored) if updateMsg { - csp.backend.UpdateMsg(csp.stored, targetCnt) - if csp.stored >= targetCnt { + csp.backend.UpdateMsg(csp.stored, targetCount) + if csp.stored >= targetCount { updateMsg = false } } @@ -125,7 +131,7 @@ func (csp *ChainSectionProcessor) updateLoop() { stored := csp.stored csp.lock.Unlock() - if targetCnt > stored { + if targetCount > stored { go func() { time.Sleep(csp.procWait) tryUpdate <- struct{}{} @@ -152,7 +158,7 @@ func (csp *ChainSectionProcessor) NewHead(headNum uint64, rollback bool) { csp.backend.SetStored(csp.stored) select { case <-csp.stop: - case csp.updateChn <- firstChanged: + case csp.updateCh <- firstChanged: } } @@ -171,7 +177,7 @@ func (csp *ChainSectionProcessor) NewHead(headNum uint64, rollback bool) { go func() { select { case <-csp.stop: - case csp.updateChn <- newCount: + case csp.updateCh <- newCount: } }() } From 3d8cb46e04c7bfc0fd7d040c6eb204ee6a759263 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Sat, 20 May 2017 19:51:45 +0200 Subject: [PATCH 4/4] core: new ChainIndex interface --- core/chain_processor.go | 176 ++++++++++++++++++++++++---------------- 1 file changed, 105 insertions(+), 71 deletions(-) diff --git a/core/chain_processor.go b/core/chain_processor.go index 9008a7d5d3..d431217137 100644 --- a/core/chain_processor.go +++ b/core/chain_processor.go @@ -18,8 +18,13 @@ package core import ( + "encoding/binary" "sync" "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethdb" ) // ChainProcessor is an external process that creates auxiliary data structures @@ -34,13 +39,15 @@ type ChainProcessor interface { NewHead(headNum uint64, rollBack bool) } -// ChainSectionProcessor is a ChainProcessor that does a post-processing job for +// ChainIndexer is a ChainProcessor that does a post-processing job for // equally sized sections of the canonical chain (like BlooomBits and CHT structures). // Further child ChainProcessors can be added which use the output of this section // processor. These child processors receive NewHead calls only after an entire section // has been finished or in case of rollbacks that might affect already finished sections. -type ChainSectionProcessor struct { - backend ChainSectionProcessorBackend +type ChainIndexer struct { + db ethdb.Database + validSectionsKey []byte + backend ChainIndex sectionSize, confirmReq uint64 stop chan struct{} updateCh chan uint64 @@ -51,39 +58,36 @@ type ChainSectionProcessor struct { childProcessors []ChainProcessor } -// ChainSectionProcessorBackend does the actual post-processing job. -// GetStored and SetStored are called under the canonical chain lock and should -// not block. Process is called without a chain lock and may take longer to -// finish. If the chain gets rolled back while Process is running, the results -// are considered invalid and SetStored is not called. -type ChainSectionProcessorBackend interface { - Process(idx uint64) bool // do the processing of section idx but do not mark it as valid yet; return false if processing failed - GetStored() uint64 // return the number of sections processed, stored and marked as valid - SetStored(count uint64) // set the number of stored and valid processed sections (called after Process returned true if no reorg happened in the meantime) +type ChainIndex interface { + Reset(section uint64) + Process(header *types.Header) + Commit(db ethdb.Database) error UpdateMsg(done, all uint64) // print a progress update message if necessary (only called when multiple sections need to be processed) } -// NewChainSectionProcessor creates a new ChainSectionProcessor -// backend: an implementation of ChainSectionProcessorBackend +// NewChainIndexer creates a new ChainIndexer +// backend: an implementation of ChainIndex // sectionSize: the size of processable sections // confirmReq: required number of confirmation blocks before a new section is being processed // procWait: waiting time between processing sections (simple way of limiting the resource usage of a db upgrade) // stop: quit channel -func NewChainSectionProcessor(backend ChainSectionProcessorBackend, sectionSize, confirmReq uint64, procWait time.Duration, stop chan struct{}) *ChainSectionProcessor { - csp := &ChainSectionProcessor{ - backend: backend, - sectionSize: sectionSize, - confirmReq: confirmReq, - stop: stop, - procWait: procWait, - updateCh: make(chan uint64, 100), - stored: backend.GetStored(), +func NewChainIndexer(db ethdb.Database, validSectionsKey []byte, backend ChainIndex, sectionSize, confirmReq uint64, procWait time.Duration, stop chan struct{}) *ChainIndexer { + c := &ChainIndexer{ + db: db, + validSectionsKey: validSectionsKey, + backend: backend, + sectionSize: sectionSize, + confirmReq: confirmReq, + stop: stop, + procWait: procWait, + updateCh: make(chan uint64, 100), } - go csp.updateLoop() - return csp + c.stored = c.ValidSections() + go c.updateLoop() + return c } -func (csp *ChainSectionProcessor) updateLoop() { +func (c *ChainIndexer) updateLoop() { tryUpdate := make(chan struct{}, 1) updating := false var targetCount uint64 @@ -91,49 +95,49 @@ func (csp *ChainSectionProcessor) updateLoop() { for { select { - case <-csp.stop: + case <-c.stop: return - case targetCount = <-csp.updateCh: + case targetCount = <-c.updateCh: if !updating { updating = true tryUpdate <- struct{}{} } case <-tryUpdate: - csp.lock.Lock() - if targetCount > csp.stored { - if !updateMsg && targetCount > csp.stored+1 { + c.lock.Lock() + if targetCount > c.stored { + if !updateMsg && targetCount > c.stored+1 { updateMsg = true - csp.backend.UpdateMsg(csp.stored, targetCount) + c.backend.UpdateMsg(c.stored, targetCount) } - csp.calcValid = true - csp.calcIdx = csp.stored + c.calcValid = true + c.calcIdx = c.stored - csp.lock.Unlock() - ok := csp.backend.Process(csp.calcIdx) - csp.lock.Lock() + c.lock.Unlock() + ok := c.processSection(c.calcIdx) + c.lock.Lock() - if ok && csp.calcValid { - csp.stored = csp.calcIdx + 1 - csp.backend.SetStored(csp.stored) + if ok && c.calcValid { + c.stored = c.calcIdx + 1 + c.setValidSections(c.stored) if updateMsg { - csp.backend.UpdateMsg(csp.stored, targetCount) - if csp.stored >= targetCount { + c.backend.UpdateMsg(c.stored, targetCount) + if c.stored >= targetCount { updateMsg = false } } - csp.lastForwarded = csp.stored*csp.sectionSize - 1 - for _, cp := range csp.childProcessors { - cp.NewHead(csp.lastForwarded, false) + c.lastForwarded = c.stored*c.sectionSize - 1 + for _, cp := range c.childProcessors { + cp.NewHead(c.lastForwarded, false) } } - csp.calcValid = false + c.calcValid = false } - stored := csp.stored - csp.lock.Unlock() + stored := c.stored + c.lock.Unlock() if targetCount > stored { go func() { - time.Sleep(csp.procWait) + time.Sleep(c.procWait) tryUpdate <- struct{}{} }() } else { @@ -143,41 +147,71 @@ func (csp *ChainSectionProcessor) updateLoop() { } } +func (c *ChainIndexer) processSection(section uint64) bool { + c.backend.Reset(section) + for i := section * c.sectionSize; i < (section+1)*c.sectionSize; i++ { + hash := GetCanonicalHash(c.db, i) + if hash == (common.Hash{}) { + return false + } + header := GetHeader(c.db, hash, i) + if header == nil { + return false + } + c.backend.Process(header) + } + return c.backend.Commit(c.db) == nil +} + +func (c *ChainIndexer) ValidSections() uint64 { + data, _ := c.db.Get(c.validSectionsKey) + if len(data) == 8 { + return binary.BigEndian.Uint64(data[:]) + } + return 0 +} + +func (c *ChainIndexer) setValidSections(cnt uint64) { + var data [8]byte + binary.BigEndian.PutUint64(data[:], cnt) + c.db.Put(c.validSectionsKey, data[:]) +} + // NewHead implements the ChainProcessor interface -func (csp *ChainSectionProcessor) NewHead(headNum uint64, rollback bool) { - csp.lock.Lock() - defer csp.lock.Unlock() +func (c *ChainIndexer) NewHead(headNum uint64, rollback bool) { + c.lock.Lock() + defer c.lock.Unlock() if rollback { - firstChanged := headNum / csp.sectionSize - if firstChanged <= csp.calcIdx { - csp.calcValid = false + firstChanged := headNum / c.sectionSize + if firstChanged <= c.calcIdx { + c.calcValid = false } - if firstChanged < csp.stored { - csp.stored = firstChanged - csp.backend.SetStored(csp.stored) + if firstChanged < c.stored { + c.stored = firstChanged + c.setValidSections(c.stored) select { - case <-csp.stop: - case csp.updateCh <- firstChanged: + case <-c.stop: + case c.updateCh <- firstChanged: } } - if headNum < csp.lastForwarded { - csp.lastForwarded = headNum - for _, cp := range csp.childProcessors { - cp.NewHead(csp.lastForwarded, true) + if headNum < c.lastForwarded { + c.lastForwarded = headNum + for _, cp := range c.childProcessors { + cp.NewHead(c.lastForwarded, true) } } } else { var newCount uint64 - if headNum >= csp.confirmReq { - newCount = (headNum + 1 - csp.confirmReq) / csp.sectionSize - if newCount > csp.stored { + if headNum >= c.confirmReq { + newCount = (headNum + 1 - c.confirmReq) / c.sectionSize + if newCount > c.stored { go func() { select { - case <-csp.stop: - case csp.updateCh <- newCount: + case <-c.stop: + case c.updateCh <- newCount: } }() } @@ -187,6 +221,6 @@ func (csp *ChainSectionProcessor) NewHead(headNum uint64, rollback bool) { // AddChildProcessor adds a child ChainProcessor that can use the output of this // section processor. -func (csp *ChainSectionProcessor) AddChildProcessor(cp ChainProcessor) { - csp.childProcessors = append(csp.childProcessors, cp) +func (c *ChainIndexer) AddChildProcessor(cp ChainProcessor) { + c.childProcessors = append(c.childProcessors, cp) }