Repository: petermattis/pebble
Pull request #164 (2019-06-18 20:08:38)

Add Pebble pacing simulation
Simulation which models the new proposed pacing mechanism for Pebble.

This should be compared with #159, which is a simulation to model the current pacing mechanism in RocksDB.

This mechanism takes inspiration from [bLSM](https://www.eecs.harvard.edu/~margo/cs165/papers/gp-lsm.pdf) and [ScyllaDB](https://www.scylladb.com/2018/12/04/worry-free-ingestion-flow-control/). In particular, the memtable pacing takes inspiration from bLSM by using a low and high watermark to throttle user writes and slow down memtable flushing, and inspiration from ScyllaDB to pace memtable flushing by looking at compaction debt to ensure that data is ingested as quickly as possible, but not any quicker.

The formula for throttling memtable flushing is:

**memtable_flush_rate = f(compaction_debt<sub>0</sub>/compaction_debt) * memtable_flush_rate<sub>0</sub>**

Where **compaction_debt<sub>0</sub>** is the compaction debt slowdown threshold and **memtable_flush_rate<sub>0</sub>** is the maximum configured memtable flush rate. Both of these will have to be configurable options in Pebble.

This formula guarantees that compaction debt does not grow indefinitely even without a hard stop limit such as `l0_stop_writes_threshold` or `compaction_debt_stop_writes_threshold`. Writes will therefore always slow down gradually instead of stopping abruptly.

An alternative approach I was thinking of was directly tying the memtable flush rate to the compaction rate. Because the memtable flushes smoothly, we could update compaction rate smoothly. We could directly tie the rate at which the memtable flushes to the rate that compaction debt is decreased, and maintain compaction debt between a low and high watermark just like we do with user writes and memtable flushing. However, the issue with this approach is that we do not have a good estimate on how much compaction debt each byte of memtable flushing adds.

If the DB only had write operations with new keys, this would be easy to calculate. The formula would be `compaction_debt_incurred = (num_levels-1) * memtable_flush_size`. However, this is hard to estimate in practice because of updates and deletes. Without a good estimate, there is no way to tie the number of bytes flushed from the memtable to the rate of compaction. This was easy to do with user writes and memtable flushing because the rate at which user writes come in compared to the rate at which the memtable is flushed can be maintained at a 1:1 rate, since 1 byte written means 1 byte flushed. However, we cannot accurately derive this ratio for memtable flushing and compactions.

Comparing this simulation with the RocksDB simulation is difficult because the RocksDB simulation relies heavily on tuning the constants depending on the workload. If the constants aren't tuned correctly, compaction debt grows indefinitely.

petermattis: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.png" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/petermattis/pebble/164)
<!-- Reviewable:end -->

petermattis: > Comparing this simulation with the RocksDB simulation is difficult because the RocksDB simulation relies heavily on tuning the constants depending on the workload. If the constants aren't tuned correctly, compaction debt grows indefinitely.

I won't be able to look at this in depth until next week. I'm wondering if you can elaborate on this paragraph in the mean time. Does the pacing mechanism in this PR appear to be better or worse than the RocksDB mechanisms? The fewer tunables seems to be a desirable property. Is there any better behavior exhibited in the simulation?
Ryanfsdf: > I won't be able to look at this in depth until next week. I'm wondering if you can elaborate on this paragraph in the mean time. Does the pacing mechanism in this PR appear to be better or worse than the RocksDB mechanisms? The fewer tunables seems to be a desirable property. Is there any better behavior exhibited in the simulation?

The pacing mechanism in this PR appears to be much better than the RocksDB mechanism. My worry is that I'm not doing the RocksDB mechanism justice because their mechanism seems much more sensitive to choices of constants.

For example, from the RocksDB simulation:
```
// From RocksDB:
// If the DB recovers from delay conditions, we reward with reducing
// double the slowdown ratio. This is to balance the long term slowdown
// increase signal.
drainLimit := db.writeLimiter.Limit() * 1.4
db.writeLimiter.SetLimit(drainLimit)
```
With the constant `1.4`, which is what RocksDB uses, compaction debt grows indefinitely in the RocksDB simulation. If this constant is changed to `1.1`, the RocksDB simulation shows better results and compaction debt doesn't grow indefinitely. However, the simulation seemed to overthrottle the write rate instead.

Better behavior exhibited in the new proposed mechanism include:
1) The system accepts writes at the maximum possible rate that the system can handle and no faster. RocksDB doesn't provide this sort of guarantee with their mechanism. In the new simulation, the write rate always stabilized at the maximum possible write rate (which I calculated). This wasn't the case for the RocksDB simulation. It would either overthrottle or underthrottle depending on the constants chosen.
2) With RocksDB's simulated mechanism, compaction debt grew indefinitely in some cases. With the new mechanism, it's not possible for the compaction debt to grow indefinitely because the throttling applied is directly proportional to the size of the compaction debt.
3) Throttling is applied much smoother in the new mechanism. RocksDB multiplies the write rate by `0.8` and `1/0.8`, which is very coarse. The write rate would be jumpy whereas the new mechanism was fairly stable.
petermattis: The early part of this week is jammed for me. I'm not going to be able to look at this PR until Wednesday or Thursday.

> The pacing mechanism in this PR appears to be much better than the RocksDB mechanism. My worry is that I'm not doing the RocksDB mechanism justice because their mechanism seems much more sensitive to choices of constants.

Sensitivity to constants is a negative in itself.

Can you make a chart or charts showing the better behavior from the mechanisms in this PR?

Concurrently, I think you can start work on the real implementation in Pebble (if you haven't done so already).
petermattis: > These days after reading Luo's comments I am less confident about throttling user writes, due to each successive throttled write experiencing a longer queueing delay. Maybe we should also be thinking about how we can prevent user write throttling conditions from being hit. But those would be separate discussions.

Can you expand on this? We have to throttle user writes at some point. Not throttling them at all could lead to either an explosion of memtables or an explosion of L0 files. At some point, the only tool we have is to delay how long a user-write takes. We either do this explicitly with a rate limit or quota pool, or implicitly by stopping user-writes completely when a memtable or L0 limit is reached. Am I missing some other tool that is available?
ajkr: > Can you expand on this? We have to throttle user writes at some point. Not throttling them at all could lead to either an explosion of memtables or an explosion of L0 files. At some point, the only tool we have is to delay how long a user-write takes. We either do this explicitly with a rate limit or quota pool, or implicitly by stopping user-writes completely when a memtable or L0 limit is reached. Am I missing some other tool that is available?

Right I agree there needs to be limits on memtables/L0 files/pending compaction bytes to prevent memory/read-amp/space-amp explosion. They can be either explicit like RocksDB's configuration options or implicit like how this simulation limits pending compaction bytes.

For gradual slowdown or pacing of user writes vs. hard stop at the limits, I am not so sure anymore. Here is the first example that came to mind to use for comparison.

Setup:
- Each memtable is 50MB, limit is two, say at t=0s we're at 50% (one memtable is full and can start flushing, the other is empty)
- Flush rate is 5MB/s
- User writing at 10MB/s

Pacing:
- At t=0s, no throttling necessary
- At t=1s, we've flushed 5MB but dirtied 10MB. Now our dirty bytes are at 1.1*memtable size so we throttle user write rate to 5MB/s
- Each second after t=1s we accumulate 5MB of slowed down writes. So at t=n seconds for n >= 1, a new write would be behind 5MB*(n-1) existing delayed writes, so the new write gets delayed n-1 seconds.

Hard limit:
- At t=0s there is no wait. The first memtable flush starts
- At t=5s the second memtable fills and we stop accepting writes.
- At t=10s the first memtable flush finishes and the second begins. There are 50MB backlogged writes accumulated that can now all be written at once to the third memtable. Those writes saw at most 5 second delays. Now we are already at the stop limit again.
- At t=20s the second memtable flush finishes and the third begins. There are 100MB backlogged writes accumulated. 50MB can be written at once to the fourth memtable. Those writes experienced at most 10 second delays. There are 50MB more still blocked.
- As long as this continues the writes between t=5*n and t=5*(n+1) seconds get written to memtable at t=10*n seconds, where n >= 1. Their delay is at most 5*n seconds.

I think this shows it's not totally obvious that pacing/slowdown is clearly better than having hard limits after considering queueing delays. Regardless of which way we go, though, I'd expect we will try to avoid the slowdown / limiting conditions at all costs. For example, allocating more resources to flush/compaction (parallelizing them), or even redirecting excess traffic to other replicas.

Another risk I thought of with pacing/gradual slowdown is it will apply in scenarios where the workload naturally wouldn't have hit the limits. That is, maybe a workload naturally almost reaches the limit then slows down and doesn't hit the limit. With pacing or gradual slowdown we'd likely apply artificial delay in this case, which hurts tail latencies but wasn't necessary.
petermattis: > Right I agree there needs to be limits on memtables/L0 files/pending compaction bytes to prevent memory/read-amp/space-amp explosion. They can be either explicit like RocksDB's configuration options or implicit like how this simulation limits pending compaction bytes.
> 
> For gradual slowdown or pacing of user writes vs. hard stop at the limits, I am not so sure anymore. Here is the first example that came to mind to use for comparison.

Ack. I've come around to this view. We need a limit for the system to be able to protect itself, but it can be a hard limit, and not a slowdown. This would be a nice simplification of the configuration options, and the corresponding implementation.

Patch:
diff --git a/compaction.go b/compaction.go
index 3d93c57..b09bebd 100644
--- a/compaction.go
+++ b/compaction.go
@@ -1037,7 +1037,35 @@ func (d *DB) runCompaction(c *compaction) (
 // 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
 // deleteObsoleteFiles must be performed.
-func (d *DB) scanObsoleteFiles(list []string) {
+//
+// d.mu must be held when calling this, but the mutex may be dropped and
+// re-acquired during the course of this method.
+func (d *DB) scanObsoleteFiles() {
+	// Release d.mu while doing I/O
+	// Note the unusual order: Unlock and then Lock.
+	d.mu.Unlock()
+	defer d.mu.Lock()
+
+	fs := d.opts.FS
+	list, err := fs.List(d.dirname)
+	if err != nil {
+		// Ignore any filesystem errors.
+		return
+	}
+
+	if d.dirname != d.walDirname {
+		list2, err := fs.List(d.walDirname)
+		if err != nil {
+			// Ignore any filesystem errors.
+		} else {
+			list = append(list, list2...)
+		}
+	}
+
+	// Grab d.mu again in order to get a snapshot of the live state. Note that we
+	// need to this after the directory list because after releasing the lock
+	// again new files can be created.
+	d.mu.Lock()
 	liveFileNums := make(map[uint64]struct{}, len(d.mu.compact.pendingOutputs))
 	for fileNum := range d.mu.compact.pendingOutputs {
 		liveFileNums[fileNum] = struct{}{}
@@ -1045,6 +1073,7 @@ func (d *DB) scanObsoleteFiles(list []string) {
 	d.mu.versions.addLiveFileNums(liveFileNums)
 	logNumber := d.mu.versions.logNumber
 	manifestFileNumber := d.mu.versions.manifestFileNumber
+	d.mu.Unlock()
 
 	var obsoleteLogs []uint64
 	var obsoleteTables []uint64
@@ -1084,10 +1113,12 @@ func (d *DB) scanObsoleteFiles(list []string) {
 		}
 	}
 
+	d.mu.Lock()
 	d.mu.log.queue = merge(d.mu.log.queue, obsoleteLogs)
 	d.mu.versions.obsoleteTables = merge(d.mu.versions.obsoleteTables, obsoleteTables)
 	d.mu.versions.obsoleteManifests = merge(d.mu.versions.obsoleteManifests, obsoleteManifests)
 	d.mu.versions.obsoleteOptions = merge(d.mu.versions.obsoleteOptions, obsoleteOptions)
+	d.mu.Unlock()
 }
 
 // deleteObsoleteFiles deletes those files that are no longer needed.
@@ -1158,40 +1189,36 @@ func (d *DB) deleteObsoleteFiles(jobID int) {
 
 			path := dbFilename(d.dirname, f.fileType, fileNum)
 			err := d.opts.FS.Remove(path)
-			if err == os.ErrNotExist {
-				continue
-			}
 
-			// TODO(peter): need to handle this errror, probably by re-adding the
-			// file that couldn't be deleted to one of the obsolete slices map.
-
-			switch f.fileType {
-			case fileTypeLog:
-				if d.opts.EventListener.WALDeleted != nil {
-					d.opts.EventListener.WALDeleted(WALDeleteInfo{
-						JobID:   jobID,
-						Path:    path,
-						FileNum: fileNum,
-						Err:     err,
-					})
-				}
-			case fileTypeManifest:
-				if d.opts.EventListener.ManifestDeleted != nil {
-					d.opts.EventListener.ManifestDeleted(ManifestDeleteInfo{
-						JobID:   jobID,
-						Path:    path,
-						FileNum: fileNum,
-						Err:     err,
-					})
-				}
-			case fileTypeTable:
-				if d.opts.EventListener.TableDeleted != nil {
-					d.opts.EventListener.TableDeleted(TableDeleteInfo{
-						JobID:   jobID,
-						Path:    path,
-						FileNum: fileNum,
-						Err:     err,
-					})
+			if err != os.ErrNotExist {
+				switch f.fileType {
+				case fileTypeLog:
+					if d.opts.EventListener.WALDeleted != nil {
+						d.opts.EventListener.WALDeleted(WALDeleteInfo{
+							JobID:   jobID,
+							Path:    path,
+							FileNum: fileNum,
+							Err:     err,
+						})
+					}
+				case fileTypeManifest:
+					if d.opts.EventListener.ManifestDeleted != nil {
+						d.opts.EventListener.ManifestDeleted(ManifestDeleteInfo{
+							JobID:   jobID,
+							Path:    path,
+							FileNum: fileNum,
+							Err:     err,
+						})
+					}
+				case fileTypeTable:
+					if d.opts.EventListener.TableDeleted != nil {
+						d.opts.EventListener.TableDeleted(TableDeleteInfo{
+							JobID:   jobID,
+							Path:    path,
+							FileNum: fileNum,
+							Err:     err,
+						})
+					}
 				}
 			}
 		}
diff --git a/internal/pacertoy/pebble/main.go b/internal/pacertoy/pebble/main.go
new file mode 100644
index 0000000..025125e
--- /dev/null
+++ b/internal/pacertoy/pebble/main.go
@@ -0,0 +1,435 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"os"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/petermattis/pebble/internal/rate"
+
+	"golang.org/x/exp/rand"
+)
+
+const (
+	// If measureLatency is true, the simulator outputs p50, p95, and p99
+	// latencies after all writes are completed. In this mode, all writes
+	// are immediately queued. If this is disabled, writes come in continually
+	// at different rates.
+	measureLatency        = false
+
+	writeAmount           = 2000 << 20 // 2 GB
+
+	// Max rate for all compactions. This is intentionally set low enough that
+	// user writes will have to be delayed.
+	maxCompactionRate     = 100 << 20 // 100 MB/s
+	minCompactionRate     = 20 << 20 // 20 MB/s
+
+	memtableSize          = 64 << 20 // 64 MB
+	maxMemtableCount      = 5
+	drainDelayThreshold   = 1.05 * memtableSize
+	maxFlushRate          = 30 << 20 // 30 MB/s
+	minFlushRate          = 4 << 20 // 4 MB/s
+
+	l0CompactionThreshold = 1
+
+	levelRatio            = 10
+	numLevels             = 7
+
+	compactionDebtSlowdownThreshold = 2 * memtableSize
+)
+
+type compactionPacer struct {
+	level      int64
+	maxDrainer *rate.Limiter
+	minDrainer *rate.Limiter
+}
+
+func newCompactionPacer() *compactionPacer {
+	p := &compactionPacer{
+		maxDrainer: rate.NewLimiter(maxCompactionRate, maxCompactionRate),
+		minDrainer: rate.NewLimiter(minCompactionRate, minCompactionRate),
+	}
+	return p
+}
+
+func (p *compactionPacer) fill(n int64) {
+	atomic.AddInt64(&p.level, n)
+}
+
+func (p *compactionPacer) drain(n int64, delay bool) bool {
+	p.maxDrainer.WaitN(context.Background(), int(n))
+
+	if delay {
+		p.minDrainer.WaitN(context.Background(), int(n))
+	}
+	level := atomic.AddInt64(&p.level, -n)
+	return level <= compactionDebtSlowdownThreshold
+}
+
+type flushPacer struct {
+	level             int64
+	drainDelayLevel   float64
+	fillCond          sync.Cond
+	// minDrainer is the drainer which sets the minimum speed of draining.
+	minDrainer        *rate.Limiter
+	// maxDrainer is the drainer which sets the maximum speed of draining.
+	maxDrainer        *rate.Limiter
+}
+
+func newFlushPacer(mu *sync.Mutex) *flushPacer {
+	p := &flushPacer{
+		drainDelayLevel: drainDelayThreshold,
+		minDrainer:      rate.NewLimiter(minFlushRate, minFlushRate),
+		maxDrainer:      rate.NewLimiter(maxFlushRate, maxFlushRate),
+	}
+	p.fillCond.L = mu
+	return p
+}
+
+func (p *flushPacer) fill(n int64) {
+	atomic.AddInt64(&p.level, n)
+	p.fillCond.Signal()
+}
+
+func (p *flushPacer) drain(n int64, delay bool) bool {
+	p.maxDrainer.WaitN(context.Background(), int(n))
+
+	if delay {
+		p.minDrainer.WaitN(context.Background(), int(n))
+	}
+	level := atomic.AddInt64(&p.level, -n)
+	p.fillCond.Signal()
+	return float64(level) <= p.drainDelayLevel
+}
+
+type DB struct {
+	mu         sync.Mutex
+	flushPacer *flushPacer
+	flushCond  sync.Cond
+	memtables  []*int64
+	fill       int64
+	drain      int64
+
+	compactionMu        sync.Mutex
+	compactionPacer     *compactionPacer
+	// L0 is represented as an array of integers whereas every other level
+	// is represented as a single integer.
+	L0                  []*int64
+	// Non-L0 sstables. sstables[0] == L1.
+	sstables            []int64
+	maxSSTableSizes     []int64
+	compactionFlushCond sync.Cond
+	prevCompactionDebt  float64
+}
+
+func newDB() *DB {
+	db := &DB{}
+	db.flushPacer = newFlushPacer(&db.mu)
+	db.flushCond.L = &db.mu
+	db.memtables = append(db.memtables, new(int64))
+
+	db.compactionFlushCond.L = &db.compactionMu
+	db.L0 = append(db.L0, new(int64))
+	db.compactionPacer = newCompactionPacer()
+
+	db.maxSSTableSizes = make([]int64, numLevels-1)
+	base := int64(levelRatio)
+	for i := uint64(0); i < numLevels-2; i++ {
+		// Each level is 10 times larger than the one above it.
+		db.maxSSTableSizes[i] = memtableSize * l0CompactionThreshold * base
+		base *= levelRatio
+
+		// Begin with each level full.
+		newLevel := db.maxSSTableSizes[i]
+
+		db.sstables = append(db.sstables, newLevel)
+	}
+	db.sstables = append(db.sstables, 0)
+	db.maxSSTableSizes[numLevels-2] = math.MaxInt64
+
+	go db.drainMemtable()
+	go db.drainCompaction()
+
+	return db
+}
+
+// drainCompaction simulates background compactions.
+func (db *DB) drainCompaction() {
+	rng := rand.New(rand.NewSource(1))
+
+	for {
+		db.compactionMu.Lock()
+
+		for len(db.L0) <= l0CompactionThreshold {
+			db.compactionFlushCond.Wait()
+		}
+		l0Table := db.L0[0]
+		db.compactionMu.Unlock()
+
+		var delay bool
+		for i, size := int64(0), int64(0); i < *l0Table; i += size {
+			size = 10000 + rng.Int63n(500)
+			if size > (*l0Table - i) {
+				size = *l0Table - i
+			}
+			delay = db.compactionPacer.drain(size, delay)
+		}
+
+		db.compactionMu.Lock()
+		db.L0 = db.L0[1:]
+		db.compactionMu.Unlock()
+
+		singleTableSize := int64(memtableSize)
+		tablesToCompact := 0
+		for i := range db.sstables {
+			newSSTableSize := atomic.AddInt64(&db.sstables[i], singleTableSize)
+			if newSSTableSize > db.maxSSTableSizes[i] {
+				atomic.AddInt64(&db.sstables[i], -singleTableSize)
+				tablesToCompact++
+			} else {
+				// Lower levels do not need compaction if level above it did not
+				// need compaction.
+				break
+			}
+		}
+
+		totalCompactionBytes := int64(tablesToCompact * memtableSize)
+
+		for t := 0; t < tablesToCompact; t++ {
+			db.compactionPacer.fill(memtableSize)
+			for i, size := int64(0), int64(0); i < memtableSize; i += size {
+				size = 10000 + rng.Int63n(500)
+				if size > (totalCompactionBytes - i) {
+					size = totalCompactionBytes - i
+				}
+				delay = db.compactionPacer.drain(size, delay)
+			}
+
+			db.delayMemtableDrain()
+		}
+	}
+}
+
+// fillCompaction fills L0 sstables.
+func (db *DB) fillCompaction(size int64) {
+	db.compactionMu.Lock()
+
+	db.compactionPacer.fill(size)
+
+	last := db.L0[len(db.L0)-1]
+	if *last+size > memtableSize {
+		last = new(int64)
+		db.L0 = append(db.L0, last)
+		db.compactionFlushCond.Signal()
+	}
+	*last += size
+
+	db.compactionMu.Unlock()
+}
+
+// drainMemtable simulates memtable flushing.
+func (db *DB) drainMemtable() {
+	rng := rand.New(rand.NewSource(2))
+
+	for {
+		db.mu.Lock()
+		for len(db.memtables) <= 1 {
+			db.flushCond.Wait()
+		}
+		memtable := db.memtables[0]
+		db.mu.Unlock()
+
+		var delay bool
+		for i, size := int64(0), int64(0); i < *memtable; i += size {
+			size = 1000 + rng.Int63n(50)
+			if size > (*memtable - i) {
+				size = *memtable - i
+			}
+			delay = db.flushPacer.drain(size, delay)
+			atomic.AddInt64(&db.drain, size)
+
+			db.fillCompaction(size)
+		}
+
+		db.delayMemtableDrain()
+
+		db.mu.Lock()
+		db.memtables = db.memtables[1:]
+		db.mu.Unlock()
+	}
+}
+
+// delayMemtableDrain applies memtable drain delays depending on compaction debt.
+func (db *DB) delayMemtableDrain() {
+	totalCompactionBytes := atomic.LoadInt64(&db.compactionPacer.level)
+	compactionDebt := math.Max(float64(totalCompactionBytes)-l0CompactionThreshold*memtableSize, 0.0)
+
+	db.mu.Lock()
+	if compactionDebt > compactionDebtSlowdownThreshold {
+		// Compaction debt is above the threshold and the debt is growing. Throttle memtable flushing.
+		drainLimit := maxFlushRate * rate.Limit(compactionDebtSlowdownThreshold/compactionDebt)
+		if drainLimit > 0 && drainLimit <= maxFlushRate {
+			db.flushPacer.maxDrainer.SetLimit(drainLimit)
+		}
+	} else {
+		// Continuously speed up memtable flushing to make sure that slowdown signal did not
+		// decrease the memtable flush rate by too much.
+		drainLimit := db.flushPacer.maxDrainer.Limit() * 1.05
+		if drainLimit > 0 && drainLimit <= maxFlushRate {
+			db.flushPacer.maxDrainer.SetLimit(drainLimit)
+		}
+	}
+
+	db.prevCompactionDebt = compactionDebt
+	db.mu.Unlock()
+}
+
+// fillMemtable simulates memtable filling.
+func (db *DB) fillMemtable(size int64) {
+	db.mu.Lock()
+
+	for len(db.memtables) > maxMemtableCount {
+		db.flushPacer.fillCond.Wait()
+	}
+	db.flushPacer.fill(size)
+	atomic.AddInt64(&db.fill, size)
+
+	last := db.memtables[len(db.memtables)-1]
+	if *last+size > memtableSize {
+		last = new(int64)
+		db.memtables = append(db.memtables, last)
+		db.flushCond.Signal()
+	}
+	*last += size
+
+	db.mu.Unlock()
+}
+
+// printLevels prints the levels.
+func (db *DB) printLevels() {
+	db.mu.Lock()
+	for i := range db.sstables {
+		fmt.Printf("Level %d: %d/%d\n", i+1, db.sstables[i]/(1024*1024), db.maxSSTableSizes[i]/(1024*1024))
+	}
+	db.mu.Unlock()
+}
+
+// simulateWrite simulates user writes.
+func simulateWrite(db *DB, measureLatencyMode bool) {
+	limiter := rate.NewLimiter(10<<20, 10<<20) // 10 MB/s
+	fmt.Printf("filling at 10 MB/sec\n")
+
+	setRate := func(mb int) {
+		fmt.Printf("filling at %d MB/sec\n", mb)
+		limiter.SetLimit(rate.Limit(mb << 20))
+	}
+
+	if !measureLatencyMode {
+		go func() {
+			rng := rand.New(rand.NewSource(3))
+			for {
+				secs := 5 + rng.Intn(5)
+				time.Sleep(time.Duration(secs) * time.Second)
+				mb := 10 + rng.Intn(20)
+				setRate(mb)
+			}
+		}()
+	}
+
+	rng := rand.New(rand.NewSource(uint64(4)))
+
+	totalWrites := int64(0)
+	percentiles := []int64{50, 95, 99}
+	percentileIndex := 0
+	percentileTimes := make([]time.Time, 0)
+
+	startTime := time.Now()
+	for totalWrites <= writeAmount {
+		size := 1000 + rng.Int63n(50)
+		if !measureLatencyMode {
+			limiter.WaitN(context.Background(), int(size))
+		}
+		db.fillMemtable(size)
+
+		// Calculate latency percentiles
+		totalWrites += size
+		if percentileIndex < len(percentiles) && totalWrites > (percentiles[percentileIndex] * writeAmount / 100) {
+			percentileTimes = append(percentileTimes, time.Now())
+			percentileIndex += 1
+		}
+	}
+
+	time.Sleep(time.Second * 10)
+	// Latency should only be measured when `limiter.WaitN` is removed.
+	if measureLatencyMode {
+		fmt.Printf("_____p50______p95______p99\n")
+		fmt.Printf("%8s %8s %8s\n",
+			time.Duration(percentileTimes[0].Sub(startTime).Seconds())*time.Second,
+			time.Duration(percentileTimes[1].Sub(startTime).Seconds())*time.Second,
+			time.Duration(percentileTimes[2].Sub(startTime).Seconds())*time.Second)
+	}
+
+	os.Exit(0)
+}
+
+func main() {
+	db := newDB()
+
+	go simulateWrite(db, measureLatency)
+
+	tick := time.NewTicker(time.Second)
+	start := time.Now()
+	lastNow := start
+	var lastFill, lastDrain int64
+
+	for i := 0; ; i++ {
+		select {
+		case <-tick.C:
+			if (i % 20) == 0 {
+				fmt.Printf("_elapsed___memtbs____dirty_____fill____drain____cdebt__l0count___max-f-rate\n")
+			}
+
+			if (i % 7) == 0 {
+				//db.printLevels()
+			}
+
+			db.mu.Lock()
+			memtableCount := len(db.memtables)
+			db.mu.Unlock()
+			dirty := atomic.LoadInt64(&db.flushPacer.level)
+			fill := atomic.LoadInt64(&db.fill)
+			drain := atomic.LoadInt64(&db.drain)
+
+			db.compactionMu.Lock()
+			compactionL0 := len(db.L0)
+			db.compactionMu.Unlock()
+			totalCompactionBytes := atomic.LoadInt64(&db.compactionPacer.level)
+			compactionDebt := math.Max(float64(totalCompactionBytes)-l0CompactionThreshold*memtableSize, 0.0)
+			maxFlushRate := db.flushPacer.maxDrainer.Limit()
+
+			now := time.Now()
+			elapsed := now.Sub(lastNow).Seconds()
+			fmt.Printf("%8s %8d %8.1f %8.1f %8.1f %8.1f %8d %12.1f\n",
+				time.Duration(now.Sub(start).Seconds()+0.5)*time.Second,
+				memtableCount,
+				float64(dirty)/(1024.0*1024.0),
+				float64(fill-lastFill)/(1024.0*1024.0*elapsed),
+				float64(drain-lastDrain)/(1024.0*1024.0*elapsed),
+				compactionDebt/(1024.0*1024.0),
+				compactionL0,
+				maxFlushRate/(1024.0*1024.0))
+
+			lastNow = now
+			lastFill = fill
+			lastDrain = drain
+		}
+	}
+}
diff --git a/internal/pacertoy/main.go b/internal/pacertoy/rocksdb/main.go
similarity index 99%
rename from internal/pacertoy/main.go
rename to internal/pacertoy/rocksdb/main.go
index f08e812..698e8e1 100644
--- a/internal/pacertoy/main.go
+++ b/internal/pacertoy/rocksdb/main.go
@@ -19,7 +19,7 @@ import (
 
 const (
 	// Max rate for all compactions. This is intentionally set low enough that
-	// user writes will have to be delayed
+	// user writes will have to be delayed.
 	maxCompactionRate     = 80 << 20 // 80 MB/s
 
 	memtableSize          = 64 << 20 // 64 MB
@@ -315,7 +315,7 @@ func (db *DB) printLevels() {
 
 // simulateWrite simulates user writes.
 func simulateWrite(db *DB) {
-	limiter := rate.NewLimiter(10<<20, 10<<20) // 10 mb
+	limiter := rate.NewLimiter(10<<20, 10<<20) // 10 MB/s
 	fmt.Printf("filling at 10 MB/sec\n")
 
 	setRate := func(mb int) {
diff --git a/open.go b/open.go
index 5d4475c..b493590 100644
--- a/open.go
+++ b/open.go
@@ -150,13 +150,6 @@ func Open(dirname string, opts *Options) (*DB, error) {
 	if err != nil {
 		return nil, err
 	}
-	if d.dirname != d.walDirname {
-		ls2, err := opts.FS.List(d.dirname)
-		if err != nil {
-			return nil, err
-		}
-		ls = append(ls, ls2...)
-	}
 
 	// Replay any newer log files than the ones named in the manifest.
 	type fileNumAndName struct {
@@ -234,7 +227,7 @@ func Open(dirname string, opts *Options) (*DB, error) {
 
 	jobID := d.mu.nextJobID
 	d.mu.nextJobID++
-	d.scanObsoleteFiles(ls)
+	d.scanObsoleteFiles()
 	d.deleteObsoleteFiles(jobID)
 	d.maybeScheduleFlush()
 	d.maybeScheduleCompaction()
diff --git a/version_set.go b/version_set.go
index f2e5b92..739bd9c 100644
--- a/version_set.go
+++ b/version_set.go
@@ -207,14 +207,6 @@ func (vs *versionSet) logAndApply(jobID int, ve *versionEdit, dir vfs.File) erro
 
 		if newManifestFileNumber != 0 {
 			if err := vs.createManifest(vs.dirname, newManifestFileNumber); err != nil {
-				if vs.opts.EventListener.ManifestCreated != nil {
-					vs.opts.EventListener.ManifestCreated(ManifestCreateInfo{
-						JobID:   jobID,
-						Path:    dbFilename(vs.dirname, fileTypeManifest, newManifestFileNumber),
-						FileNum: newManifestFileNumber,
-						Err:     err,
-					})
-				}
 				return err
 			}
 		}
@@ -223,37 +215,28 @@ func (vs *versionSet) logAndApply(jobID int, ve *versionEdit, dir vfs.File) erro
 		if err != nil {
 			return err
 		}
-		// NB: Any error from this point on is considered fatal as we don't now if
-		// the MANIFEST write occurred or not. Trying to determine that is
-		// fraught. Instead we rely on the standard recovery mechanism run when a
-		// database is open. In particular, that mechanism generates a new MANIFEST
-		// and ensures it is synced.
 		if err := ve.encode(w); err != nil {
-			vs.opts.Logger.Fatalf("MANIFEST write failed: %v", err)
 			return err
 		}
 		if err := vs.manifest.Flush(); err != nil {
-			vs.opts.Logger.Fatalf("MANIFEST flush failed: %v", err)
 			return err
 		}
 		if err := vs.manifestFile.Sync(); err != nil {
-			vs.opts.Logger.Fatalf("MANIFEST sync failed: %v", err)
 			return err
 		}
 		if newManifestFileNumber != 0 {
 			if err := setCurrentFile(vs.dirname, vs.fs, newManifestFileNumber); err != nil {
-				vs.opts.Logger.Fatalf("MANIFEST set current failed: %v", err)
 				return err
 			}
 			if err := dir.Sync(); err != nil {
-				vs.opts.Logger.Fatalf("MANIFEST dirsync failed: %v", err)
 				return err
 			}
-			if vs.opts.EventListener.ManifestCreated != nil {
+			if vs.opts.EventListener.ManifestDeleted != nil {
 				vs.opts.EventListener.ManifestCreated(ManifestCreateInfo{
 					JobID:   jobID,
 					Path:    dbFilename(vs.dirname, fileTypeManifest, newManifestFileNumber),
 					FileNum: newManifestFileNumber,
+					Err:     err,
 				})
 			}
 		}
@@ -307,6 +290,7 @@ func (vs *versionSet) createManifest(dirname string, fileNum uint64) (err error)
 			manifest.Close()
 		}
 		if manifestFile != nil {
+			manifestFile.Sync()
 			manifestFile.Close()
 		}
 		if err != nil {


Test patch:
diff --git a/error_test.go b/error_test.go
deleted file mode 100644
index 710de8b..0000000
--- a/error_test.go
+++ /dev/null
@@ -1,242 +0,0 @@
-// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
-// of this source code is governed by a BSD-style license that can be found in
-// the LICENSE file.
-
-package pebble
-
-import (
-	"fmt"
-	"io"
-	"os"
-	"sync/atomic"
-	"testing"
-
-	"github.com/petermattis/pebble/vfs"
-)
-
-type panicLogger struct{}
-
-func (l panicLogger) Infof(format string, args ...interface{}) {
-}
-
-func (l panicLogger) Fatalf(format string, args ...interface{}) {
-	panic(fmt.Errorf("fatal: "+format, args...))
-}
-
-type errorFS struct {
-	fs    vfs.FS
-	index int32
-}
-
-func newErrorFS(index int32) *errorFS {
-	return &errorFS{
-		fs:    vfs.NewMem(),
-		index: index,
-	}
-}
-
-func (fs *errorFS) maybeError() error {
-	if atomic.AddInt32(&fs.index, -1) == -1 {
-		return fmt.Errorf("injected error")
-	}
-	return nil
-}
-
-func (fs *errorFS) Create(name string) (vfs.File, error) {
-	if err := fs.maybeError(); err != nil {
-		return nil, err
-	}
-	f, err := fs.fs.Create(name)
-	if err != nil {
-		return nil, err
-	}
-	return errorFile{f, fs}, nil
-}
-
-func (fs *errorFS) Link(oldname, newname string) error {
-	if err := fs.maybeError(); err != nil {
-		return err
-	}
-	return fs.fs.Link(oldname, newname)
-}
-
-func (fs *errorFS) Open(name string) (vfs.File, error) {
-	if err := fs.maybeError(); err != nil {
-		return nil, err
-	}
-	f, err := fs.fs.Open(name)
-	if err != nil {
-		return nil, err
-	}
-	return errorFile{f, fs}, nil
-}
-
-func (fs *errorFS) OpenDir(name string) (vfs.File, error) {
-	if err := fs.maybeError(); err != nil {
-		return nil, err
-	}
-	f, err := fs.fs.OpenDir(name)
-	if err != nil {
-		return nil, err
-	}
-	return errorFile{f, fs}, nil
-}
-
-func (fs *errorFS) Remove(name string) error {
-	if _, err := fs.fs.Stat(name); os.IsNotExist(err) {
-		return nil
-	}
-
-	if err := fs.maybeError(); err != nil {
-		return err
-	}
-	return fs.fs.Remove(name)
-}
-
-func (fs *errorFS) Rename(oldname, newname string) error {
-	if err := fs.maybeError(); err != nil {
-		return err
-	}
-	return fs.fs.Rename(oldname, newname)
-}
-
-func (fs *errorFS) MkdirAll(dir string, perm os.FileMode) error {
-	if err := fs.maybeError(); err != nil {
-		return err
-	}
-	return fs.fs.MkdirAll(dir, perm)
-}
-
-func (fs *errorFS) Lock(name string) (io.Closer, error) {
-	if err := fs.maybeError(); err != nil {
-		return nil, err
-	}
-	return fs.fs.Lock(name)
-}
-
-func (fs *errorFS) List(dir string) ([]string, error) {
-	if err := fs.maybeError(); err != nil {
-		return nil, err
-	}
-	return fs.fs.List(dir)
-}
-
-func (fs *errorFS) Stat(name string) (os.FileInfo, error) {
-	if err := fs.maybeError(); err != nil {
-		return nil, err
-	}
-	return fs.fs.Stat(name)
-}
-
-type errorFile struct {
-	file vfs.File
-	fs   *errorFS
-}
-
-func (f errorFile) Close() error {
-	// We don't inject errors during close as those calls should never fail in
-	// practice.
-	return f.file.Close()
-}
-
-func (f errorFile) Read(p []byte) (int, error) {
-	if err := f.fs.maybeError(); err != nil {
-		return 0, err
-	}
-	return f.file.Read(p)
-}
-
-func (f errorFile) ReadAt(p []byte, off int64) (int, error) {
-	if err := f.fs.maybeError(); err != nil {
-		return 0, err
-	}
-	return f.file.ReadAt(p, off)
-}
-
-func (f errorFile) Write(p []byte) (int, error) {
-	if err := f.fs.maybeError(); err != nil {
-		return 0, err
-	}
-	return f.file.Write(p)
-}
-
-func (f errorFile) Stat() (os.FileInfo, error) {
-	if err := f.fs.maybeError(); err != nil {
-		return nil, err
-	}
-	return f.file.Stat()
-}
-
-func (f errorFile) Sync() error {
-	if err := f.fs.maybeError(); err != nil {
-		return err
-	}
-	return f.file.Sync()
-}
-
-// TestErrors repeatedly runs a short sequence of operations, injecting FS
-// errors at different points, until success is achieved.
-func TestErrors(t *testing.T) {
-	run := func(fs *errorFS) (err error) {
-		defer func() {
-			if r := recover(); r != nil {
-				if e, ok := r.(error); ok {
-					err = e
-				} else {
-					err = fmt.Errorf("%v", r)
-				}
-			}
-		}()
-
-		d, err := Open("", &Options{
-			FS:     fs,
-			Logger: panicLogger{},
-		})
-		if err != nil {
-			return err
-		}
-
-		key := []byte("a")
-		value := []byte("b")
-		if err := d.Set(key, value, nil); err != nil {
-			return err
-		}
-		if err := d.Flush(); err != nil {
-			return err
-		}
-		if err := d.Compact(nil, nil); err != nil {
-			return err
-		}
-
-		iter := d.NewIter(nil)
-		for valid := iter.First(); valid; valid = iter.Next() {
-		}
-		if err := iter.Close(); err != nil {
-			return err
-		}
-		return d.Close()
-	}
-
-	errorCounts := make(map[string]int)
-	for i := int32(0); ; i++ {
-		fs := newErrorFS(i)
-		err := run(fs)
-		if err == nil {
-			t.Logf("success %d\n", i)
-			break
-		}
-		errorCounts[err.Error()]++
-	}
-
-	expectedErrors := []string{
-		"fatal: MANIFEST flush failed: injected error",
-		"fatal: MANIFEST sync failed: injected error",
-		"fatal: MANIFEST set current failed: injected error",
-		"fatal: MANIFEST dirsync failed: injected error",
-	}
-	for _, expected := range expectedErrors {
-		if errorCounts[expected] == 0 {
-			t.Errorf("expected error %q did not occur", expected)
-		}
-	}
-}
diff --git a/open_test.go b/open_test.go
index 34abe1e..e094d5a 100644
--- a/open_test.go
+++ b/open_test.go
@@ -83,9 +83,9 @@ func testOpenCloseOpenClose(t *testing.T, fs vfs.FS, root string) {
 	for _, startFromEmpty := range []bool{false, true} {
 		for _, walDirname := range []string{"", "wal"} {
 			for _, length := range []int{-1, 0, 1, 1000, 10000, 100000} {
-				dirname := "sharedDatabase" + walDirname
+				dirname := "sharedDatabase"
 				if startFromEmpty {
-					dirname = "startFromEmpty" + walDirname + strconv.Itoa(length)
+					dirname = "startFromEmpty" + strconv.Itoa(length)
 				}
 				dirname = filepath.Join(root, dirname)
 				if walDirname == "" {
@@ -101,7 +101,7 @@ func testOpenCloseOpenClose(t *testing.T, fs vfs.FS, root string) {
 
 				d0, err := Open(dirname, opts)
 				if err != nil {
-					t.Fatalf("sfe=%t, length=%d: Open #0: %v",
+					t.Errorf("sfe=%t, length=%d: Open #0: %v",
 						startFromEmpty, length, err)
 					continue
 				}