Merge pull request #311 from lyh169/kvmvcc_mavl

plugin/store/init/init.go

@@ -3,5 +3,6 @@ package init
 import (
 	_ "github.com/33cn/plugin/plugin/store/kvdb"       //auto gen
 	_ "github.com/33cn/plugin/plugin/store/kvmvcc"     //auto gen
+	_ "github.com/33cn/plugin/plugin/store/kvmvccmavl" //auto gen
 	_ "github.com/33cn/plugin/plugin/store/mpt"        //auto gen
 )

plugin/store/kvmvccmavl/kvmvcc_mavl.go

+// Copyright Fuzamei Corp. 2018 All Rights Reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package kvmvccmavl kvmvcc+mavl接口
+package kvmvccmavl
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+
+	"time"
+
+	dbm "github.com/33cn/chain33/common/db"
+	clog "github.com/33cn/chain33/common/log"
+	log "github.com/33cn/chain33/common/log/log15"
+	"github.com/33cn/chain33/queue"
+	drivers "github.com/33cn/chain33/system/store"
+	"github.com/33cn/chain33/types"
+	"github.com/hashicorp/golang-lru"
+)
+
+var (
+	kmlog = log.New("module", "kvmvccMavl")
+	// ErrStateHashLost ...
+	ErrStateHashLost        = errors.New("ErrStateHashLost")
+	kvmvccMavlFork    int64 = 200 * 10000
+	isDelMavlData           = false
+	delMavlDataHeight       = kvmvccMavlFork + 10000
+	delMavlDataState  int32
+	wg                sync.WaitGroup
+	quit              bool
+)
+
+const (
+	cacheSize         = 2048 //可以缓存2048个roothash, height对
+	batchDataSize     = 1024 * 1024 * 1
+	delMavlStateStart = 1
+	delMavlStateEnd   = 0
+)
+
+// SetLogLevel set log level
+func SetLogLevel(level string) {
+	clog.SetLogLevel(level)
+}
+
+// DisableLog disable log output
+func DisableLog() {
+	kmlog.SetHandler(log.DiscardHandler())
+}
+
+func init() {
+	drivers.Reg("kvmvccmavl", New)
+}
+
+// KVmMavlStore provide kvmvcc and mavl store interface implementation
+type KVmMavlStore struct {
+	*drivers.BaseStore
+	*KVMVCCStore
+	*MavlStore
+	cache *lru.Cache
+}
+
+type subKVMVCCConfig struct {
+	EnableMVCCIter  bool  `json:"enableMVCCIter"`
+	EnableMavlPrune bool  `json:"enableMavlPrune"`
+	PruneHeight     int32 `json:"pruneHeight"`
+}
+
+type subMavlConfig struct {
+	EnableMavlPrefix bool  `json:"enableMavlPrefix"`
+	EnableMVCC       bool  `json:"enableMVCC"`
+	EnableMavlPrune  bool  `json:"enableMavlPrune"`
+	PruneHeight      int32 `json:"pruneHeight"`
+}
+
+type subConfig struct {
+	EnableMVCCIter   bool  `json:"enableMVCCIter"`
+	EnableMavlPrefix bool  `json:"enableMavlPrefix"`
+	EnableMVCC       bool  `json:"enableMVCC"`
+	EnableMavlPrune  bool  `json:"enableMavlPrune"`
+	PruneHeight      int32 `json:"pruneHeight"`
+}
+
+// New construct KVMVCCStore module
+func New(cfg *types.Store, sub []byte) queue.Module {
+	bs := drivers.NewBaseStore(cfg)
+	var kvms *KVmMavlStore
+	var subcfg subConfig
+	var subKVMVCCcfg subKVMVCCConfig
+	var subMavlcfg subMavlConfig
+	if sub != nil {
+		types.MustDecode(sub, &subcfg)
+		subKVMVCCcfg.EnableMVCCIter = subcfg.EnableMVCCIter
+		subKVMVCCcfg.EnableMavlPrune = subcfg.EnableMavlPrune
+		subKVMVCCcfg.PruneHeight = subcfg.PruneHeight
+
+		subMavlcfg.EnableMavlPrefix = subcfg.EnableMavlPrefix
+		subMavlcfg.EnableMVCC = subcfg.EnableMVCC
+		subMavlcfg.EnableMavlPrune = subcfg.EnableMavlPrune
+		subMavlcfg.PruneHeight = subcfg.PruneHeight
+	}
+	cache, err := lru.New(cacheSize)
+	if err != nil {
+		panic("new KVmMavlStore fail")
+	}
+
+	kvms = &KVmMavlStore{bs, NewKVMVCC(&subKVMVCCcfg, bs.GetDB()),
+		NewMavl(&subMavlcfg, bs.GetDB()), cache}
+	// 查询是否已经删除mavl
+	_, err = bs.GetDB().Get(genDelMavlKey(mvccPrefix))
+	if err == nil {
+		isDelMavlData = true
+	}
+	bs.SetChild(kvms)
+	return kvms
+}
+
+// Close the KVmMavlStore module
+func (kvmMavls *KVmMavlStore) Close() {
+	quit = true
+	wg.Wait()
+	kvmMavls.KVMVCCStore.Close()
+	kvmMavls.MavlStore.Close()
+	kvmMavls.BaseStore.Close()
+	kmlog.Info("store kvmMavls closed")
+}
+
+// Set kvs with statehash to KVmMavlStore
+func (kvmMavls *KVmMavlStore) Set(datas *types.StoreSet, sync bool) ([]byte, error) {
+	if datas.Height < kvmvccMavlFork {
+		hash, err := kvmMavls.MavlStore.Set(datas, sync)
+		if err != nil {
+			return hash, err
+		}
+		_, err = kvmMavls.KVMVCCStore.Set(datas, hash, sync)
+		if err != nil {
+			return hash, err
+		}
+		if err == nil {
+			kvmMavls.cache.Add(string(hash), datas.Height)
+		}
+		return hash, err
+	}
+	// 仅仅做kvmvcc
+	hash, err := kvmMavls.KVMVCCStore.Set(datas, nil, sync)
+	if err == nil {
+		kvmMavls.cache.Add(string(hash), datas.Height)
+	}
+	// 删除Mavl数据
+	if datas.Height > delMavlDataHeight && !isDelMavlData && !isDelMavling() {
+		wg.Add(1)
+		go DelMavl(kvmMavls.GetDB())
+	}
+	return hash, err
+}
+
+// Get kvs with statehash from KVmMavlStore
+func (kvmMavls *KVmMavlStore) Get(datas *types.StoreGet) [][]byte {
+	return kvmMavls.KVMVCCStore.Get(datas)
+}
+
+// MemSet set kvs to the mem of KVmMavlStore module and return the StateHash
+func (kvmMavls *KVmMavlStore) MemSet(datas *types.StoreSet, sync bool) ([]byte, error) {
+	if datas.Height < kvmvccMavlFork {
+		hash, err := kvmMavls.MavlStore.MemSet(datas, sync)
+		if err != nil {
+			return hash, err
+		}
+		_, err = kvmMavls.KVMVCCStore.MemSet(datas, hash, sync)
+		if err != nil {
+			return hash, err
+		}
+		if err == nil {
+			kvmMavls.cache.Add(string(hash), datas.Height)
+		}
+		return hash, err
+	}
+	// 仅仅做kvmvcc
+	hash, err := kvmMavls.KVMVCCStore.MemSet(datas, nil, sync)
+	if err == nil {
+		kvmMavls.cache.Add(string(hash), datas.Height)
+	}
+	// 删除Mavl数据
+	if datas.Height > delMavlDataHeight && !isDelMavlData && !isDelMavling() {
+		wg.Add(1)
+		go DelMavl(kvmMavls.GetDB())
+	}
+	return hash, err
+}
+
+// Commit kvs in the mem of KVmMavlStore module to state db and return the StateHash
+func (kvmMavls *KVmMavlStore) Commit(req *types.ReqHash) ([]byte, error) {
+	if value, ok := kvmMavls.cache.Get(string(req.Hash)); ok {
+		if value.(int64) < kvmvccMavlFork {
+			hash, err := kvmMavls.MavlStore.Commit(req)
+			if err != nil {
+				return hash, err
+			}
+			_, err = kvmMavls.KVMVCCStore.Commit(req)
+			return hash, err
+		}
+		return kvmMavls.KVMVCCStore.Commit(req)
+	}
+	return kvmMavls.KVMVCCStore.Commit(req)
+}
+
+// Rollback kvs in the mem of KVmMavlStore module and return the StateHash
+func (kvmMavls *KVmMavlStore) Rollback(req *types.ReqHash) ([]byte, error) {
+	if value, ok := kvmMavls.cache.Get(string(req.Hash)); ok {
+		if value.(int64) < kvmvccMavlFork {
+			hash, err := kvmMavls.MavlStore.Rollback(req)
+			if err != nil {
+				return hash, err
+			}
+			_, err = kvmMavls.KVMVCCStore.Rollback(req)
+			return hash, err
+		}
+		return kvmMavls.KVMVCCStore.Rollback(req)
+	}
+	return kvmMavls.KVMVCCStore.Rollback(req)
+}
+
+// IterateRangeByStateHash travel with Prefix by StateHash  to get the latest version kvs.
+func (kvmMavls *KVmMavlStore) IterateRangeByStateHash(statehash []byte, start []byte, end []byte, ascending bool, fn func(key, value []byte) bool) {
+	if value, ok := kvmMavls.cache.Get(string(statehash)); ok {
+		if value.(int64) < kvmvccMavlFork {
+			kvmMavls.MavlStore.IterateRangeByStateHash(statehash, start, end, ascending, fn)
+			return
+		}
+		kvmMavls.KVMVCCStore.IterateRangeByStateHash(statehash, start, end, ascending, fn)
+		return
+	}
+	kvmMavls.KVMVCCStore.IterateRangeByStateHash(statehash, start, end, ascending, fn)
+}
+
+// ProcEvent handles supported events
+func (kvmMavls *KVmMavlStore) ProcEvent(msg *queue.Message) {
+	msg.ReplyErr("KVmMavlStore", types.ErrActionNotSupport)
+}
+
+// Del set kvs to nil with StateHash
+func (kvmMavls *KVmMavlStore) Del(req *types.StoreDel) ([]byte, error) {
+	if req.Height < kvmvccMavlFork {
+		hash, err := kvmMavls.MavlStore.Del(req)
+		if err != nil {
+			return hash, err
+		}
+		_, err = kvmMavls.KVMVCCStore.Del(req)
+		if err != nil {
+			return hash, err
+		}
+		if err == nil {
+			kvmMavls.cache.Remove(string(req.StateHash))
+		}
+		return hash, err
+	}
+	// 仅仅做kvmvcc
+	hash, err := kvmMavls.KVMVCCStore.Del(req)
+	if err == nil {
+		kvmMavls.cache.Remove(string(req.StateHash))
+	}
+	return hash, err
+}
+
+// DelMavl 数据库中mavl数据清除
+// 达到kvmvccMavlFork + 100000 后触发清除
+func DelMavl(db dbm.DB) {
+	defer wg.Done()
+	setDelMavl(delMavlStateStart)
+	defer setDelMavl(delMavlStateEnd)
+	isDel := delMavlData(db)
+	if isDel {
+		isDelMavlData = true
+		kmlog.Info("DelMavl success")
+	}
+}
+
+func delMavlData(db dbm.DB) bool {
+	it := db.Iterator(nil, nil, true)
+	defer it.Close()
+	batch := db.NewBatch(true)
+	for it.Rewind(); it.Valid(); it.Next() {
+		if quit {
+			return false
+		}
+		if !bytes.HasPrefix(it.Key(), mvccPrefix) { // 将非mvcc的mavl数据全部删除
+			batch.Delete(it.Key())
+			if batch.ValueSize() > batchDataSize {
+				batch.Write()
+				batch.Reset()
+				time.Sleep(time.Millisecond * 100)
+			}
+		}
+	}
+	batch.Set(genDelMavlKey(mvccPrefix), []byte(""))
+	batch.Write()
+	return true
+}
+
+func genDelMavlKey(prefix []byte) []byte {
+	delMavl := "--delMavlData--"
+	return []byte(fmt.Sprintf("%s%s", string(prefix), delMavl))
+}
+
+func isDelMavling() bool {
+	return atomic.LoadInt32(&delMavlDataState) == 1
+}
+
+func setDelMavl(state int32) {
+	atomic.StoreInt32(&delMavlDataState, state)
+}

plugin/store/kvmvccmavl/mavl.go

+// Copyright Fuzamei Corp. 2018 All Rights Reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package kvmvccmavl
+
+import (
+	"sync"
+
+	"github.com/33cn/chain33/common"
+	dbm "github.com/33cn/chain33/common/db"
+	"github.com/33cn/chain33/queue"
+	"github.com/33cn/chain33/system/store/mavl/db"
+	"github.com/33cn/chain33/types"
+)
+
+// MavlStore mavl store struct
+type MavlStore struct {
+	db               dbm.DB
+	trees            *sync.Map
+	enableMavlPrefix bool
+	enableMVCC       bool
+	enableMavlPrune  bool
+	pruneHeight      int32
+}
+
+// NewMavl new mavl store module
+func NewMavl(sub *subMavlConfig, db dbm.DB) *MavlStore {
+	var subcfg subMavlConfig
+	if sub != nil {
+		subcfg.EnableMavlPrefix = sub.EnableMavlPrefix
+		subcfg.EnableMVCC = sub.EnableMVCC
+		subcfg.EnableMavlPrune = sub.EnableMavlPrune
+		subcfg.PruneHeight = sub.PruneHeight
+	}
+	mavls := &MavlStore{db, &sync.Map{}, subcfg.EnableMavlPrefix, subcfg.EnableMVCC, subcfg.EnableMavlPrune, subcfg.PruneHeight}
+	mavl.EnableMavlPrefix(subcfg.EnableMavlPrefix)
+	mavl.EnableMVCC(subcfg.EnableMVCC)
+	mavl.EnablePrune(subcfg.EnableMavlPrune)
+	mavl.SetPruneHeight(int(subcfg.PruneHeight))
+	return mavls
+}
+
+// Close close mavl store
+func (mavls *MavlStore) Close() {
+	mavl.ClosePrune()
+	kmlog.Info("store mavl closed")
+}
+
+// Set set k v to mavl store db; sync is true represent write sync
+func (mavls *MavlStore) Set(datas *types.StoreSet, sync bool) ([]byte, error) {
+	return mavl.SetKVPair(mavls.db, datas, sync)
+}
+
+// Get get values by keys
+func (mavls *MavlStore) Get(datas *types.StoreGet) [][]byte {
+	var tree *mavl.Tree
+	var err error
+	values := make([][]byte, len(datas.Keys))
+	search := string(datas.StateHash)
+	if data, ok := mavls.trees.Load(search); ok {
+		tree = data.(*mavl.Tree)
+	} else {
+		tree = mavl.NewTree(mavls.db, true)
+		//get接口也应该传入高度
+		//tree.SetBlockHeight(datas.Height)
+		err = tree.Load(datas.StateHash)
+		kmlog.Debug("store mavl get tree", "err", err, "StateHash", common.ToHex(datas.StateHash))
+	}
+	if err == nil {
+		for i := 0; i < len(datas.Keys); i++ {
+			_, value, exit := tree.Get(datas.Keys[i])
+			if exit {
+				values[i] = value
+			}
+		}
+	}
+	return values
+}
+
+// MemSet set keys values to memcory mavl, return root hash and error
+func (mavls *MavlStore) MemSet(datas *types.StoreSet, sync bool) ([]byte, error) {
+	beg := types.Now()
+	defer func() {
+		kmlog.Info("MemSet", "cost", types.Since(beg))
+	}()
+	if len(datas.KV) == 0 {
+		kmlog.Info("store mavl memset,use preStateHash as stateHash for kvset is null")
+		mavls.trees.Store(string(datas.StateHash), nil)
+		return datas.StateHash, nil
+	}
+	tree := mavl.NewTree(mavls.db, sync)
+	tree.SetBlockHeight(datas.Height)
+	err := tree.Load(datas.StateHash)
+	if err != nil {
+		return nil, err
+	}
+	for i := 0; i < len(datas.KV); i++ {
+		tree.Set(datas.KV[i].Key, datas.KV[i].Value)
+	}
+	hash := tree.Hash()
+	mavls.trees.Store(string(hash), tree)
+	return hash, nil
+}
+
+// Commit convert memcory mavl to storage db
+func (mavls *MavlStore) Commit(req *types.ReqHash) ([]byte, error) {
+	beg := types.Now()
+	defer func() {
+		kmlog.Info("Commit", "cost", types.Since(beg))
+	}()
+	tree, ok := mavls.trees.Load(string(req.Hash))
+	if !ok {
+		kmlog.Error("store mavl commit", "err", types.ErrHashNotFound)
+		return nil, types.ErrHashNotFound
+	}
+	if tree == nil {
+		kmlog.Info("store mavl commit,do nothing for kvset is null")
+		mavls.trees.Delete(string(req.Hash))
+		return req.Hash, nil
+	}
+	hash := tree.(*mavl.Tree).Save()
+	if hash == nil {
+		kmlog.Error("store mavl commit", "err", types.ErrHashNotFound)
+		return nil, types.ErrDataBaseDamage
+	}
+	mavls.trees.Delete(string(req.Hash))
+	return req.Hash, nil
+}
+
+// Rollback 回退将缓存的mavl树删除掉
+func (mavls *MavlStore) Rollback(req *types.ReqHash) ([]byte, error) {
+	beg := types.Now()
+	defer func() {
+		kmlog.Info("Rollback", "cost", types.Since(beg))
+	}()
+	_, ok := mavls.trees.Load(string(req.Hash))
+	if !ok {
+		kmlog.Error("store mavl rollback", "err", types.ErrHashNotFound)
+		return nil, types.ErrHashNotFound
+	}
+	mavls.trees.Delete(string(req.Hash))
+	return req.Hash, nil
+}
+
+// IterateRangeByStateHash 迭代实现功能； statehash：当前状态hash, start：开始查找的key, end: 结束的key, ascending：升序，降序, fn 迭代回调函数
+func (mavls *MavlStore) IterateRangeByStateHash(statehash []byte, start []byte, end []byte, ascending bool, fn func(key, value []byte) bool) {
+	mavl.IterateRangeByStateHash(mavls.db, statehash, start, end, ascending, fn)
+}
+
+// ProcEvent not support message
+func (mavls *MavlStore) ProcEvent(msg queue.Message) {
+	msg.ReplyErr("Store", types.ErrActionNotSupport)
+}
+
+// Del ...
+func (mavls *MavlStore) Del(req *types.StoreDel) ([]byte, error) {
+	//not support
+	return nil, nil
+}