leveldb: throttle transaction if there is to many tables in level-0

leveldb/db_transaction.go

@@ -201,14 +201,14 @@ func (tr *Transaction) Commit() error {
 		// Committing transaction.
 		tr.rec.setSeqNum(tr.seq)
 		tr.db.compCommitLk.Lock()
-		defer tr.db.compCommitLk.Unlock()
 		for retry := 0; retry < 3; retry++ {
 			if err := tr.db.s.commit(&tr.rec); err != nil {
 				tr.db.logf("transaction@commit error R·%d %q", retry, err)
 				select {
 				case <-time.After(time.Second):
 				case _, _ = <-tr.db.closeC:
 					tr.db.logf("transaction@commit exiting")
+					tr.db.compCommitLk.Unlock()
 					return err
 				}
 			} else {
@@ -219,6 +219,12 @@ func (tr *Transaction) Commit() error {
 		}
 		// Trigger table auto-compaction.
 		tr.db.compTrigger(tr.db.tcompCmdC)
+		tr.db.compCommitLk.Unlock()
+
+		// Additionally, wait compaction when certain threshold reached.
+		if err := tr.db.waitCompaction(); err != nil {
+			return err
+		}
 	}
 	return nil
 }
@@ -245,10 +251,19 @@ func (tr *Transaction) Discard() {
 	tr.lk.Unlock()
 }
 
+func (db *DB) waitCompaction() error {
+	if db.s.tLen(0) >= db.s.o.GetWriteL0PauseTrigger() {
+		return db.compTriggerWait(db.tcompCmdC)
+	}
+	return nil
+}
+
 // OpenTransaction opens an atomic DB transaction. Only one transaction can be
 // opened at a time. Subsequent call to Write and OpenTransaction will be blocked
 // until in-flight transaction is committed or discarded.
-// The returned transaction handle is goroutine-safe.
+// The returned transaction handle is safe for concurrent use.
+//
+// Transaction is expensive and put restrains on the compaction.
 //
 // The transaction must be closed once done, either by committing or discarding
 // the transaction.
@@ -278,6 +293,11 @@ func (db *DB) OpenTransaction() (*Transaction, error) {
 		}
 	}
 
+	// Wait compaction when certain threshold reached.
+	if err := db.waitCompaction(); err != nil {
+		return nil, err
+	}
+
 	tr := &Transaction{
 		db:  db,
 		seq: db.seq,

leveldb/db_write.go

@@ -55,9 +55,9 @@ func (db *DB) rotateMem(n int, wait bool) (mem *memDB, err error) {
 
 func (db *DB) flush(n int) (mdb *memDB, mdbFree int, err error) {
 	delayed := false
+	slowdownTrigger := db.s.o.GetWriteL0SlowdownTrigger()
+	pauseTrigger := db.s.o.GetWriteL0PauseTrigger()
 	flush := func() (retry bool) {
-		v := db.s.version()
-		defer v.release()
 		mdb = db.getEffectiveMem()
 		if mdb == nil {
 			err = ErrClosed
@@ -69,14 +69,15 @@ func (db *DB) flush(n int) (mdb *memDB, mdbFree int, err error) {
 				mdb = nil
 			}
 		}()
+		tLen := db.s.tLen(0)
 		mdbFree = mdb.Free()
 		switch {
-		case v.tLen(0) >= db.s.o.GetWriteL0SlowdownTrigger() && !delayed:
+		case tLen >= slowdownTrigger && !delayed:
 			delayed = true
 			time.Sleep(time.Millisecond)
 		case mdbFree >= n:
 			return false
-		case v.tLen(0) >= db.s.o.GetWriteL0PauseTrigger():
+		case tLen >= pauseTrigger:
 			delayed = true
 			err = db.compTriggerWait(db.tcompCmdC)
 			if err != nil {

leveldb/session_util.go

@@ -50,6 +50,12 @@ func (s *session) version() *version {
 	return s.stVersion
 }
 
+func (s *session) tLen(level int) int {
+	s.vmu.Lock()
+	defer s.vmu.Unlock()
+	return s.stVersion.tLen(level)
+}
+
 // Set current version to v.
 func (s *session) setVersion(v *version) {
 	s.vmu.Lock()