Repository: petermattis/pebble (go)

Issue:
db: DB.{Flush,Compact} are slow in the absence of write traffic
Performing a manual flush/compact via `DB.{Flush,Compact}` can be very slow with the new flush and compaction pacing heuristics. Those heuristics limit how fast flush and compaction occur when there are no user writes. So if a benchmark workloads writes a bunch of data, and then calls `DB.Flush`, that flush operation will only write at 1MB/s by default. 

We could try to automatically increase the flush/compaction rate if user writes remain zero for a period of time. Alternately, we could adjust the pacing heuristics when a manual flush/compaction is being performed to use a different minimum rate.

Pull request description:
Don't pace flushes/compactions if there are no user writes
In the absence of user writes, allow flushes and compactions to proceed
as fast as possible, but lower the pacer recalculation threshold in
order to be nimble to new user writes.

Fixes #248

Patch:
diff --git a/compaction.go b/compaction.go
index b2c28c9..a6dc2a5 100644
--- a/compaction.go
+++ b/compaction.go
@@ -615,15 +615,22 @@ func (d *DB) getCompactionPacerInfo() compactionPacerInfo {
 		slowdownThreshold:   uint64(estimatedMaxWAmp * float64(d.opts.MemTableSize)),
 		totalCompactionDebt: d.mu.versions.picker.estimatedCompactionDebt(bytesFlushed),
 	}
+	for _, m := range d.mu.mem.queue {
+		pacerInfo.totalDirtyBytes += m.totalBytes()
+	}
 	d.mu.Unlock()
 
 	return pacerInfo
 }
 
 func (d *DB) getFlushPacerInfo() flushPacerInfo {
-	return flushPacerInfo{
-		totalBytes: d.memTableTotalBytes(),
+	var pacerInfo flushPacerInfo
+	d.mu.Lock()
+	for _, m := range d.mu.mem.queue {
+		pacerInfo.totalBytes += m.totalBytes()
 	}
+	d.mu.Unlock()
+	return pacerInfo
 }
 
 // maybeScheduleFlush schedules a flush if necessary.
@@ -1119,17 +1126,6 @@ func (d *DB) runCompaction(jobID int, c *compaction, pacer pacer) (
 	return ve, pendingOutputs, nil
 }
 
-// memTableTotalBytes returns the total number of bytes in the memtables. Note
-// that this includes the mutable memtable as well.
-func (d *DB) memTableTotalBytes() (totalBytes uint64) {
-	d.mu.Lock()
-	for _, m := range d.mu.mem.queue {
-		totalBytes += m.totalBytes()
-	}
-	d.mu.Unlock()
-	return totalBytes
-}
-
 // scanObsoleteFiles scans the filesystem for files that are no longer needed
 // and adds those to the internal lists of obsolete files. Note that he files
 // are not actually deleted by this method. A subsequent call to
diff --git a/pacer.go b/pacer.go
index 4b42d9c..d2e1dcb 100644
--- a/pacer.go
+++ b/pacer.go
@@ -73,6 +73,10 @@ type compactionPacerInfo struct {
 	// the threshold.
 	slowdownThreshold   uint64
 	totalCompactionDebt uint64
+	// totalDirtyBytes is the number of dirty bytes in memtables. The compaction
+	// pacer can monitor changes to this value to determine if user writes have
+	// stopped.
+	totalDirtyBytes uint64
 }
 
 // compactionPacerEnv defines the environment in which the compaction rate limiter
@@ -92,6 +96,7 @@ type compactionPacer struct {
 	internalPacer
 	env                 compactionPacerEnv
 	totalCompactionDebt uint64
+	totalDirtyBytes     uint64
 }
 
 func newCompactionPacer(env compactionPacerEnv) *compactionPacer {
@@ -121,8 +126,18 @@ func (p *compactionPacer) maybeThrottle(bytesIterated uint64) error {
 		pacerInfo := p.env.getInfo()
 		p.slowdownThreshold = pacerInfo.slowdownThreshold
 		p.totalCompactionDebt = pacerInfo.totalCompactionDebt
-		p.refreshBytesThreshold = bytesIterated + (p.env.memTableSize*5/100)
+		p.refreshBytesThreshold = bytesIterated + (p.env.memTableSize * 5 / 100)
 		p.iterCount = 1000
+		if p.totalDirtyBytes == pacerInfo.totalDirtyBytes {
+			// The total dirty bytes in the memtables have not changed since the
+			// previous call: user writes have completely stopped. Allow the
+			// compaction to proceed as fast as possible until the next
+			// recalculation. We adjust the recalculation threshold so that we can be
+			// nimble in the face of new user writes.
+			p.totalCompactionDebt += p.slowdownThreshold
+			p.iterCount = 100
+		}
+		p.totalDirtyBytes = pacerInfo.totalDirtyBytes
 	}
 	p.iterCount--
 
@@ -161,8 +176,9 @@ type flushPacerEnv struct {
 // rate limit is applied.
 type flushPacer struct {
 	internalPacer
-	env        flushPacerEnv
-	totalBytes uint64
+	env                flushPacerEnv
+	totalBytes         uint64
+	adjustedTotalBytes uint64
 }
 
 func newFlushPacer(env flushPacerEnv) *flushPacer {
@@ -170,7 +186,7 @@ func newFlushPacer(env flushPacerEnv) *flushPacer {
 		env: env,
 		internalPacer: internalPacer{
 			limiter:           env.limiter,
-			slowdownThreshold: env.memTableSize*105/100,
+			slowdownThreshold: env.memTableSize * 105 / 100,
 		},
 	}
 }
@@ -191,16 +207,26 @@ func (p *flushPacer) maybeThrottle(bytesIterated uint64) error {
 	// byte count requires grabbing DB.mu which is expensive.
 	if p.iterCount == 0 || bytesIterated > p.refreshBytesThreshold {
 		pacerInfo := p.env.getInfo()
-		p.totalBytes = pacerInfo.totalBytes
-		p.refreshBytesThreshold = bytesIterated + (p.env.memTableSize*5/100)
 		p.iterCount = 1000
+		p.refreshBytesThreshold = bytesIterated + (p.env.memTableSize * 5 / 100)
+		p.adjustedTotalBytes = pacerInfo.totalBytes
+		if p.totalBytes == pacerInfo.totalBytes {
+			// The total bytes in the memtables have not changed since the previous
+			// call: user writes have completely stopped. Allow the flush to proceed
+			// as fast as possible until the next recalculation. We adjust the
+			// recalculation threshold so that we can be nimble in the face of new
+			// user writes.
+			p.adjustedTotalBytes += p.slowdownThreshold
+			p.iterCount = 100
+		}
+		p.totalBytes = pacerInfo.totalBytes
 	}
 	p.iterCount--
 
 	// dirtyBytes is the total number of bytes in the memtables minus the number of
 	// bytes flushed. It represents unflushed bytes in all the memtables, even the
 	// ones which aren't being flushed such as the mutable memtable.
-	dirtyBytes := p.totalBytes - bytesIterated
+	dirtyBytes := p.adjustedTotalBytes - bytesIterated
 	flushAmount := bytesIterated - p.prevBytesIterated
 	p.prevBytesIterated = bytesIterated
 
@@ -212,7 +238,7 @@ func (p *flushPacer) maybeThrottle(bytesIterated uint64) error {
 	return p.limit(flushAmount, dirtyBytes)
 }
 
-type noopPacer struct {}
+type noopPacer struct{}
 
 func (p *noopPacer) maybeThrottle(_ uint64) error {
 	return nil


Test patch:
diff --git a/pacer_test.go b/pacer_test.go
index fd2a6f9..94e69e0 100644
--- a/pacer_test.go
+++ b/pacer_test.go
@@ -65,6 +65,7 @@ func TestCompactionPacerMaybeThrottle(t *testing.T) {
 				}
 
 				burst := uint64(1)
+				dirtyBytes := uint64(1)
 				var bytesIterated uint64
 				var currentTotal uint64
 				var slowdownThreshold uint64
@@ -87,6 +88,8 @@ func TestCompactionPacerMaybeThrottle(t *testing.T) {
 							bytesIterated = varValue
 						case "currentTotal":
 							currentTotal = varValue
+						case "dirtyBytes":
+							dirtyBytes = varValue
 						case "slowdownThreshold":
 							slowdownThreshold = varValue
 						default:
@@ -102,6 +105,7 @@ func TestCompactionPacerMaybeThrottle(t *testing.T) {
 						return compactionPacerInfo{
 							slowdownThreshold:   slowdownThreshold,
 							totalCompactionDebt: currentTotal,
+							totalDirtyBytes:     dirtyBytes,
 						}
 					}
 					compactionPacer := newCompactionPacer(compactionPacerEnv{
diff --git a/testdata/compaction_pacer_maybe_throttle b/testdata/compaction_pacer_maybe_throttle
index 63ccf16..db6cba4 100644
--- a/testdata/compaction_pacer_maybe_throttle
+++ b/testdata/compaction_pacer_maybe_throttle
@@ -119,6 +119,15 @@ slowdownThreshold: 10
 ----
 wait: 2
 
+init compaction
+burst: 10
+bytesIterated: 2
+currentTotal: 12
+dirtyBytes: 0
+slowdownThreshold: 10
+----
+allow: 2
+
 init compaction
 burst: 10
 bytesIterated: 3