core: implement ChainIndexer

1 files changed, 294 insertions, 0 deletions

diff --git a/core/chain_indexer.go b/core/chain_indexer.go
new file mode 100644
index 000000000..f1ead526d
--- /dev/null
+++ b/core/chain_indexer.go
@@ -0,0 +1,294 @@
+// 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[:]...))
+}