Merge pull request #557 from lochjin/dev1.2

Optimize update utxo about batch

consensus/model/database.go

@@ -20,6 +20,7 @@ type DataBase interface {
 	PutUtxo(key []byte, data []byte) error
 	DeleteUtxo(key []byte) error
 	ForeachUtxo(fn func(key []byte, data []byte) error) error
+	UpdateUtxo(opts []*common.UtxoOpt) error
 	GetTokenState(blockID uint) ([]byte, error)
 	PutTokenState(blockID uint, data []byte) error
 	DeleteTokenState(blockID uint) error

core/blockchain/process.go

@@ -434,7 +434,7 @@ func (b *BlockChain) connectBlock(node meerdag.IBlock, blockNode *BlockNode, vie
 		// Update the utxo set using the state of the utxo view.  This
 		// entails removing all of the utxos spent and adding the new
 		// ones created by the block.
-		err = b.dbPutUtxoView(view)
+		err = b.dbUpdateUtxoView(view)
 		if err != nil {
 			return err
 		}
@@ -478,7 +478,7 @@ func (b *BlockChain) disconnectBlock(ib meerdag.IBlock, block *types.SerializedB
 	// Update the utxo set using the state of the utxo view.  This
 	// entails restoring all of the utxos spent and removing the new
 	// ones created by the block.
-	err := b.dbPutUtxoView(view)
+	err := b.dbUpdateUtxoView(view)
 	if err != nil {
 		return err
 	}

core/blockchain/utxo.go

@@ -5,6 +5,7 @@ import (
 	"github.com/Qitmeer/qng/common/hash"
 	"github.com/Qitmeer/qng/core/blockchain/utxo"
 	"github.com/Qitmeer/qng/core/types"
+	"github.com/Qitmeer/qng/database/common"
 	"github.com/Qitmeer/qng/database/legacydb"
 	"github.com/Qitmeer/qng/meerdag"
 )
@@ -238,5 +239,45 @@ func (b *BlockChain) dbPutUtxoViewByBlock(block *types.SerializedBlock) error {
 	for _, tx := range block.Transactions() {
 		view.AddTxOuts(tx, block.Hash())
 	}
-	return b.dbPutUtxoView(view)
+	return b.dbUpdateUtxoView(view)
+}
+
+func (b *BlockChain) dbUpdateUtxoView(view *utxo.UtxoViewpoint) error {
+	opts := []*common.UtxoOpt{}
+	for op, en := range view.Entries() {
+		outpoint := op
+		entry := en
+		// No need to update the database if the entry was not modified.
+		if entry == nil || !entry.IsModified() {
+			continue
+		}
+		// Remove the utxo entry if it is spent.
+		if entry.IsSpent() {
+			key := utxo.OutpointKey(outpoint)
+			opts = append(opts, &common.UtxoOpt{Add: false, Key: *key})
+			utxo.RecycleOutpointKey(key)
+			if b.Acct != nil {
+				err := b.Acct.Apply(false, &outpoint, entry)
+				if err != nil {
+					log.Error(err.Error())
+				}
+			}
+			continue
+		}
+
+		// Serialize and store the utxo entry.
+		serialized, err := utxo.SerializeUtxoEntry(entry)
+		if err != nil {
+			return err
+		}
+		key := utxo.OutpointKey(outpoint)
+		opts = append(opts, &common.UtxoOpt{Add: true, Key: *key, Data: serialized})
+		if b.Acct != nil {
+			err = b.Acct.Apply(true, &outpoint, entry)
+			if err != nil {
+				log.Error(err.Error())
+			}
+		}
+	}
+	return b.DB().UpdateUtxo(opts)
 }

database/chaindb/chaindb.go

@@ -257,6 +257,27 @@ func (cdb *ChainDB) ForeachUtxo(fn func(key []byte, data []byte) error) error {
 	return rawdb.ForeachUtxo(cdb.db, fn)
 }
 
+func (cdb *ChainDB) UpdateUtxo(opts []*common.UtxoOpt) error {
+	if len(opts) <= 0 {
+		return nil
+	}
+	if cdb.diff != nil {
+		return cdb.diff.UpdateUtxo(opts)
+	}
+	batch := cdb.db.NewBatch()
+	for _, opt := range opts {
+		if opt.Add {
+			err := rawdb.WriteUtxo(batch, opt.Key, opt.Data)
+			if err != nil {
+				return err
+			}
+		} else {
+			rawdb.DeleteUtxo(batch, opt.Key)
+		}
+	}
+	return batch.Write()
+}
+
 func (cdb *ChainDB) GetTokenState(blockID uint) ([]byte, error) {
 	if cdb.diff != nil {
 		return cdb.diff.GetTokenState(blockID)

database/chaindb/chaindb_test.go

@@ -1,10 +1,19 @@
 package chaindb
 
 import (
+	"bytes"
+	"crypto/rand"
+	"github.com/Qitmeer/qng/common/util"
+	l "github.com/Qitmeer/qng/log"
+	"github.com/Qitmeer/qng/meerevm/eth"
 	"github.com/Qitmeer/qng/services/common"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/ethdb/dbtest"
+	elog "github.com/ethereum/go-ethereum/log"
+	"golang.org/x/exp/slices"
+	"os"
 	"testing"
+	"time"
 )
 
 func TestChainCloseClosesDB(t *testing.T) {
@@ -92,3 +101,151 @@ func TestChainDBBatch(t *testing.T) {
 		t.Fatalf("want exist,but absent")
 	}
 }
+
+func BenchmarkIOLevelDB(b *testing.B) {
+	doBenchmarkIO(b, "leveldb")
+}
+
+func BenchmarkIOPebbleDB(b *testing.B) {
+	doBenchmarkIO(b, "pebble")
+}
+
+func doBenchmarkIO(b *testing.B, dbtype string) {
+	dataDir, err := os.MkdirTemp("", "data_"+dbtype+"_*")
+	if err != nil {
+		b.Fatal(err)
+	}
+	log.Info("benchmark", "dbtype", dbtype, "datadir", dataDir)
+
+	cfg := common.DefaultConfig("")
+	l.Glogger().Verbosity(l.LvlCrit)
+	eth.InitLog(elog.LvlCrit.String(), cfg.DebugPrintOrigins)
+	BenchDatabaseSuite(b, func() ethdb.KeyValueStore {
+		cfg.DataDir = dataDir
+		cfg.DbType = dbtype
+		cdb, err := NewNaked(cfg)
+		if err != nil {
+			b.Fatal(err)
+		}
+		return cdb.DB()
+	})
+	if util.FileExists(dataDir) {
+		err = os.RemoveAll(dataDir)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+// BenchDatabaseSuite runs a suite of benchmarks against a KeyValueStore database
+// implementation.
+func BenchDatabaseSuite(b *testing.B, New func() ethdb.KeyValueStore) {
+	var (
+		keys, vals   = makeDataset(1000, 32, 32, false)
+		sKeys, sVals = makeDataset(1000, 32, 32, true)
+	)
+	// Run benchmarks sequentially
+	b.Run("Write", func(b *testing.B) {
+		benchWrite := func(b *testing.B, keys, vals [][]byte) {
+			b.ResetTimer()
+
+			db := New()
+			defer db.Close()
+
+			for i := 0; i < len(keys); i++ {
+				db.Put(keys[i], vals[i])
+			}
+		}
+		b.Run("WriteSorted", func(b *testing.B) {
+			benchWrite(b, sKeys, sVals)
+		})
+		b.Run("WriteRandom", func(b *testing.B) {
+			benchWrite(b, keys, vals)
+		})
+	})
+	b.Run("Read", func(b *testing.B) {
+		benchRead := func(b *testing.B, keys, vals [][]byte) {
+			db := New()
+			defer db.Close()
+
+			for i := 0; i < len(keys); i++ {
+				db.Put(keys[i], vals[i])
+			}
+			b.ResetTimer()
+
+			for i := 0; i < len(keys); i++ {
+				db.Get(keys[i])
+			}
+		}
+		b.Run("ReadSorted", func(b *testing.B) {
+			benchRead(b, sKeys, sVals)
+		})
+		b.Run("ReadRandom", func(b *testing.B) {
+			benchRead(b, keys, vals)
+		})
+	})
+	b.Run("Iteration", func(b *testing.B) {
+		benchIteration := func(b *testing.B, keys, vals [][]byte) {
+			db := New()
+			defer db.Close()
+
+			for i := 0; i < len(keys); i++ {
+				db.Put(keys[i], vals[i])
+			}
+			b.ResetTimer()
+
+			it := db.NewIterator(nil, nil)
+			for it.Next() {
+			}
+			it.Release()
+		}
+		b.Run("IterationSorted", func(b *testing.B) {
+			benchIteration(b, sKeys, sVals)
+		})
+		b.Run("IterationRandom", func(b *testing.B) {
+			benchIteration(b, keys, vals)
+		})
+	})
+	b.Run("BatchWrite", func(b *testing.B) {
+		benchBatchWrite := func(b *testing.B, keys, vals [][]byte) {
+			b.ResetTimer()
+
+			db := New()
+			defer db.Close()
+
+			batch := db.NewBatch()
+			for i := 0; i < len(keys); i++ {
+				batch.Put(keys[i], vals[i])
+			}
+			batch.Write()
+		}
+		b.Run("BenchWriteSorted", func(b *testing.B) {
+			benchBatchWrite(b, sKeys, sVals)
+		})
+		b.Run("BenchWriteRandom", func(b *testing.B) {
+			benchBatchWrite(b, keys, vals)
+		})
+	})
+}
+
+// randomHash generates a random blob of data and returns it as a hash.
+func randBytes(len int) []byte {
+	buf := make([]byte, len)
+	if n, err := rand.Read(buf); n != len || err != nil {
+		panic(err)
+	}
+	return buf
+}
+
+func makeDataset(size, ksize, vsize int, order bool) ([][]byte, [][]byte) {
+	var keys [][]byte
+	var vals [][]byte
+	for i := 0; i < size; i += 1 {
+		keys = append(keys, randBytes(ksize))
+		vals = append(vals, randBytes(vsize))
+	}
+	if order {
+		slices.SortFunc(keys, func(a, b []byte) int { return bytes.Compare(a, b) })
+	}
+	return keys, vals
+}

database/chaindb/difflayer.go

@@ -5,6 +5,7 @@ import (
 	"github.com/Qitmeer/qng/common/hash"
 	"github.com/Qitmeer/qng/consensus/model"
 	"github.com/Qitmeer/qng/core/types"
+	com "github.com/Qitmeer/qng/database/common"
 	"github.com/Qitmeer/qng/database/rawdb"
 	"github.com/Qitmeer/qng/meerdag"
 	"github.com/ethereum/go-ethereum/common"
@@ -349,6 +350,26 @@ func (dl *diffLayer) ForeachUtxo(fn func(key []byte, data []byte) error) error {
 	return rawdb.ForeachUtxo(dl.db, fun)
 }
 
+func (dl *diffLayer) UpdateUtxo(opts []*com.UtxoOpt) error {
+	dl.lock.Lock()
+	defer dl.lock.Unlock()
+
+	if dl.utxo == nil {
+		dl.utxo = map[string][]byte{}
+	}
+
+	for _, opt := range opts {
+		if opt.Add {
+			dl.utxo[string(opt.Key)] = opt.Data
+			dl.memory.Add(uint64(len(opt.Data) + len(opt.Key)))
+		} else {
+			dl.utxo[string(opt.Key)] = nil
+			dl.memory.Add(uint64(len(opt.Key)))
+		}
+	}
+	return nil
+}
+
 func (dl *diffLayer) GetTokenState(blockID uint) ([]byte, error) {
 	dl.lock.RLock()
 	defer dl.lock.RUnlock()

database/common/utxo.go

@@ -0,0 +1,7 @@
+package common
+
+type UtxoOpt struct {
+	Key  []byte
+	Data []byte
+	Add  bool
+}

database/legacychaindb/legacychaindb.go

@@ -10,6 +10,7 @@ import (
 	"github.com/Qitmeer/qng/core/dbnamespace"
 	"github.com/Qitmeer/qng/core/shutdown"
 	"github.com/Qitmeer/qng/core/types"
+	"github.com/Qitmeer/qng/database/common"
 	"github.com/Qitmeer/qng/database/legacydb"
 	"github.com/Qitmeer/qng/database/rawdb"
 	"github.com/Qitmeer/qng/meerdag"
@@ -249,6 +250,28 @@ func (cdb *LegacyChainDB) ForeachUtxo(fn func(key []byte, data []byte) error) er
 	})
 }
 
+func (cdb *LegacyChainDB) UpdateUtxo(opts []*common.UtxoOpt) error {
+	if len(opts) <= 0 {
+		return nil
+	}
+
+	return cdb.db.Update(func(dbTx legacydb.Tx) error {
+		bucket := dbTx.Metadata().Bucket(dbnamespace.UtxoSetBucketName)
+		var err error
+		for _, opt := range opts {
+			if opt.Add {
+				err = bucket.Put(opt.Key, opt.Data)
+			} else {
+				err = bucket.Delete(opt.Key)
+			}
+			if err != nil {
+				return err
+			}
+		}
+		return err
+	})
+}
+
 func (cdb *LegacyChainDB) GetTokenState(blockID uint) ([]byte, error) {
 	var data []byte
 	err := cdb.db.View(func(dbTx legacydb.Tx) error {