e17d8ddbeb
Added PutBlockReceipts; storing receipts by blocks. Eventually this will require pruning during some cleanup cycle. During forks the receipts by block are used to get the new canonical receipts and transactions. This PR fixes #1473 by rewriting transactions and receipts from the point of where the fork occured.
835 lines
22 KiB
Go
835 lines
22 KiB
Go
// Copyright 2014 The go-ethereum Authors
|
|
// This file is part of go-ethereum.
|
|
//
|
|
// go-ethereum 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.
|
|
//
|
|
// go-ethereum 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 go-ethereum. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
// Package core implements the Ethereum consensus protocol.
|
|
package core
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"math/big"
|
|
"runtime"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/common"
|
|
"github.com/ethereum/go-ethereum/core/state"
|
|
"github.com/ethereum/go-ethereum/core/types"
|
|
"github.com/ethereum/go-ethereum/event"
|
|
"github.com/ethereum/go-ethereum/logger"
|
|
"github.com/ethereum/go-ethereum/logger/glog"
|
|
"github.com/ethereum/go-ethereum/metrics"
|
|
"github.com/ethereum/go-ethereum/pow"
|
|
"github.com/hashicorp/golang-lru"
|
|
)
|
|
|
|
var (
|
|
chainlogger = logger.NewLogger("CHAIN")
|
|
jsonlogger = logger.NewJsonLogger()
|
|
|
|
blockInsertTimer = metrics.NewTimer("chain/inserts")
|
|
|
|
ErrNoGenesis = errors.New("Genesis not found in chain")
|
|
)
|
|
|
|
const (
|
|
blockCacheLimit = 256
|
|
maxFutureBlocks = 256
|
|
maxTimeFutureBlocks = 30
|
|
checkpointLimit = 200
|
|
)
|
|
|
|
type ChainManager struct {
|
|
//eth EthManager
|
|
blockDb common.Database
|
|
stateDb common.Database
|
|
extraDb common.Database
|
|
processor types.BlockProcessor
|
|
eventMux *event.TypeMux
|
|
genesisBlock *types.Block
|
|
// Last known total difficulty
|
|
mu sync.RWMutex
|
|
chainmu sync.RWMutex
|
|
tsmu sync.RWMutex
|
|
|
|
checkpoint int // checkpoint counts towards the new checkpoint
|
|
td *big.Int
|
|
currentBlock *types.Block
|
|
lastBlockHash common.Hash
|
|
currentGasLimit *big.Int
|
|
|
|
transState *state.StateDB
|
|
txState *state.ManagedState
|
|
|
|
cache *lru.Cache // cache is the LRU caching
|
|
futureBlocks *lru.Cache // future blocks are blocks added for later processing
|
|
|
|
quit chan struct{}
|
|
// procInterrupt must be atomically called
|
|
procInterrupt int32 // interrupt signaler for block processing
|
|
wg sync.WaitGroup
|
|
|
|
pow pow.PoW
|
|
}
|
|
|
|
func NewChainManager(blockDb, stateDb, extraDb common.Database, pow pow.PoW, mux *event.TypeMux) (*ChainManager, error) {
|
|
cache, _ := lru.New(blockCacheLimit)
|
|
bc := &ChainManager{
|
|
blockDb: blockDb,
|
|
stateDb: stateDb,
|
|
extraDb: extraDb,
|
|
eventMux: mux,
|
|
quit: make(chan struct{}),
|
|
cache: cache,
|
|
pow: pow,
|
|
}
|
|
|
|
bc.genesisBlock = bc.GetBlockByNumber(0)
|
|
if bc.genesisBlock == nil {
|
|
// XXX Uncomment me before Frontier
|
|
//return nil, ErrNoGenesis
|
|
genesis, err := WriteTestNetGenesisBlock(bc.stateDb, bc.blockDb, 42)
|
|
if err != nil {
|
|
glog.Fatalln("genisis err", err)
|
|
}
|
|
bc.genesisBlock = genesis
|
|
}
|
|
|
|
if err := bc.setLastState(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Check the current state of the block hashes and make sure that we do not have any of the bad blocks in our chain
|
|
for hash, _ := range BadHashes {
|
|
if block := bc.GetBlock(hash); block != nil {
|
|
glog.V(logger.Error).Infof("Found bad hash. Reorganising chain to state %x\n", block.ParentHash().Bytes()[:4])
|
|
block = bc.GetBlock(block.ParentHash())
|
|
if block == nil {
|
|
glog.Fatal("Unable to complete. Parent block not found. Corrupted DB?")
|
|
}
|
|
bc.SetHead(block)
|
|
|
|
glog.V(logger.Error).Infoln("Chain reorg was successfull. Resuming normal operation")
|
|
}
|
|
}
|
|
|
|
bc.transState = bc.State().Copy()
|
|
// Take ownership of this particular state
|
|
bc.txState = state.ManageState(bc.State().Copy())
|
|
|
|
bc.futureBlocks, _ = lru.New(maxFutureBlocks)
|
|
bc.makeCache()
|
|
|
|
go bc.update()
|
|
|
|
return bc, nil
|
|
}
|
|
|
|
func (bc *ChainManager) SetHead(head *types.Block) {
|
|
bc.mu.Lock()
|
|
defer bc.mu.Unlock()
|
|
|
|
for block := bc.currentBlock; block != nil && block.Hash() != head.Hash(); block = bc.GetBlock(block.ParentHash()) {
|
|
bc.removeBlock(block)
|
|
}
|
|
|
|
bc.cache, _ = lru.New(blockCacheLimit)
|
|
bc.currentBlock = head
|
|
bc.makeCache()
|
|
|
|
statedb := state.New(head.Root(), bc.stateDb)
|
|
bc.txState = state.ManageState(statedb)
|
|
bc.transState = statedb.Copy()
|
|
bc.setTotalDifficulty(head.Td)
|
|
bc.insert(head)
|
|
bc.setLastState()
|
|
}
|
|
|
|
func (self *ChainManager) Td() *big.Int {
|
|
self.mu.RLock()
|
|
defer self.mu.RUnlock()
|
|
|
|
return new(big.Int).Set(self.td)
|
|
}
|
|
|
|
func (self *ChainManager) GasLimit() *big.Int {
|
|
self.mu.RLock()
|
|
defer self.mu.RUnlock()
|
|
|
|
return self.currentBlock.GasLimit()
|
|
}
|
|
|
|
func (self *ChainManager) LastBlockHash() common.Hash {
|
|
self.mu.RLock()
|
|
defer self.mu.RUnlock()
|
|
|
|
return self.lastBlockHash
|
|
}
|
|
|
|
func (self *ChainManager) CurrentBlock() *types.Block {
|
|
self.mu.RLock()
|
|
defer self.mu.RUnlock()
|
|
|
|
return self.currentBlock
|
|
}
|
|
|
|
func (self *ChainManager) Status() (td *big.Int, currentBlock common.Hash, genesisBlock common.Hash) {
|
|
self.mu.RLock()
|
|
defer self.mu.RUnlock()
|
|
|
|
return new(big.Int).Set(self.td), self.currentBlock.Hash(), self.genesisBlock.Hash()
|
|
}
|
|
|
|
func (self *ChainManager) SetProcessor(proc types.BlockProcessor) {
|
|
self.processor = proc
|
|
}
|
|
|
|
func (self *ChainManager) State() *state.StateDB {
|
|
return state.New(self.CurrentBlock().Root(), self.stateDb)
|
|
}
|
|
|
|
func (self *ChainManager) TransState() *state.StateDB {
|
|
self.tsmu.RLock()
|
|
defer self.tsmu.RUnlock()
|
|
|
|
return self.transState
|
|
}
|
|
|
|
func (self *ChainManager) setTransState(statedb *state.StateDB) {
|
|
self.transState = statedb
|
|
}
|
|
|
|
func (bc *ChainManager) recover() bool {
|
|
data, _ := bc.blockDb.Get([]byte("checkpoint"))
|
|
if len(data) != 0 {
|
|
block := bc.GetBlock(common.BytesToHash(data))
|
|
if block != nil {
|
|
err := bc.blockDb.Put([]byte("LastBlock"), block.Hash().Bytes())
|
|
if err != nil {
|
|
glog.Fatalln("db write err:", err)
|
|
}
|
|
|
|
bc.currentBlock = block
|
|
bc.lastBlockHash = block.Hash()
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (bc *ChainManager) setLastState() error {
|
|
data, _ := bc.blockDb.Get([]byte("LastBlock"))
|
|
if len(data) != 0 {
|
|
block := bc.GetBlock(common.BytesToHash(data))
|
|
if block != nil {
|
|
bc.currentBlock = block
|
|
bc.lastBlockHash = block.Hash()
|
|
} else {
|
|
glog.Infof("LastBlock (%x) not found. Recovering...\n", data)
|
|
if bc.recover() {
|
|
glog.Infof("Recover successful")
|
|
} else {
|
|
glog.Fatalf("Recover failed. Please report")
|
|
}
|
|
}
|
|
} else {
|
|
bc.Reset()
|
|
}
|
|
bc.td = bc.currentBlock.Td
|
|
bc.currentGasLimit = CalcGasLimit(bc.currentBlock)
|
|
|
|
if glog.V(logger.Info) {
|
|
glog.Infof("Last block (#%v) %x TD=%v\n", bc.currentBlock.Number(), bc.currentBlock.Hash(), bc.td)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (bc *ChainManager) makeCache() {
|
|
bc.cache, _ = lru.New(blockCacheLimit)
|
|
// load in last `blockCacheLimit` - 1 blocks. Last block is the current.
|
|
bc.cache.Add(bc.genesisBlock.Hash(), bc.genesisBlock)
|
|
for _, block := range bc.GetBlocksFromHash(bc.currentBlock.Hash(), blockCacheLimit) {
|
|
bc.cache.Add(block.Hash(), block)
|
|
}
|
|
}
|
|
|
|
func (bc *ChainManager) Reset() {
|
|
bc.mu.Lock()
|
|
defer bc.mu.Unlock()
|
|
|
|
for block := bc.currentBlock; block != nil; block = bc.GetBlock(block.ParentHash()) {
|
|
bc.removeBlock(block)
|
|
}
|
|
|
|
bc.cache, _ = lru.New(blockCacheLimit)
|
|
|
|
// Prepare the genesis block
|
|
err := WriteBlock(bc.blockDb, bc.genesisBlock)
|
|
if err != nil {
|
|
glog.Fatalln("db err:", err)
|
|
}
|
|
|
|
bc.insert(bc.genesisBlock)
|
|
bc.currentBlock = bc.genesisBlock
|
|
bc.makeCache()
|
|
|
|
bc.setTotalDifficulty(common.Big("0"))
|
|
}
|
|
|
|
func (bc *ChainManager) removeBlock(block *types.Block) {
|
|
bc.blockDb.Delete(append(blockHashPre, block.Hash().Bytes()...))
|
|
}
|
|
|
|
func (bc *ChainManager) ResetWithGenesisBlock(gb *types.Block) {
|
|
bc.mu.Lock()
|
|
defer bc.mu.Unlock()
|
|
|
|
for block := bc.currentBlock; block != nil; block = bc.GetBlock(block.ParentHash()) {
|
|
bc.removeBlock(block)
|
|
}
|
|
|
|
// Prepare the genesis block
|
|
gb.Td = gb.Difficulty()
|
|
bc.genesisBlock = gb
|
|
|
|
err := WriteBlock(bc.blockDb, bc.genesisBlock)
|
|
if err != nil {
|
|
glog.Fatalln("db err:", err)
|
|
}
|
|
|
|
bc.insert(bc.genesisBlock)
|
|
bc.currentBlock = bc.genesisBlock
|
|
bc.makeCache()
|
|
bc.td = gb.Difficulty()
|
|
}
|
|
|
|
// Export writes the active chain to the given writer.
|
|
func (self *ChainManager) Export(w io.Writer) error {
|
|
if err := self.ExportN(w, uint64(0), self.currentBlock.NumberU64()); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ExportN writes a subset of the active chain to the given writer.
|
|
func (self *ChainManager) ExportN(w io.Writer, first uint64, last uint64) error {
|
|
self.mu.RLock()
|
|
defer self.mu.RUnlock()
|
|
|
|
if first > last {
|
|
return fmt.Errorf("export failed: first (%d) is greater than last (%d)", first, last)
|
|
}
|
|
|
|
glog.V(logger.Info).Infof("exporting %d blocks...\n", last-first+1)
|
|
|
|
for nr := first; nr <= last; nr++ {
|
|
block := self.GetBlockByNumber(nr)
|
|
if block == nil {
|
|
return fmt.Errorf("export failed on #%d: not found", nr)
|
|
}
|
|
|
|
if err := block.EncodeRLP(w); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// insert injects a block into the current chain block chain. Note, this function
|
|
// assumes that the `mu` mutex is held!
|
|
func (bc *ChainManager) insert(block *types.Block) {
|
|
err := WriteHead(bc.blockDb, block)
|
|
if err != nil {
|
|
glog.Fatal("db write fail:", err)
|
|
}
|
|
|
|
bc.checkpoint++
|
|
if bc.checkpoint > checkpointLimit {
|
|
err = bc.blockDb.Put([]byte("checkpoint"), block.Hash().Bytes())
|
|
if err != nil {
|
|
glog.Fatal("db write fail:", err)
|
|
}
|
|
|
|
bc.checkpoint = 0
|
|
}
|
|
|
|
bc.currentBlock = block
|
|
bc.lastBlockHash = block.Hash()
|
|
}
|
|
|
|
// Accessors
|
|
func (bc *ChainManager) Genesis() *types.Block {
|
|
return bc.genesisBlock
|
|
}
|
|
|
|
// Block fetching methods
|
|
func (bc *ChainManager) HasBlock(hash common.Hash) bool {
|
|
if bc.cache.Contains(hash) {
|
|
return true
|
|
}
|
|
|
|
data, _ := bc.blockDb.Get(append(blockHashPre, hash[:]...))
|
|
return len(data) != 0
|
|
}
|
|
|
|
func (self *ChainManager) GetBlockHashesFromHash(hash common.Hash, max uint64) (chain []common.Hash) {
|
|
block := self.GetBlock(hash)
|
|
if block == nil {
|
|
return
|
|
}
|
|
// XXX Could be optimised by using a different database which only holds hashes (i.e., linked list)
|
|
for i := uint64(0); i < max; i++ {
|
|
block = self.GetBlock(block.ParentHash())
|
|
if block == nil {
|
|
break
|
|
}
|
|
|
|
chain = append(chain, block.Hash())
|
|
if block.Number().Cmp(common.Big0) <= 0 {
|
|
break
|
|
}
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
func (self *ChainManager) GetBlock(hash common.Hash) *types.Block {
|
|
if block, ok := self.cache.Get(hash); ok {
|
|
return block.(*types.Block)
|
|
}
|
|
|
|
block := GetBlockByHash(self.blockDb, hash)
|
|
if block == nil {
|
|
return nil
|
|
}
|
|
|
|
// Add the block to the cache
|
|
self.cache.Add(hash, (*types.Block)(block))
|
|
|
|
return (*types.Block)(block)
|
|
}
|
|
|
|
func (self *ChainManager) GetBlockByNumber(num uint64) *types.Block {
|
|
self.mu.RLock()
|
|
defer self.mu.RUnlock()
|
|
|
|
return self.getBlockByNumber(num)
|
|
|
|
}
|
|
|
|
// GetBlocksFromHash returns the block corresponding to hash and up to n-1 ancestors.
|
|
func (self *ChainManager) GetBlocksFromHash(hash common.Hash, n int) (blocks []*types.Block) {
|
|
for i := 0; i < n; i++ {
|
|
block := self.GetBlock(hash)
|
|
if block == nil {
|
|
break
|
|
}
|
|
blocks = append(blocks, block)
|
|
hash = block.ParentHash()
|
|
}
|
|
return
|
|
}
|
|
|
|
// non blocking version
|
|
func (self *ChainManager) getBlockByNumber(num uint64) *types.Block {
|
|
return GetBlockByNumber(self.blockDb, num)
|
|
}
|
|
|
|
func (self *ChainManager) GetUnclesInChain(block *types.Block, length int) (uncles []*types.Header) {
|
|
for i := 0; block != nil && i < length; i++ {
|
|
uncles = append(uncles, block.Uncles()...)
|
|
block = self.GetBlock(block.ParentHash())
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
// setTotalDifficulty updates the TD of the chain manager. Note, this function
|
|
// assumes that the `mu` mutex is held!
|
|
func (bc *ChainManager) setTotalDifficulty(td *big.Int) {
|
|
bc.td = new(big.Int).Set(td)
|
|
}
|
|
|
|
func (bc *ChainManager) Stop() {
|
|
close(bc.quit)
|
|
atomic.StoreInt32(&bc.procInterrupt, 1)
|
|
|
|
bc.wg.Wait()
|
|
|
|
glog.V(logger.Info).Infoln("Chain manager stopped")
|
|
}
|
|
|
|
type queueEvent struct {
|
|
queue []interface{}
|
|
canonicalCount int
|
|
sideCount int
|
|
splitCount int
|
|
}
|
|
|
|
func (self *ChainManager) procFutureBlocks() {
|
|
blocks := make([]*types.Block, self.futureBlocks.Len())
|
|
for i, hash := range self.futureBlocks.Keys() {
|
|
block, _ := self.futureBlocks.Get(hash)
|
|
blocks[i] = block.(*types.Block)
|
|
}
|
|
if len(blocks) > 0 {
|
|
types.BlockBy(types.Number).Sort(blocks)
|
|
self.InsertChain(blocks)
|
|
}
|
|
}
|
|
|
|
type writeStatus byte
|
|
|
|
const (
|
|
NonStatTy writeStatus = iota
|
|
CanonStatTy
|
|
SplitStatTy
|
|
SideStatTy
|
|
)
|
|
|
|
// WriteBlock writes the block to the chain (or pending queue)
|
|
func (self *ChainManager) WriteBlock(block *types.Block, queued bool) (status writeStatus, err error) {
|
|
self.wg.Add(1)
|
|
defer self.wg.Done()
|
|
|
|
cblock := self.currentBlock
|
|
// Compare the TD of the last known block in the canonical chain to make sure it's greater.
|
|
// At this point it's possible that a different chain (fork) becomes the new canonical chain.
|
|
if block.Td.Cmp(self.Td()) > 0 {
|
|
// chain fork
|
|
if block.ParentHash() != cblock.Hash() {
|
|
// during split we merge two different chains and create the new canonical chain
|
|
err := self.merge(cblock, block)
|
|
if err != nil {
|
|
return NonStatTy, err
|
|
}
|
|
|
|
status = SplitStatTy
|
|
}
|
|
|
|
self.mu.Lock()
|
|
self.setTotalDifficulty(block.Td)
|
|
self.insert(block)
|
|
self.mu.Unlock()
|
|
|
|
self.setTransState(state.New(block.Root(), self.stateDb))
|
|
self.txState.SetState(state.New(block.Root(), self.stateDb))
|
|
|
|
status = CanonStatTy
|
|
} else {
|
|
status = SideStatTy
|
|
}
|
|
|
|
err = WriteBlock(self.blockDb, block)
|
|
if err != nil {
|
|
glog.Fatalln("db err:", err)
|
|
}
|
|
// Delete from future blocks
|
|
self.futureBlocks.Remove(block.Hash())
|
|
|
|
return
|
|
}
|
|
|
|
// InsertChain will attempt to insert the given chain in to the canonical chain or, otherwise, create a fork. It an error is returned
|
|
// it will return the index number of the failing block as well an error describing what went wrong (for possible errors see core/errors.go).
|
|
func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
|
|
self.wg.Add(1)
|
|
defer self.wg.Done()
|
|
|
|
self.chainmu.Lock()
|
|
defer self.chainmu.Unlock()
|
|
|
|
// A queued approach to delivering events. This is generally
|
|
// faster than direct delivery and requires much less mutex
|
|
// acquiring.
|
|
var (
|
|
queue = make([]interface{}, len(chain))
|
|
queueEvent = queueEvent{queue: queue}
|
|
stats struct{ queued, processed, ignored int }
|
|
tstart = time.Now()
|
|
|
|
nonceDone = make(chan nonceResult, len(chain))
|
|
nonceQuit = make(chan struct{})
|
|
nonceChecked = make([]bool, len(chain))
|
|
)
|
|
|
|
// Start the parallel nonce verifier.
|
|
go verifyNonces(self.pow, chain, nonceQuit, nonceDone)
|
|
defer close(nonceQuit)
|
|
|
|
txcount := 0
|
|
for i, block := range chain {
|
|
if atomic.LoadInt32(&self.procInterrupt) == 1 {
|
|
glog.V(logger.Debug).Infoln("Premature abort during chain processing")
|
|
break
|
|
}
|
|
|
|
bstart := time.Now()
|
|
// Wait for block i's nonce to be verified before processing
|
|
// its state transition.
|
|
for !nonceChecked[i] {
|
|
r := <-nonceDone
|
|
nonceChecked[r.i] = true
|
|
if !r.valid {
|
|
block := chain[r.i]
|
|
return r.i, &BlockNonceErr{Hash: block.Hash(), Number: block.Number(), Nonce: block.Nonce()}
|
|
}
|
|
}
|
|
|
|
if BadHashes[block.Hash()] {
|
|
err := fmt.Errorf("Found known bad hash in chain %x", block.Hash())
|
|
blockErr(block, err)
|
|
return i, err
|
|
}
|
|
|
|
// Setting block.Td regardless of error (known for example) prevents errors down the line
|
|
// in the protocol handler
|
|
block.Td = new(big.Int).Set(CalcTD(block, self.GetBlock(block.ParentHash())))
|
|
|
|
// Call in to the block processor and check for errors. It's likely that if one block fails
|
|
// all others will fail too (unless a known block is returned).
|
|
logs, receipts, err := self.processor.Process(block)
|
|
if err != nil {
|
|
if IsKnownBlockErr(err) {
|
|
stats.ignored++
|
|
continue
|
|
}
|
|
|
|
if err == BlockFutureErr {
|
|
// Allow up to MaxFuture second in the future blocks. If this limit
|
|
// is exceeded the chain is discarded and processed at a later time
|
|
// if given.
|
|
if max := uint64(time.Now().Unix()) + maxTimeFutureBlocks; block.Time() > max {
|
|
return i, fmt.Errorf("%v: BlockFutureErr, %v > %v", BlockFutureErr, block.Time(), max)
|
|
}
|
|
|
|
self.futureBlocks.Add(block.Hash(), block)
|
|
stats.queued++
|
|
continue
|
|
}
|
|
|
|
if IsParentErr(err) && self.futureBlocks.Contains(block.ParentHash()) {
|
|
self.futureBlocks.Add(block.Hash(), block)
|
|
stats.queued++
|
|
continue
|
|
}
|
|
|
|
blockErr(block, err)
|
|
|
|
go ReportBlock(block, err)
|
|
|
|
return i, err
|
|
}
|
|
|
|
txcount += len(block.Transactions())
|
|
|
|
// write the block to the chain and get the status
|
|
status, err := self.WriteBlock(block, true)
|
|
if err != nil {
|
|
return i, err
|
|
}
|
|
switch status {
|
|
case CanonStatTy:
|
|
if glog.V(logger.Debug) {
|
|
glog.Infof("[%v] inserted block #%d (%d TXs %d UNCs) (%x...). Took %v\n", time.Now().UnixNano(), block.Number(), len(block.Transactions()), len(block.Uncles()), block.Hash().Bytes()[0:4], time.Since(bstart))
|
|
}
|
|
queue[i] = ChainEvent{block, block.Hash(), logs}
|
|
queueEvent.canonicalCount++
|
|
|
|
// This puts transactions in a extra db for rpc
|
|
PutTransactions(self.extraDb, block, block.Transactions())
|
|
// store the receipts
|
|
PutReceipts(self.extraDb, receipts)
|
|
case SideStatTy:
|
|
if glog.V(logger.Detail) {
|
|
glog.Infof("inserted forked block #%d (TD=%v) (%d TXs %d UNCs) (%x...). Took %v\n", block.Number(), block.Difficulty(), len(block.Transactions()), len(block.Uncles()), block.Hash().Bytes()[0:4], time.Since(bstart))
|
|
}
|
|
queue[i] = ChainSideEvent{block, logs}
|
|
queueEvent.sideCount++
|
|
case SplitStatTy:
|
|
queue[i] = ChainSplitEvent{block, logs}
|
|
queueEvent.splitCount++
|
|
}
|
|
PutBlockReceipts(self.extraDb, block, receipts)
|
|
|
|
stats.processed++
|
|
}
|
|
|
|
if (stats.queued > 0 || stats.processed > 0 || stats.ignored > 0) && bool(glog.V(logger.Info)) {
|
|
tend := time.Since(tstart)
|
|
start, end := chain[0], chain[len(chain)-1]
|
|
glog.Infof("imported %d block(s) (%d queued %d ignored) including %d txs in %v. #%v [%x / %x]\n", stats.processed, stats.queued, stats.ignored, txcount, tend, end.Number(), start.Hash().Bytes()[:4], end.Hash().Bytes()[:4])
|
|
}
|
|
|
|
go self.eventMux.Post(queueEvent)
|
|
|
|
return 0, nil
|
|
}
|
|
|
|
// diff takes two blocks, an old chain and a new chain and will reconstruct the blocks and inserts them
|
|
// to be part of the new canonical chain.
|
|
func (self *ChainManager) diff(oldBlock, newBlock *types.Block) (types.Blocks, error) {
|
|
var (
|
|
newChain types.Blocks
|
|
commonBlock *types.Block
|
|
oldStart = oldBlock
|
|
newStart = newBlock
|
|
)
|
|
|
|
// first reduce whoever is higher bound
|
|
if oldBlock.NumberU64() > newBlock.NumberU64() {
|
|
// reduce old chain
|
|
for oldBlock = oldBlock; oldBlock != nil && oldBlock.NumberU64() != newBlock.NumberU64(); oldBlock = self.GetBlock(oldBlock.ParentHash()) {
|
|
}
|
|
} else {
|
|
// reduce new chain and append new chain blocks for inserting later on
|
|
for newBlock = newBlock; newBlock != nil && newBlock.NumberU64() != oldBlock.NumberU64(); newBlock = self.GetBlock(newBlock.ParentHash()) {
|
|
newChain = append(newChain, newBlock)
|
|
}
|
|
}
|
|
if oldBlock == nil {
|
|
return nil, fmt.Errorf("Invalid old chain")
|
|
}
|
|
if newBlock == nil {
|
|
return nil, fmt.Errorf("Invalid new chain")
|
|
}
|
|
|
|
numSplit := newBlock.Number()
|
|
for {
|
|
if oldBlock.Hash() == newBlock.Hash() {
|
|
commonBlock = oldBlock
|
|
break
|
|
}
|
|
newChain = append(newChain, newBlock)
|
|
|
|
oldBlock, newBlock = self.GetBlock(oldBlock.ParentHash()), self.GetBlock(newBlock.ParentHash())
|
|
if oldBlock == nil {
|
|
return nil, fmt.Errorf("Invalid old chain")
|
|
}
|
|
if newBlock == nil {
|
|
return nil, fmt.Errorf("Invalid new chain")
|
|
}
|
|
}
|
|
|
|
if glog.V(logger.Debug) {
|
|
commonHash := commonBlock.Hash()
|
|
glog.Infof("Chain split detected @ %x. Reorganising chain from #%v %x to %x", commonHash[:4], numSplit, oldStart.Hash().Bytes()[:4], newStart.Hash().Bytes()[:4])
|
|
}
|
|
|
|
return newChain, nil
|
|
}
|
|
|
|
// merge merges two different chain to the new canonical chain
|
|
func (self *ChainManager) merge(oldBlock, newBlock *types.Block) error {
|
|
newChain, err := self.diff(oldBlock, newBlock)
|
|
if err != nil {
|
|
return fmt.Errorf("chain reorg failed: %v", err)
|
|
}
|
|
|
|
// insert blocks. Order does not matter. Last block will be written in ImportChain itself which creates the new head properly
|
|
self.mu.Lock()
|
|
for _, block := range newChain {
|
|
// insert the block in the canonical way, re-writing history
|
|
self.insert(block)
|
|
// write canonical receipts and transactions
|
|
PutTransactions(self.extraDb, block, block.Transactions())
|
|
PutReceipts(self.extraDb, GetBlockReceipts(self.extraDb, block.Hash()))
|
|
|
|
}
|
|
self.mu.Unlock()
|
|
|
|
return nil
|
|
}
|
|
|
|
func (self *ChainManager) update() {
|
|
events := self.eventMux.Subscribe(queueEvent{})
|
|
futureTimer := time.Tick(5 * time.Second)
|
|
out:
|
|
for {
|
|
select {
|
|
case ev := <-events.Chan():
|
|
switch ev := ev.(type) {
|
|
case queueEvent:
|
|
for _, event := range ev.queue {
|
|
switch event := event.(type) {
|
|
case ChainEvent:
|
|
// We need some control over the mining operation. Acquiring locks and waiting for the miner to create new block takes too long
|
|
// and in most cases isn't even necessary.
|
|
if self.lastBlockHash == event.Hash {
|
|
self.currentGasLimit = CalcGasLimit(event.Block)
|
|
self.eventMux.Post(ChainHeadEvent{event.Block})
|
|
}
|
|
}
|
|
|
|
self.eventMux.Post(event)
|
|
}
|
|
}
|
|
case <-futureTimer:
|
|
self.procFutureBlocks()
|
|
case <-self.quit:
|
|
break out
|
|
}
|
|
}
|
|
}
|
|
|
|
func blockErr(block *types.Block, err error) {
|
|
h := block.Header()
|
|
glog.V(logger.Error).Infof("Bad block #%v (%x)\n", h.Number, h.Hash().Bytes())
|
|
glog.V(logger.Error).Infoln(err)
|
|
glog.V(logger.Debug).Infoln(verifyNonces)
|
|
}
|
|
|
|
type nonceResult struct {
|
|
i int
|
|
valid bool
|
|
}
|
|
|
|
// block verifies nonces of the given blocks in parallel and returns
|
|
// an error if one of the blocks nonce verifications failed.
|
|
func verifyNonces(pow pow.PoW, blocks []*types.Block, quit <-chan struct{}, done chan<- nonceResult) {
|
|
// Spawn a few workers. They listen for blocks on the in channel
|
|
// and send results on done. The workers will exit in the
|
|
// background when in is closed.
|
|
var (
|
|
in = make(chan int)
|
|
nworkers = runtime.GOMAXPROCS(0)
|
|
)
|
|
defer close(in)
|
|
if len(blocks) < nworkers {
|
|
nworkers = len(blocks)
|
|
}
|
|
for i := 0; i < nworkers; i++ {
|
|
go func() {
|
|
for i := range in {
|
|
done <- nonceResult{i: i, valid: pow.Verify(blocks[i])}
|
|
}
|
|
}()
|
|
}
|
|
// Feed block indices to the workers.
|
|
for i := range blocks {
|
|
select {
|
|
case in <- i:
|
|
continue
|
|
case <-quit:
|
|
return
|
|
}
|
|
}
|
|
}
|