path: root/core/chain_indexer.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 implements the Ethereum consensus protocol.
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"
    "github.com/ethereum/go-ethereum/event"
)

// 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, indexDb                            ethdb.Database
    backend                                     ChainIndexerBackend
    sectionSize, confirmReq                     uint64
    stop                                        chan struct{}
    lock                                        sync.Mutex
    procWait                                    time.Duration
    tryUpdate                                   chan struct{}
    stored, targetCount, calcIdx, lastForwarded uint64
    updating                                    bool
    children                                    []*ChainIndexer
}

// ChainIndexerBackend interface is a backend for the indexer doing the actual post-processing job
type ChainIndexerBackend interface {
    Reset(section uint64)           // start processing a new section
    Process(header *types.Header)   // process a single block (called for each block in the section)
    Commit(db ethdb.Database) error // do some more processing if necessary and store the results in the database
    UpdateMsg(done, all uint64)     // print a progress update message if necessary (only called when multiple sections need to be processed)
}

// NewChainIndexer creates a new  ChainIndexer
//  db:             database where the index of available processed sections is stored (the index is stored by the
//                  indexer, the actual processed chain data is stored by the backend)
//  dbKey:          key prefix where the index is stored
//  backend:        an implementation of ChainIndexerBackend
//  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 NewChainIndexer(chainDb, indexDb ethdb.Database, backend ChainIndexerBackend, sectionSize, confirmReq uint64, procWait time.Duration, stop chan struct{}) *ChainIndexer {
    c := &ChainIndexer{
        chainDb:     chainDb,
        indexDb:     indexDb,
        backend:     backend,
        sectionSize: sectionSize,
        confirmReq:  confirmReq,
        tryUpdate:   make(chan struct{}, 1),
        stop:        stop,
        procWait:    procWait,
    }
    c.stored = c.getValidSections()
    go c.updateLoop()
    return c
}

// updateLoop is the main event loop of the indexer
func (c *ChainIndexer) updateLoop() {
    updateMsg := false

    for {
        select {
        case <-c.stop:
            return
        case <-c.tryUpdate:
            c.lock.Lock()
            if c.targetCount > c.stored {
                if !updateMsg && c.targetCount > c.stored+1 {
                    updateMsg = true
                    c.backend.UpdateMsg(c.stored, c.targetCount)
                }
                c.calcIdx = c.stored

                var lastSectionHead common.Hash
                if c.calcIdx > 0 {
                    lastSectionHead = c.getSectionHead(c.calcIdx - 1)
                }

                c.lock.Unlock()
                sectionHead, ok := c.processSection(c.calcIdx, lastSectionHead)
                c.lock.Lock()

                if ok && lastSectionHead == c.getSectionHead(c.calcIdx-1) {
                    c.stored = c.calcIdx + 1
                    c.setSectionHead(c.calcIdx, sectionHead)
                    c.setValidSections(c.stored)
                    if updateMsg {
                        c.backend.UpdateMsg(c.stored, c.targetCount)
                        if c.stored >= c.targetCount {
                            updateMsg = false
                        }
                    }
                    c.lastForwarded = c.stored*c.sectionSize - 1
                    for _, cp := range c.children {
                        cp.newHead(c.lastForwarded, false)
                    }
                } else {
                    // if processing has failed, do not retry until further notification
                    c.targetCount = c.stored
                }
            }

            if c.targetCount > c.stored {
                go func() {
                    time.Sleep(c.procWait)
                    c.tryUpdate <- struct{}{}
                }()
            } else {
                c.updating = false
            }
            c.lock.Unlock()
        }
    }
}

// ChainEventLoop runs in a goroutine and feeds blockchain events to the indexer by calling newHead
// (not needed for child indexers where the parent calls newHead)
func (c *ChainIndexer) ChainEventLoop(currentHeader *types.Header, eventMux *event.TypeMux) {
    sub := eventMux.Subscribe(ChainEvent{})
    c.newHead(currentHeader.Number.Uint64(), false)
    lastHead := currentHeader.Hash()
    for {
        select {
        case <-c.stop:
            return
        case ev := <-sub.Chan():
            header := ev.Data.(ChainEvent).Block.Header()
            c.newHead(header.Number.Uint64(), header.ParentHash != lastHead)
            lastHead = header.Hash()
        }
    }
}

// AddChildIndexer adds a child ChainIndexer that can use the output of this one
func (c *ChainIndexer) AddChildIndexer(ci *ChainIndexer) {
    c.children = append(c.children, ci)
}

// newHead notifies the indexer about new chain heads or rollbacks
func (c *ChainIndexer) newHead(headNum uint64, rollback bool) {
    c.lock.Lock()
    defer c.lock.Unlock()

    if rollback {
        firstChanged := headNum / c.sectionSize
        if firstChanged < c.targetCount {
            c.targetCount = firstChanged
        }
        if firstChanged < c.stored {
            c.stored = firstChanged
            c.setValidSections(c.stored)
        }
        headNum = firstChanged * c.sectionSize

        if headNum < c.lastForwarded {
            c.lastForwarded = headNum
            for _, cp := range c.children {
                cp.newHead(c.lastForwarded, true)
            }
        }

    } else {
        var newCount uint64
        if headNum >= c.confirmReq {
            newCount = (headNum + 1 - c.confirmReq) / c.sectionSize
            if newCount > c.targetCount {
                c.targetCount = newCount
                if !c.updating {
                    c.updating = true
                    c.tryUpdate <- struct{}{}
                }
            }
        }
    }
}

// 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 ok == false.
func (c *ChainIndexer) processSection(section uint64, lastSectionHead common.Hash) (sectionHead common.Hash, ok bool) {
    c.backend.Reset(section)

    head := lastSectionHead
    for i := section * c.sectionSize; i < (section+1)*c.sectionSize; i++ {
        hash := GetCanonicalHash(c.chainDb, i)
        if hash == (common.Hash{}) {
            return common.Hash{}, false
        }
        header := GetHeader(c.chainDb, hash, i)
        if header == nil || header.ParentHash != head {
            return common.Hash{}, false
        }
        c.backend.Process(header)
        head = header.Hash()
    }
    if err := c.backend.Commit(c.chainDb); err != nil {
        return common.Hash{}, false
    }
    return head, true
}

// CanonicalSections returns the number of processed sections that are consistent with
// the current canonical chain
func (c *ChainIndexer) CanonicalSections() uint64 {
    c.lock.Lock()
    defer c.lock.Unlock()

    cnt := c.getValidSections()
    for cnt > 0 {
        if c.getSectionHead(cnt-1) == GetCanonicalHash(c.chainDb, cnt*c.sectionSize-1) {
            break
        }
        cnt--
        c.setValidSections(cnt)
    }
    return cnt
}

// getValidSections reads the number of valid sections from the index database
func (c *ChainIndexer) getValidSections() uint64 {
    data, _ := c.indexDb.Get([]byte("count"))
    if len(data) == 8 {
        return binary.BigEndian.Uint64(data[:])
    }
    return 0
}

// setValidSections writes the number of valid sections to the index database
func (c *ChainIndexer) setValidSections(cnt uint64) {
    oldCnt := c.getValidSections()
    if cnt < oldCnt {
        for i := cnt; i < oldCnt; i++ {
            c.removeSectionHead(i)
        }
    }

    var data [8]byte
    binary.BigEndian.PutUint64(data[:], cnt)
    c.indexDb.Put([]byte("count"), data[:])
}

// getSectionHead reads the last block hash of a processed section from the index database
func (c *ChainIndexer) getSectionHead(idx uint64) common.Hash {
    var data [8]byte
    binary.BigEndian.PutUint64(data[:], idx)

    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(idx uint64, shead common.Hash) {
    var data [8]byte
    binary.BigEndian.PutUint64(data[:], idx)

    c.indexDb.Put(append([]byte("shead"), data[:]...), shead.Bytes())
}

// removeSectionHead removes the reference to a processed section from the index database
func (c *ChainIndexer) removeSectionHead(idx uint64) {
    var data [8]byte
    binary.BigEndian.PutUint64(data[:], idx)

    c.indexDb.Delete(append([]byte("shead"), data[:]...))
}