Repository: petermattis/pebble
Pull request #82 (2019-04-20 01:07:37)

Batch, memTable, and DB.Ingest doc comments

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


Patch:
diff --git a/batch.go b/batch.go
index 07be5b6..0847242 100644
--- a/batch.go
+++ b/batch.go
@@ -78,7 +78,100 @@ func (s *batchStorage) Compare(a []byte, b uint32) int {
 	return 1
 }
 
-// Batch is a sequence of Sets and/or Deletes that are applied atomically.
+// A Batch is a sequence of Sets, Merges, Deletes, and/or DeleteRanges that are
+// applied atomically. Batch implements the Reader interface, but only an
+// indexed batch supports reading (without error) via Get or NewIter. A
+// non-indexed batch will return ErrNotIndexed when read from .
+//
+// Indexing
+//
+// Batches can be optionally indexed (see DB.NewIndexedBatch). An indexed batch
+// allows iteration via an Iterator (see Batch.NewIter). The iterator provides
+// a merged view of the operations in the batch and the underlying
+// database. This is implemented by treating the batch as an additional layer
+// in the LSM where every entry in the batch is considered newer than any entry
+// in the underlying database (batch entries have the InternalKeySeqNumBatch
+// bit set). By treating the batch as an additional layer in the LSM, iteration
+// supports all batch operations (i.e. Set, Merge, Delete, and DeleteRange)
+// with minimal effort.
+//
+// The same key can be operated on multiple times in a batch, though only the
+// latest operation will be visible. For example, Put("a", "b"), Delete("a")
+// will cause the key "a" to not be visible in the batch. Put("a", "b"),
+// Put("a", "c") will cause a read of "a" to return the value "c".
+//
+// The batch index is implemented via an skiplist (internal/batchskl). While
+// the skiplist implementation is very fast, inserting into an indexed batch is
+// significantly slower than inserting into a non-indexed batch. Only use an
+// indexed batch if you require reading from it.
+//
+// Atomic commit
+//
+// The operations in a batch are persisted by calling Batch.Commit which is
+// equivalent to calling DB.Apply(batch). A batch is committed atomically by
+// writing the internal batch representation to the WAL, adding all of the
+// batch operations to the memtable associated with the WAL, and then
+// incrementing the visible sequence number so that subsequent reads can see
+// the effects of the batch operations. If WriteOptions.Sync is true, a call to
+// Batch.Commit will guarantee that the batch is persisted to disk before
+// returning. See commitPipeline for more on the implementation details.
+//
+// Large batches
+//
+// The size of a batch is limited only by available memory (be aware that
+// indexed batches require considerably additional memory for the skiplist
+// structure). A given WAL file has a single memtable associated with it (this
+// restriction could be removed, but doing so is onerous and complex). And a
+// memtable has a fixed size due to the underlying fixed size arena. Note that
+// this differs from RocksDB where a memtable can grow arbitrarily large using
+// a list of arena chunks. In RocksDB this is accomplished by storing pointers
+// in the arena memory, but that isn't possible in Go.
+//
+// During Batch.Commit, a batch which is larger than a threshold (>
+// MemTableSize/2) is wrapped in a flushableBatch and inserted into the queue
+// of memtables. A flushableBatch forces WAL to be rotated, but that happens
+// anyways when the memtable becomes full so this does not cause significant
+// WAL churn. Because the flushableBatch is readable as another layer in the
+// LSM, Batch.Commit returns as soon as the flushableBatch has been added to
+// the queue of memtables.
+//
+// Internally, a flushableBatch provides Iterator support by sorting the batch
+// contents (the batch is sorted once, when it is added to the memtable
+// queue). Sorting the batch contents and insertion of the contents into a
+// memtable have the same big-O time, but the constant factor dominates
+// here. Sorting is significantly faster and uses significantly less memory.
+//
+// Internal representation
+//
+// The internal batch representation is a contiguous byte buffer with a fixed
+// 12-byte header, followed by a series of records.
+//
+//   +-------------+------------+--- ... ---+
+//   | SeqNum (8B) | Count (4B) |  Entries  |
+//   +-------------+------------+--- ... ---+
+//
+// Each record has a 1-byte kind tag prefix, followed by 1 or 2 length prefixed
+// strings (varstring):
+//
+//   +-----------+-----------------+-------------------+
+//   | Kind (1B) | Key (varstring) | Value (varstring) |
+//   +-----------+-----------------+-------------------+
+//
+// A varstring is a varint32 followed by N bytes of data. The Kind tags are
+// exactly those specified by db.InternalKeyKind. The following table shows the
+// format for records of each kind:
+//
+//   InternalKeyKindDelete       varstring
+//   InternalKeyKindSet          varstring varstring
+//   InternalKeyKindMerge        varstring varstring
+//   InternalKeyKindRangeDelete  varstring varstring
+//
+// The intuitive understanding here are that the arguments to Delete(), Set(),
+// Merge(), and DeleteRange() are encoded into the batch.
+//
+// The internal batch representation is the on disk format for a batch in the
+// WAL, and thus stable. New record kinds may be added, but the existing ones
+// will not be modified.
 type Batch struct {
 	storage batchStorage
 
diff --git a/compaction.go b/compaction.go
index f22b71c..3b137f8 100644
--- a/compaction.go
+++ b/compaction.go
@@ -480,7 +480,6 @@ func (d *DB) flush1() error {
 			close(d.mu.mem.queue[i].flushed())
 		}
 		d.mu.mem.queue = d.mu.mem.queue[n:]
-		d.updateReadStateLocked()
 		return nil
 	}
 
@@ -507,7 +506,6 @@ func (d *DB) flush1() error {
 		close(d.mu.mem.queue[i].flushed())
 	}
 	d.mu.mem.queue = d.mu.mem.queue[n:]
-	d.updateReadStateLocked()
 
 	// var newDirty int
 	// for _, mem := range d.mu.mem.queue {
@@ -749,7 +747,6 @@ func (d *DB) compact1() (err error) {
 	if err != nil {
 		return err
 	}
-	d.updateReadStateLocked()
 	d.deleteObsoleteFiles(jobID)
 	return nil
 }
diff --git a/db.go b/db.go
index db1bd0b..616f070 100644
--- a/db.go
+++ b/db.go
@@ -137,13 +137,6 @@ type DB struct {
 	largeBatchThreshold int
 	optionsFileNum      uint64
 
-	// readState provides access to the state needed for reading without needing
-	// to acquire DB.mu.
-	readState struct {
-		sync.RWMutex
-		val *readState
-	}
-
 	// TODO(peter): describe exactly what this mutex protects. So far: every
 	// field in the struct.
 	mu struct {
@@ -199,10 +192,14 @@ func (d *DB) Get(key []byte) ([]byte, error) {
 }
 
 func (d *DB) getInternal(key []byte, b *Batch, s *Snapshot) ([]byte, error) {
-	// 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().
-	readState := d.loadReadState()
+	d.mu.Lock()
+	// Grab and reference the current version to prevent its underlying files
+	// from being deleted if we have a concurrent compaction. Note that
+	// version.unref() can be called without holding DB.mu.
+	current := d.mu.versions.currentVersion()
+	current.ref()
+	memtables := d.mu.mem.queue
+	d.mu.Unlock()
 
 	// Determine the seqnum to read at after grabbing the read state (current and
 	// memtables) above.
@@ -225,16 +222,16 @@ func (d *DB) getInternal(key []byte, b *Batch, s *Snapshot) ([]byte, error) {
 	get.snapshot = seqNum
 	get.key = key
 	get.batch = b
-	get.mem = readState.memtables
-	get.l0 = readState.current.files[0]
-	get.version = readState.current
+	get.mem = memtables
+	get.l0 = current.files[0]
+	get.version = current
 
 	i := &buf.dbi
 	i.cmp = d.cmp
 	i.equal = d.equal
 	i.merge = d.merge
 	i.iter = get
-	i.readState = readState
+	i.version = current
 
 	defer i.Close()
 	if !i.Next() {
@@ -382,10 +379,14 @@ func (d *DB) newIterInternal(
 	s *Snapshot,
 	o *db.IterOptions,
 ) *Iterator {
-	// 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().
-	readState := d.loadReadState()
+	d.mu.Lock()
+	// Grab and reference the current version to prevent its underlying files
+	// from being deleted if we have a concurrent compaction. Note that
+	// version.unref() can be called without holding DB.mu.
+	current := d.mu.versions.currentVersion()
+	current.ref()
+	memtables := d.mu.mem.queue
+	d.mu.Unlock()
 
 	// Determine the seqnum to read at after grabbing the read state (current and
 	// memtables) above.
@@ -412,7 +413,7 @@ func (d *DB) newIterInternal(
 	dbi.cmp = d.cmp
 	dbi.equal = d.equal
 	dbi.merge = d.merge
-	dbi.readState = readState
+	dbi.version = current
 
 	iters := buf.iters[:0]
 	rangeDelIters := buf.rangeDelIters[:0]
@@ -426,7 +427,6 @@ func (d *DB) newIterInternal(
 	// TODO(peter): We only need to add memtables which contain sequence numbers
 	// older than seqNum. Unfortunately, memtables don't track their oldest
 	// sequence number currently.
-	memtables := readState.memtables
 	for i := len(memtables) - 1; i >= 0; i-- {
 		mem := memtables[i]
 		iters = append(iters, mem.newIter(o))
@@ -435,7 +435,6 @@ func (d *DB) newIterInternal(
 	}
 
 	// The level 0 files need to be added from newest to oldest.
-	current := readState.current
 	for i := len(current.files[0]) - 1; i >= 0; i-- {
 		f := &current.files[0][i]
 		iter, rangeDelIter, err := d.newIters(f)
@@ -558,8 +557,6 @@ func (d *DB) Close() error {
 	d.mu.closed = true
 
 	if err == nil {
-		d.readState.val.unrefLocked()
-
 		current := d.mu.versions.currentVersion()
 		for v := d.mu.versions.versions.front(); true; v = v.next {
 			refs := atomic.LoadInt32(&v.refs)
@@ -791,7 +788,6 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			d.mu.mem.mutable = newMemTable(d.opts)
 		}
 		d.mu.mem.queue = append(d.mu.mem.queue, d.mu.mem.mutable)
-		d.updateReadStateLocked()
 		if (imm != nil && imm.unref()) || scheduleFlush {
 			d.maybeScheduleFlush()
 		}
diff --git a/ingest.go b/ingest.go
index 54796bf..6a3e802 100644
--- a/ingest.go
+++ b/ingest.go
@@ -213,6 +213,34 @@ func ingestTargetLevel(cmp db.Compare, v *version, meta *fileMetadata) int {
 // flushed. The ingested sstable files are moved into the DB and must reside on
 // the same filesystem as the DB. Sstables can be created for ingestion using
 // sstable.Writer.
+//
+// Ingestion loads each sstable into the lowest level of the LSM which it
+// doesn't overlap (see ingestTargetLevel). If an sstable overlaps a memtable,
+// ingestion forces the memtable to flush, and then waits for the flush to
+// occur.
+//
+// The steps for ingestion are:
+//
+//   1. Allocate file numbers for every sstable beign ingested.
+//   2. Load the metadata for all sstables being ingest.
+//   3. Sort the sstables by smallest key, verifying non overlap.
+//   4. Hard link the sstables into the DB directory.
+//   5. Allocate a sequence number to use for all of the entries in the
+//      sstables. This is the step where overlap with memtables is
+//      determined. If there is overlap, we remember the most recent memtable
+//      that overlaps.
+//   6. Update the sequence number in the ingested sstables.
+//   7. Wait for the most recent memtable that overlaps to flush (if any).
+//   8. Add the ingested sstables to the version (DB.ingestApply).
+//   9. Publish the ingestion sequence number.
+//
+// Note that if the mutable memtable overlaps with ingestion, a flush of the
+// memtable is forced equivalent to DB.Flush. Additionally, subsequent
+// mutations that get sequence numbers larger than the ingestion sequence
+// number get queued up behind the ingestion waiting for it to complete. This
+// can produce a noticeable hiccup in performance. See
+// https://github.com/petermattis/pebble/issues/25 for an idea for how to fix
+// this hiccup.
 func (d *DB) Ingest(paths []string) error {
 	// Allocate file numbers for all of the files being ingested and mark them as
 	// pending in order to prevent them from being deleted. Note that this causes
@@ -357,6 +385,5 @@ func (d *DB) ingestApply(meta []*fileMetadata) (*versionEdit, error) {
 	if err := d.mu.versions.logAndApply(ve); err != nil {
 		return nil, err
 	}
-	d.updateReadStateLocked()
 	return ve, nil
 }
diff --git a/iterator.go b/iterator.go
index 391c3a0..4dbfdb6 100644
--- a/iterator.go
+++ b/iterator.go
@@ -36,7 +36,7 @@ type Iterator struct {
 	equal     db.Equal
 	merge     db.Merge
 	iter      internalIterator
-	readState *readState
+	version   *version
 	err       error
 	key       []byte
 	keyBuf    []byte
@@ -373,9 +373,9 @@ func (i *Iterator) Error() error {
 // It is valid to call Close multiple times. Other methods should not be
 // called after the iterator has been closed.
 func (i *Iterator) Close() error {
-	if i.readState != nil {
-		i.readState.unref()
-		i.readState = nil
+	if i.version != nil {
+		i.version.unref()
+		i.version = nil
 	}
 	if err := i.iter.Close(); err != nil && i.err != nil {
 		i.err = err
diff --git a/mem_table.go b/mem_table.go
index 616644a..116d3f1 100644
--- a/mem_table.go
+++ b/mem_table.go
@@ -18,14 +18,31 @@ func memTableEntrySize(keyBytes, valueBytes int) uint32 {
 	return arenaskl.MaxNodeSize(uint32(keyBytes)+8, uint32(valueBytes))
 }
 
-// memTable is a memory-backed implementation of the db.Reader interface.
+// A memTable implements an in-memory layer of the LSM. A memTable is mutable,
+// but append-only. Records are added, but never removed. Deletion is supported
+// via tombstones, but it is up to higher level code (see Iterator) to support
+// processing those tombstones.
 //
-// It is safe to call Get, Set, and Find concurrently.
+// A memTable is implemented on top of a lock-free arena-backed skiplist. An
+// arena is a fixed size contiguous chunk of memory (see
+// db.Options.MemTableSize). A memTable's memory consumtion is thus fixed at
+// the time of creation (with the exception of the cached fragmented range
+// tombstones). The arena-backed skiplist provides both forward and reverse
+// links which makes forward and reverse iteration the same speed.
 //
-// A memTable's memory consumption increases monotonically, even if keys are
-// deleted or values are updated with shorter slices. Users are responsible for
-// explicitly compacting a memTable into a separate DB (whether in-memory or
-// on-disk) when appropriate.
+// A batch is "applied" to a memTable in a two step process: prepare(batch) ->
+// apply(batch). memTable.prepare() is not thread-safe and must be called with
+// external sychronization. Preparation reserves space in the memTable for the
+// batch. Note that we pessimistically compute how much space a batch will
+// consume in the memTable (see memTableEntrySize and
+// Batch.memTableSize). Preparation is an O(1) operation. Applying a batch to
+// the memTable can be performed concurrently with other apply
+// operations. Applying a batch is an O(n logm) operation where N is the number
+// of records in the batch and M is the number of records in the memtable. The
+// commitPipeline serializes batch preparation, and allows batch application to
+// proceed concurrently.
+//
+// It is safe to call get, apply, newIter, and newRangeDelIter concurrently.
 type memTable struct {
 	cmp         db.Compare
 	equal       db.Equal
diff --git a/open.go b/open.go
index dae415c..57b6f13 100644
--- a/open.go
+++ b/open.go
@@ -174,7 +174,6 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	if err := d.mu.versions.logAndApply(&ve); err != nil {
 		return nil, err
 	}
-	d.updateReadStateLocked()
 
 	// Write the current options to disk.
 	d.optionsFileNum = d.mu.versions.nextFileNum()
diff --git a/read_state.go b/read_state.go
deleted file mode 100644
index e507a3b..0000000
--- a/read_state.go
+++ /dev/null
@@ -1,82 +0,0 @@
-// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
-// of this source code is governed by a BSD-style license that can be found in
-// the LICENSE file.
-
-package pebble
-
-import (
-	"sync/atomic"
-)
-
-// readState encapsulates the state needed for reading (the current version and
-// list of memtables). Loading the readState is done without grabbing
-// DB.mu. Instead, a separate DB.readState.RWMutex is used for
-// synchronization. This mutex solely covers the current readState object which
-// means it is rarely or ever contended.
-//
-// Note that various fancy lock-free mechanisms can be imagined for loading the
-// readState, but benchmarking showed the ones considered to purely be
-// pessimizations. The RWMutex version is a single atomic increment for the
-// RLock and an atomic decrement for the RUnlock. It is difficult to do better
-// than that without something like thread-local storage which isn't available
-// in Go.
-type readState struct {
-	refcnt    int32
-	current   *version
-	memtables []flushable
-}
-
-// ref adds a reference to the readState.
-func (s *readState) ref() {
-	atomic.AddInt32(&s.refcnt, 1)
-}
-
-// unref removes a reference to the readState. If this was the last reference,
-// the reference the readState holds on the version is released. Requires DB.mu
-// is NOT held as version.unref() will acquire it. See unrefLocked() if DB.mu
-// is held by the caller.
-func (s *readState) unref() {
-	if atomic.AddInt32(&s.refcnt, -1) == 0 {
-		s.current.unref()
-	}
-}
-
-// unrefLocked removes a reference to the readState. If this was the last
-// reference, the reference the readState holds on the version is
-// released. Requires DB.mu is held as version.unrefLocked() requires it. See
-// unref() if DB.mu is NOT held by the caller.
-func (s *readState) unrefLocked() {
-	if atomic.AddInt32(&s.refcnt, -1) == 0 {
-		s.current.unrefLocked()
-	}
-}
-
-// loadReadState returns the current readState. The returned readState must be
-// unreferenced when the caller is finished with it.
-func (d *DB) loadReadState() *readState {
-	d.readState.RLock()
-	state := d.readState.val
-	state.ref()
-	d.readState.RUnlock()
-	return state
-}
-
-// updateReadStateLocked creates a new readState from the current version and
-// list of memtables. Requires DB.mu is held.
-func (d *DB) updateReadStateLocked() {
-	s := &readState{
-		refcnt:    1,
-		current:   d.mu.versions.currentVersion(),
-		memtables: d.mu.mem.queue,
-	}
-	s.current.ref()
-
-	d.readState.Lock()
-	old := d.readState.val
-	d.readState.val = s
-	d.readState.Unlock()
-
-	if old != nil {
-		old.unrefLocked()
-	}
-}


Test patch:
diff --git a/data_test.go b/data_test.go
index da48407..dac8627 100644
--- a/data_test.go
+++ b/data_test.go
@@ -273,7 +273,6 @@ func runDBDefineCmd(td *datadriven.TestData) (*DB, error) {
 				if !d.mu.mem.mutable.empty() {
 					d.mu.mem.mutable = newMemTable(d.opts)
 					d.mu.mem.queue = append(d.mu.mem.queue, d.mu.mem.mutable)
-					d.updateReadStateLocked()
 				}
 				mem = d.mu.mem.mutable
 				fields = fields[1:]
@@ -308,7 +307,6 @@ func runDBDefineCmd(td *datadriven.TestData) (*DB, error) {
 		if err := d.mu.versions.logAndApply(ve); err != nil {
 			return nil, err
 		}
-		d.updateReadStateLocked()
 		for i := range ve.newFiles {
 			meta := &ve.newFiles[i].meta
 			delete(d.mu.compact.pendingOutputs, meta.fileNum)
diff --git a/read_state_test.go b/read_state_test.go
deleted file mode 100644
index d694093..0000000
--- a/read_state_test.go
+++ /dev/null
@@ -1,47 +0,0 @@
-// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
-// of this source code is governed by a BSD-style license that can be found in
-// the LICENSE file.
-
-package pebble
-
-import (
-	"fmt"
-	"math/rand"
-	"testing"
-	"time"
-
-	"github.com/petermattis/pebble/db"
-	"github.com/petermattis/pebble/storage"
-)
-
-func BenchmarkReadState(b *testing.B) {
-	d, err := Open("", &db.Options{
-		Storage: storage.NewMem(),
-	})
-	if err != nil {
-		b.Fatal(err)
-	}
-
-	for _, updateFrac := range []float32{0, 0.1, 0.5} {
-		b.Run(fmt.Sprintf("updates=%.0f", updateFrac*100), func(b *testing.B) {
-			b.RunParallel(func(pb *testing.PB) {
-				rng := rand.New(rand.NewSource(time.Now().UnixNano()))
-
-				for pb.Next() {
-					if rng.Float32() < updateFrac {
-						d.mu.Lock()
-						d.updateReadStateLocked()
-						d.mu.Unlock()
-					} else {
-						s := d.loadReadState()
-						s.unref()
-					}
-				}
-			})
-		})
-	}
-
-	if err := d.Close(); err != nil {
-		b.Fatal(err)
-	}
-}