Repository: petermattis/pebble
Pull request #159 (2019-06-04 22:52:57)

Add RocksDB simulation
Simulation which models RocksDB heuristics for pacing user writes.

The "levels" of the LSM tree are initially filled to full capacity, except for the bottom-most level. This means that every flush results in an increase in compaction debt of (num_levels * size_of_sstable), where compaction debt is defined as the total sum of all the bytes at each level which are over the maximum number of bytes at that level.

We see the sawtooth-like pattern when we look at compaction debt but it may be due to an error in my modeling. If we look from `10s` below, we see that when a flush happens which then triggers a compaction, compaction debt grows very large immediately and tapers down. Because of how this pattern interacts with the RocksDB heuristics, compaction debt and the number of L0 sstables grow indefinitely because this model doesn't have a hard stop limit on compaction debt and L0 sstables.
```
_elapsed___memtbs____dirty_____fill____drain____cdebt__l0count___max-w-rate
filling at 10 MB/sec
      1s        1     20.0     20.0      0.0      0.0        1         30.0
      2s        1     30.0     10.0      0.0      0.0        1         30.0
      3s        1     40.0     10.0      0.0      0.0        1         30.0
      4s        1     50.0     10.0      0.0      0.0        1         30.0
      5s        1     60.0     10.0      0.0      0.0        1         30.0
      6s        1      6.0     10.0     64.0      0.0        1         30.0
      7s        1     16.0     10.0      0.0      0.0        1         30.0
filling at 20 MB/sec
      8s        1     36.0     20.0      0.0      0.0        1         30.0
      9s        1     56.0     20.0      0.0      0.0        1         30.0
     10s        1     12.0     20.0     64.0    256.0        1         30.0
     11s        1     32.0     20.0      0.0    176.0        1         30.0
     12s        1     52.0     20.0      0.0     96.0        1         30.0
     13s        1      8.0     20.0     64.0     80.0        2         30.0
     14s        1     28.0     20.0      0.0      0.0        2         30.0
```

Like in RocksDB, the memtable is cleared "as fast as possible", and user writes are paced directly by an increase in compaction debt. User write rate limits are computed at the granularity of flushes and compaction of each level.

I have a different simulation which incorporates the "smooth" memtable flushing, where the memtable flushing speed is paced by an increase in compaction debt, and user writes are paced by memtable flushing speed.

I couldn't do any fair comparison right now because the results mostly depend on the tuning.

Is this experiment headed in the right direction, and do you see anything which might affect the modeling?

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/159)
<!-- Reviewable:end -->


Patch:
diff --git a/.travis.yml b/.travis.yml
index f375f86..6df30c9 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -9,12 +9,12 @@ matrix:
     - name: "go1.11.x"
       go: 1.11.x
       gobuild_args: -v
+    - name: "go1.11.x race"
+      go: 1.11.x
+      gobuild_args: -v -race
     - name: "go1.12.x"
       go: 1.12.x
       gobuild_args: -v
-    - name: "go1.12.x race"
-      go: 1.12.x
-      gobuild_args: -v -race
 
 notifications:
   email:
diff --git a/Makefile b/Makefile
index af0de5f..82e4bee 100644
--- a/Makefile
+++ b/Makefile
@@ -4,7 +4,6 @@ PKG := ./...
 BENCH_PKGS := internal/arenaskl internal/batchskl internal/record sstable .
 STRESSFLAGS :=
 TESTS := .
-BUILDER := ../../cockroachdb/cockroach/build/builder.sh
 
 .PHONY: all
 all:
@@ -18,11 +17,7 @@ all:
 
 .PHONY: test
 test:
-	${GO} test ${GOFLAGS} -run ${TESTS} ${PKG}
-
-.PHONY: test
-test-linux:
-	${BUILDER} ${GO} test ${GOFLAGS} -run ${TESTS} $(shell go list ${PKG})
+	$(GO) test ${GOFLAGS} -run ${TESTS} ${PKG}
 
 .PHONY: testrace
 testrace: GOFLAGS += -race
@@ -31,7 +26,7 @@ testrace: test
 .PHONY: stress stressrace
 stressrace: GOFLAGS += -race
 stress stressrace:
-	${GO} test -v ${GOFLAGS} -exec 'stress ${STRESSFLAGS}' -run "${TESTS}" -timeout 0 ${PKG}
+	$(GO) test -v ${GOFLAGS} -exec 'stress ${STRESSFLAGS}' -run "${TESTS}" -timeout 0 ${PKG}
 
 .PHONY: bench
 bench: GOFLAGS += -timeout 1h
@@ -40,7 +35,7 @@ bench: $(patsubst %,%.bench,$(if $(findstring ./...,${PKG}),${BENCH_PKGS},${PKG}
 internal/arenaskl.bench: GOFLAGS += -cpu 1,8
 
 %.bench:
-	${GO} test -run - -bench . -count 10 ${GOFLAGS} ./$* 2>&1 | tee $*/bench.txt.new
+	$(GO) test -run - -bench . -count 10 ${GOFLAGS} ./$* 2>&1 | tee $*/bench.txt.new
 
 .PHONY: clean
 clean:
diff --git a/batch.go b/batch.go
index 2a66836..0988639 100644
--- a/batch.go
+++ b/batch.go
@@ -258,8 +258,8 @@ func (b *Batch) release() {
 
 func (b *Batch) refreshMemTableSize() {
 	b.memTableSize = 0
-	for r := b.Reader(); ; {
-		_, key, value, ok := r.Next()
+	for iter := b.iter(); ; {
+		_, key, value, ok := iter.next()
 		if !ok {
 			break
 		}
@@ -288,9 +288,9 @@ func (b *Batch) Apply(batch *Batch, _ *WriteOptions) error {
 	count := binary.LittleEndian.Uint32(batch.storage.data[8:12])
 	b.setCount(b.count() + count)
 
-	for iter := BatchReader(b.storage.data[offset:]); len(iter) > 0; {
+	for iter := batchReader(b.storage.data[offset:]); len(iter) > 0; {
 		offset := uintptr(unsafe.Pointer(&iter[0])) - uintptr(unsafe.Pointer(&b.storage.data[0]))
-		kind, key, value, ok := iter.Next()
+		kind, key, value, ok := iter.next()
 		if !ok {
 			break
 		}
@@ -448,6 +448,8 @@ func (b *Batch) DeleteRange(start, end []byte, _ *WriteOptions) error {
 // which makes it useful for testing WAL performance.
 //
 // It is safe to modify the contents of the argument after LogData returns.
+//
+// TODO(peter): untested.
 func (b *Batch) LogData(data []byte, _ *WriteOptions) error {
 	if len(b.storage.data) == 0 {
 		b.init(len(data) + binary.MaxVarintLen64 + batchHeaderLen)
@@ -649,9 +651,7 @@ func (b *Batch) count() uint32 {
 	return binary.LittleEndian.Uint32(b.countData())
 }
 
-// Reader returns a BatchReader for the current batch contents. If the batch is
-// mutated, the new entries will not be visible to the reader.
-func (b *Batch) Reader() BatchReader {
+func (b *Batch) iter() batchReader {
 	return b.storage.data[batchHeaderLen:]
 }
 
@@ -690,18 +690,11 @@ func batchDecodeStr(data []byte) (odata []byte, s []byte, ok bool) {
 	return data[v:], data[:v], true
 }
 
-// BatchReader iterates over the entries contained in a batch.
-type BatchReader []byte
-
-// MakeBatchReader constructs a BatchReader from a batch representation. The
-// header (containing the batch count and seqnum) is ignored.
-func MakeBatchReader(repr []byte) BatchReader {
-	return repr[batchHeaderLen:]
-}
+type batchReader []byte
 
-// Next returns the next entry in this batch. The final return value is false
-// if the batch is corrupt. The end of batch is reached when len(r)==0.
-func (r *BatchReader) Next() (kind InternalKeyKind, ukey []byte, value []byte, ok bool) {
+// next returns the next operation in this batch.
+// The final return value is false if the batch is corrupi.
+func (r *batchReader) next() (kind InternalKeyKind, ukey []byte, value []byte, ok bool) {
 	p := *r
 	if len(p) == 0 {
 		return 0, nil, nil, false
@@ -724,7 +717,7 @@ func (r *BatchReader) Next() (kind InternalKeyKind, ukey []byte, value []byte, o
 	return kind, ukey, value, true
 }
 
-func (r *BatchReader) nextStr() (s []byte, ok bool) {
+func (r *batchReader) nextStr() (s []byte, ok bool) {
 	p := *r
 	u, numBytes := binary.Uvarint(p)
 	if numBytes <= 0 {
@@ -881,9 +874,9 @@ func newFlushableBatch(batch *Batch, comparer *Comparer) *flushableBatch {
 	}
 
 	var index uint32
-	for iter := BatchReader(b.data[batchHeaderLen:]); len(iter) > 0; index++ {
+	for iter := batchReader(b.data[batchHeaderLen:]); len(iter) > 0; index++ {
 		offset := uintptr(unsafe.Pointer(&iter[0])) - uintptr(unsafe.Pointer(&b.data[0]))
-		kind, key, _, ok := iter.Next()
+		kind, key, _, ok := iter.next()
 		if !ok {
 			break
 		}
@@ -944,8 +937,6 @@ func (b *flushableBatch) newIter(o *IterOptions) internalIterator {
 		offsets: b.offsets,
 		cmp:     b.cmp,
 		index:   -1,
-		lower:   o.GetLowerBound(),
-		upper:   o.GetUpperBound(),
 	}
 }
 
@@ -1006,8 +997,6 @@ type flushableBatchIter struct {
 	index   int
 	key     InternalKey
 	err     error
-	lower   []byte
-	upper   []byte
 }
 
 // flushableBatchIter implements the internalIterator interface.
@@ -1022,10 +1011,6 @@ func (i *flushableBatchIter) SeekGE(key []byte) (*InternalKey, []byte) {
 		return nil, nil
 	}
 	i.key = i.getKey(i.index)
-	if i.upper != nil && i.cmp(i.key.UserKey, i.upper) >= 0 {
-		i.index = len(i.offsets)
-		return nil, nil
-	}
 	return &i.key, i.Value()
 }
 
@@ -1043,10 +1028,6 @@ func (i *flushableBatchIter) SeekLT(key []byte) (*InternalKey, []byte) {
 		return nil, nil
 	}
 	i.key = i.getKey(i.index)
-	if i.lower != nil && i.cmp(i.key.UserKey, i.lower) < 0 {
-		i.index = -1
-		return nil, nil
-	}
 	return &i.key, i.Value()
 }
 
@@ -1056,10 +1037,6 @@ func (i *flushableBatchIter) First() (*InternalKey, []byte) {
 	}
 	i.index = 0
 	i.key = i.getKey(i.index)
-	if i.upper != nil && i.cmp(i.key.UserKey, i.upper) >= 0 {
-		i.index = len(i.offsets)
-		return nil, nil
-	}
 	return &i.key, i.Value()
 }
 
@@ -1069,10 +1046,6 @@ func (i *flushableBatchIter) Last() (*InternalKey, []byte) {
 	}
 	i.index = len(i.offsets) - 1
 	i.key = i.getKey(i.index)
-	if i.lower != nil && i.cmp(i.key.UserKey, i.lower) < 0 {
-		i.index = -1
-		return nil, nil
-	}
 	return &i.key, i.Value()
 }
 
@@ -1085,10 +1058,6 @@ func (i *flushableBatchIter) Next() (*InternalKey, []byte) {
 		return nil, nil
 	}
 	i.key = i.getKey(i.index)
-	if i.upper != nil && i.cmp(i.key.UserKey, i.upper) >= 0 {
-		i.index = len(i.offsets)
-		return nil, nil
-	}
 	return &i.key, i.Value()
 }
 
@@ -1101,10 +1070,6 @@ func (i *flushableBatchIter) Prev() (*InternalKey, []byte) {
 		return nil, nil
 	}
 	i.key = i.getKey(i.index)
-	if i.lower != nil && i.cmp(i.key.UserKey, i.lower) < 0 {
-		i.index = -1
-		return nil, nil
-	}
 	return &i.key, i.Value()
 }
 
@@ -1141,6 +1106,6 @@ func (i *flushableBatchIter) Close() error {
 }
 
 func (i *flushableBatchIter) SetBounds(lower, upper []byte) {
-	i.lower = lower
-	i.upper = upper
+	// This should not be called as bounds are not used for this iterator.
+	panic("TODO(ryan): pebble: SetBounds unimplemented")
 }
diff --git a/compaction.go b/compaction.go
index b09bebd..98447aa 100644
--- a/compaction.go
+++ b/compaction.go
@@ -11,7 +11,6 @@ import (
 	"math"
 	"os"
 	"sort"
-	"sync/atomic"
 	"unsafe"
 
 	"github.com/petermattis/pebble/internal/base"
@@ -577,7 +576,7 @@ type manualCompaction struct {
 //
 // d.mu must be held when calling this.
 func (d *DB) maybeScheduleFlush() {
-	if d.mu.compact.flushing || atomic.LoadInt32(&d.closed) != 0 {
+	if d.mu.compact.flushing || d.mu.closed {
 		return
 	}
 	if len(d.mu.mem.queue) <= 1 {
@@ -702,7 +701,7 @@ func (d *DB) flush1() error {
 //
 // d.mu must be held when calling this.
 func (d *DB) maybeScheduleCompaction() {
-	if d.mu.compact.compacting || atomic.LoadInt32(&d.closed) != 0 {
+	if d.mu.compact.compacting || d.mu.closed {
 		return
 	}
 
diff --git a/compaction_picker.go b/compaction_picker.go
index 2d75ea0..f313fe0 100644
--- a/compaction_picker.go
+++ b/compaction_picker.go
@@ -49,14 +49,14 @@ func (p *compactionPicker) compactionNeeded() bool {
 func (p *compactionPicker) initLevelMaxBytes(v *version, opts *Options) {
 	// Determine the first non-empty level and the maximum size of any level.
 	firstNonEmptyLevel := -1
-	var bottomLevelSize int64
+	var maxLevelSize int64
 	for level := 1; level < numLevels; level++ {
 		levelSize := int64(totalSize(v.files[level]))
-		if levelSize > 0 {
-			if firstNonEmptyLevel == -1 {
-				firstNonEmptyLevel = level
-			}
-			bottomLevelSize = levelSize
+		if levelSize > 0 && firstNonEmptyLevel == -1 {
+			firstNonEmptyLevel = level
+		}
+		if maxLevelSize < levelSize {
+			maxLevelSize = levelSize
 		}
 	}
 
@@ -67,7 +67,7 @@ func (p *compactionPicker) initLevelMaxBytes(v *version, opts *Options) {
 		p.levelMaxBytes[level] = math.MaxInt64
 	}
 
-	if bottomLevelSize == 0 {
+	if maxLevelSize == 0 {
 		// No levels for L1 and up contain any data. Target L0 compactions for the
 		// last level.
 		p.baseLevel = numLevels - 1
@@ -79,15 +79,17 @@ func (p *compactionPicker) initLevelMaxBytes(v *version, opts *Options) {
 	baseBytesMax := opts.LBaseMaxBytes
 	baseBytesMin := int64(float64(baseBytesMax) / levelMultiplier)
 
-	curLevelSize := bottomLevelSize
+	curLevelSize := maxLevelSize
 	for level := numLevels - 2; level >= firstNonEmptyLevel; level-- {
 		curLevelSize = int64(float64(curLevelSize) / levelMultiplier)
 	}
 
+	var baseLevelSize int64
 	if curLevelSize <= baseBytesMin {
-		// If we make target size of last level to be bottomLevelSize, target size of
+		// If we make target size of last level to be maxLevelSize, target size of
 		// the first non-empty level would be smaller than baseBytesMin. We set it
 		// be baseBytesMin.
+		baseLevelSize = baseBytesMin + 1
 		p.baseLevel = firstNonEmptyLevel
 	} else {
 		// Compute base level (where L0 data is compacted to).
@@ -96,29 +98,23 @@ func (p *compactionPicker) initLevelMaxBytes(v *version, opts *Options) {
 			p.baseLevel--
 			curLevelSize = int64(float64(curLevelSize) / levelMultiplier)
 		}
+		if curLevelSize > baseBytesMax {
+			baseLevelSize = baseBytesMax
+		} else {
+			baseLevelSize = curLevelSize
+		}
 	}
 
-	var smoothedLevelMultiplier float64
-	if p.baseLevel < numLevels-1 {
-		smoothedLevelMultiplier = math.Pow(
-			float64(bottomLevelSize)/float64(baseBytesMax),
-			1.0/float64(numLevels-p.baseLevel-1))
-	} else {
-		smoothedLevelMultiplier = 1.0
-	}
-
-	levelSize := float64(baseBytesMax)
+	levelSize := baseLevelSize
 	for level := p.baseLevel; level < numLevels; level++ {
-		if level > p.baseLevel && levelSize > 0 {
-			levelSize *= smoothedLevelMultiplier
+		if level > p.baseLevel {
+			if levelSize > 0 && float64(math.MaxInt64/levelSize) >= levelMultiplier {
+				levelSize = int64(float64(levelSize) * levelMultiplier)
+			}
 		}
-		// Round the result since test cases use small target level sizes, which
-		// can be impacted by floating-point imprecision + integer truncation.
-		roundedLevelSize := math.Round(levelSize)
-		if roundedLevelSize > float64(math.MaxInt64) {
-			p.levelMaxBytes[level] = math.MaxInt64
-		} else {
-			p.levelMaxBytes[level] = int64(roundedLevelSize)
+		p.levelMaxBytes[level] = levelSize
+		if p.levelMaxBytes[level] < baseBytesMax {
+			p.levelMaxBytes[level] = baseBytesMax
 		}
 	}
 }
diff --git a/db.go b/db.go
index 002ca8e..6f1e522 100644
--- a/db.go
+++ b/db.go
@@ -32,9 +32,6 @@ var (
 	// ErrNotFound is returned when a get operation does not find the requested
 	// key.
 	ErrNotFound = base.ErrNotFound
-	// ErrClosed is returned when an operation is performed on a closed snapshot
-	// or DB.
-	ErrClosed = errors.New("pebble: closed")
 )
 
 type flushable interface {
@@ -168,13 +165,12 @@ type DB struct {
 
 	logRecycler logRecycler
 
-	closed int32 // updated atomically
-
 	// TODO(peter): describe exactly what this mutex protects. So far: every
 	// field in the struct.
 	mu struct {
 		sync.Mutex
 
+		closed    bool
 		nextJobID int
 
 		versions versionSet
@@ -231,10 +227,6 @@ func (d *DB) Get(key []byte) ([]byte, error) {
 }
 
 func (d *DB) getInternal(key []byte, b *Batch, s *Snapshot) ([]byte, error) {
-	if atomic.LoadInt32(&d.closed) != 0 {
-		panic(ErrClosed)
-	}
-
 	// Grab and reference the current readState. This prevents the underlying
 	// files in the associated version from being deleted if there is a current
 	// compaction. The readState is unref'd by Iterator.Close().
@@ -274,7 +266,7 @@ func (d *DB) getInternal(key []byte, b *Batch, s *Snapshot) ([]byte, error) {
 	i.readState = readState
 
 	defer i.Close()
-	if !i.First() {
+	if !i.Next() {
 		err := i.Error()
 		if err != nil {
 			return nil, err
@@ -351,10 +343,6 @@ func (d *DB) LogData(data []byte, opts *WriteOptions) error {
 //
 // It is safe to modify the contents of the arguments after Apply returns.
 func (d *DB) Apply(batch *Batch, opts *WriteOptions) error {
-	if atomic.LoadInt32(&d.closed) != 0 {
-		panic(ErrClosed)
-	}
-
 	sync := opts.GetSync()
 	if sync && d.opts.DisableWAL {
 		return errors.New("pebble: WAL disabled")
@@ -449,10 +437,6 @@ func (d *DB) newIterInternal(
 	s *Snapshot,
 	o *IterOptions,
 ) *Iterator {
-	if atomic.LoadInt32(&d.closed) != 0 {
-		panic(ErrClosed)
-	}
-
 	// Grab and reference the current readState. This prevents the underlying
 	// files in the associated version from being deleted if there is a current
 	// compaction. The readState is unref'd by Iterator.Close().
@@ -594,10 +578,6 @@ func (d *DB) NewIter(o *IterOptions) *Iterator {
 // deleted. Instead, a snapshot prevents deletion of sequence numbers
 // referenced by the snapshot.
 func (d *DB) NewSnapshot() *Snapshot {
-	if atomic.LoadInt32(&d.closed) != 0 {
-		panic(ErrClosed)
-	}
-
 	s := &Snapshot{
 		db:     d,
 		seqNum: atomic.LoadUint64(&d.mu.versions.visibleSeqNum),
@@ -616,10 +596,9 @@ func (d *DB) NewSnapshot() *Snapshot {
 func (d *DB) Close() error {
 	d.mu.Lock()
 	defer d.mu.Unlock()
-	if atomic.LoadInt32(&d.closed) != 0 {
-		panic(ErrClosed)
+	if d.mu.closed {
+		return nil
 	}
-	atomic.StoreInt32(&d.closed, 1)
 	for d.mu.compact.compacting || d.mu.compact.flushing {
 		d.mu.compact.cond.Wait()
 	}
@@ -627,6 +606,7 @@ func (d *DB) Close() error {
 	err = firstError(err, d.mu.log.Close())
 	err = firstError(err, d.fileLock.Close())
 	d.commit.Close()
+	d.mu.closed = true
 
 	err = firstError(err, d.dataDir.Close())
 
@@ -652,10 +632,6 @@ func (d *DB) Close() error {
 
 // Compact the specified range of keys in the database.
 func (d *DB) Compact(start, end []byte /* CompactionOptions */) error {
-	if atomic.LoadInt32(&d.closed) != 0 {
-		panic(ErrClosed)
-	}
-
 	iStart := base.MakeInternalKey(start, InternalKeySeqNumMax, InternalKeyKindMax)
 	iEnd := base.MakeInternalKey(end, 0, 0)
 	meta := []*fileMetadata{&fileMetadata{smallest: iStart, largest: iEnd}}
@@ -727,10 +703,6 @@ func (d *DB) manualCompact(manual *manualCompaction) error {
 
 // Flush the memtable to stable storage.
 func (d *DB) Flush() error {
-	if atomic.LoadInt32(&d.closed) != 0 {
-		panic(ErrClosed)
-	}
-
 	d.mu.Lock()
 	mem := d.mu.mem.mutable
 	err := d.makeRoomForWrite(nil)
@@ -746,10 +718,6 @@ func (d *DB) Flush() error {
 //
 // TODO(peter): untested
 func (d *DB) AsyncFlush() error {
-	if atomic.LoadInt32(&d.closed) != 0 {
-		panic(ErrClosed)
-	}
-
 	d.mu.Lock()
 	err := d.makeRoomForWrite(nil)
 	d.mu.Unlock()
diff --git a/internal/pacertoy/main.go b/internal/pacertoy/main.go
new file mode 100644
index 0000000..f08e812
--- /dev/null
+++ b/internal/pacertoy/main.go
@@ -0,0 +1,398 @@
+// 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"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/petermattis/pebble/internal/rate"
+
+	"golang.org/x/exp/rand"
+)
+
+const (
+	// Max rate for all compactions. This is intentionally set low enough that
+	// user writes will have to be delayed
+	maxCompactionRate     = 80 << 20 // 80 MB/s
+
+	memtableSize          = 64 << 20 // 64 MB
+	memtableStopThreshold = 2 * memtableSize
+	maxWriteRate          = 30 << 20 // 30 MB/s
+	startingWriteRate     = 30 << 20 // 30 MB/s
+
+	l0SlowdownThreshold   = 4
+	l0CompactionThreshold = 1
+
+	levelRatio            = 10
+	numLevels             = 7
+
+	// Slowdown threshold is set at the compaction debt incurred by the largest
+	// possible compaction.
+	compactionDebtSlowdownThreshold = memtableSize*(numLevels-2)
+)
+
+type compactionPacer struct {
+	level             int64
+	drainer           *rate.Limiter
+}
+
+func newCompactionPacer() *compactionPacer {
+	p := &compactionPacer{
+		drainer: rate.NewLimiter(maxCompactionRate, maxCompactionRate),
+	}
+	return p
+}
+
+func (p *compactionPacer) fill(n int64) {
+	atomic.AddInt64(&p.level, n)
+}
+
+func (p *compactionPacer) drain(n int64) {
+	p.drainer.WaitN(context.Background(), int(n))
+
+	atomic.AddInt64(&p.level, -n)
+}
+
+type flushPacer struct {
+	level                 int64
+	memtableStopThreshold float64
+	fillCond              sync.Cond
+}
+
+func newFlushPacer(mu *sync.Mutex) *flushPacer {
+	p := &flushPacer{
+		memtableStopThreshold: memtableStopThreshold,
+	}
+	p.fillCond.L = mu
+	return p
+}
+
+func (p *flushPacer) fill(n int64) {
+	for float64(atomic.LoadInt64(&p.level)) >= p.memtableStopThreshold {
+		p.fillCond.Wait()
+	}
+	atomic.AddInt64(&p.level, n)
+	p.fillCond.Signal()
+}
+
+func (p *flushPacer) drain(n int64) {
+	atomic.AddInt64(&p.level, -n)
+}
+
+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
+	previouslyInDebt    bool
+
+	writeLimiter        *rate.Limiter
+}
+
+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
+
+	db.writeLimiter = rate.NewLimiter(startingWriteRate, startingWriteRate)
+
+	go db.drainMemtable()
+	go db.drainCompaction()
+
+	return db
+}
+
+// drainCompaction simulates background compactions.
+func (db *DB) drainCompaction() {
+	rng := rand.New(rand.NewSource(uint64(time.Now().UnixNano())))
+
+	for {
+		db.compactionMu.Lock()
+
+		for len(db.L0) <= l0CompactionThreshold {
+			db.compactionFlushCond.Wait()
+		}
+		l0Table := db.L0[0]
+		db.compactionMu.Unlock()
+
+		for i, size := int64(0), int64(0); i < *l0Table; i += size {
+			size = 10000 + rng.Int63n(500)
+			if size > (*l0Table - i) {
+				size = *l0Table - i
+			}
+			db.compactionPacer.drain(size)
+		}
+
+		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)
+		db.compactionPacer.fill(totalCompactionBytes)
+
+		for t := 0; t < tablesToCompact; t++ {
+			for i, size := int64(0), int64(0); i < memtableSize; i += size {
+				size = 10000 + rng.Int63n(500)
+				if size > (totalCompactionBytes - i) {
+					size = totalCompactionBytes - i
+				}
+				db.compactionPacer.drain(size)
+			}
+
+			db.delayUserWrites()
+		}
+	}
+}
+
+// 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(uint64(time.Now().UnixNano())))
+
+	for {
+		db.mu.Lock()
+		for len(db.memtables) <= 1 {
+			db.flushCond.Wait()
+		}
+		memtable := db.memtables[0]
+		db.mu.Unlock()
+
+		for i, size := int64(0), int64(0); i < *memtable; i += size {
+			size = 1000 + rng.Int63n(50)
+			if size > (*memtable - i) {
+				size = *memtable - i
+			}
+			db.flushPacer.drain(size)
+			atomic.AddInt64(&db.drain, size)
+
+			db.fillCompaction(size)
+		}
+
+		db.delayUserWrites()
+
+		db.mu.Lock()
+		db.memtables = db.memtables[1:]
+		db.mu.Unlock()
+	}
+}
+
+// delayUserWrites applies write delays depending on compaction debt.
+func (db *DB) delayUserWrites() {
+	totalCompactionBytes := atomic.LoadInt64(&db.compactionPacer.level)
+	compactionDebt := math.Max(float64(totalCompactionBytes)-l0CompactionThreshold*memtableSize, 0.0)
+
+	db.mu.Lock()
+	if len(db.L0) > l0SlowdownThreshold || compactionDebt > compactionDebtSlowdownThreshold {
+		db.previouslyInDebt = true
+		if compactionDebt > db.prevCompactionDebt {
+			// Debt is growing.
+			drainLimit := db.writeLimiter.Limit() * 0.8
+			if drainLimit > 0 {
+				db.writeLimiter.SetLimit(drainLimit)
+			}
+		} else {
+			// Debt is shrinking.
+			drainLimit := db.writeLimiter.Limit() * 1/0.8
+			if drainLimit <= maxWriteRate {
+				db.writeLimiter.SetLimit(drainLimit)
+			}
+		}
+	} else if db.previouslyInDebt {
+		// If compaction was previously delayed and has recovered, RocksDB
+		// "rewards" the rate by double the slowdown ratio.
+
+		// 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
+		if drainLimit <= maxWriteRate {
+			db.writeLimiter.SetLimit(drainLimit)
+		}
+		db.previouslyInDebt = false
+	}
+
+	db.prevCompactionDebt = compactionDebt
+	db.mu.Unlock()
+}
+
+// fillMemtable simulates memtable filling.
+func (db *DB) fillMemtable(size int64) {
+	db.mu.Lock()
+
+	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) {
+	limiter := rate.NewLimiter(10<<20, 10<<20) // 10 mb
+	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))
+	}
+
+	go func() {
+		rng := rand.New(rand.NewSource(uint64(time.Now().UnixNano())))
+		for {
+			secs := 5 + rng.Intn(5)
+			time.Sleep(time.Duration(secs) * time.Second)
+			mb := 11 + rng.Intn(20)
+			setRate(mb)
+		}
+	}()
+
+	rng := rand.New(rand.NewSource(uint64(time.Now().UnixNano())))
+
+	for {
+		size := 1000 + rng.Int63n(50)
+		limiter.WaitN(context.Background(), int(size))
+		db.writeLimiter.WaitN(context.Background(), int(size))
+		db.fillMemtable(size)
+	}
+}
+
+func main() {
+	db := newDB()
+
+	go simulateWrite(db)
+
+	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-w-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)
+			maxWriteRate := db.writeLimiter.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,
+				maxWriteRate/(1024.0*1024.0))
+
+			lastNow = now
+			lastFill = fill
+			lastDrain = drain
+		}
+	}
+}
diff --git a/internal/randvar/zipf.go b/internal/randvar/zipf.go
index 7925248..f04628f 100644
--- a/internal/randvar/zipf.go
+++ b/internal/randvar/zipf.go
@@ -46,7 +46,6 @@ type Zipf struct {
 	min   uint64
 	// Internally computed constants.
 	alpha, zeta2 float64
-	halfPowTheta float64
 	// Mutable state.
 	mu struct {
 		sync.Mutex
@@ -81,7 +80,6 @@ func NewZipf(rng *rand.Rand, min, max uint64, theta float64) (*Zipf, error) {
 
 	// Compute hidden parameters.
 	z.zeta2 = computeZetaFromScratch(2, theta)
-	z.halfPowTheta = 1.0 + math.Pow(0.5, z.theta)
 	z.mu.zetaN = computeZetaFromScratch(max+1-min, theta)
 	z.alpha = 1.0 / (1.0 - theta)
 	z.mu.eta = (1 - math.Pow(2.0/float64(z.mu.max+1-z.min), 1.0-theta)) / (1.0 - z.zeta2/z.mu.zetaN)
@@ -92,7 +90,7 @@ func NewZipf(rng *rand.Rand, min, max uint64, theta float64) (*Zipf, error) {
 // zeta(oldMax, theta). Returns zeta(max, theta), computed incrementally.
 func computeZetaIncrementally(oldMax, max uint64, theta float64, sum float64) float64 {
 	if max < oldMax {
-		panic("unable to decrement max!")
+		panic(fmt.Errorf("unable to increment max backwards!"))
 	}
 	for i := oldMax + 1; i <= max; i++ {
 		sum += 1.0 / math.Pow(float64(i), theta)
@@ -117,8 +115,8 @@ func (z *Zipf) IncMax(delta int) {
 	z.mu.Lock()
 	oldMax := z.mu.max
 	z.mu.max += uint64(delta)
-	z.mu.zetaN = computeZetaIncrementally(oldMax+1-z.min, z.mu.max+1-z.min, z.theta, z.mu.zetaN)
-	z.mu.eta = (1 - math.Pow(2.0/float64(z.mu.max+1-z.min), 1.0-z.theta)) / (1.0 - z.zeta2/z.mu.zetaN)
+	z.mu.zetaN = computeZetaIncrementally(oldMax, z.mu.max, z.theta, z.mu.zetaN)
+	z.mu.eta = (1 - math.Pow(2.0/float64(z.mu.max-z.min), 1.0-z.theta)) / (1.0 - z.zeta2/z.mu.zetaN)
 	z.mu.Unlock()
 }
 
@@ -131,11 +129,11 @@ func (z *Zipf) Uint64() uint64 {
 	var result uint64
 	if uz < 1.0 {
 		result = z.min
-	} else if uz < z.halfPowTheta {
+	} else if uz < 1.0+math.Pow(0.5, z.theta) {
 		result = z.min + 1
 	} else {
 		spread := float64(z.mu.max + 1 - z.min)
-		result = z.min + uint64(spread*math.Pow(z.mu.eta*u-z.mu.eta+1.0, z.alpha))
+		result = z.min + uint64(int64(spread*math.Pow(z.mu.eta*u-z.mu.eta+1.0, z.alpha)))
 	}
 	z.mu.Unlock()
 	return result
diff --git a/iterator.go b/iterator.go
index 1697105..c0947f4 100644
--- a/iterator.go
+++ b/iterator.go
@@ -95,20 +95,21 @@ func (i *Iterator) findNextEntry() bool {
 }
 
 func (i *Iterator) nextUserKey() {
-	if i.iterKey == nil {
-		return
-	}
-	done := i.iterKey.SeqNum() == 0
-	if !i.valid {
-		i.keyBuf = append(i.keyBuf[:0], i.iterKey.UserKey...)
-		i.key = i.keyBuf
-	}
-	for {
-		i.iterKey, i.iterValue = i.iter.Next()
-		if done || i.iterKey == nil || !i.equal(i.key, i.iterKey.UserKey) {
-			break
+	if i.iterKey != nil {
+		done := i.iterKey.SeqNum() == 0
+		if !i.valid {
+			i.keyBuf = append(i.keyBuf[:0], i.iterKey.UserKey...)
+			i.key = i.keyBuf
 		}
-		done = i.iterKey.SeqNum() == 0
+		for {
+			i.iterKey, i.iterValue = i.iter.Next()
+			if done || i.iterKey == nil || !i.equal(i.key, i.iterKey.UserKey) {
+				break
+			}
+			done = i.iterKey.SeqNum() == 0
+		}
+	} else {
+		i.iterKey, i.iterValue = i.iter.First()
 	}
 }
 
@@ -188,20 +189,21 @@ func (i *Iterator) findPrevEntry() bool {
 }
 
 func (i *Iterator) prevUserKey() {
-	if i.iterKey == nil {
-		return
-	}
-	if !i.valid {
-		// If we're going to compare against the prev key, we need to save the
-		// current key.
-		i.keyBuf = append(i.keyBuf[:0], i.iterKey.UserKey...)
-		i.key = i.keyBuf
-	}
-	for {
-		i.iterKey, i.iterValue = i.iter.Prev()
-		if i.iterKey == nil || !i.equal(i.key, i.iterKey.UserKey) {
-			break
+	if i.iterKey != nil {
+		if !i.valid {
+			// If we're going to compare against the prev key, we need to save the
+			// current key.
+			i.keyBuf = append(i.keyBuf[:0], i.iterKey.UserKey...)
+			i.key = i.keyBuf
+		}
+		for {
+			i.iterKey, i.iterValue = i.iter.Prev()
+			if i.iterKey == nil || !i.equal(i.key, i.iterKey.UserKey) {
+				break
+			}
 		}
+	} else {
+		i.iterKey, i.iterValue = i.iter.Last()
 	}
 }
 
@@ -279,14 +281,14 @@ func (i *Iterator) SeekGE(key []byte) bool {
 // the Comparer. Also note that the iterator will not observe keys not matching
 // the prefix.
 func (i *Iterator) SeekPrefixGE(key []byte) bool {
-	if i.err != nil {
-		return false
-	}
-
 	if i.split == nil {
 		panic("pebble: split must be provided for SeekPrefixGE")
 	}
 
+	if i.err != nil {
+		return false
+	}
+
 	// Make a copy of the prefix so that modifications to the key after
 	// SeekPrefixGE returns does not affect the stored prefix.
 	prefixLen := i.split(key)
@@ -372,17 +374,7 @@ func (i *Iterator) Next() bool {
 		// TODO(peter): This is subtle. Perhaps we should provide a special version
 		// of nextUserKey which advances until a larger user key is found.
 		i.valid = false
-		if i.iterKey == nil {
-			// We're positioned before the first key. Need to reposition to point to
-			// the first key.
-			if lowerBound := i.opts.GetLowerBound(); lowerBound != nil {
-				i.iterKey, i.iterValue = i.iter.SeekGE(lowerBound)
-			} else {
-				i.iterKey, i.iterValue = i.iter.First()
-			}
-		} else {
-			i.nextUserKey()
-		}
+		i.nextUserKey()
 		i.nextUserKey()
 	case iterPosNext:
 	}
@@ -407,17 +399,7 @@ func (i *Iterator) Prev() bool {
 		// TODO(peter): This is subtle. Perhaps we should provide a special version
 		// of prevUserKey which advances until a smaller user key is found.
 		i.valid = false
-		if i.iterKey == nil {
-			// We're positioned after the last key. Need to reposition to point to
-			// the last key.
-			if upperBound := i.opts.GetUpperBound(); upperBound != nil {
-				i.iterKey, i.iterValue = i.iter.SeekLT(upperBound)
-			} else {
-				i.iterKey, i.iterValue = i.iter.Last()
-			}
-		} else {
-			i.prevUserKey()
-		}
+		i.prevUserKey()
 		i.prevUserKey()
 	case iterPosPrev:
 	}
diff --git a/mem_table.go b/mem_table.go
index 0ba7458..51f1cb1 100644
--- a/mem_table.go
+++ b/mem_table.go
@@ -144,8 +144,8 @@ func (m *memTable) apply(batch *Batch, seqNum uint64) error {
 	var ins arenaskl.Inserter
 	var tombstoneCount uint32
 	startSeqNum := seqNum
-	for r := batch.Reader(); ; seqNum++ {
-		kind, ukey, value, ok := r.Next()
+	for iter := batch.iter(); ; seqNum++ {
+		kind, ukey, value, ok := iter.next()
 		if !ok {
 			break
 		}
diff --git a/snapshot.go b/snapshot.go
index 7834640..6d889e3 100644
--- a/snapshot.go
+++ b/snapshot.go
@@ -25,9 +25,6 @@ var _ Reader = (*Snapshot)(nil)
 // The caller should not modify the contents of the returned slice, but it is
 // safe to modify the contents of the argument after Get returns.
 func (s *Snapshot) Get(key []byte) ([]byte, error) {
-	if s.db == nil {
-		panic(ErrClosed)
-	}
 	return s.db.getInternal(key, nil /* batch */, s)
 }
 
@@ -35,9 +32,6 @@ func (s *Snapshot) Get(key []byte) ([]byte, error) {
 // return false). The iterator can be positioned via a call to SeekGE,
 // SeekLT, First or Last.
 func (s *Snapshot) NewIter(o *IterOptions) *Iterator {
-	if s.db == nil {
-		panic(ErrClosed)
-	}
 	return s.db.newIterInternal(nil /* batchIter */, nil /* batchRangeDelIter */, s, o)
 }
 
@@ -46,13 +40,9 @@ func (s *Snapshot) NewIter(o *IterOptions) *Iterator {
 // leak of resources on disk due to the entries the snapshot is preventing from
 // being deleted.
 func (s *Snapshot) Close() error {
-	if s.db == nil {
-		panic(ErrClosed)
-	}
 	s.db.mu.Lock()
 	s.db.mu.snapshots.remove(s)
 	s.db.mu.Unlock()
-	s.db = nil
 	return nil
 }
 
diff --git a/vfs/syncing_file_linux.go b/vfs/syncing_file_linux.go
index c5f8d4b..cf724f5 100644
--- a/vfs/syncing_file_linux.go
+++ b/vfs/syncing_file_linux.go
@@ -11,17 +11,6 @@ import (
 	"syscall"
 )
 
-type syncFileRange func(fd int, off int64, n int64, flags int) (err error)
-
-// sync_file_range depends on both the filesystem, and the broader kernel
-// support. In particular, Windows Subsystem for Linux does not support
-// sync_file_range, even when used with ext{2,3,4}. syncRangeSmokeTest performs
-// a test of of sync_file_range, returning false on ENOSYS, and true otherwise.
-func syncRangeSmokeTest(fd uintptr, fn syncFileRange) bool {
-	err := fn(int(fd), 0 /* offset */, 0 /* nbytes */, 0 /* flags */)
-	return err != syscall.ENOSYS
-}
-
 func isSyncRangeSupported(fd uintptr) bool {
 	var stat syscall.Statfs_t
 	if err := syscall.Fstatfs(int(fd), &stat); err != nil {
@@ -36,7 +25,7 @@ func isSyncRangeSupported(fd uintptr) bool {
 	const extMagic = 0xef53
 	switch stat.Type {
 	case extMagic:
-		return syncRangeSmokeTest(fd, syscall.SyncFileRange)
+		return true
 	}
 	return false
 }


Test patch:
diff --git a/batch_test.go b/batch_test.go
index 2884740..cf0f1a4 100644
--- a/batch_test.go
+++ b/batch_test.go
@@ -35,31 +35,21 @@ func TestBatch(t *testing.T) {
 		{InternalKeyKindSet, "binarydata", "\x00"},
 		{InternalKeyKindSet, "binarydata", "\xff"},
 		{InternalKeyKindMerge, "merge", "mergedata"},
-		{InternalKeyKindMerge, "merge", ""},
-		{InternalKeyKindMerge, "", ""},
-		{InternalKeyKindRangeDelete, "a", "b"},
-		{InternalKeyKindRangeDelete, "", ""},
-		{InternalKeyKindLogData, "logdata", ""},
-		{InternalKeyKindLogData, "", ""},
 	}
 	var b Batch
 	for _, tc := range testCases {
 		switch tc.kind {
 		case InternalKeyKindSet:
-			_ = b.Set([]byte(tc.key), []byte(tc.value), nil)
+			b.Set([]byte(tc.key), []byte(tc.value), nil)
 		case InternalKeyKindMerge:
-			_ = b.Merge([]byte(tc.key), []byte(tc.value), nil)
+			b.Merge([]byte(tc.key), []byte(tc.value), nil)
 		case InternalKeyKindDelete:
-			_ = b.Delete([]byte(tc.key), nil)
-		case InternalKeyKindRangeDelete:
-			_ = b.DeleteRange([]byte(tc.key), []byte(tc.value), nil)
-		case InternalKeyKindLogData:
-			_ = b.LogData([]byte(tc.key), nil)
+			b.Delete([]byte(tc.key), nil)
 		}
 	}
-	r := b.Reader()
+	iter := b.iter()
 	for _, tc := range testCases {
-		kind, k, v, ok := r.Next()
+		kind, k, v, ok := iter.next()
 		if !ok {
 			t.Fatalf("next returned !ok: test case = %v", tc)
 		}
@@ -69,8 +59,8 @@ func TestBatch(t *testing.T) {
 				kind, key, value, tc.kind, tc.key, tc.value)
 		}
 	}
-	if len(r) != 0 {
-		t.Errorf("reader was not exhausted: remaining bytes = %q", r)
+	if len(iter) != 0 {
+		t.Errorf("iterator was not exhausted: remaining bytes = %q", iter)
 	}
 }
 
@@ -295,7 +285,7 @@ func TestFlushableBatchIter(t *testing.T) {
 	})
 }
 
-func TestFlushableBatch(t *testing.T) {
+func TestFlushableBatchSeqNum(t *testing.T) {
 	var b *flushableBatch
 	datadriven.RunTest(t, "testdata/flushable_batch", func(d *datadriven.TestData) string {
 		switch d.Cmd {
@@ -317,26 +307,6 @@ func TestFlushableBatch(t *testing.T) {
 			b = newFlushableBatch(batch, DefaultComparer)
 			return ""
 
-		case "iter":
-			var opts IterOptions
-			for _, arg := range d.CmdArgs {
-				if len(arg.Vals) != 1 {
-					return fmt.Sprintf("%s: %s=<value>", d.Cmd, arg.Key)
-				}
-				switch arg.Key {
-				case "lower":
-					opts.LowerBound = []byte(arg.Vals[0])
-				case "upper":
-					opts.UpperBound = []byte(arg.Vals[0])
-				default:
-					return fmt.Sprintf("%s: unknown arg: %s", d.Cmd, arg.Key)
-				}
-			}
-
-			iter := b.newIter(&opts)
-			defer iter.Close()
-			return runInternalIterCmd(d, iter)
-
 		case "dump":
 			if len(d.CmdArgs) != 1 || len(d.CmdArgs[0].Vals) != 1 || d.CmdArgs[0].Key != "seq" {
 				return fmt.Sprintf("dump seq=<value>\n")
diff --git a/db_test.go b/db_test.go
index 2ba3a4b..569b448 100644
--- a/db_test.go
+++ b/db_test.go
@@ -690,45 +690,3 @@ func TestRollManifest(t *testing.T) {
 		t.Fatal(err)
 	}
 }
-
-func TestDBClosed(t *testing.T) {
-	d, err := Open("", &Options{
-		FS: vfs.NewMem(),
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-	if err := d.Close(); err != nil {
-		t.Fatal(err)
-	}
-
-	catch := func(f func()) (err error) {
-		defer func() {
-			if r := recover(); r != nil {
-				err = r.(error)
-			}
-		}()
-		f()
-		return nil
-	}
-
-	require.EqualValues(t, ErrClosed, catch(func() { _ = d.Close() }))
-
-	require.EqualValues(t, ErrClosed, catch(func() { _ = d.Compact(nil, nil) }))
-	require.EqualValues(t, ErrClosed, catch(func() { _ = d.Flush() }))
-	require.EqualValues(t, ErrClosed, catch(func() { _ = d.AsyncFlush() }))
-
-	require.EqualValues(t, ErrClosed, catch(func() { _, _ = d.Get(nil) }))
-	require.EqualValues(t, ErrClosed, catch(func() { _ = d.Delete(nil, nil) }))
-	require.EqualValues(t, ErrClosed, catch(func() { _ = d.DeleteRange(nil, nil, nil) }))
-	require.EqualValues(t, ErrClosed, catch(func() { _ = d.LogData(nil, nil) }))
-	require.EqualValues(t, ErrClosed, catch(func() { _ = d.Merge(nil, nil, nil) }))
-	require.EqualValues(t, ErrClosed, catch(func() { _ = d.Set(nil, nil, nil) }))
-
-	require.EqualValues(t, ErrClosed, catch(func() { _ = d.NewSnapshot() }))
-
-	b := d.NewIndexedBatch()
-	require.EqualValues(t, ErrClosed, catch(func() { _ = b.Commit(nil) }))
-	require.EqualValues(t, ErrClosed, catch(func() { _ = d.Apply(b, nil) }))
-	require.EqualValues(t, ErrClosed, catch(func() { _ = b.NewIter(nil) }))
-}
diff --git a/get_iter_test.go b/get_iter_test.go
index 91c437d..0666511 100644
--- a/get_iter_test.go
+++ b/get_iter_test.go
@@ -522,7 +522,7 @@ func TestGetIter(t *testing.T) {
 			i.iter = get
 
 			defer i.Close()
-			if !i.First() {
+			if !i.Next() {
 				err := i.Error()
 				if err != nil {
 					return nil, err
diff --git a/internal/randvar/zipf_test.go b/internal/randvar/zipf_test.go
index 8da8980..df655f2 100644
--- a/internal/randvar/zipf_test.go
+++ b/internal/randvar/zipf_test.go
@@ -71,34 +71,6 @@ func TestZeta(t *testing.T) {
 	})
 }
 
-func TestZetaIncMax(t *testing.T) {
-	// Construct a zipf generator covering the range [0,10] incrementally.
-	z0, err := NewZipf(nil, 0, 0, 0.99)
-	if err != nil {
-		t.Fatal(err)
-	}
-	for i := 0; i < 10; i++ {
-		z0.IncMax(1)
-	}
-
-	// Contruct a zipf generator covering the range [0,10] via the constructor.
-	z10, err := NewZipf(nil, 0, 10, 0.99)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	z0.mu.Lock()
-	defer z0.mu.Unlock()
-	z10.mu.Lock()
-	defer z10.mu.Unlock()
-	if z0.mu.zetaN != z10.mu.zetaN {
-		t.Fatalf("expected zetaN %v, but found %v", z10.mu.zetaN, z0.mu.zetaN)
-	}
-	if z0.mu.eta != z10.mu.eta {
-		t.Fatalf("expected eta %v, but found %v", z10.mu.eta, z0.mu.eta)
-	}
-}
-
 func TestNewZipf(t *testing.T) {
 	var gens = []struct {
 		min, max uint64
diff --git a/snapshot_test.go b/snapshot_test.go
index b3b2f37..f4ca494 100644
--- a/snapshot_test.go
+++ b/snapshot_test.go
@@ -13,7 +13,6 @@ import (
 
 	"github.com/petermattis/pebble/internal/datadriven"
 	"github.com/petermattis/pebble/vfs"
-	"github.com/stretchr/testify/require"
 )
 
 func TestSnapshotListToSlice(t *testing.T) {
@@ -164,30 +163,3 @@ func TestSnapshot(t *testing.T) {
 		}
 	})
 }
-
-func TestSnapshotClosed(t *testing.T) {
-	d, err := Open("", &Options{
-		FS: vfs.NewMem(),
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	catch := func(f func()) (err error) {
-		defer func() {
-			if r := recover(); r != nil {
-				err = r.(error)
-			}
-		}()
-		f()
-		return nil
-	}
-
-	snap := d.NewSnapshot()
-	if err := snap.Close(); err != nil {
-		t.Fatal(err)
-	}
-	require.EqualValues(t, ErrClosed, catch(func() { _ = snap.Close() }))
-	require.EqualValues(t, ErrClosed, catch(func() { _, _ = snap.Get(nil) }))
-	require.EqualValues(t, ErrClosed, catch(func() { snap.NewIter(nil) }))
-}
diff --git a/testdata/compaction_picker_level_max_bytes b/testdata/compaction_picker_level_max_bytes
index d610e5c..a6b74f4 100644
--- a/testdata/compaction_picker_level_max_bytes
+++ b/testdata/compaction_picker_level_max_bytes
@@ -11,13 +11,13 @@ init 1
 6: 2
 ----
 5: 1
-6: 2
+6: 1
 
 init 1
 6: 2
 ----
 5: 1
-6: 2
+6: 1
 
 init 1
 3: 1
@@ -26,9 +26,9 @@ init 1
 6: 1
 ----
 3: 1
-4: 1
-5: 1
-6: 1
+4: 10
+5: 100
+6: 1000
 
 init 1
 1: 1
@@ -39,11 +39,11 @@ init 1
 6: 1
 ----
 1: 1
-2: 1
-3: 1
-4: 1
-5: 1
-6: 1
+2: 10
+3: 100
+4: 1000
+5: 10000
+6: 100000
 
 init 1
 1: 1
@@ -100,45 +100,25 @@ init 1
 5: 10000
 6: 100000
 
-# Smoothing multiplier is
-# `(size(Lbottom)/size(Lbase))^(1/(Lbottom-Lbase)) = (1000000/1)^(1/(6-1)) = 1000000^(1/5)`
-#
-# size(L1) = size(Lbase) = 1
-# size(L2) = size(L1) * 1000000^(1/5) ~= 16
-# size(L3) = size(L2) * 1000000^(1/5) ~= 251
-# size(L4) = size(L3) * 1000000^(1/5) ~= 3981
-# size(L5) = size(L4) * 1000000^(1/5) ~= 63096
-# size(L6) = size(L5) * 1000000^(1/5) ~= 1000000
-
 init 1
 6: 1000000
 ----
 1: 1
-2: 16
-3: 251
-4: 3981
-5: 63096
-6: 1000000
-
-# Smoothing multiplier is
-# `(size(Lbottom)/size(Lbase))^(Lbottom-Lbase) = (64000000/64)^(1/(6-1)) = 1000000^(1/5)`
-#
-# size(L1) = size(Lbase) = 64
-# size(L2) = size(L1) * 1000000^(1/5) ~= 1014
-# size(L3) = size(L2) * 1000000^(1/5) ~= 16076
-# size(L4) = size(L3) * 1000000^(1/5) ~= 254789
-# size(L5) = size(L4) * 1000000^(1/5) ~= 4038127
-# size(L6) = size(L5) * 1000000^(1/5) ~= 64000000
+2: 10
+3: 100
+4: 1000
+5: 10000
+6: 100000
 
 init 64
 6: 64000000
 ----
 1: 64
-2: 1014
-3: 16076
-4: 254789
-5: 4038127
-6: 64000000
+2: 640
+3: 6400
+4: 64000
+5: 640000
+6: 6400000
 
 init 1
 0: 4
diff --git a/testdata/compaction_picker_target_level b/testdata/compaction_picker_target_level
index 3d22bd0..ab0df71 100644
--- a/testdata/compaction_picker_target_level
+++ b/testdata/compaction_picker_target_level
@@ -24,18 +24,11 @@ pick 1
 ----
 5: 2.0
 
-# Smoothing multiplier is
-# `(size(Lbottom)/size(Lbase))^(Lbottom-Lbase) = (30/1)^(1/(6-4)) = 30^(1/2)`
-#
-# size(L4) = size(Lbase) = 1
-# size(L5) = size(L4) * 30^(1/2) ~= 5
-# size(L6) = size(L5) * 30^(1/2) = 30
-
 pick 1
 5: 2
 6: 30
 ----
-5: 0.4
+5: 2.0
 
 pick 1
 4: 2
diff --git a/testdata/event_listener b/testdata/event_listener
index 28f3577..cb768a3 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -101,5 +101,5 @@ level__files____size___score______in__ingest____move____read___write___w-amp
     3      0     0 B    0.00     0 B     0 B     0 B     0 B     0 B     0.0
     4      0     0 B    0.00     0 B     0 B     0 B     0 B     0 B     0.0
     5      1   959 B    0.00     0 B   959 B     0 B     0 B     0 B     0.0
-    6      1   959 B    1.00   1.9 K     0 B     0 B   1.9 K   959 B     0.5
+    6      1   959 B    0.00   1.9 K     0 B     0 B   1.9 K   959 B     0.5
 total      2   1.9 K    0.00   1.0 K   959 B     0 B   1.9 K   3.8 K     3.8
diff --git a/testdata/flushable_batch b/testdata/flushable_batch
index 2f84099..315750c 100644
--- a/testdata/flushable_batch
+++ b/testdata/flushable_batch
@@ -36,107 +36,3 @@ dump seq=1000
 a#1002,1:1
 b#1001,1:1
 c#1000,1:1
-
-iter
-first
-next
-next
-next
-----
-a:1
-b:1
-c:1
-.
-
-iter
-last
-prev
-prev
-prev
-----
-c:1
-b:1
-a:1
-.
-
-iter lower=b
-seek-ge b
-prev
-----
-b:1
-.
-
-iter lower=c
-seek-lt b
-----
-.
-
-iter lower=e
-last
-----
-.
-
-iter upper=1
-first
-----
-.
-
-iter upper=b
-first
-next
-----
-a:1
-.
-
-iter upper=b
-seek-ge b
-----
-.
-
-iter
-set-bounds lower=b
-seek-ge b
-prev
-----
-b:1
-.
-
-iter
-set-bounds lower=c
-seek-lt b
-----
-.
-
-iter
-set-bounds lower=e
-last
-----
-.
-
-iter
-set-bounds upper=1
-first
-----
-.
-
-iter
-set-bounds upper=b
-first
-next
-----
-a:1
-.
-
-iter
-set-bounds upper=b
-seek-ge b
-----
-.
-
-iter
-first
-set-bounds upper=b
-seek-ge b
-----
-a:1
-.
diff --git a/testdata/iterator b/testdata/iterator
index 3b13ea6..f7b0ff6 100644
--- a/testdata/iterator
+++ b/testdata/iterator
@@ -9,7 +9,7 @@ prev
 ----
 a:b
 .
-.
+a:b
 
 iter seq=2
 seek-ge b
@@ -34,7 +34,7 @@ prev
 ----
 a:b
 .
-.
+a:b
 
 iter seq=3
 seek-ge a
@@ -43,7 +43,7 @@ prev
 ----
 a:c
 .
-.
+a:c
 
 iter seq=2
 seek-prefix-ge a
@@ -52,7 +52,7 @@ prev
 ----
 a:b
 .
-.
+a:b
 
 iter seq=3
 seek-prefix-ge a
@@ -61,7 +61,7 @@ prev
 ----
 a:c
 .
-.
+a:c
 
 
 define
@@ -93,7 +93,7 @@ next
 ----
 a:b
 .
-.
+a:b
 
 iter seq=3
 seek-prefix-ge a
@@ -185,7 +185,7 @@ next
 ----
 a:a
 .
-.
+a:a
 
 iter seq=4
 seek-lt c
@@ -196,7 +196,7 @@ next
 b:b
 a:a
 .
-.
+a:a
 
 
 iter seq=4
@@ -210,7 +210,7 @@ c:c
 b:b
 a:a
 .
-.
+a:a
 
 iter seq=4
 seek-prefix-ge a
@@ -253,7 +253,7 @@ prev
 ----
 a:b
 .
-.
+a:b
 
 iter seq=2
 seek-ge b
@@ -272,7 +272,7 @@ next
 ----
 a:b
 .
-.
+a:b
 
 iter seq=2
 seek-lt c
@@ -281,7 +281,7 @@ next
 ----
 a:b
 .
-.
+a:b
 
 iter seq=2
 seek-prefix-ge a
@@ -290,7 +290,7 @@ prev
 ----
 a:b
 .
-.
+a:b
 
 iter seq=2
 seek-prefix-ge b
@@ -488,7 +488,7 @@ prev
 a:bcd
 b:ab
 .
-.
+b:ab
 
 iter seq=3
 seek-ge a
@@ -513,7 +513,7 @@ next
 b:ab
 a:bcd
 .
-.
+a:bcd
 
 iter seq=3
 seek-lt c
@@ -995,46 +995,6 @@ last
 .
 a:a
 
-iter seq=2
-last
-next
-set-bounds upper=c
-prev
-----
-d:d
-.
-.
-.
-
-iter seq=2
-first
-prev
-set-bounds lower=b
-next
-----
-a:a
-.
-.
-.
-
-iter seq=2
-set-bounds lower=b
-seek-lt c
-next
-----
-.
-b:b
-c:c
-
-iter seq=2
-set-bounds upper=d
-seek-ge c
-prev
-----
-.
-c:c
-b:b
-
 define
 a.SET.1:a
 aa.SET.1:aa
diff --git a/testdata/snapshot b/testdata/snapshot
index 42a9661..aa93e38 100644
--- a/testdata/snapshot
+++ b/testdata/snapshot
@@ -14,18 +14,20 @@ prev
 ----
 a:1
 .
-.
+a:1
 
 iter snapshot=2
 first
 next
 next
 prev
+prev
 ----
 a:1
 b:2
 .
-.
+b:2
+a:1
 
 iter snapshot=3
 first
@@ -33,12 +35,16 @@ next
 next
 next
 prev
+prev
+prev
 ----
 a:1
 b:2
 c:3
 .
-.
+c:3
+b:2
+a:1
 
 define
 set a 1
@@ -56,7 +62,7 @@ prev
 ----
 a:1
 .
-.
+a:1
 
 iter snapshot=2
 first
@@ -65,7 +71,7 @@ prev
 ----
 a:2
 .
-.
+a:2
 
 iter snapshot=3
 first
@@ -74,7 +80,7 @@ prev
 ----
 a:3
 .
-.
+a:3
 
 define
 set a 1
@@ -93,7 +99,7 @@ prev
 ----
 a:1
 .
-.
+a:1
 
 iter snapshot=2
 first
@@ -102,7 +108,7 @@ prev
 ----
 a:2
 .
-.
+a:2
 
 iter snapshot=3
 first
@@ -111,7 +117,7 @@ prev
 ----
 a:3
 .
-.
+a:3
 
 define
 merge a 1
@@ -130,7 +136,7 @@ prev
 ----
 a:1
 .
-.
+a:1
 
 iter snapshot=2
 first
@@ -139,7 +145,7 @@ prev
 ----
 a:21
 .
-.
+a:21
 
 iter snapshot=3
 first
@@ -148,4 +154,4 @@ prev
 ----
 a:321
 .
-.
+a:321
diff --git a/vfs/syncing_file_linux_test.go b/vfs/syncing_file_linux_test.go
index 0de5d2d..142d618 100644
--- a/vfs/syncing_file_linux_test.go
+++ b/vfs/syncing_file_linux_test.go
@@ -15,31 +15,6 @@ import (
 	"unsafe"
 )
 
-func TestSyncRangeSmokeTest(t *testing.T) {
-	testCases := []struct {
-		err      error
-		expected bool
-	}{
-		{nil, true},
-		{syscall.EINVAL, true},
-		{syscall.ENOSYS, false},
-	}
-	for i, c := range testCases {
-		t.Run("", func(t *testing.T) {
-			ok := syncRangeSmokeTest(uintptr(i),
-				func(fd int, off int64, n int64, flags int) (err error) {
-					if i != fd {
-						t.Fatalf("expected fd %d, but got %d", i, fd)
-					}
-					return c.err
-				})
-			if c.expected != ok {
-				t.Fatalf("expected %t, but got %t: %v", c.expected, ok, c.err)
-			}
-		})
-	}
-}
-
 func BenchmarkDirectIOWrite(b *testing.B) {
 	const targetSize = 16 << 20
 	const alignment = 4096