Repository: cockroachdb/pebble
Pull request #677 (2020-05-18 00:40:47)

cmd/pebble: miscellaneous ycsb benchmark improvements
* internal/randvar: reduce exclusive locks in randvar
* cmd/pebble: miscellaneous ycsb benchmark improvements
* internal/rate: change Reserve{,N} to return a value
* cmd/pebble: take CPU profiles every 10 seconds
* cmd/pebble: pull in the current CRDB version of MVCC routines
* cmd/pebble: fix badger/boltdb/rocksdb build problems
* cmd/pebble: match CRDB Pebble options
* cmd/pebble: fix reverse scan keys metric


Patch:
diff --git a/cmd/pebble/badger.go b/cmd/pebble/badger.go
index 581dfb6a..5a1040ad 100644
--- a/cmd/pebble/badger.go
+++ b/cmd/pebble/badger.go
@@ -11,7 +11,7 @@ import (
 	"log"
 
 	"github.com/cockroachdb/pebble"
-	"github.com/dgraph-io/badger"
+	"github.com/dgraph-io/badger/v2"
 )
 
 // Adapters for Badger.
@@ -20,7 +20,7 @@ type badgerDB struct {
 }
 
 func newBadgerDB(dir string) DB {
-	db, err := badger.Open(badger.DefaultOptions(dir))
+	db, err := badger.Open(badger.DefaultOptions(dir).WithMaxCacheSize(cacheSize))
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -49,8 +49,8 @@ func (b badgerDB) Scan(key []byte, count int64, reverse bool) error {
 	panic("badgerDB.Scan: unimplemented")
 }
 
-func (b badgerDB) Metrics() *pebble.VersionMetrics {
-	return &pebble.VersionMetrics{}
+func (b badgerDB) Metrics() *pebble.Metrics {
+	return &pebble.Metrics{}
 }
 
 func (b badgerDB) Flush() error {
@@ -137,7 +137,3 @@ func (b badgerBatch) Set(key, value []byte, _ *pebble.WriteOptions) error {
 func (b badgerBatch) LogData(data []byte, _ *pebble.WriteOptions) error {
 	panic("badgerBatch.logData: unimplemented")
 }
-
-func (b badgerBatch) Repr() []byte {
-	return nil
-}
diff --git a/cmd/pebble/boltdb.go b/cmd/pebble/boltdb.go
index 3f065e69..2c879b24 100644
--- a/cmd/pebble/boltdb.go
+++ b/cmd/pebble/boltdb.go
@@ -10,8 +10,8 @@ import (
 	"bytes"
 	"log"
 
+	"github.com/boltdb/bolt"
 	"github.com/cockroachdb/pebble"
-	"github.com/ngaut/bolt"
 )
 
 // Adapters for BoltDB
@@ -63,8 +63,8 @@ func (b boltDB) Scan(key []byte, count int64, reverse bool) error {
 	panic("boltDB.Scan: unimplemented")
 }
 
-func (b boltDB) Metrics() *pebble.VersionMetrics {
-	return &pebble.VersionMetrics{}
+func (b boltDB) Metrics() *pebble.Metrics {
+	return &pebble.Metrics{}
 }
 
 func (b boltDB) Flush() error {
@@ -160,7 +160,3 @@ func (b boltDBBatch) Set(key, value []byte, _ *pebble.WriteOptions) error {
 func (b boltDBBatch) LogData(data []byte, _ *pebble.WriteOptions) error {
 	panic("boltDBBatch.logData: unimplemented")
 }
-
-func (b boltDBBatch) Repr() []byte {
-	return nil
-}
diff --git a/cmd/pebble/db.go b/cmd/pebble/db.go
index 2f13e03b..1d777f5c 100644
--- a/cmd/pebble/db.go
+++ b/cmd/pebble/db.go
@@ -8,6 +8,7 @@ import (
 	"log"
 
 	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/bloom"
 	"github.com/cockroachdb/pebble/internal/bytealloc"
 )
 
@@ -37,13 +38,13 @@ type batch interface {
 	Commit(opts *pebble.WriteOptions) error
 	Set(key, value []byte, opts *pebble.WriteOptions) error
 	LogData(data []byte, opts *pebble.WriteOptions) error
-	Repr() []byte
 }
 
 // Adapters for Pebble. Since the interfaces above are based on Pebble's
 // interfaces, it can simply forward calls for everything.
 type pebbleDB struct {
-	d *pebble.DB
+	d       *pebble.DB
+	ballast []byte
 }
 
 func newPebbleDB(dir string) DB {
@@ -53,21 +54,34 @@ func newPebbleDB(dir string) DB {
 		Cache:                       cache,
 		Comparer:                    mvccComparer,
 		DisableWAL:                  disableWAL,
-		MemTableSize:                64 << 20,
-		MemTableStopWritesThreshold: 4,
-		MaxConcurrentCompactions:    2,
-		MinCompactionRate:           4 << 20, // 4 MB/s
-		MinFlushRate:                1 << 20, // 1 MB/s
 		L0CompactionThreshold:       2,
-		L0StopWritesThreshold:       400,
+		L0StopWritesThreshold:       1000,
 		LBaseMaxBytes:               64 << 20, // 64 MB
-		Levels: []pebble.LevelOptions{{
-			BlockSize: 32 << 10,
-		}},
+		Levels:                      make([]pebble.LevelOptions, 7),
+		MaxConcurrentCompactions:    3,
+		MaxOpenFiles:                16384,
+		MemTableSize:                64 << 20,
+		MemTableStopWritesThreshold: 4,
 		Merger: &pebble.Merger{
 			Name: "cockroach_merge_operator",
 		},
+		MinCompactionRate: 4 << 20, // 4 MB/s
+		MinFlushRate:      4 << 20, // 4 MB/s
 	}
+
+	for i := 0; i < len(opts.Levels); i++ {
+		l := &opts.Levels[i]
+		l.BlockSize = 32 << 10       // 32 KB
+		l.IndexBlockSize = 256 << 10 // 256 KB
+		l.FilterPolicy = bloom.FilterPolicy(10)
+		l.FilterType = pebble.TableFilter
+		if i > 0 {
+			l.TargetFileSize = opts.Levels[i-1].TargetFileSize * 2
+		}
+		l.EnsureDefaults()
+	}
+	opts.Levels[6].FilterPolicy = nil
+
 	opts.EnsureDefaults()
 
 	if verbose {
@@ -82,7 +96,10 @@ func newPebbleDB(dir string) DB {
 	if err != nil {
 		log.Fatal(err)
 	}
-	return pebbleDB{p}
+	return pebbleDB{
+		d:       p,
+		ballast: make([]byte, 1<<30),
+	}
 }
 
 func (p pebbleDB) Flush() error {
diff --git a/cmd/pebble/main.go b/cmd/pebble/main.go
index 1acd71ac..f3870eb4 100644
--- a/cmd/pebble/main.go
+++ b/cmd/pebble/main.go
@@ -19,7 +19,7 @@ var (
 	disableWAL      bool
 	duration        time.Duration
 	engineType      string
-	maxOpsPerSec    = newRateFlag("1000000")
+	maxOpsPerSec    = newRateFlag("")
 	verbose         bool
 	waitCompactions bool
 	wipe            bool
diff --git a/cmd/pebble/mvcc.go b/cmd/pebble/mvcc.go
index 9cc0ac60..0e388de1 100644
--- a/cmd/pebble/mvcc.go
+++ b/cmd/pebble/mvcc.go
@@ -30,11 +30,62 @@ var mvccComparer = &pebble.Comparer{
 	},
 
 	Separator: func(dst, a, b []byte) []byte {
-		return append(dst, a...)
+		aKey, _, ok := mvccSplitKey(a)
+		if !ok {
+			return append(dst, a...)
+		}
+		bKey, _, ok := mvccSplitKey(b)
+		if !ok {
+			return append(dst, a...)
+		}
+		// If the keys are the same just return a.
+		if bytes.Equal(aKey, bKey) {
+			return append(dst, a...)
+		}
+		n := len(dst)
+		// MVCC key comparison uses bytes.Compare on the roachpb.Key, which is the same semantics as
+		// pebble.DefaultComparer, so reuse the latter's Separator implementation.
+		dst = pebble.DefaultComparer.Separator(dst, aKey, bKey)
+		// Did it pick a separator different than aKey -- if it did not we can't do better than a.
+		buf := dst[n:]
+		if bytes.Equal(aKey, buf) {
+			return append(dst[:n], a...)
+		}
+		// The separator is > aKey, so we only need to add the timestamp sentinel.
+		return append(dst, 0)
 	},
 
 	Successor: func(dst, a []byte) []byte {
-		return append(dst, a...)
+		aKey, _, ok := mvccSplitKey(a)
+		if !ok {
+			return append(dst, a...)
+		}
+		n := len(dst)
+		// MVCC key comparison uses bytes.Compare on the roachpb.Key, which is the same semantics as
+		// pebble.DefaultComparer, so reuse the latter's Successor implementation.
+		dst = pebble.DefaultComparer.Successor(dst, aKey)
+		// Did it pick a successor different than aKey -- if it did not we can't do better than a.
+		buf := dst[n:]
+		if bytes.Equal(aKey, buf) {
+			return append(dst[:n], a...)
+		}
+		// The successor is > aKey, so we only need to add the timestamp sentinel.
+		return append(dst, 0)
+	},
+
+	Split: func(k []byte) int {
+		key, _, ok := mvccSplitKey(k)
+		if !ok {
+			return len(k)
+		}
+		// This matches the behavior of libroach/KeyPrefix. RocksDB requires that
+		// keys generated via a SliceTransform be comparable with normal encoded
+		// MVCC keys. Encoded MVCC keys have a suffix indicating the number of
+		// bytes of timestamp data. MVCC keys without a timestamp have a suffix of
+		// 0. We're careful in EncodeKey to make sure that the user-key always has
+		// a trailing 0. If there is no timestamp this falls out naturally. If
+		// there is a timestamp we prepend a 0 to the encoded timestamp data.
+		return len(key) + 1
 	},
 
 	Name: "cockroach_comparator",
@@ -57,23 +108,44 @@ func mvccSplitKey(mvccKey []byte) (key []byte, ts []byte, ok bool) {
 }
 
 func mvccCompare(a, b []byte) int {
-	aKey, aTS, aOK := mvccSplitKey(a)
-	bKey, bTS, bOK := mvccSplitKey(b)
-	if !aOK || !bOK {
+	// NB: For performance, this routine manually splits the key into the
+	// user-key and timestamp components rather than using SplitMVCCKey. Don't
+	// try this at home kids: use SplitMVCCKey.
+
+	aEnd := len(a) - 1
+	bEnd := len(b) - 1
+	if aEnd < 0 || bEnd < 0 {
+		// This should never happen unless there is some sort of corruption of
+		// the keys. This is a little bizarre, but the behavior exactly matches
+		// engine/db.cc:DBComparator.
+		return bytes.Compare(a, b)
+	}
+
+	// Compute the index of the separator between the key and the timestamp.
+	aSep := aEnd - int(a[aEnd])
+	bSep := bEnd - int(b[bEnd])
+	if aSep < 0 || bSep < 0 {
 		// This should never happen unless there is some sort of corruption of
-		// the keys.
+		// the keys. This is a little bizarre, but the behavior exactly matches
+		// engine/db.cc:DBComparator.
 		return bytes.Compare(a, b)
 	}
-	if c := bytes.Compare(aKey, bKey); c != 0 {
+
+	// Compare the "user key" part of the key.
+	if c := bytes.Compare(a[:aSep], b[:bSep]); c != 0 {
 		return c
 	}
+
+	// Compare the timestamp part of the key.
+	aTS := a[aSep:aEnd]
+	bTS := b[bSep:bEnd]
 	if len(aTS) == 0 {
 		if len(bTS) == 0 {
 			return 0
 		}
 		return -1
 	} else if len(bTS) == 0 {
-		return +1
+		return 1
 	}
 	return bytes.Compare(bTS, aTS)
 }
diff --git a/cmd/pebble/random.go b/cmd/pebble/random.go
index 0d038222..0d713648 100644
--- a/cmd/pebble/random.go
+++ b/cmd/pebble/random.go
@@ -44,6 +44,15 @@ func (f *rateFlag) Type() string {
 
 // Set implements the Flag.Value interface.
 func (f *rateFlag) Set(spec string) error {
+	if spec == "" {
+		if err := f.Flag.Set("0"); err != nil {
+			return err
+		}
+		f.fluctuateDuration = time.Duration(0)
+		f.spec = spec
+		return nil
+	}
+
 	parts := strings.Split(spec, "/")
 	if len(parts) == 0 || len(parts) > 2 {
 		return errors.Errorf("invalid ratevar spec: %s", errors.Safe(spec))
@@ -65,18 +74,34 @@ func (f *rateFlag) Set(spec string) error {
 }
 
 func (f *rateFlag) newRateLimiter() *rate.Limiter {
-	limiter := rate.NewLimiter(rate.Limit(f.Uint64()), 1)
+	if f.spec == "" {
+		return nil
+	}
+	rng := randvar.NewRand()
+	limiter := rate.NewLimiter(rate.Limit(f.Uint64(rng)), 1)
 	if f.fluctuateDuration != 0 {
 		go func(limiter *rate.Limiter) {
 			ticker := time.NewTicker(f.fluctuateDuration)
 			for range ticker.C {
-				limiter.SetLimit(rate.Limit(f.Uint64()))
+				limiter.SetLimit(rate.Limit(f.Uint64(rng)))
 			}
 		}(limiter)
 	}
 	return limiter
 }
 
+func wait(l *rate.Limiter) {
+	if l == nil {
+		return
+	}
+
+	now := time.Now()
+	r := l.ReserveN(now, 1)
+	if d := r.DelayFrom(now); d > 0 {
+		time.Sleep(d)
+	}
+}
+
 type sequence struct {
 	val         int64
 	cycleLength int64
diff --git a/cmd/pebble/rocksdb.go b/cmd/pebble/rocksdb.go
index 3a0c328b..625d678d 100644
--- a/cmd/pebble/rocksdb.go
+++ b/cmd/pebble/rocksdb.go
@@ -7,30 +7,28 @@
 package main
 
 import (
+	"context"
 	"log"
 	"strconv"
 	"strings"
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
-	"github.com/cockroachdb/cockroach/pkg/storage/engine"
+	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/pebble"
 )
 
 // Adapters for rocksDB
 type rocksDB struct {
-	d       *engine.RocksDB
+	d       *storage.RocksDB
 	ballast []byte
 }
 
 func newRocksDB(dir string) DB {
 	// TODO: match Pebble / Rocks options
-	r, err := engine.NewRocksDB(
-		engine.RocksDBConfig{
-			Dir: dir,
-		},
-		engine.NewRocksDBCache(cacheSize),
-	)
+	cfg := storage.RocksDBConfig{}
+	cfg.Dir = dir
+	r, err := storage.NewRocksDB(cfg, storage.NewRocksDBCache(cacheSize))
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -41,13 +39,13 @@ func newRocksDB(dir string) DB {
 }
 
 type rocksDBIterator struct {
-	iter       engine.Iterator
+	iter       storage.Iterator
 	lowerBound []byte
 	upperBound []byte
 }
 
 type rocksDBBatch struct {
-	batch engine.Batch
+	batch storage.Batch
 }
 
 func (i rocksDBIterator) SeekGE(key []byte) bool {
@@ -56,7 +54,7 @@ func (i rocksDBIterator) SeekGE(key []byte) bool {
 	if !ok {
 		panic("mvccSplitKey failed")
 	}
-	i.iter.Seek(engine.MVCCKey{
+	i.iter.SeekGE(storage.MVCCKey{
 		Key: userKey,
 	})
 	return i.Valid()
@@ -92,7 +90,7 @@ func (i rocksDBIterator) Last() bool {
 	if !ok {
 		panic("mvccSplitKey failed")
 	}
-	i.iter.SeekReverse(engine.MVCCKey{
+	i.iter.SeekLT(storage.MVCCKey{
 		Key: userKey,
 	})
 	return i.Valid()
@@ -120,23 +118,19 @@ func (b rocksDBBatch) Set(key, value []byte, _ *pebble.WriteOptions) error {
 		panic("mvccSplitKey failed")
 	}
 	ts := hlc.Timestamp{WallTime: 1}
-	return b.batch.Put(engine.MVCCKey{Key: userKey, Timestamp: ts}, value)
+	return b.batch.Put(storage.MVCCKey{Key: userKey, Timestamp: ts}, value)
 }
 
 func (b rocksDBBatch) LogData(data []byte, _ *pebble.WriteOptions) error {
 	return b.batch.LogData(data)
 }
 
-func (b rocksDBBatch) Repr() []byte {
-	return b.batch.Repr()
-}
-
 func (r rocksDB) Flush() error {
 	return r.d.Flush()
 }
 
 func (r rocksDB) NewIter(opts *pebble.IterOptions) iterator {
-	ropts := engine.IterOptions{}
+	ropts := storage.IterOptions{}
 	if opts != nil {
 		ropts.LowerBound = opts.LowerBound
 		ropts.UpperBound = opts.UpperBound
@@ -162,37 +156,31 @@ func (r rocksDB) Scan(key []byte, count int64, reverse bool) error {
 		panic("mvccSplitKey failed")
 	}
 	endKey := roachpb.KeyMax
-	ropts := engine.IterOptions{
-		LowerBound: key,
-	}
 	if reverse {
 		endKey = beginKey
 		beginKey = roachpb.KeyMin
-		ropts.UpperBound = key
-		ropts.LowerBound = nil
 	}
 
-	iter := r.d.NewIterator(ropts)
-	defer iter.Close()
 	// We hard code a timestamp with walltime=1 in the data, so we just have to
 	// use a larger timestamp here (walltime=10).
 	ts := hlc.Timestamp{WallTime: 10}
-	_, numKVs, _, intents, err := iter.MVCCScan(
-		beginKey, endKey, count, ts, engine.MVCCScanOptions{Reverse: reverse},
+	res, err := storage.MVCCScanToBytes(
+		context.Background(), r.d, beginKey, endKey, ts,
+		storage.MVCCScanOptions{Reverse: reverse, MaxKeys: count},
 	)
-	if numKVs > count {
+	if res.NumKeys > count {
 		panic("MVCCScan returned too many keys")
 	}
-	if len(intents) > 0 {
+	if len(res.Intents) > 0 {
 		panic("MVCCScan found intents")
 	}
 	return err
 }
 
-func (r rocksDB) Metrics() *pebble.VersionMetrics {
+func (r rocksDB) Metrics() *pebble.Metrics {
 	stats := r.d.GetCompactionStats()
 	var inLevelsSection bool
-	var vMetrics pebble.VersionMetrics
+	var vMetrics pebble.Metrics
 	for _, line := range strings.Split(stats, "\n") {
 		if strings.HasPrefix(line, "-----") {
 			continue
@@ -247,8 +235,197 @@ func (r rocksDB) Metrics() *pebble.VersionMetrics {
 			bytesReadGB, _ := strconv.ParseFloat(fields[5], 64)
 			vMetrics.Levels[level].BytesRead = uint64(1024.0 * 1024.0 * 1024.0 * bytesReadGB)
 			bytesWrittenGB, _ := strconv.ParseFloat(fields[8], 64)
-			vMetrics.Levels[level].BytesWritten = uint64(1024.0 * 1024.0 * 1024.0 * bytesWrittenGB)
+			vMetrics.Levels[level].BytesCompacted = uint64(1024.0 * 1024.0 * 1024.0 * bytesWrittenGB)
 		}
 	}
 	return &vMetrics
 }
+
+type crdbPebbleDB struct {
+	d       *storage.Pebble
+	ballast []byte
+}
+
+func newCRDBPebbleDB(dir string) DB {
+	cfg := storage.PebbleConfig{}
+	cfg.Dir = dir
+	// TODO(peter): We can't create a pebble.Cache here because cfg.Opts.Cache is
+	// the vendored type and we can't import that vendored directory here. The
+	// following diff can be applied to github.com/cockroachdb/cockroach/pkg/storage/pebble.go:
+	//
+	// --- a/pkg/storage/pebble.go
+	// +++ b/pkg/storage/pebble.go
+	// @@ -377,6 +377,7 @@ func (l pebbleLogger) Fatalf(format string, args ...interface{}) {
+	//  type PebbleConfig struct {
+	//         // StorageConfig contains storage configs for all storage engines.
+	//         base.StorageConfig
+	// +       CacheSize int64
+	//         // Pebble specific options.
+	//         Opts *pebble.Options
+	//  }
+	// @@ -461,6 +462,11 @@ func ResolveEncryptedEnvOptions(
+
+	//  // NewPebble creates a new Pebble instance, at the specified path.
+	//  func NewPebble(ctx context.Context, cfg PebbleConfig) (*Pebble, error) {
+	// +       if cfg.Opts.Cache == nil {
+	// +               cfg.Opts.Cache = pebble.NewCache(cfg.CacheSize)
+	// +               defer cfg.Opts.Cache.Unref()
+	// +       }
+	// +
+	//
+	// With the above diff applied, we can uncomment the line below:
+	//
+	// cfg.CacheSize = cacheSize
+	cfg.Opts = storage.DefaultPebbleOptions()
+	r, err := storage.NewPebble(context.Background(), cfg)
+	if err != nil {
+		log.Fatal(err)
+	}
+	return crdbPebbleDB{
+		d:       r,
+		ballast: make([]byte, 1<<30),
+	}
+}
+
+type crdbPebbleDBIterator struct {
+	iter       storage.Iterator
+	lowerBound []byte
+	upperBound []byte
+}
+
+type crdbPebbleDBBatch struct {
+	batch storage.Batch
+}
+
+func (i crdbPebbleDBIterator) SeekGE(key []byte) bool {
+	// TODO: unnecessary overhead here. Change the interface.
+	userKey, _, ok := mvccSplitKey(key)
+	if !ok {
+		panic("mvccSplitKey failed")
+	}
+	i.iter.SeekGE(storage.MVCCKey{
+		Key: userKey,
+	})
+	return i.Valid()
+}
+
+func (i crdbPebbleDBIterator) Valid() bool {
+	valid, _ := i.iter.Valid()
+	return valid
+}
+
+func (i crdbPebbleDBIterator) Key() []byte {
+	key := i.iter.Key()
+	return []byte(key.Key)
+}
+
+func (i crdbPebbleDBIterator) Value() []byte {
+	return i.iter.Value()
+}
+
+func (i crdbPebbleDBIterator) First() bool {
+	return i.SeekGE(i.lowerBound)
+}
+
+func (i crdbPebbleDBIterator) Next() bool {
+	i.iter.Next()
+	valid, _ := i.iter.Valid()
+	return valid
+}
+
+func (i crdbPebbleDBIterator) Last() bool {
+	// TODO: unnecessary overhead here. Change the interface.
+	userKey, _, ok := mvccSplitKey(i.upperBound)
+	if !ok {
+		panic("mvccSplitKey failed")
+	}
+	i.iter.SeekLT(storage.MVCCKey{
+		Key: userKey,
+	})
+	return i.Valid()
+}
+
+func (i crdbPebbleDBIterator) Prev() bool {
+	i.iter.Prev()
+	valid, _ := i.iter.Valid()
+	return valid
+}
+
+func (i crdbPebbleDBIterator) Close() error {
+	i.iter.Close()
+	return nil
+}
+
+func (b crdbPebbleDBBatch) Commit(opts *pebble.WriteOptions) error {
+	return b.batch.Commit(opts.Sync)
+}
+
+func (b crdbPebbleDBBatch) Set(key, value []byte, _ *pebble.WriteOptions) error {
+	// TODO: unnecessary overhead here. Change the interface.
+	userKey, _, ok := mvccSplitKey(key)
+	if !ok {
+		panic("mvccSplitKey failed")
+	}
+	ts := hlc.Timestamp{WallTime: 1}
+	return b.batch.Put(storage.MVCCKey{Key: userKey, Timestamp: ts}, value)
+}
+
+func (b crdbPebbleDBBatch) LogData(data []byte, _ *pebble.WriteOptions) error {
+	return b.batch.LogData(data)
+}
+
+func (r crdbPebbleDB) Flush() error {
+	return r.d.Flush()
+}
+
+func (r crdbPebbleDB) NewIter(opts *pebble.IterOptions) iterator {
+	ropts := storage.IterOptions{}
+	if opts != nil {
+		ropts.LowerBound = opts.LowerBound
+		ropts.UpperBound = opts.UpperBound
+	} else {
+		ropts.UpperBound = roachpb.KeyMax
+	}
+	iter := r.d.NewIterator(ropts)
+	return crdbPebbleDBIterator{
+		iter:       iter,
+		lowerBound: ropts.LowerBound,
+		upperBound: ropts.UpperBound,
+	}
+}
+
+func (r crdbPebbleDB) NewBatch() batch {
+	return crdbPebbleDBBatch{r.d.NewBatch()}
+}
+
+func (r crdbPebbleDB) Scan(key []byte, count int64, reverse bool) error {
+	// TODO: unnecessary overhead here. Change the interface.
+	beginKey, _, ok := mvccSplitKey(key)
+	if !ok {
+		panic("mvccSplitKey failed")
+	}
+	endKey := roachpb.KeyMax
+	if reverse {
+		endKey = beginKey
+		beginKey = roachpb.KeyMin
+	}
+
+	// We hard code a timestamp with walltime=1 in the data, so we just have to
+	// use a larger timestamp here (walltime=10).
+	ts := hlc.Timestamp{WallTime: 10}
+	res, err := storage.MVCCScanToBytes(
+		context.Background(), r.d, beginKey, endKey, ts,
+		storage.MVCCScanOptions{Reverse: reverse, MaxKeys: count},
+	)
+	if res.NumKeys > count {
+		panic("MVCCScan returned too many keys")
+	}
+	if len(res.Intents) > 0 {
+		panic("MVCCScan found intents")
+	}
+	return err
+}
+
+func (r crdbPebbleDB) Metrics() *pebble.Metrics {
+	return &pebble.Metrics{}
+}
diff --git a/cmd/pebble/rocksdb_disabled.go b/cmd/pebble/rocksdb_disabled.go
index 9d657bc5..e6889e0e 100644
--- a/cmd/pebble/rocksdb_disabled.go
+++ b/cmd/pebble/rocksdb_disabled.go
@@ -12,3 +12,8 @@ func newRocksDB(dir string) DB {
 	log.Fatalf("pebble not compiled with RocksDB support: recompile with \"-tags rocksdb\"\n")
 	return nil
 }
+
+func newCRDBPebbleDB(dir string) DB {
+	log.Fatalf("pebble not compiled with CRDB support: recompile with \"-tags rocksdb\"\n")
+	return nil
+}
diff --git a/cmd/pebble/scan.go b/cmd/pebble/scan.go
index ebcacf9e..51886e13 100644
--- a/cmd/pebble/scan.go
+++ b/cmd/pebble/scan.go
@@ -5,7 +5,6 @@
 package main
 
 import (
-	"context"
 	"fmt"
 	"log"
 	"math"
@@ -71,9 +70,10 @@ func runScan(cmd *cobra.Command, args []string) {
 
 			for i := 0; i < count; {
 				b := d.NewBatch()
+				var value []byte
 				for end := i + batch; i < end; i++ {
 					keys[i] = mvccEncode(nil, encodeUint32Ascending([]byte("key-"), uint32(i)), uint64(i+1), 0)
-					value := scanConfig.values.Bytes(rng)
+					value = scanConfig.values.Bytes(rng, value)
 					if err := b.Set(keys[i], value, nil); err != nil {
 						log.Fatal(err)
 					}
@@ -100,8 +100,9 @@ func runScan(cmd *cobra.Command, args []string) {
 					minTS := encodeUint64Ascending(nil, math.MaxUint64)
 
 					for {
-						limiter.Wait(context.Background())
-						rows := int(rowDist.Uint64())
+						wait(limiter)
+
+						rows := int(rowDist.Uint64(rng))
 						startIdx := rng.Int31n(int32(len(keys) - rows))
 						startKey := encodeUint32Ascending(startKeyBuf[:4], uint32(startIdx))
 						endKey := encodeUint32Ascending(endKeyBuf[:4], uint32(startIdx+int32(rows)))
diff --git a/cmd/pebble/sync.go b/cmd/pebble/sync.go
index ad46b52d..5edc5a10 100644
--- a/cmd/pebble/sync.go
+++ b/cmd/pebble/sync.go
@@ -5,7 +5,6 @@
 package main
 
 import (
-	"context"
 	"fmt"
 	"log"
 	"sync"
@@ -69,14 +68,16 @@ func runSync(cmd *cobra.Command, args []string) {
 					rand := rand.New(rand.NewSource(uint64(time.Now().UnixNano())))
 					var raw []byte
 					var buf []byte
+					var block []byte
 					for {
-						limiter.Wait(context.Background())
+						wait(limiter)
+
 						start := time.Now()
 						b := d.NewBatch()
 						var n uint64
-						count := int(batchDist.Uint64())
+						count := int(batchDist.Uint64(rand))
 						for j := 0; j < count; j++ {
-							block := syncConfig.values.Bytes(rand)
+							block = syncConfig.values.Bytes(rand, block)
 
 							if syncConfig.walOnly {
 								if err := b.LogData(block, nil); err != nil {
diff --git a/cmd/pebble/ycsb.go b/cmd/pebble/ycsb.go
index f1bd252a..94debd10 100644
--- a/cmd/pebble/ycsb.go
+++ b/cmd/pebble/ycsb.go
@@ -5,7 +5,6 @@
 package main
 
 import (
-	"context"
 	"fmt"
 	"log"
 	"strconv"
@@ -32,6 +31,14 @@ const (
 	ycsbNumOps
 )
 
+var ycsbOpsMap = map[string]int{
+	"insert": ycsbInsert,
+	"read":   ycsbRead,
+	"rscan":  ycsbReverseScan,
+	"scan":   ycsbScan,
+	"update": ycsbUpdate,
+}
+
 var ycsbConfig struct {
 	batch            *randvar.Flag
 	keys             string
@@ -195,11 +202,11 @@ func ycsbParseKeyDist(d string) (randvar.Dynamic, error) {
 	totalKeys := uint64(ycsbConfig.initialKeys + ycsbConfig.prepopulatedKeys)
 	switch strings.ToLower(d) {
 	case "latest":
-		return randvar.NewDefaultSkewedLatest(nil)
+		return randvar.NewDefaultSkewedLatest()
 	case "uniform":
-		return randvar.NewUniform(nil, 1, totalKeys), nil
+		return randvar.NewUniform(1, totalKeys), nil
 	case "zipf":
-		return randvar.NewZipf(nil, 1, totalKeys, 0.99)
+		return randvar.NewZipf(1, totalKeys, 0.99)
 	default:
 		return nil, errors.Errorf("unknown distribution: %s", errors.Safe(d))
 	}
@@ -236,8 +243,15 @@ func runYcsb(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
+type ycsbBuf struct {
+	rng      *rand.Rand
+	keyBuf   []byte
+	valueBuf []byte
+}
+
 type ycsb struct {
 	writeOpts   *pebble.WriteOptions
+	weights     ycsbWeights
 	reg         *histogramRegistry
 	ops         *randvar.Weighted
 	keyDist     randvar.Dynamic
@@ -248,8 +262,6 @@ type ycsb struct {
 	numOps      uint64
 	numKeys     [ycsbNumOps]uint64
 	prevNumKeys [ycsbNumOps]uint64
-	opsMap      map[string]int
-	latency     [ycsbNumOps]*namedHistogram
 	limiter     *rate.Limiter
 }
 
@@ -261,6 +273,7 @@ func newYcsb(
 ) *ycsb {
 	y := &ycsb{
 		reg:       newHistogramRegistry(),
+		weights:   weights,
 		ops:       randvar.NewWeighted(nil, weights...),
 		keyDist:   keyDist,
 		batchDist: batchDist,
@@ -271,44 +284,51 @@ func newYcsb(
 	if disableWAL {
 		y.writeOpts = pebble.NoSync
 	}
+	return y
+}
 
-	y.opsMap = make(map[string]int)
-	maybeRegister := func(op int, name string) *namedHistogram {
-		w := weights.get(op)
-		if w == 0 {
-			return nil
-		}
-		wstr := fmt.Sprint(int(100 * w))
-		fill := strings.Repeat("_", 3-len(wstr))
-		if fill == "" {
-			fill = "_"
-		}
-		fullName := fmt.Sprintf("%s%s%s", name, fill, wstr)
-		y.opsMap[fullName] = op
-		return y.reg.Register(fullName)
+func (y *ycsb) maybeRegister(op int, name string) *namedHistogram {
+	w := y.weights.get(op)
+	if w == 0 {
+		return nil
 	}
-
-	y.latency[ycsbInsert] = maybeRegister(ycsbInsert, "insert")
-	y.latency[ycsbRead] = maybeRegister(ycsbRead, "read")
-	y.latency[ycsbScan] = maybeRegister(ycsbScan, "scan")
-	y.latency[ycsbReverseScan] = maybeRegister(ycsbReverseScan, "rscan")
-	y.latency[ycsbUpdate] = maybeRegister(ycsbUpdate, "update")
-	return y
+	wstr := fmt.Sprint(int(100 * w))
+	fill := strings.Repeat("_", 3-len(wstr))
+	if fill == "" {
+		fill = "_"
+	}
+	fullName := fmt.Sprintf("%s%s%s", name, fill, wstr)
+	return y.reg.Register(fullName)
 }
 
 func (y *ycsb) init(db DB, wg *sync.WaitGroup) {
 	if ycsbConfig.initialKeys > 0 {
-		rng := randvar.NewRand()
+		buf := &ycsbBuf{rng: randvar.NewRand()}
 
 		b := db.NewBatch()
+		size := 0
+		start := time.Now()
+		last := start
 		for i := 1; i <= ycsbConfig.initialKeys; i++ {
-			if len(b.Repr()) >= 1<<20 {
+			if now := time.Now(); now.Sub(last) >= time.Second {
+				fmt.Printf("%5s inserted %d keys (%0.1f%%)\n",
+					time.Duration(now.Sub(start).Seconds()+0.5)*time.Second,
+					i-1, 100*float64(i-1)/float64(ycsbConfig.initialKeys))
+				last = now
+			}
+			if size >= 1<<20 {
 				if err := b.Commit(y.writeOpts); err != nil {
 					log.Fatal(err)
 				}
 				b = db.NewBatch()
+				size = 0
+			}
+			key := y.makeKey(uint64(i+ycsbConfig.prepopulatedKeys), buf)
+			value := y.randBytes(buf)
+			if err := b.Set(key, value, nil); err != nil {
+				log.Fatal(err)
 			}
-			_ = b.Set(y.makeKey(uint64(i+ycsbConfig.prepopulatedKeys)), y.randBytes(rng), nil)
+			size += len(key) + len(value)
 		}
 		if err := b.Commit(y.writeOpts); err != nil {
 			log.Fatal(err)
@@ -330,28 +350,35 @@ func (y *ycsb) init(db DB, wg *sync.WaitGroup) {
 func (y *ycsb) run(db DB, wg *sync.WaitGroup) {
 	defer wg.Done()
 
-	rng := randvar.NewRand()
+	var latency [ycsbNumOps]*namedHistogram
+	for name, op := range ycsbOpsMap {
+		latency[op] = y.maybeRegister(op, name)
+	}
+
+	buf := &ycsbBuf{rng: randvar.NewRand()}
+
 	for {
-		y.limiter.Wait(context.Background())
+		wait(y.limiter)
+
 		start := time.Now()
 
 		op := y.ops.Int()
 		switch op {
 		case ycsbInsert:
-			y.insert(db, rng)
+			y.insert(db, buf)
 		case ycsbRead:
-			y.read(db, rng)
+			y.read(db, buf)
 		case ycsbScan:
-			y.scan(db, rng, false /* reverse */)
+			y.scan(db, buf, false /* reverse */)
 		case ycsbReverseScan:
-			y.scan(db, rng, true /* reverse */)
+			y.scan(db, buf, true /* reverse */)
 		case ycsbUpdate:
-			y.update(db, rng)
+			y.update(db, buf)
 		default:
 			panic("not reached")
 		}
 
-		y.latency[op].Record(time.Since(start))
+		latency[op].Record(time.Since(start))
 		if ycsbConfig.numOps > 0 &&
 			atomic.AddUint64(&y.numOps, 1) >= ycsbConfig.numOps {
 			break
@@ -373,36 +400,42 @@ func (y *ycsb) hashKey(key uint64) uint64 {
 	return h
 }
 
-func (y *ycsb) makeKey(keyNum uint64) []byte {
-	key := make([]byte, 4, 24+10)
+func (y *ycsb) makeKey(keyNum uint64, buf *ycsbBuf) []byte {
+	const size = 24 + 10
+	if cap(buf.keyBuf) < size {
+		buf.keyBuf = make([]byte, size)
+	}
+	key := buf.keyBuf[:4]
 	copy(key, "user")
 	key = strconv.AppendUint(key, y.hashKey(keyNum), 10)
 	// Use the MVCC encoding for keys. This appends a timestamp with
 	// walltime=1. That knowledge is utilized by rocksDB.Scan.
 	key = append(key, '\x00', '\x00', '\x00', '\x00', '\x00',
 		'\x00', '\x00', '\x00', '\x01', '\x09')
+	buf.keyBuf = key
 	return key
 }
 
-func (y *ycsb) nextReadKey() []byte {
+func (y *ycsb) nextReadKey(buf *ycsbBuf) []byte {
 	// NB: the range of values returned by keyDist is tied to the range returned
 	// by keyNum.Base. See how these are both incremented by ycsb.insert().
-	keyNum := y.keyDist.Uint64()
-	return y.makeKey(keyNum)
+	keyNum := y.keyDist.Uint64(buf.rng)
+	return y.makeKey(keyNum, buf)
 }
 
-func (y *ycsb) randBytes(rng *rand.Rand) []byte {
-	return y.valueDist.Bytes(rng)
+func (y *ycsb) randBytes(buf *ycsbBuf) []byte {
+	buf.valueBuf = y.valueDist.Bytes(buf.rng, buf.valueBuf)
+	return buf.valueBuf
 }
 
-func (y *ycsb) insert(db DB, rng *rand.Rand) {
-	count := y.batchDist.Uint64()
+func (y *ycsb) insert(db DB, buf *ycsbBuf) {
+	count := y.batchDist.Uint64(buf.rng)
 	keyNums := make([]uint64, count)
 
 	b := db.NewBatch()
 	for i := range keyNums {
 		keyNums[i] = y.keyNum.Next()
-		_ = b.Set(y.makeKey(keyNums[i]), y.randBytes(rng), nil)
+		_ = b.Set(y.makeKey(keyNums[i], buf), y.randBytes(buf), nil)
 	}
 	if err := b.Commit(y.writeOpts); err != nil {
 		log.Fatal(err)
@@ -420,30 +453,38 @@ func (y *ycsb) insert(db DB, rng *rand.Rand) {
 	}
 }
 
-func (y *ycsb) read(db DB, rng *rand.Rand) {
-	key := y.nextReadKey()
+func (y *ycsb) read(db DB, buf *ycsbBuf) {
+	key := y.nextReadKey(buf)
 	iter := db.NewIter(nil)
 	iter.SeekGE(key)
+	if iter.Valid() {
+		_ = iter.Key()
+		_ = iter.Value()
+	}
 	if err := iter.Close(); err != nil {
 		log.Fatal(err)
 	}
 	atomic.AddUint64(&y.numKeys[ycsbRead], 1)
 }
 
-func (y *ycsb) scan(db DB, rng *rand.Rand, reverse bool) {
-	count := y.scanDist.Uint64()
-	key := y.nextReadKey()
+func (y *ycsb) scan(db DB, buf *ycsbBuf, reverse bool) {
+	count := y.scanDist.Uint64(buf.rng)
+	key := y.nextReadKey(buf)
 	if err := db.Scan(key, int64(count), reverse); err != nil {
 		log.Fatal(err)
 	}
-	atomic.AddUint64(&y.numKeys[ycsbScan], count)
+	if reverse {
+		atomic.AddUint64(&y.numKeys[ycsbReverseScan], count)
+	} else {
+		atomic.AddUint64(&y.numKeys[ycsbScan], count)
+	}
 }
 
-func (y *ycsb) update(db DB, rng *rand.Rand) {
-	count := int(y.batchDist.Uint64())
+func (y *ycsb) update(db DB, buf *ycsbBuf) {
+	count := int(y.batchDist.Uint64(buf.rng))
 	b := db.NewBatch()
 	for i := 0; i < count; i++ {
-		_ = b.Set(y.nextReadKey(), y.randBytes(rng), nil)
+		_ = b.Set(y.nextReadKey(buf), y.randBytes(buf), nil)
 	}
 	if err := b.Commit(y.writeOpts); err != nil {
 		log.Fatal(err)
@@ -456,7 +497,7 @@ func (y *ycsb) tick(elapsed time.Duration, i int) {
 		fmt.Println("____optype__elapsed____ops/sec___keys/sec__p50(ms)__p95(ms)__p99(ms)_pMax(ms)")
 	}
 	y.reg.Tick(func(tick histogramTick) {
-		op := y.opsMap[tick.Name]
+		op := ycsbOpsMap[tick.Name]
 		numKeys := atomic.LoadUint64(&y.numKeys[op])
 		h := tick.Hist
 
@@ -478,7 +519,7 @@ func (y *ycsb) tick(elapsed time.Duration, i int) {
 func (y *ycsb) done(elapsed time.Duration) {
 	fmt.Println("\n____optype__elapsed_____ops(total)___ops/sec(cum)__keys/sec(cum)__avg(ms)__p50(ms)__p95(ms)__p99(ms)_pMax(ms)")
 	y.reg.Tick(func(tick histogramTick) {
-		op := y.opsMap[tick.Name]
+		op := ycsbOpsMap[tick.Name]
 		numKeys := atomic.LoadUint64(&y.numKeys[op])
 		h := tick.Cumulative
 
diff --git a/internal/randvar/flag.go b/internal/randvar/flag.go
index f36d1fba..d962b77a 100644
--- a/internal/randvar/flag.go
+++ b/internal/randvar/flag.go
@@ -62,15 +62,15 @@ func (f *Flag) Set(spec string) error {
 
 	switch strings.ToLower(m[1]) {
 	case "", "uniform":
-		f.Static = NewUniform(nil, uint64(min), uint64(max))
+		f.Static = NewUniform(uint64(min), uint64(max))
 	case "latest":
-		f.Static, err = NewSkewedLatest(nil, uint64(min), uint64(max), 0.99)
+		f.Static, err = NewSkewedLatest(uint64(min), uint64(max), 0.99)
 		if err != nil {
 			return err
 		}
 	case "zipf":
 		var err error
-		f.Static, err = NewZipf(nil, uint64(min), uint64(max), 0.99)
+		f.Static, err = NewZipf(uint64(min), uint64(max), 0.99)
 		if err != nil {
 			return err
 		}
@@ -131,13 +131,16 @@ func (f *BytesFlag) Set(spec string) error {
 
 // Bytes returns random bytes. The length of the random bytes comes from the
 // internal sizeFlag.
-func (f *BytesFlag) Bytes(r *rand.Rand) []byte {
-	size := int(f.sizeFlag.Uint64())
+func (f *BytesFlag) Bytes(r *rand.Rand, buf []byte) []byte {
+	size := int(f.sizeFlag.Uint64(r))
 	uniqueSize := int(float64(size) / f.targetCompression)
 	if uniqueSize < 1 {
 		uniqueSize = 1
 	}
-	data := make([]byte, size)
+	if cap(buf) < size {
+		buf = make([]byte, size)
+	}
+	data := buf[:size]
 	offset := 0
 	for offset+8 <= uniqueSize {
 		binary.LittleEndian.PutUint64(data[offset:], r.Uint64())
diff --git a/internal/randvar/randvar.go b/internal/randvar/randvar.go
index db7f7633..e5998978 100644
--- a/internal/randvar/randvar.go
+++ b/internal/randvar/randvar.go
@@ -4,10 +4,12 @@
 
 package randvar
 
+import "golang.org/x/exp/rand"
+
 // Static models a random variable that pulls from a distribution with static
 // bounds
 type Static interface {
-	Uint64() uint64
+	Uint64(rng *rand.Rand) uint64
 }
 
 // Dynamic models a random variable that pulls from a distribution with an
diff --git a/internal/randvar/uniform.go b/internal/randvar/uniform.go
index 6ced2e0f..4c4c5ccf 100644
--- a/internal/randvar/uniform.go
+++ b/internal/randvar/uniform.go
@@ -26,8 +26,7 @@ import (
 type Uniform struct {
 	min uint64
 	mu  struct {
-		sync.Mutex
-		rng *rand.Rand
+		sync.RWMutex
 		max uint64
 	}
 }
@@ -35,9 +34,8 @@ type Uniform struct {
 // NewUniform constructs a new Uniform generator with the given
 // parameters. Returns an error if the parameters are outside the accepted
 // range.
-func NewUniform(rng *rand.Rand, min, max uint64) *Uniform {
+func NewUniform(min, max uint64) *Uniform {
 	g := &Uniform{min: min}
-	g.mu.rng = ensureRand(rng)
 	g.mu.max = max
 	return g
 }
@@ -51,9 +49,9 @@ func (g *Uniform) IncMax(delta int) {
 
 // Uint64 returns a random Uint64 between min and max, drawn from a uniform
 // distribution.
-func (g *Uniform) Uint64() uint64 {
-	g.mu.Lock()
-	result := g.mu.rng.Uint64n(g.mu.max-g.min+1) + g.min
-	g.mu.Unlock()
+func (g *Uniform) Uint64(rng *rand.Rand) uint64 {
+	g.mu.RLock()
+	result := rng.Uint64n(g.mu.max-g.min+1) + g.min
+	g.mu.RUnlock()
 	return result
 }
diff --git a/internal/randvar/zipf.go b/internal/randvar/zipf.go
index 3b8e2e85..95316979 100644
--- a/internal/randvar/zipf.go
+++ b/internal/randvar/zipf.go
@@ -49,8 +49,7 @@ type Zipf struct {
 	halfPowTheta float64
 	// Mutable state.
 	mu struct {
-		sync.Mutex
-		rng   *rand.Rand
+		sync.RWMutex
 		max   uint64
 		eta   float64
 		zetaN float64
@@ -58,13 +57,13 @@ type Zipf struct {
 }
 
 // NewDefaultZipf constructs a new Zipf generator with the default parameters.
-func NewDefaultZipf(rng *rand.Rand) (*Zipf, error) {
-	return NewZipf(rng, 1, defaultMax, defaultTheta)
+func NewDefaultZipf() (*Zipf, error) {
+	return NewZipf(1, defaultMax, defaultTheta)
 }
 
 // NewZipf constructs a new Zipf generator with the given parameters.  Returns
 // an error if the parameters are outside the accepted range.
-func NewZipf(rng *rand.Rand, min, max uint64, theta float64) (*Zipf, error) {
+func NewZipf(min, max uint64, theta float64) (*Zipf, error) {
 	if min > max {
 		return nil, errors.Errorf("min %d > max %d", errors.Safe(min), errors.Safe(max))
 	}
@@ -76,7 +75,6 @@ func NewZipf(rng *rand.Rand, min, max uint64, theta float64) (*Zipf, error) {
 		min:   min,
 		theta: theta,
 	}
-	z.mu.rng = ensureRand(rng)
 	z.mu.max = max
 
 	// Compute hidden parameters.
@@ -124,9 +122,9 @@ func (z *Zipf) IncMax(delta int) {
 
 // Uint64 draws a new value between min and max, with probabilities according
 // to the Zipf distribution.
-func (z *Zipf) Uint64() uint64 {
-	z.mu.Lock()
-	u := z.mu.rng.Float64()
+func (z *Zipf) Uint64(rng *rand.Rand) uint64 {
+	u := rng.Float64()
+	z.mu.RLock()
 	uz := u * z.mu.zetaN
 	var result uint64
 	if uz < 1.0 {
@@ -137,6 +135,6 @@ func (z *Zipf) Uint64() uint64 {
 		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))
 	}
-	z.mu.Unlock()
+	z.mu.RUnlock()
 	return result
 }
diff --git a/internal/rate/rate.go b/internal/rate/rate.go
index 6bd8ad70..50db8180 100644
--- a/internal/rate/rate.go
+++ b/internal/rate/rate.go
@@ -188,7 +188,7 @@ func (r *Reservation) CancelAt(now time.Time) {
 }
 
 // Reserve is shorthand for ReserveN(time.Now(), 1).
-func (lim *Limiter) Reserve() *Reservation {
+func (lim *Limiter) Reserve() Reservation {
 	return lim.ReserveN(time.Now(), 1)
 }
 
@@ -206,9 +206,8 @@ func (lim *Limiter) Reserve() *Reservation {
 // Use this method if you wish to wait and slow down in accordance with the rate limit without dropping events.
 // If you need to respect a deadline or cancel the delay, use Wait instead.
 // To drop or skip events exceeding rate limit, use Allow instead.
-func (lim *Limiter) ReserveN(now time.Time, n int) *Reservation {
-	r := lim.reserveN(now, n, InfDuration)
-	return &r
+func (lim *Limiter) ReserveN(now time.Time, n int) Reservation {
+	return lim.reserveN(now, n, InfDuration)
 }
 
 // Wait is shorthand for WaitN(ctx, 1).


Test patch:
diff --git a/cmd/pebble/test.go b/cmd/pebble/test.go
index 88689d30..3c5e1a27 100644
--- a/cmd/pebble/test.go
+++ b/cmd/pebble/test.go
@@ -9,6 +9,7 @@ import (
 	"log"
 	"os"
 	"os/signal"
+	"runtime"
 	"runtime/pprof"
 	"sort"
 	"sync"
@@ -25,19 +26,57 @@ const (
 )
 
 func startCPUProfile() func() {
-	f, err := os.Create("cpu.prof")
-	if err != nil {
-		log.Fatal(err)
-	}
-	// runtime.SetBlockProfileRate(1)
-	// runtime.SetMutexProfileFraction(1)
+	runtime.SetMutexProfileFraction(1000)
+
+	doneCh := make(chan struct{})
+	var doneWG sync.WaitGroup
+	doneWG.Add(1)
+
+	go func() {
+		defer doneWG.Done()
+
+		start := time.Now()
+		t := time.NewTicker(10 * time.Second)
+		defer t.Stop()
+
+		var currentProfile *os.File
+		defer func() {
+			if currentProfile != nil {
+				pprof.StopCPUProfile()
+				currentProfile.Close()
+			}
+		}()
+
+		for {
+			if currentProfile != nil {
+				pprof.StopCPUProfile()
+				currentProfile.Close()
+				currentProfile = nil
+			}
+			path := fmt.Sprintf("cpu.%04d.prof", int(time.Since(start).Seconds()+0.5))
+			f, err := os.Create(path)
+			if err != nil {
+				log.Fatalf("unable to create cpu profile: %s", err)
+				return
+			}
+			if err := pprof.StartCPUProfile(f); err != nil {
+				log.Fatalf("unable to start cpu profile: %v", err)
+				f.Close()
+				return
+			}
+			currentProfile = f
+
+			select {
+			case <-doneCh:
+				return
+			case <-t.C:
+			}
+		}
+	}()
 
-	if err := pprof.StartCPUProfile(f); err != nil {
-		log.Fatal(err)
-	}
 	return func() {
-		pprof.StopCPUProfile()
-		f.Close()
+		close(doneCh)
+		doneWG.Wait()
 
 		if p := pprof.Lookup("heap"); p != nil {
 			f, err := os.Create("heap.prof")
@@ -50,27 +89,16 @@ func startCPUProfile() func() {
 			f.Close()
 		}
 
-		// if p := pprof.Lookup("mutex"); p != nil {
-		// 	f, err := os.Create("mutex.prof")
-		// 	if err != nil {
-		// 		log.Fatal(err)
-		// 	}
-		// 	if err := p.WriteTo(f, 0); err != nil {
-		// 		log.Fatal(err)
-		// 	}
-		// 	f.Close()
-		// }
-
-		// if p := pprof.Lookup("block"); p != nil {
-		// 	f, err := os.Create("block.prof")
-		// 	if err != nil {
-		// 		log.Fatal(err)
-		// 	}
-		// 	if err := p.WriteTo(f, 0); err != nil {
-		// 		log.Fatal(err)
-		// 	}
-		// 	f.Close()
-		// }
+		if p := pprof.Lookup("mutex"); p != nil {
+			f, err := os.Create("mutex.prof")
+			if err != nil {
+				log.Fatal(err)
+			}
+			if err := p.WriteTo(f, 0); err != nil {
+				log.Fatal(err)
+			}
+			f.Close()
+		}
 	}
 }
 
@@ -232,6 +260,8 @@ func runTest(dir string, t test) {
 		db = newBoltDB(dir)
 	case "pebble":
 		db = newPebbleDB(dir)
+	case "crdb-pebble":
+		db = newCRDBPebbleDB(dir)
 	case "rocksdb":
 		db = newRocksDB(dir)
 	}
diff --git a/internal/metamorphic/generator_test.go b/internal/metamorphic/generator_test.go
index 7340daa8..293e0e06 100644
--- a/internal/metamorphic/generator_test.go
+++ b/internal/metamorphic/generator_test.go
@@ -7,6 +7,7 @@ package metamorphic
 import (
 	"testing"
 
+	"github.com/cockroachdb/pebble/internal/randvar"
 	"github.com/stretchr/testify/require"
 )
 
@@ -120,7 +121,7 @@ func TestGenerator(t *testing.T) {
 }
 
 func TestGeneratorRandom(t *testing.T) {
-	ops := generate(ops.Uint64(), defaultConfig)
+	ops := generate(ops.Uint64(randvar.NewRand()), defaultConfig)
 	if testing.Verbose() {
 		t.Logf("\n%s", formatOps(ops))
 	}
diff --git a/internal/metamorphic/meta_test.go b/internal/metamorphic/meta_test.go
index 744c2615..94ded260 100644
--- a/internal/metamorphic/meta_test.go
+++ b/internal/metamorphic/meta_test.go
@@ -166,7 +166,7 @@ func TestMeta(t *testing.T) {
 
 	// Generate a new set of random ops, writing them to <dir>/ops. These will be
 	// read by the child processes when performing a test run.
-	ops := generate(ops.Uint64(), defaultConfig)
+	ops := generate(ops.Uint64(randvar.NewRand()), defaultConfig)
 	opsPath := filepath.Join(metaDir, "ops")
 	formattedOps := formatOps(ops)
 	require.NoError(t, ioutil.WriteFile(opsPath, []byte(formattedOps), 0644))
diff --git a/internal/randvar/skewed_latest.go b/internal/randvar/skewed_latest.go
index 59eda3cd..f7f45c48 100644
--- a/internal/randvar/skewed_latest.go
+++ b/internal/randvar/skewed_latest.go
@@ -26,7 +26,7 @@ import (
 // distribution.
 type SkewedLatest struct {
 	mu struct {
-		sync.Mutex
+		sync.RWMutex
 		max  uint64
 		zipf *Zipf
 	}
@@ -34,17 +34,17 @@ type SkewedLatest struct {
 
 // NewDefaultSkewedLatest constructs a new SkewedLatest generator with the
 // default parameters.
-func NewDefaultSkewedLatest(rng *rand.Rand) (*SkewedLatest, error) {
-	return NewSkewedLatest(rng, 1, defaultMax, defaultTheta)
+func NewDefaultSkewedLatest() (*SkewedLatest, error) {
+	return NewSkewedLatest(1, defaultMax, defaultTheta)
 }
 
 // NewSkewedLatest constructs a new SkewedLatest generator with the given
 // parameters. It returns an error if the parameters are outside the accepted
 // range.
-func NewSkewedLatest(rng *rand.Rand, min, max uint64, theta float64) (*SkewedLatest, error) {
+func NewSkewedLatest(min, max uint64, theta float64) (*SkewedLatest, error) {
 	z := &SkewedLatest{}
 	z.mu.max = max
-	zipf, err := NewZipf(rng, 0, max-min, theta)
+	zipf, err := NewZipf(0, max-min, theta)
 	if err != nil {
 		return nil, err
 	}
@@ -62,9 +62,9 @@ func (z *SkewedLatest) IncMax(delta int) {
 
 // Uint64 returns a random Uint64 between min and max, where keys near max are
 // most likely to be drawn.
-func (z *SkewedLatest) Uint64() uint64 {
-	z.mu.Lock()
-	result := z.mu.max - z.mu.zipf.Uint64()
-	z.mu.Unlock()
+func (z *SkewedLatest) Uint64(rng *rand.Rand) uint64 {
+	z.mu.RLock()
+	result := z.mu.max - z.mu.zipf.Uint64(rng)
+	z.mu.RUnlock()
 	return result
 }
diff --git a/internal/randvar/skewed_latest_test.go b/internal/randvar/skewed_latest_test.go
index c59ffb10..c9cf20e3 100644
--- a/internal/randvar/skewed_latest_test.go
+++ b/internal/randvar/skewed_latest_test.go
@@ -53,12 +53,13 @@ func dumpSamples(x []int) {
 }
 
 func TestSkewedLatest(t *testing.T) {
-	z, err := NewSkewedLatest(nil, 0, 99, 0.99)
+	rng := NewRand()
+	z, err := NewSkewedLatest(0, 99, 0.99)
 	require.NoError(t, err)
 
 	x := make([]int, 10000)
 	for i := range x {
-		x[i] = int(z.Uint64())
+		x[i] = int(z.Uint64(rng))
 	}
 
 	if testing.Verbose() {
diff --git a/internal/randvar/zipf_test.go b/internal/randvar/zipf_test.go
index 1cd98841..49cb9b7f 100644
--- a/internal/randvar/zipf_test.go
+++ b/internal/randvar/zipf_test.go
@@ -75,7 +75,7 @@ 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)
+	z0, err := NewZipf(0, 0, 0.99)
 	require.NoError(t, err)
 
 	for i := 0; i < 10; i++ {
@@ -83,7 +83,7 @@ func TestZetaIncMax(t *testing.T) {
 	}
 
 	// Contruct a zipf generator covering the range [0,10] via the constructor.
-	z10, err := NewZipf(nil, 0, 10, 0.99)
+	z10, err := NewZipf(0, 10, 0.99)
 	require.NoError(t, err)
 
 	z0.mu.Lock()
@@ -108,18 +108,19 @@ func TestNewZipf(t *testing.T) {
 	}
 
 	for _, gen := range gens {
-		_, err := NewZipf(nil, gen.min, gen.max, gen.theta)
+		_, err := NewZipf(gen.min, gen.max, gen.theta)
 		require.NoError(t, err)
 	}
 }
 
 func TestZipf(t *testing.T) {
-	z, err := NewZipf(nil, 0, 99, 0.99)
+	rng := NewRand()
+	z, err := NewZipf(0, 99, 0.99)
 	require.NoError(t, err)
 
 	x := make([]int, 10000)
 	for i := range x {
-		x[i] = int(z.Uint64())
+		x[i] = int(z.Uint64(rng))
 	}
 
 	if testing.Verbose() {