Skip to content

P
plugin
Commit e2c6f375 authored by liuyuhang's avatar liuyuhang Committed by vipwzw
Browse files
Options

add empty block handle

parent 05324258
Hide whitespace changes
Inline Side-by-side
Showing with 499 additions and 23 deletions
chain33.para.toml
View file @ e2c6f375
......@@ -148,6 +148,8 @@ enableMemTree=true
enableMemVal=true
# 缓存close ticket数目，该缓存越大同步速度越快，最大设置到1500000
tkCloseCacheLen=100000
# 该参数针对平行链，如果平行链的ForkKvmvccmavl高度不为0,需要开启此功能,开启此功能需要从0开始执行区块
enableEmptyBlockHandle=false
[wallet]
minFee=100000
......
chain33.toml
View file @ e2c6f375
......@@ -180,6 +180,8 @@ enableMemTree=true
enableMemVal=true
# 缓存close ticket数目，该缓存越大同步速度越快，最大设置到1500000
tkCloseCacheLen=100000
# 该参数针对平行链，主链无需开启此功能
enableEmptyBlockHandle=false
[wallet]
minFee=100000
......
plugin/store/kvmvccmavl/kvmvcc_mavl.go
View file @ e2c6f375
......@@ -80,6 +80,7 @@ type subKVMVCCConfig struct {
EnableMVCCIter bool `json:"enableMVCCIter"`
EnableMVCCPrune bool `json:"enableMVCCPrune"`
PruneHeight int32 `json:"pruneHeight"`
EnableEmptyBlockHandle bool `json:"enableEmptyBlockHandle"`
}
type subMavlConfig struct {
......@@ -109,6 +110,8 @@ type subConfig struct {
EnableMemVal bool `json:"enableMemVal"`
// 缓存close ticket数目
TkCloseCacheLen int32 `json:"tkCloseCacheLen"`
// 使能空块处理
EnableEmptyBlockHandle bool `json:"enableEmptyBlockHandle"`
}
// New construct KVMVCCStore module
......@@ -122,6 +125,7 @@ func New(cfg *types.Store, sub []byte, chain33cfg *types.Chain33Config) queue.Mo
subKVMVCCcfg.EnableMVCCIter = subcfg.EnableMVCCIter
subKVMVCCcfg.EnableMVCCPrune = subcfg.EnableMVCCPrune
subKVMVCCcfg.PruneHeight = subcfg.PruneMVCCHeight
subKVMVCCcfg.EnableEmptyBlockHandle = subcfg.EnableEmptyBlockHandle
subMavlcfg.EnableMavlPrefix = subcfg.EnableMavlPrefix
subMavlcfg.EnableMVCC = subcfg.EnableMVCC
......@@ -173,9 +177,16 @@ func (kvmMavls *KVmMavlStore) Set(datas *types.StoreSet, sync bool) ([]byte, err
if err != nil {
return hash, err
}
_, err = kvmMavls.KVMVCCStore.Set(datas, hash, sync)
if err != nil {
return hash, err
if kvmMavls.kvmvccCfg.EnableEmptyBlockHandle {
_, err = kvmMavls.KVMVCCStore.SetRdm(datas, hash, sync)
if err != nil {
return hash, err
}
} else {
_, err = kvmMavls.KVMVCCStore.Set(datas, hash, sync)
if err != nil {
return hash, err
}
}
if err == nil {
kvmMavls.cache.Add(string(hash), datas.Height)
......@@ -183,7 +194,14 @@ func (kvmMavls *KVmMavlStore) Set(datas *types.StoreSet, sync bool) ([]byte, err
return hash, err
}
// 仅仅做kvmvcc
hash, err := kvmMavls.KVMVCCStore.Set(datas, nil, sync)
var hash []byte
var err error
if kvmMavls.kvmvccCfg.EnableEmptyBlockHandle && datas.Height == kvmvccMavlFork { // kvmvccMavlFork高度下前一个区块需要映射
hash, err = kvmMavls.KVMVCCStore.SetRdm(datas, nil, sync)
} else {
hash, err = kvmMavls.KVMVCCStore.Set(datas, nil, sync)
}
if err == nil {
kvmMavls.cache.Add(string(hash), datas.Height)
}
......@@ -197,6 +215,19 @@ func (kvmMavls *KVmMavlStore) Set(datas *types.StoreSet, sync bool) ([]byte, err
// Get kvs with statehash from KVmMavlStore
func (kvmMavls *KVmMavlStore) Get(datas *types.StoreGet) [][]byte {
if kvmMavls.kvmvccCfg.EnableEmptyBlockHandle {
// 空块情况下只有第一个hash的为非空块
mvccHash, err := kvmMavls.KVMVCCStore.GetFirstHashRdm(datas.StateHash)
if err == nil {
nData := &types.StoreGet{
StateHash: mvccHash,
Keys: datas.Keys,
}
return kvmMavls.KVMVCCStore.Get(nData)
}
// ForkKvmvccmavl之后无mavl，即不需映射，直接获取
return kvmMavls.KVMVCCStore.Get(datas)
}
return kvmMavls.KVMVCCStore.Get(datas)
}
......@@ -207,17 +238,31 @@ func (kvmMavls *KVmMavlStore) MemSet(datas *types.StoreSet, sync bool) ([]byte,
if err != nil {
return hash, err
}
_, err = kvmMavls.KVMVCCStore.MemSet(datas, hash, sync)
if err != nil {
return hash, err
if kvmMavls.kvmvccCfg.EnableEmptyBlockHandle {
_, err = kvmMavls.KVMVCCStore.MemSetRdm(datas, hash, sync)
if err != nil {
return hash, err
}
} else {
_, 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)
var hash []byte
var err error
if kvmMavls.kvmvccCfg.EnableEmptyBlockHandle && datas.Height == kvmvccMavlFork { // kvmvccMavlFork高度下前一个区块需要映射
hash, err = kvmMavls.KVMVCCStore.MemSetRdm(datas, nil, sync)
} else {
hash, err = kvmMavls.KVMVCCStore.MemSet(datas, nil, sync)
}
if err == nil {
kvmMavls.cache.Add(string(hash), datas.Height)
}
......@@ -255,7 +300,20 @@ func (kvmMavls *KVmMavlStore) Rollback(req *types.ReqHash) ([]byte, error) {
if err != nil {
return hash, err
}
_, err = kvmMavls.KVMVCCStore.Rollback(req)
realReq := &types.ReqHash{
Hash: req.Hash,
Upgrade: req.Upgrade,
}
// 获取kvmvcc的实际statehash
if kvmMavls.kvmvccCfg.EnableEmptyBlockHandle {
if value, ok := kvmMavls.cache.Get(string(realReq.Hash)); ok {
mvccHash, err := kvmMavls.KVMVCCStore.GetHashRdm(realReq.Hash, value.(int64))
if err == nil {
realReq.Hash = mvccHash
}
}
}
_, err = kvmMavls.KVMVCCStore.Rollback(realReq)
return hash, err
}
return kvmMavls.KVMVCCStore.Rollback(req)
......@@ -265,15 +323,18 @@ func (kvmMavls *KVmMavlStore) Rollback(req *types.ReqHash) ([]byte, error) {
// 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)
if value, ok := kvmMavls.cache.Get(string(statehash)); ok && value.(int64) < kvmvccMavlFork {
kvmMavls.MavlStore.IterateRangeByStateHash(statehash, start, end, ascending, fn)
return
}
kvmMavls.KVMVCCStore.IterateRangeByStateHash(statehash, start, end, ascending, fn)
hash := statehash
if kvmMavls.kvmvccCfg.EnableEmptyBlockHandle {
mvccHash, err := kvmMavls.KVMVCCStore.GetFirstHashRdm(statehash)
if err == nil {
hash = mvccHash
}
}
kvmMavls.KVMVCCStore.IterateRangeByStateHash(hash, start, end, ascending, fn)
}
// ProcEvent handles supported events
......@@ -301,17 +362,32 @@ func (kvmMavls *KVmMavlStore) MemSetUpgrade(datas *types.StoreSet, sync bool) ([
return hash, err
}
}
_, err = kvmMavls.KVMVCCStore.MemSet(datas, hash, sync)
if err != nil {
return hash, err
if kvmMavls.kvmvccCfg.EnableEmptyBlockHandle {
_, err = kvmMavls.KVMVCCStore.MemSetRdm(datas, hash, sync)
if err != nil {
return hash, err
}
} else {
_, 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)
var hash []byte
var err error
if kvmMavls.kvmvccCfg.EnableEmptyBlockHandle && datas.Height == kvmvccMavlFork { // kvmvccMavlFork高度下前一个区块需要映射
hash, err = kvmMavls.KVMVCCStore.MemSetRdm(datas, nil, sync)
} else {
hash, err = kvmMavls.KVMVCCStore.MemSet(datas, nil, sync)
}
if err == nil {
kvmMavls.cache.Add(string(hash), datas.Height)
}
......@@ -341,14 +417,25 @@ func (kvmMavls *KVmMavlStore) Del(req *types.StoreDel) ([]byte, error) {
if err != nil {
return hash, err
}
hash, err = kvmMavls.KVMVCCStore.Del(req)
storeDel := &types.StoreDel{
StateHash: req.StateHash,
Height: req.Height,
}
// 获取kvmvcc的实际statehash
if kvmMavls.kvmvccCfg.EnableEmptyBlockHandle {
mvccHash, err := kvmMavls.KVMVCCStore.GetHashRdm(req.StateHash, req.Height)
if err == nil {
storeDel.StateHash = mvccHash
}
}
_, err = kvmMavls.KVMVCCStore.Del(storeDel)
if err != nil {
return hash, err
return req.StateHash, err
}
if err == nil {
kvmMavls.cache.Remove(string(req.StateHash))
}
return hash, err
return req.StateHash, err
}
// 仅仅做kvmvcc
hash, err := kvmMavls.KVMVCCStore.Del(req)
......
plugin/store/kvmvccmavl/kvmvcc_mavl_test.go
View file @ e2c6f375
......@@ -767,6 +767,283 @@ func TestDeletePrunedMavl(t *testing.T) {
wg.Wait()
}
func TestEmptyBlock(t *testing.T) {
dir, err := ioutil.TempDir("", "example")
assert.Nil(t, err)
defer os.RemoveAll(dir) // clean up
os.RemoveAll(dir) //删除已存在目录
storeCfg := newStoreCfg(dir)
subcfg := &subConfig{
EnableMVCCIter: true,
PruneMavlHeight: 10000,
PruneMVCCHeight: 10000,
// 使能空块处理
EnableEmptyBlockHandle: true,
}
sub, err := json.Marshal(subcfg)
assert.NoError(t, err)
store := New(storeCfg, sub, nil).(*KVmMavlStore)
assert.NotNil(t, store)
var kv []*types.KeyValue
var key string
var value string
var keys [][]byte
for i := 0; i < 2; i++ {
key = GetRandomString(MaxKeylenth)
value = fmt.Sprintf("v%d", i)
keys = append(keys, []byte(key))
kv = append(kv, &types.KeyValue{Key: []byte(key), Value: []byte(value)})
}
datas := &types.StoreSet{
StateHash: drivers.EmptyRoot[:],
KV: kv,
Height: 0}
var hash []byte
for i := 0; i < 10; i++ {
datas.Height = int64(i)
hash, err = store.MemSet(datas, true)
require.NoError(t, err)
req := &types.ReqHash{
Hash: hash,
}
_, err = store.Commit(req)
require.NoError(t, err)
datas.StateHash = hash
}
// 从第10个开始是空块
for i := 10; i < 16; i++ {
eData := &types.StoreSet{
StateHash: datas.StateHash,
KV: nil,
Height: int64(i),
}
hash = testStore(t, eData, store)
gdatas := &types.StoreGet{
StateHash: hash,
Keys: keys,
}
values := store.Get(gdatas)
for i, da := range kv {
require.Equal(t, values[i], da.Value)
}
}
// 回退4个块
eData := &types.StoreSet{
StateHash: datas.StateHash,
KV: nil,
Height: 12,
}
testStore(t, eData, store)
// check
gdatas := &types.StoreGet{
StateHash: hash,
Keys: keys,
}
values := store.Get(gdatas)
for i, da := range kv {
require.Equal(t, values[i], da.Value)
}
// 在加入高度为13的块
eData = &types.StoreSet{
StateHash: datas.StateHash,
KV: nil,
Height: 13,
}
testStore(t, eData, store)
// check
gdatas = &types.StoreGet{
StateHash: hash,
Keys: keys,
}
values = store.Get(gdatas)
for i, da := range kv {
require.Equal(t, values[i], da.Value)
}
// 加入实块
newK := []byte("tk111")
newV := []byte("tv111")
newKV := []*types.KeyValue{{Key: newK, Value: newV}}
eData = &types.StoreSet{
StateHash: datas.StateHash,
KV: newKV,
Height: 14,
}
newHash, err := store.MemSet(eData, true)
require.NoError(t, err)
req := &types.ReqHash{
Hash: newHash,
}
_, err = store.Commit(req)
require.NoError(t, err)
// check1 check hash高度下
gdatas = &types.StoreGet{
StateHash: hash,
Keys: append(keys, newK),
}
values = store.Get(gdatas)
require.Equal(t, len(values), len(kv)+1)
for i, da := range kv {
require.Equal(t, values[i], da.Value)
}
require.Nil(t, values[len(values)-1])
// check2 chek newHash
gdatas = &types.StoreGet{
StateHash: newHash,
Keys: append(keys, newK),
}
values = store.Get(gdatas)
kv = append(kv, newKV...)
for i, da := range kv {
require.Equal(t, values[i], da.Value)
}
// test SetRdm
}
func testStore(t *testing.T, eData *types.StoreSet, store *KVmMavlStore) []byte {
hash, err := store.MemSet(eData, true)
require.NoError(t, err)
require.Equal(t, hash, eData.StateHash)
req := &types.ReqHash{
Hash: hash,
}
_, err = store.Commit(req)
require.NoError(t, err)
return hash
}
func TestEmptyBlockForkSet(t *testing.T) {
dir, err := ioutil.TempDir("", "example")
assert.Nil(t, err)
defer os.RemoveAll(dir) // clean up
os.RemoveAll(dir) //删除已存在目录
storeCfg := newStoreCfg(dir)
subcfg := &subConfig{
EnableMVCCIter: true,
PruneMavlHeight: 10000,
PruneMVCCHeight: 10000,
// 使能空块处理
EnableEmptyBlockHandle: true,
}
sub, err := json.Marshal(subcfg)
assert.NoError(t, err)
store := New(storeCfg, sub, nil).(*KVmMavlStore)
assert.NotNil(t, store)
var kv []*types.KeyValue
var key string
var value string
var keys [][]byte
for i := 0; i < 2; i++ {
key = GetRandomString(MaxKeylenth)
value = fmt.Sprintf("v%d", i)
keys = append(keys, []byte(key))
kv = append(kv, &types.KeyValue{Key: []byte(key), Value: []byte(value)})
}
// kvmvccMavlFork = 5 加一个分叉高度测试
kvmvccMavlFork = 5
defer func() {
kvmvccMavlFork = 200 * 10000
}()
datas := &types.StoreSet{
StateHash: drivers.EmptyRoot[:],
KV: kv,
Height: 0}
var hash []byte
for i := 0; i < 10; i++ {
datas.Height = int64(i)
hash, err = store.Set(datas, true)
require.NoError(t, err)
datas.StateHash = hash
}
// check
gdatas := &types.StoreGet{
StateHash: hash,
Keys: keys,
}
values := store.Get(gdatas)
for i, da := range kv {
require.Equal(t, values[i], da.Value)
}
}
func TestEmptyBlockForkMemSet(t *testing.T) {
dir, err := ioutil.TempDir("", "example")
assert.Nil(t, err)
defer os.RemoveAll(dir) // clean up
os.RemoveAll(dir) //删除已存在目录
storeCfg := newStoreCfg(dir)
subcfg := &subConfig{
EnableMVCCIter: true,
PruneMavlHeight: 10000,
PruneMVCCHeight: 10000,
// 使能空块处理
EnableEmptyBlockHandle: true,
}
sub, err := json.Marshal(subcfg)
assert.NoError(t, err)
store := New(storeCfg, sub, nil).(*KVmMavlStore)
assert.NotNil(t, store)
var kv []*types.KeyValue
var key string
var value string
var keys [][]byte
for i := 0; i < 2; i++ {
key = GetRandomString(MaxKeylenth)
value = fmt.Sprintf("v%d", i)
keys = append(keys, []byte(key))
kv = append(kv, &types.KeyValue{Key: []byte(key), Value: []byte(value)})
}
// kvmvccMavlFork = 5 加一个分叉高度测试
kvmvccMavlFork = 5
defer func() {
kvmvccMavlFork = 200 * 10000
}()
datas := &types.StoreSet{
StateHash: drivers.EmptyRoot[:],
KV: kv,
Height: 0}
var hash []byte
for i := 0; i < 10; i++ {
datas.Height = int64(i)
hash, err = store.MemSet(datas, true)
require.NoError(t, err)
req := &types.ReqHash{
Hash: hash,
}
_, err = store.Commit(req)
require.NoError(t, err)
datas.StateHash = hash
}
// check
gdatas := &types.StoreGet{
StateHash: hash,
Keys: keys,
}
values := store.Get(gdatas)
for i, da := range kv {
require.Equal(t, values[i], da.Value)
}
}
func BenchmarkGetkmvccMavl(b *testing.B) { benchmarkGet(b, false) }
func BenchmarkGetkmvcc(b *testing.B) { benchmarkGet(b, true) }
......
plugin/store/kvmvccmavl/kvmvccdb.go
View file @ e2c6f375
......@@ -41,6 +41,9 @@ var (
//mvccLast = append(mvccPrefix, []byte("l.")...)
//mvccMetaVersion = append(mvccMeta, []byte("version.")...)
//mvccMetaVersionKeyList = append(mvccMeta, []byte("versionkl.")...)
// for empty block
rdmHashPrefix = append(mvccPrefix, []byte("rdm.")...)
)
// KVMCCCConfig KVMCCC config
......@@ -48,6 +51,7 @@ type KVMCCCConfig struct {
EnableMVCCIter bool
EnableMVCCPrune bool
PruneHeight int32
EnableEmptyBlockHandle bool
}
// KVMVCCStore provide kvmvcc store interface implementation
......@@ -72,6 +76,7 @@ func NewKVMVCC(sub *subKVMVCCConfig, db dbm.DB) *KVMVCCStore {
EnableMVCCIter: sub.EnableMVCCIter,
EnableMVCCPrune: sub.EnableMVCCPrune,
PruneHeight: sub.PruneHeight,
EnableEmptyBlockHandle: sub.EnableEmptyBlockHandle,
}
if kvmvccCfg.EnableMVCCIter {
kvs = &KVMVCCStore{db, dbm.NewMVCCIter(db), make(map[string][]*types.KeyValue), false, kvmvccCfg}
......@@ -343,6 +348,109 @@ func calcHash(datas proto.Message) []byte {
return common.Sha256(b)
}
func (mvccs *KVMVCCStore) SetRdm(datas *types.StoreSet, mavlHash []byte, sync bool) ([]byte, error) {
mvccHash := calcHash(datas)
// 取出前一个hash映射
var preMvccHash []byte
var err error
if datas.Height > 0 {
preMvccHash, err = mvccs.GetHashRdm(datas.StateHash, datas.Height-1)
if err != nil {
kmlog.Error("kvmvcc GetHashRdm", "error", err, "height", datas.Height-1)
return nil, err
}
}
kvlist, err := mvccs.mvcc.AddMVCC(datas.KV, mvccHash, preMvccHash, datas.Height)
if err != nil {
return nil, err
}
hash := mvccHash
if mavlHash != nil {
hash = mavlHash
// add rdm
key := calcRdmKey(mavlHash, datas.Height)
kvlist = append(kvlist, &types.KeyValue{Key: key, Value: mvccHash})
}
mvccs.saveKVSets(kvlist, sync)
return hash, nil
}
func (mvccs *KVMVCCStore) MemSetRdm(datas *types.StoreSet, mavlHash []byte, sync bool) ([]byte, error) {
beg := types.Now()
defer func() {
kmlog.Debug("kvmvcc MemSetRdm", "cost", types.Since(beg))
}()
kvset, err := mvccs.checkVersion(datas.Height)
if err != nil {
return nil, err
}
//kmlog.Debug("KVMVCCStore MemSet AddMVCC", "prestatehash", common.ToHex(datas.StateHash), "hash", common.ToHex(hash), "height", datas.Height)
mvcchash := calcHash(datas)
// 取出前一个hash映射
var preMvccHash []byte
if datas.Height > 0 {
preMvccHash, err = mvccs.GetHashRdm(datas.StateHash, datas.Height-1)
if err != nil {
kmlog.Error("kvmvcc GetHashRdm", "error", err, "height", datas.Height-1)
return nil, err
}
}
kvlist, err := mvccs.mvcc.AddMVCC(datas.KV, mvcchash, preMvccHash, datas.Height)
if err != nil {
return nil, err
}
if len(kvlist) > 0 {
kvset = append(kvset, kvlist...)
}
hash := mvcchash
if mavlHash != nil {
// 如果mavlHash是nil,即返回mvcchash
hash = mavlHash
// add rdm
kv := &types.KeyValue{Key: calcRdmKey(mavlHash, datas.Height), Value: mvcchash}
kvset = append(kvset, kv)
}
mvccs.kvsetmap[string(hash)] = kvset
mvccs.sync = sync
// 进行裁剪
if mvccs.kvmvccCfg != nil && mvccs.kvmvccCfg.EnableMVCCPrune &&
!isPruning() && mvccs.kvmvccCfg.PruneHeight != 0 &&
datas.Height%int64(mvccs.kvmvccCfg.PruneHeight) == 0 &&
datas.Height/int64(mvccs.kvmvccCfg.PruneHeight) > 1 {
wg.Add(1)
go pruning(mvccs.db, datas.Height, mvccs.kvmvccCfg)
}
return hash, nil
}
func (mvccs *KVMVCCStore) GetHashRdm(hash []byte, height int64) ([]byte, error) {
key := calcRdmKey(hash, height)
return mvccs.db.Get(key)
}
func (mvccs *KVMVCCStore) GetFirstHashRdm(hash []byte) ([]byte, error) {
prefix := append(rdmHashPrefix, hash...)
list := dbm.NewListHelper(mvccs.db)
values := list.IteratorScanFromFirst(prefix, 1)
if len(values) == 1 {
return values[0], nil
}
return nil, types.ErrNotFound
}
func calcRdmKey(hash []byte, height int64) []byte {
key := append(rdmHashPrefix, hash...)
key = append(key, []byte(".")...)
key = append(key, pad(height)...)
return key
}
/*裁剪-------------------------------------------*/
func pruning(db dbm.DB, height int64, KVmvccCfg *KVMCCCConfig) {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment