make batcher configurable

batch.go

@@ -10,18 +10,18 @@ import (
 type BatchWithFlusher struct {
 	db    dbm.DB    // This is only used to create new batch
 	batch dbm.Batch // Batched writing buffer.
+
+	flushThreshold int // The threshold to flush the batch to disk.
 }
 
 var _ dbm.Batch = &BatchWithFlusher{}
 
-// Ethereum has found that commit of 100KB is optimal, ref ethereum/go-ethereum#15115
-var flushThreshold = 100000
-
 // NewBatchWithFlusher returns new BatchWithFlusher wrapping the passed in batch
-func NewBatchWithFlusher(db dbm.DB) *BatchWithFlusher {
+func NewBatchWithFlusher(db dbm.DB, flushThreshold int) *BatchWithFlusher {
 	return &BatchWithFlusher{
-		db:    db,
-		batch: db.NewBatchWithSize(flushThreshold),
+		db:             db,
+		batch:          db.NewBatchWithSize(flushThreshold),
+		flushThreshold: flushThreshold,
 	}
 }
 
@@ -50,7 +50,7 @@ func (b *BatchWithFlusher) Set(key []byte, value []byte) error {
 	if err != nil {
 		return err
 	}
-	if batchSizeAfter > flushThreshold {
+	if batchSizeAfter > b.flushThreshold {
 		err = b.batch.Write()
 		if err != nil {
 			return err
@@ -59,7 +59,7 @@ func (b *BatchWithFlusher) Set(key []byte, value []byte) error {
 		if err != nil {
 			return err
 		}
-		b.batch = b.db.NewBatchWithSize(flushThreshold)
+		b.batch = b.db.NewBatchWithSize(b.flushThreshold)
 	}
 	err = b.batch.Set(key, value)
 	if err != nil {
@@ -77,7 +77,7 @@ func (b *BatchWithFlusher) Delete(key []byte) error {
 	if err != nil {
 		return err
 	}
-	if batchSizeAfter > flushThreshold {
+	if batchSizeAfter > b.flushThreshold {
 		err = b.batch.Write()
 		if err != nil {
 			return err
@@ -86,7 +86,7 @@ func (b *BatchWithFlusher) Delete(key []byte) error {
 		if err != nil {
 			return err
 		}
-		b.batch = b.db.NewBatchWithSize(flushThreshold)
+		b.batch = b.db.NewBatchWithSize(b.flushThreshold)
 	}
 	err = b.batch.Delete(key)
 	if err != nil {

batch_test.go

@@ -43,7 +43,7 @@ func testBatchWithFlusher(t *testing.T, backend dbm.BackendType) {
 	require.NoError(t, err)
 	defer cleanupDBDir(dir, name)
 
-	batchWithFlusher := NewBatchWithFlusher(db)
+	batchWithFlusher := NewBatchWithFlusher(db, DefaultOptions().FlushThreshold)
 
 	// we'll try to to commit 10MBs (1000 * 10KBs each entries) of data into the db
 	for keyNonce := uint16(0); keyNonce < 1000; keyNonce++ {

mutable_tree_test.go

@@ -84,6 +84,7 @@ func TestIteratorConcurrency(t *testing.T) {
 		}
 		itr, _ := tree.Iterator(nil, nil, true)
 		for ; itr.Valid(); itr.Next() {
+			// do nothing
 		}
 	}
 	wg.Wait()
@@ -107,6 +108,7 @@ func TestNewIteratorConcurrency(t *testing.T) {
 			}(i, j)
 		}
 		for ; it.Valid(); it.Next() {
+			// do nothing
 		}
 		wg.Wait()
 	}

nodedb.go

@@ -100,7 +100,7 @@ func newNodeDB(db dbm.DB, cacheSize int, opts *Options, lg log.Logger) *nodeDB {
 	return &nodeDB{
 		logger:              lg,
 		db:                  db,
-		batch:               NewBatchWithFlusher(db),
+		batch:               NewBatchWithFlusher(db, opts.FlushThreshold),
 		opts:                *opts,
 		firstVersion:        0,
 		latestVersion:       0, // initially invalid
@@ -382,7 +382,7 @@ func (ndb *nodeDB) writeBatch() error {
 		return err
 	}
 
-	ndb.batch = NewBatchWithFlusher(ndb.db)
+	ndb.batch = NewBatchWithFlusher(ndb.db, ndb.opts.FlushThreshold)
 
 	return nil
 }
@@ -877,7 +877,7 @@ func (ndb *nodeDB) Commit() error {
 	}
 
 	ndb.batch.Close()
-	ndb.batch = NewBatchWithFlusher(ndb.db)
+	ndb.batch = NewBatchWithFlusher(ndb.db, ndb.opts.FlushThreshold)
 
 	return nil
 }

options.go

@@ -81,9 +81,12 @@ type Options struct {
 
 	// When Stat is not nil, statistical logic needs to be executed
 	Stat *Statistics
+
+	// Ethereum has found that commit of 100KB is optimal, ref ethereum/go-ethereum#15115
+	FlushThreshold int
 }
 
 // DefaultOptions returns the default options for IAVL.
 func DefaultOptions() Options {
-	return Options{}
+	return Options{FlushThreshold: 100000}
 }