state_sync.go

// Copyright 2015 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 state

import (
	"bytes"
	"math/big"

	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/crypto/sha3"
	"github.com/ethereum/go-ethereum/ethdb"
	"github.com/ethereum/go-ethereum/logger/glog"
	"github.com/ethereum/go-ethereum/rlp"
	"github.com/ethereum/go-ethereum/trie"
)

type StateSync struct {
	db          ethdb.Database
	sync        *trie.TrieSync
	codeReqs    map[common.Hash]struct{} // requested but not yet written to database
	codeReqList []common.Hash            // requested since last GetHashes
}

var sha3_nil = common.BytesToHash(sha3.NewKeccak256().Sum(nil))

func NewStateSync(root common.Hash, db ethdb.Database) *StateSync {
	ss := &StateSync{
		db:       db,
		codeReqs: make(map[common.Hash]struct{}),
	}
	ss.codeReqs[sha3_nil] = struct{}{} // never request the nil hash
	ss.sync = trie.NewTrieSync(root, db, ss.KeyValueCallBack)
	return ss
}

func (self *StateSync) KeyValueCallBack(key, value []byte, leafnode common.Hash) {
	var obj struct {
		Nonce    uint64
		Balance  *big.Int
		Root     common.Hash
		CodeHash []byte
	}
	err := rlp.Decode(bytes.NewReader(value), &obj)
	if err != nil {
		glog.Errorf("can't decode state object %x: %v", key, err)
		return
	}
	self.sync.AddRoot(obj.Root, 64, leafnode, nil)

	codehash := common.BytesToHash(obj.CodeHash)
	if _, ok := self.codeReqs[codehash]; !ok {
		code, _ := self.db.Get(obj.CodeHash)
		if code == nil {
			self.codeReqs[codehash] = struct{}{}
			self.codeReqList = append(self.codeReqList, codehash)
		}
	}
}

func (self *StateSync) GetHashes(max int) []common.Hash {
	cr := len(self.codeReqList)
	gh := 0
	if max != 0 {
		if cr > max {
			cr = max
		}
		gh = max - cr
	}
	list := append(self.sync.GetHashes(gh), self.codeReqList[:cr]...)
	self.codeReqList = self.codeReqList[cr:]
	return list
}

func (self *StateSync) ProcessSyncData(list []trie.RawSyncData) {
	for i := 0; i < len(list); i++ {
		if _, ok := self.codeReqs[list[i].Hash]; ok { // code data, not a node
			self.db.Put(list[i].Hash[:], list[i].Data)
			delete(self.codeReqs, list[i].Hash)
			list[i] = list[len(list)-1]
			list = list[:len(list)-1]
			i--
		}
	}
	self.sync.ProcessSyncData(list)
}

trie_sync.go

// Copyright 2015 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 trie

import "github.com/ethereum/go-ethereum/common"

type requestQueue struct {
	req  []common.Hash
	sent int
}

func (self *requestQueue) add(req common.Hash) {
	self.req = append(self.req, req)
}

func (self *requestQueue) get(max int) []common.Hash {
	rem := len(self.req) - self.sent
	cnt := max
	if rem < max || max == 0 {
		cnt = rem
	}
	ret := self.req[self.sent : self.sent+cnt]
	self.sent += cnt
	if self.sent > 100 {
		self.req = self.req[self.sent:]
		self.sent = 0
	}
	return ret
}

type syncTrieId struct {
	trie      *Trie
	kvcb      TrieSyncKeyValueCallBack
	rootDepth int
}

type nodeReq struct {
	nodeRef            *node // hash node is replaced with actual node when received
	hash               common.Hash
	rlp                []byte
	parent             *nodeReq
	prefix             []byte // nibbles
	depth, reqChildCnt int
	id                 *syncTrieId
}

type TrieSyncKeyValueCallBack func(key, value []byte, leafnode common.Hash)

type TrieSync struct {
	db       Database
	rqList   []requestQueue
	maxDepth int
	reqs     map[common.Hash]*nodeReq
}

func NewTrieSync(root common.Hash, db Database, kvcb TrieSyncKeyValueCallBack) *TrieSync {
	ts := &TrieSync{
		db:   db,
		reqs: make(map[common.Hash]*nodeReq),
	}
	ts.AddRoot(root, 0, common.Hash{}, kvcb)
	return ts
}

func (self *TrieSync) AddRoot(root common.Hash, depth int, parent common.Hash, kvcb TrieSyncKeyValueCallBack) {
	// Short circuit if the trie is empty
	if root == emptyRoot {
		return
	}
	id := &syncTrieId{
		trie: &Trie{
			db:   self.db,
			root: hashNode(root.Bytes()),
		},
		kvcb:      kvcb,
		rootDepth: depth,
	}
	req := &nodeReq{
		nodeRef: &id.trie.root,
		hash:    root,
		id:      id,
	}
	pr, ok := self.reqs[parent]
	if ok {
		req.parent = pr
		pr.reqChildCnt++
	}
	self.addReq(req)
}

func (self *TrieSync) addReq(req *nodeReq) {
	for req.depth >= len(self.rqList) {
		self.rqList = append(self.rqList, requestQueue{})
	}
	self.rqList[req.depth].add(req.hash)
	if self.maxDepth < req.depth {
		self.maxDepth = req.depth
	}
	self.reqs[req.hash] = req
}

func (self *TrieSync) GetHashes(max int) []common.Hash {
	res := []common.Hash{}
	for self.rqList != nil && self.maxDepth >= 0 && (max == 0 || len(res) < max) {
		mm := 0
		if max != 0 {
			mm = max - len(res)
		}
		r := self.rqList[self.maxDepth].get(mm)
		if len(r) == 0 {
			self.maxDepth--
		} else {
			res = append(res, r...)
		}
	}
	return res
}

type RawSyncData struct {
	Hash common.Hash
	Data []byte
}

func (self *TrieSync) ProcessSyncData(nodes []RawSyncData) {
	for _, nd := range nodes {
		nr := self.reqs[nd.Hash]
		if nr == nil {
			continue
		}
		node := mustDecodeNode(nd.Hash[:], nd.Data)
		*nr.nodeRef = node
		nr.rlp = nd.Data
		crs := self.createChildReqs(nr)
		nr.reqChildCnt += len(crs)
		if nr.reqChildCnt == 0 {
			self.subTreeFinished(nr)
		}
		for _, cr := range crs {
			self.addReq(cr)
		}
	}
}

func (self *TrieSync) createChildReqs(nr *nodeReq) []*nodeReq {
	tn := *nr.nodeRef
	var cnodes [](*node)
	var cdepth int
	var nibbles [][]byte
	switch n := tn.(type) {
	case shortNode:
		cnodes = [](*node){&n.Val}
		cdepth = nr.depth + len(n.Key)
		nibbles = [][]byte{n.Key}
	case fullNode:
		for i := 0; i < 17; i++ {
			if n[i] != nil {
				cnodes = append(cnodes, &n[i])
				nibbles = append(nibbles, []byte{byte(i)})
			}
		}
		cdepth = nr.depth + 1
	default:
		panic(nil)
	}

	var reqs []*nodeReq
	for i, cn := range cnodes {
		prefix := append(nr.prefix, nibbles[i]...)
		if nr.id.kvcb != nil {
			value, isvalue := (*cn).(valueNode)
			if isvalue {
				nr.id.kvcb(prefix, value, nr.hash)
			}
		}
		hash, ishash := (*cn).(hashNode)
		if ishash {
			blob, _ := nr.id.trie.db.Get(hash)
			if nn, err := decodeNode(blob); nn != nil && err == nil {
				ishash = false
				*cn = nn
			}
		}
		if ishash && (self.reqs[common.BytesToHash(hash)] == nil) { // node not found in db, subtree is missing or incomplete
			req := &nodeReq{
				nodeRef: cn,
				hash:    common.BytesToHash(hash),
				parent:  nr,
				depth:   cdepth,
				prefix:  prefix,
				id:      nr.id,
			}
			reqs = append(reqs, req)
		}
	}
	return reqs
}

func (self *TrieSync) subTreeFinished(nr *nodeReq) {
	// write node to disk
	self.db.Put(nr.hash[:], nr.rlp)
	// check if parent is finished too
	delete(self.reqs, nr.hash)
	pr := nr.parent
	if pr != nil {
		pr.reqChildCnt--
		if pr.reqChildCnt == 0 {
			self.subTreeFinished(pr)
		}
	}
}