397 lines
13 KiB
Go
397 lines
13 KiB
Go
// 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 <http://www.gnu.org/licenses/>.
|
|
|
|
package core
|
|
|
|
import (
|
|
"encoding/binary"
|
|
"fmt"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/common"
|
|
"github.com/ethereum/go-ethereum/core/types"
|
|
"github.com/ethereum/go-ethereum/ethdb"
|
|
"github.com/ethereum/go-ethereum/event"
|
|
"github.com/ethereum/go-ethereum/log"
|
|
)
|
|
|
|
// ChainIndexerBackend defines the methods needed to process chain segments in
|
|
// the background and write the segment results into the database. These can be
|
|
// used to create filter blooms or CHTs.
|
|
type ChainIndexerBackend interface {
|
|
// Reset initiates the processing of a new chain segment, potentially terminating
|
|
// any partially completed operations (in case of a reorg).
|
|
Reset(section uint64)
|
|
|
|
// Process crunches through the next header in the chain segment. The caller
|
|
// will ensure a sequential order of headers.
|
|
Process(header *types.Header)
|
|
|
|
// Commit finalizes the section metadata and stores it into the database. This
|
|
// interface will usually be a batch writer.
|
|
Commit(db ethdb.Database) error
|
|
}
|
|
|
|
// ChainIndexer does a post-processing job for equally sized sections of the
|
|
// canonical chain (like BlooomBits and CHT structures). A ChainIndexer is
|
|
// connected to the blockchain through the event system by starting a
|
|
// ChainEventLoop in a goroutine.
|
|
//
|
|
// Further child ChainIndexers can be added which use the output of the parent
|
|
// section indexer. These child indexers receive new head notifications only
|
|
// after an entire section has been finished or in case of rollbacks that might
|
|
// affect already finished sections.
|
|
type ChainIndexer struct {
|
|
chainDb ethdb.Database // Chain database to index the data from
|
|
indexDb ethdb.Database // Prefixed table-view of the db to write index metadata into
|
|
backend ChainIndexerBackend // Background processor generating the index data content
|
|
children []*ChainIndexer // Child indexers to cascade chain updates to
|
|
|
|
active uint32 // Flag whether the event loop was started
|
|
update chan struct{} // Notification channel that headers should be processed
|
|
quit chan chan error // Quit channel to tear down running goroutines
|
|
|
|
sectionSize uint64 // Number of blocks in a single chain segment to process
|
|
confirmsReq uint64 // Number of confirmations before processing a completed segment
|
|
|
|
storedSections uint64 // Number of sections successfully indexed into the database
|
|
knownSections uint64 // Number of sections known to be complete (block wise)
|
|
cascadedHead uint64 // Block number of the last completed section cascaded to subindexers
|
|
|
|
throttling time.Duration // Disk throttling to prevent a heavy upgrade from hogging resources
|
|
|
|
log log.Logger
|
|
lock sync.RWMutex
|
|
}
|
|
|
|
// NewChainIndexer creates a new chain indexer to do background processing on
|
|
// chain segments of a given size after certain number of confirmations passed.
|
|
// The throttling parameter might be used to prevent database thrashing.
|
|
func NewChainIndexer(chainDb, indexDb ethdb.Database, backend ChainIndexerBackend, section, confirm uint64, throttling time.Duration, kind string) *ChainIndexer {
|
|
c := &ChainIndexer{
|
|
chainDb: chainDb,
|
|
indexDb: indexDb,
|
|
backend: backend,
|
|
update: make(chan struct{}, 1),
|
|
quit: make(chan chan error),
|
|
sectionSize: section,
|
|
confirmsReq: confirm,
|
|
throttling: throttling,
|
|
log: log.New("type", kind),
|
|
}
|
|
// Initialize database dependent fields and start the updater
|
|
c.loadValidSections()
|
|
go c.updateLoop()
|
|
|
|
return c
|
|
}
|
|
|
|
// Start creates a goroutine to feed chain head events into the indexer for
|
|
// cascading background processing.
|
|
func (c *ChainIndexer) Start(currentHeader *types.Header, eventMux *event.TypeMux) {
|
|
go c.eventLoop(currentHeader, eventMux)
|
|
}
|
|
|
|
// Close tears down all goroutines belonging to the indexer and returns any error
|
|
// that might have occurred internally.
|
|
func (c *ChainIndexer) Close() error {
|
|
var errs []error
|
|
|
|
// Tear down the primary update loop
|
|
errc := make(chan error)
|
|
c.quit <- errc
|
|
if err := <-errc; err != nil {
|
|
errs = append(errs, err)
|
|
}
|
|
// If needed, tear down the secondary event loop
|
|
if atomic.LoadUint32(&c.active) != 0 {
|
|
c.quit <- errc
|
|
if err := <-errc; err != nil {
|
|
errs = append(errs, err)
|
|
}
|
|
}
|
|
// Return any failures
|
|
switch {
|
|
case len(errs) == 0:
|
|
return nil
|
|
|
|
case len(errs) == 1:
|
|
return errs[0]
|
|
|
|
default:
|
|
return fmt.Errorf("%v", errs)
|
|
}
|
|
}
|
|
|
|
// eventLoop is a secondary - optional - event loop of the indexer which is only
|
|
// started for the outermost indexer to push chain head events into a processing
|
|
// queue.
|
|
func (c *ChainIndexer) eventLoop(currentHeader *types.Header, eventMux *event.TypeMux) {
|
|
// Mark the chain indexer as active, requiring an additional teardown
|
|
atomic.StoreUint32(&c.active, 1)
|
|
|
|
// Subscribe to chain head events
|
|
sub := eventMux.Subscribe(ChainEvent{})
|
|
defer sub.Unsubscribe()
|
|
|
|
// Fire the initial new head event to start any outstanding processing
|
|
c.newHead(currentHeader.Number.Uint64(), false)
|
|
|
|
var (
|
|
prevHeader = currentHeader
|
|
prevHash = currentHeader.Hash()
|
|
)
|
|
for {
|
|
select {
|
|
case errc := <-c.quit:
|
|
// Chain indexer terminating, report no failure and abort
|
|
errc <- nil
|
|
return
|
|
|
|
case ev, ok := <-sub.Chan():
|
|
// Received a new event, ensure it's not nil (closing) and update
|
|
if !ok {
|
|
errc := <-c.quit
|
|
errc <- nil
|
|
return
|
|
}
|
|
header := ev.Data.(ChainEvent).Block.Header()
|
|
if header.ParentHash != prevHash {
|
|
c.newHead(FindCommonAncestor(c.chainDb, prevHeader, header).Number.Uint64(), true)
|
|
}
|
|
c.newHead(header.Number.Uint64(), false)
|
|
|
|
prevHeader, prevHash = header, header.Hash()
|
|
}
|
|
}
|
|
}
|
|
|
|
// newHead notifies the indexer about new chain heads and/or reorgs.
|
|
func (c *ChainIndexer) newHead(head uint64, reorg bool) {
|
|
c.lock.Lock()
|
|
defer c.lock.Unlock()
|
|
|
|
// 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
|
|
}
|
|
// Revert the stored sections from the database to the reorg point
|
|
if changed < c.storedSections {
|
|
c.setValidSections(changed)
|
|
}
|
|
// Update the new head number to te finalized section end and notify children
|
|
head = changed * c.sectionSize
|
|
|
|
if head < c.cascadedHead {
|
|
c.cascadedHead = head
|
|
for _, child := range c.children {
|
|
child.newHead(c.cascadedHead, true)
|
|
}
|
|
}
|
|
return
|
|
}
|
|
// No reorg, calculate the number of newly known sections and update if high enough
|
|
var sections uint64
|
|
if head >= c.confirmsReq {
|
|
sections = (head + 1 - c.confirmsReq) / c.sectionSize
|
|
if sections > c.knownSections {
|
|
c.knownSections = sections
|
|
|
|
select {
|
|
case c.update <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// updateLoop is the main event loop of the indexer which pushes chain segments
|
|
// down into the processing backend.
|
|
func (c *ChainIndexer) updateLoop() {
|
|
var updated time.Time
|
|
|
|
for {
|
|
select {
|
|
case errc := <-c.quit:
|
|
// Chain indexer terminating, report no failure and abort
|
|
errc <- nil
|
|
return
|
|
|
|
case <-c.update:
|
|
// Section headers completed (or rolled back), update the index
|
|
c.lock.Lock()
|
|
if c.knownSections > c.storedSections {
|
|
// Periodically print an upgrade log message to the user
|
|
if time.Since(updated) > 8*time.Second {
|
|
if c.knownSections > c.storedSections+1 {
|
|
c.log.Info("Upgrading chain index", "percentage", c.storedSections*100/c.knownSections)
|
|
}
|
|
updated = time.Now()
|
|
}
|
|
// Cache the current section count and head to allow unlocking the mutex
|
|
section := c.storedSections
|
|
var oldHead common.Hash
|
|
if section > 0 {
|
|
oldHead = c.sectionHead(section - 1)
|
|
}
|
|
// Process the newly defined section in the background
|
|
c.lock.Unlock()
|
|
newHead, err := c.processSection(section, oldHead)
|
|
c.lock.Lock()
|
|
|
|
// If processing succeeded and no reorgs occcurred, mark the section completed
|
|
if err == nil && oldHead == c.sectionHead(section-1) {
|
|
c.setSectionHead(section, newHead)
|
|
c.setValidSections(section + 1)
|
|
|
|
c.cascadedHead = c.storedSections*c.sectionSize - 1
|
|
for _, child := range c.children {
|
|
c.log.Trace("Cascading chain index update", "head", c.cascadedHead)
|
|
child.newHead(c.cascadedHead, false)
|
|
}
|
|
} else {
|
|
// If processing failed, don't retry until further notification
|
|
c.log.Debug("Chain index processing failed", "section", section, "err", err)
|
|
c.knownSections = c.storedSections
|
|
}
|
|
}
|
|
// If there are still further sections to process, reschedule
|
|
if c.knownSections > c.storedSections {
|
|
time.AfterFunc(c.throttling, func() {
|
|
select {
|
|
case c.update <- struct{}{}:
|
|
default:
|
|
}
|
|
})
|
|
}
|
|
c.lock.Unlock()
|
|
}
|
|
}
|
|
}
|
|
|
|
// processSection processes an entire section by calling backend functions while
|
|
// ensuring the continuity of the passed headers. Since the chain mutex is not
|
|
// held while processing, the continuity can be broken by a long reorg, in which
|
|
// case the function returns with an error.
|
|
func (c *ChainIndexer) processSection(section uint64, lastHead common.Hash) (common.Hash, error) {
|
|
c.log.Trace("Processing new chain section", "section", section)
|
|
|
|
// Reset and partial processing
|
|
c.backend.Reset(section)
|
|
|
|
for number := section * c.sectionSize; number < (section+1)*c.sectionSize; number++ {
|
|
hash := GetCanonicalHash(c.chainDb, number)
|
|
if hash == (common.Hash{}) {
|
|
return common.Hash{}, fmt.Errorf("canonical block #%d unknown", number)
|
|
}
|
|
header := GetHeader(c.chainDb, hash, number)
|
|
if header == nil {
|
|
return common.Hash{}, fmt.Errorf("block #%d [%x…] not found", number, hash[:4])
|
|
} else if header.ParentHash != lastHead {
|
|
return common.Hash{}, fmt.Errorf("chain reorged during section processing")
|
|
}
|
|
c.backend.Process(header)
|
|
lastHead = header.Hash()
|
|
}
|
|
if err := c.backend.Commit(c.chainDb); err != nil {
|
|
return common.Hash{}, err
|
|
}
|
|
return lastHead, nil
|
|
}
|
|
|
|
// Sections returns the number of processed sections maintained by the indexer
|
|
// and also the information about the last header indexed for potential canonical
|
|
// verifications.
|
|
func (c *ChainIndexer) Sections() (uint64, uint64, common.Hash) {
|
|
c.lock.Lock()
|
|
defer c.lock.Unlock()
|
|
|
|
return c.storedSections, c.storedSections*c.sectionSize - 1, c.sectionHead(c.storedSections - 1)
|
|
}
|
|
|
|
// AddChildIndexer adds a child ChainIndexer that can use the output of this one
|
|
func (c *ChainIndexer) AddChildIndexer(indexer *ChainIndexer) {
|
|
c.lock.Lock()
|
|
defer c.lock.Unlock()
|
|
|
|
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)
|
|
}
|
|
}
|
|
|
|
// loadValidSections reads the number of valid sections from the index database
|
|
// and caches is into the local state.
|
|
func (c *ChainIndexer) loadValidSections() {
|
|
data, _ := c.indexDb.Get([]byte("count"))
|
|
if len(data) == 8 {
|
|
c.storedSections = binary.BigEndian.Uint64(data[:])
|
|
}
|
|
}
|
|
|
|
// setValidSections writes the number of valid sections to the index database
|
|
func (c *ChainIndexer) setValidSections(sections uint64) {
|
|
// Set the current number of valid sections in the database
|
|
var data [8]byte
|
|
binary.BigEndian.PutUint64(data[:], sections)
|
|
c.indexDb.Put([]byte("count"), data[:])
|
|
|
|
// Remove any reorged sections, caching the valids in the mean time
|
|
for c.storedSections > sections {
|
|
c.storedSections--
|
|
c.removeSectionHead(c.storedSections)
|
|
}
|
|
c.storedSections = sections // needed if new > old
|
|
}
|
|
|
|
// sectionHead retrieves the last block hash of a processed section from the
|
|
// index database.
|
|
func (c *ChainIndexer) sectionHead(section uint64) common.Hash {
|
|
var data [8]byte
|
|
binary.BigEndian.PutUint64(data[:], section)
|
|
|
|
hash, _ := c.indexDb.Get(append([]byte("shead"), data[:]...))
|
|
if len(hash) == len(common.Hash{}) {
|
|
return common.BytesToHash(hash)
|
|
}
|
|
return common.Hash{}
|
|
}
|
|
|
|
// setSectionHead writes the last block hash of a processed section to the index
|
|
// database.
|
|
func (c *ChainIndexer) setSectionHead(section uint64, hash common.Hash) {
|
|
var data [8]byte
|
|
binary.BigEndian.PutUint64(data[:], section)
|
|
|
|
c.indexDb.Put(append([]byte("shead"), data[:]...), hash.Bytes())
|
|
}
|
|
|
|
// removeSectionHead removes the reference to a processed section from the index
|
|
// database.
|
|
func (c *ChainIndexer) removeSectionHead(section uint64) {
|
|
var data [8]byte
|
|
binary.BigEndian.PutUint64(data[:], section)
|
|
|
|
c.indexDb.Delete(append([]byte("shead"), data[:]...))
|
|
}
|