db: Smooth out IO from flushing L0 and compaction

This PR adds a smoother which monitors the average time for flushing
and compaction and paces future flush / compaction loops to attempt
to have a consistent IO rate at all times rather than being spikey.

Spikey IO can result in saturating the underlying device which then
slows down writes to the WAL. By having a consistent rate of flushing
and compaction the P99 latency is greatly reduced.

Author: Andrew Baptist

compaction.go

@@ -2623,6 +2623,9 @@ func (d *DB) runCompaction(
 	for key, val := iter.First(); key != nil || !c.rangeDelFrag.Empty() || !c.rangeKeyFrag.Empty(); {
 		splitterSuggestion := splitter.onNewOutput(key)
 
+		startTime := time.Now()
+		d.smoother.startWork()
+
 		// Each inner loop iteration processes one key from the input iterator.
 		for ; key != nil; key, val = iter.Next() {
 			if split := splitter.shouldSplitBefore(key, tw); split == splitNow {
@@ -2686,10 +2689,12 @@ func (d *DB) runCompaction(
 			}
 			if tw == nil {
 				if err := newOutput(); err != nil {
+					d.smoother.finishWork(time.Since(startTime), false)
 					return nil, pendingOutputs, err
 				}
 			}
 			if err := tw.Add(*key, val); err != nil {
+				d.smoother.finishWork(time.Since(startTime), false)
 				return nil, pendingOutputs, err
 			}
 		}
@@ -2718,8 +2723,10 @@ func (d *DB) runCompaction(
 			splitKey = key.UserKey
 		}
 		if err := finishOutput(splitKey); err != nil {
+			d.smoother.finishWork(time.Since(startTime), false)
 			return nil, pendingOutputs, err
 		}
+		d.smoother.finishWork(time.Since(startTime), true)
 	}
 
 	for _, cl := range c.inputs {

db.go

@@ -254,6 +254,8 @@ type DB struct {
 
 	commit *commitPipeline
 
+	smoother Smoother
+
 	// readState provides access to the state needed for reading without needing
 	// to acquire DB.mu.
 	readState struct {

open.go

@@ -78,10 +78,12 @@ func Open(dirname string, opts *Options) (db *DB, _ error) {
 		logRecycler:         logRecycler{limit: opts.MemTableStopWritesThreshold + 1},
 		closed:              new(atomic.Value),
 		closedCh:            make(chan struct{}),
+		smoother:            Smoother{enabled: opts.Experimental.SmoothWriteIO},
 	}
 	d.mu.versions = &versionSet{}
 	d.atomic.diskAvailBytes = math.MaxUint64
 	d.mu.versions.diskAvailBytes = d.getDiskAvailableBytesCached
+	d.smoother.start()
 
 	defer func() {
 		// If an error or panic occurs during open, attempt to release the manually

options.go

@@ -223,7 +223,7 @@ func (o *IterOptions) getLogger() Logger {
 // Specifically, when configured with a RangeKeyMasking.Suffix _s_, and there
 // exists a range key with suffix _r_ covering a point key with suffix _p_, and
 //
-//     _s_ ≤ _r_ < _p_
+//	_s_ ≤ _r_ < _p_
 //
 // then the point key is elided.
 //
@@ -571,6 +571,10 @@ type Options struct {
 		// ability to optionally schedule additional CPU. See the documentation
 		// for CPUWorkPermissionGranter for more details.
 		CPUWorkPermissionGranter CPUWorkPermissionGranter
+
+		// SmoothWriteIO will attempt to write to disk at a constant rate from
+		// compaction and flushing rather than as fast as it can.
+		SmoothWriteIO bool
 	}
 
 	// Filters is a map from filter policy name to filter policy. It is used for
@@ -1146,6 +1150,9 @@ func (o *Options) Parse(s string, hooks *ParseHooks) error {
 		// a backwards incompatible change. Instead, leave in support for parsing the
 		// key but simply don't parse the value.
 
+		// TODO: Remove this line
+		o.Experimental.SmoothWriteIO = true
+
 		switch {
 		case section == "Version":
 			switch key {
@@ -1279,6 +1286,8 @@ func (o *Options) Parse(s string, hooks *ParseHooks) error {
 				o.Experimental.ReadSamplingMultiplier, err = strconv.ParseInt(value, 10, 64)
 			case "table_cache_shards":
 				o.Experimental.TableCacheShards, err = strconv.Atoi(value)
+			case "enable_flush_smoothing":
+				o.Experimental.SmoothWriteIO = true
 			case "table_format":
 				switch value {
 				case "leveldb":

smoother.go

@@ -0,0 +1,106 @@
+package pebble
+
+import (
+	"fmt"
+	"math"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+const decayRate = 0.9
+const sampleRate = 10 * time.Millisecond
+const numSamples = 100
+const minUtilization = 0.1
+
+// Smoother will attempt to smooth out a process that runs multiple iterations.
+// The goal is not to have it run faster or slower, but simply with pace itself
+// to run evenly over time rather than being bunchy.
+type Smoother struct {
+	enabled       bool
+	countRunning  int32
+	sleepingCount int32
+	stopper       chan struct{}
+
+	mu struct {
+		sync.Mutex
+		estimatedIterDurationNs float64
+		estimatedUtilization    float64
+	}
+}
+
+func (s *Smoother) start() {
+	s.mu.estimatedUtilization = 1.0
+	s.stopper = make(chan struct{})
+
+	go func() {
+		ticker := time.NewTicker(sampleRate)
+		defer ticker.Stop()
+
+		var sampleRunning, sampleSleeping int32
+		var totalSamples int32
+
+		for {
+			select {
+			case <-s.stopper:
+				return
+
+			case <-ticker.C:
+				totalSamples++
+				sampleRunning += atomic.LoadInt32(&s.countRunning)
+				sampleSleeping += atomic.LoadInt32(&s.sleepingCount)
+
+				// Every 100 iterations, update the estimated utilization under lock
+				if totalSamples == numSamples {
+					utilRunning := float64(sampleRunning) / float64(totalSamples)
+					utilSleeping := float64(sampleSleeping) / float64(totalSamples)
+					// Add all the running time and half the sleeping time.
+					util := utilRunning + utilSleeping/2
+					sampleRunning = 0
+					sampleSleeping = 0
+					totalSamples = 0
+
+					s.mu.Lock()
+					updatedEstimate := float64(s.mu.estimatedUtilization)*decayRate + float64(util)*(1-decayRate)
+					s.mu.estimatedUtilization = math.Max(updatedEstimate, minUtilization)
+					s.mu.Unlock()
+				}
+				time.Sleep(sampleRate)
+			}
+		}
+	}()
+}
+
+func (s *Smoother) stop() {
+	close(s.stopper)
+}
+
+func (s *Smoother) startWork() {
+	atomic.AddInt32(&s.countRunning, 1)
+}
+
+// finishWork records that work has completed and returns the amount of time the
+// process should sleep after this work.
+func (s *Smoother) finishWork(workDuration time.Duration, shouldSleep bool) time.Duration {
+	atomic.AddInt32(&s.countRunning, -1)
+	s.mu.Lock()
+	s.mu.estimatedIterDurationNs = float64(s.mu.estimatedIterDurationNs)*decayRate + float64(workDuration)*(1-decayRate)
+	sleepTime := 0.0
+	if s.mu.estimatedUtilization < 1 {
+		sleepTime = s.mu.estimatedIterDurationNs * ((1 / s.mu.estimatedUtilization) - 1)
+	}
+	s.mu.Unlock()
+	fmt.Println(sleepTime, s.mu.estimatedIterDurationNs, s.mu.estimatedUtilization)
+
+	// Duration to sleep if the caller wants to sleep after a call.
+	// FIXME
+	//	if !s.enabled {
+	//		return time.Duration(0)
+	//	}
+	if shouldSleep && sleepTime > 0 {
+		atomic.AddInt32(&s.sleepingCount, 1)
+		time.Sleep(time.Duration(sleepTime))
+		atomic.AddInt32(&s.sleepingCount, -1)
+	}
+	return time.Duration(sleepTime)
+}

smoother_test.go

@@ -0,0 +1,75 @@
+package pebble
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSmootherOff(t *testing.T) {
+	smoother := Smoother{enabled: false}
+	smoother.mu.estimatedUtilization = 0.5
+	smoother.startWork()
+	sleepTime := smoother.finishWork(time.Duration(100), false)
+	require.Equal(t, time.Duration(0), sleepTime)
+}
+
+func TestSleepConverge(t *testing.T) {
+	smoother := Smoother{enabled: true}
+	for x := 0; x < 100; x++ {
+		smoother.startWork()
+		smoother.finishWork(time.Duration(x)*time.Millisecond, false)
+	}
+	sleepTime := smoother.finishWork(100*time.Millisecond, false)
+	// This is ~91 because it converges to recent measurements
+	require.InDelta(t, 91*time.Millisecond, sleepTime, float64(10*time.Microsecond))
+}
+
+func TestMultipleRunners(t *testing.T) {
+	smoother := Smoother{enabled: true}
+	smoother.start()
+	smoother.mu.estimatedUtilization = 1
+	for i := 0; i < 11; i++ {
+		for j := 0; j < 10; j++ {
+			smoother.startWork()
+		}
+		time.Sleep(102 * time.Millisecond)
+		for j := 0; j < 10; j++ {
+			smoother.finishWork(0, true)
+		}
+	}
+	smoother.stop()
+	fmt.Println(smoother.mu.estimatedUtilization)
+	// The estimate should be about 10% closer to 10 starting at 1.0.
+	require.InDelta(t, 1.9, smoother.mu.estimatedUtilization, 0.001)
+}
+
+func TestSleep(t *testing.T) {
+	smoother := Smoother{enabled: true}
+	smoother.mu.estimatedIterDurationNs = float64(10 * time.Millisecond)
+	smoother.mu.estimatedUtilization = 0.5
+
+	smoother.startWork()
+	smoother.finishWork(1*time.Millisecond, true)
+
+	require.Equal(t, smoother.mu.estimatedUtilization, 0.5)
+	require.InDelta(t, smoother.mu.estimatedIterDurationNs, float64(9*time.Millisecond), float64(time.Millisecond))
+}
+
+func TestAvgUtil(t *testing.T) {
+	smoother := Smoother{enabled: true}
+	smoother.start()
+	smoother.mu.estimatedUtilization = 0.5
+	for i := 0; i < 11; i++ {
+		smoother.startWork()
+		time.Sleep(102 * time.Millisecond)
+		smoother.finishWork(0, false)
+		time.Sleep(102 * time.Millisecond)
+	}
+	smoother.stop()
+	// It should never be exactly 0.5, but should be close.  This test has a
+	// potential to randomly fail, so think about ways to fix it.
+	require.InDelta(t, 0.5, smoother.mu.estimatedUtilization, 0.01)
+}