[db] improve indexer write speed using common batch (iotexproject#2159)

Author: dustinxie

action/protocol/staking/protocol.go

@@ -265,7 +265,7 @@ func (p *Protocol) handle(ctx context.Context, act action.Action, csm CandidateS
 	}
 
 	if receiptErr, ok := err.(ReceiptError); ok {
-		log.L().Info("Non-critical error when processing staking action", zap.Error(err))
+		log.L().Debug("Non-critical error when processing staking action", zap.Error(err))
 		return p.settleAction(ctx, csm, receiptErr.ReceiptStatus(), rLog.Build(ctx, err))
 	}
 	return nil, err

blockindex/indexer.go

@@ -66,6 +66,7 @@ type (
 		genesisHash hash.Hash256
 		kvStore     db.KVStoreWithRange
 		batch       batch.KVStoreBatch
+		cBatch      batch.KVStoreBatch
 		dirtyAddr   addrIndex
 		tbk         db.CountingIndex
 		tac         db.CountingIndex
@@ -84,6 +85,7 @@ func NewIndexer(kv db.KVStore, genesisHash hash.Hash256) (Indexer, error) {
 	x := blockIndexer{
 		kvStore:     kvRange,
 		batch:       batch.NewBatch(),
+		cBatch:      batch.NewBatch(),
 		dirtyAddr:   make(addrIndex),
 		genesisHash: genesisHash,
 	}
@@ -172,7 +174,7 @@ func (x *blockIndexer) DeleteTipBlock(blk *block.Block) error {
 	if err := x.tac.Revert(uint64(len(blk.Actions))); err != nil {
 		return err
 	}
-	return x.commit()
+	return x.kvStore.WriteBatch(x.batch)
 }
 
 // Height return the blockchain height
@@ -293,21 +295,29 @@ func (x *blockIndexer) putBlock(blk *block.Block) error {
 	if height != x.tbk.Size() {
 		return errors.Wrapf(db.ErrInvalid, "wrong block height %d, expecting %d", height, x.tbk.Size())
 	}
+
 	// index hash --> height
 	hash := blk.HashBlock()
 	x.batch.Put(blockHashToHeightNS, hash[hashOffset:], byteutil.Uint64ToBytesBigEndian(height), "failed to put hash -> height mapping")
+
 	// index height --> block hash, number of actions, and total transfer amount
 	bd := &blockIndex{
 		hash:      hash[:],
 		numAction: uint32(len(blk.Actions)),
 		tsfAmount: blk.CalculateTransferAmount()}
+	if err := x.tbk.UseBatch(x.cBatch); err != nil {
+		return err
+	}
 	if err := x.tbk.Add(bd.Serialize(), true); err != nil {
 		return errors.Wrapf(err, "failed to put block %d index", height)
 	}
 
 	// store height of the block, so getReceiptByActionHash() can use height to directly pull receipts
 	ad := (&actionIndex{
 		blkHeight: blk.Height()}).Serialize()
+	if err := x.tac.UseBatch(x.cBatch); err != nil {
+		return err
+	}
 	// index actions in the block
 	for _, selp := range blk.Actions {
 		actHash := selp.Hash()
@@ -328,7 +338,7 @@ func (x *blockIndexer) commit() error {
 	var commitErr error
 	for k, v := range x.dirtyAddr {
 		if commitErr == nil {
-			if err := v.Commit(); err != nil {
+			if err := v.AddTotalSize(); err != nil {
 				commitErr = err
 			}
 		}
@@ -338,10 +348,13 @@ func (x *blockIndexer) commit() error {
 		return commitErr
 	}
 	// total block and total action index
-	if err := x.tbk.Commit(); err != nil {
+	if err := x.tbk.AddTotalSize(); err != nil {
+		return err
+	}
+	if err := x.tac.AddTotalSize(); err != nil {
 		return err
 	}
-	if err := x.tac.Commit(); err != nil {
+	if err := db.CommitWithFillPercent(x.kvStore, x.cBatch, 1.0); err != nil {
 		return err
 	}
 	return x.kvStore.WriteBatch(x.batch)
@@ -362,6 +375,9 @@ func (x *blockIndexer) getIndexerForAddr(addr []byte, batch bool) (db.CountingIn
 		if err != nil {
 			return nil, err
 		}
+		if err := indexer.UseBatch(x.cBatch); err != nil {
+			return nil, err
+		}
 		x.dirtyAddr[address] = indexer
 	}
 	return indexer, nil

db/counting_index.go

@@ -41,6 +41,10 @@ type (
 		Close()
 		// Commit commits the batch
 		Commit() error
+		// UseBatch
+		UseBatch(batch.KVStoreBatch) error
+		// AddTotalSize
+		AddTotalSize() error
 	}
 
 	// countingIndex is CountingIndex implementation based on KVStore
@@ -117,7 +121,7 @@ func (c *countingIndex) Add(value []byte, inBatch bool) error {
 	b := batch.NewBatch()
 	b.Put(c.bucket, byteutil.Uint64ToBytesBigEndian(c.size), value, "failed to add %d-th item", c.size+1)
 	b.Put(c.bucket, CountKey, byteutil.Uint64ToBytesBigEndian(c.size+1), "failed to update size = %d", c.size+1)
-	if err := c.commit(b); err != nil {
+	if err := CommitWithFillPercent(c.kvStore, b, 1.0); err != nil {
 		return err
 	}
 	c.size++
@@ -164,7 +168,7 @@ func (c *countingIndex) Revert(count uint64) error {
 		b.Delete(c.bucket, byteutil.Uint64ToBytesBigEndian(start+i), "failed to delete %d-th item", start+i)
 	}
 	b.Put(c.bucket, CountKey, byteutil.Uint64ToBytesBigEndian(start), "failed to update size = %d", start)
-	if err := c.commit(b); err != nil {
+	if err := CommitWithFillPercent(c.kvStore, b, 1.0); err != nil {
 		return err
 	}
 	c.size = start
@@ -184,18 +188,28 @@ func (c *countingIndex) Commit() error {
 		return nil
 	}
 	c.batch.Put(c.bucket, CountKey, byteutil.Uint64ToBytesBigEndian(c.size), "failed to update size = %d", c.size)
-	if err := c.commit(c.batch); err != nil {
+	if err := CommitWithFillPercent(c.kvStore, c.batch, 1.0); err != nil {
 		return err
 	}
 	c.batch = nil
 	return nil
 }
 
-func (c *countingIndex) commit(b batch.KVStoreBatch) error {
-	if kvFillPercent, ok := c.kvStore.(KVStoreWithBucketFillPercent); ok {
-		// set an aggressive fill percent
-		// b/c counting index only appends, further inserts to the bucket would never split the page
-		return kvFillPercent.WriteBatchWithFillPercent(b, 1.0)
+// UseBatch sets a (usually common) batch for the counting index to use
+func (c *countingIndex) UseBatch(b batch.KVStoreBatch) error {
+	if b == nil {
+		return ErrInvalid
 	}
-	return c.kvStore.WriteBatch(b)
+	c.batch = b
+	return nil
+}
+
+// AddTotalSize updates the total size before committing the (usually common) batch
+func (c *countingIndex) AddTotalSize() error {
+	if c.batch == nil {
+		return ErrInvalid
+	}
+	c.batch.Put(c.bucket, CountKey, byteutil.Uint64ToBytesBigEndian(c.size), "failed to update size = %d", c.size)
+	c.batch = nil
+	return nil
 }

db/counting_index_test.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/iotexproject/go-pkgs/hash"
 
+	"github.com/iotexproject/iotex-core/db/batch"
 	"github.com/iotexproject/iotex-core/pkg/log"
 	"github.com/iotexproject/iotex-core/testutil"
 )
@@ -71,10 +72,18 @@ func TestCountingIndex(t *testing.T) {
 		index, err = GetCountingIndex(kv, bucket)
 		require.NoError(err)
 		// write another 100 entries
-		for i := 200; i < 300; i++ {
+		for i := 200; i < 250; i++ {
 			h := hash.Hash160b([]byte(strconv.Itoa(i)))
 			require.NoError(index.Add(h[:], false))
 		}
+		require.EqualValues(250, index.Size())
+		// use external batch
+		require.NoError(index.UseBatch(batch.NewBatch()))
+		for i := 250; i < 300; i++ {
+			h := hash.Hash160b([]byte(strconv.Itoa(i)))
+			require.NoError(index.Add(h[:], true))
+		}
+		require.NoError(index.Commit())
 		require.EqualValues(300, index.Size())
 
 		_, err = index.Range(248, 0)

db/kvstore_impl.go

@@ -144,3 +144,13 @@ func (m *memKVStore) GetBucketByPrefix(namespace []byte) ([][]byte, error) {
 func (m *memKVStore) GetKeyByPrefix(namespace, prefix []byte) ([][]byte, error) {
 	return nil, nil
 }
+
+// CommitWithFillPercent commits the batch using fill percent, if underlying KVStore is capable
+func CommitWithFillPercent(kvstore KVStore, b batch.KVStoreBatch, percent float64) error {
+	if kvFillPercent, ok := kvstore.(KVStoreWithBucketFillPercent); ok {
+		// set an aggressive fill percent
+		// b/c counting index only appends, further inserts to the bucket would never split the page
+		return kvFillPercent.WriteBatchWithFillPercent(b, percent)
+	}
+	return kvstore.WriteBatch(b)
+}

state/factory/factory.go

@@ -518,7 +518,7 @@ func (sf *factory) PutBlock(ctx context.Context, blk *block.Block) error {
 		// regenerate workingset
 		_, err = ws.Process(ctx, blk.RunnableActions().Actions())
 		if err != nil {
-			log.L().Panic("Failed to update state.", zap.Error(err))
+			log.L().Error("Failed to update state.", zap.Error(err))
 			return err
 		}
 	}

state/factory/statedb.go

@@ -356,7 +356,7 @@ func (sdb *stateDB) PutBlock(ctx context.Context, blk *block.Block) error {
 	if !isExist {
 		_, err = ws.Process(ctx, blk.RunnableActions().Actions())
 		if err != nil {
-			log.L().Panic("Failed to update state.", zap.Error(err))
+			log.L().Error("Failed to update state.", zap.Error(err))
 			return err
 		}
 	}