Repository: petermattis/pebble
Pull request #116 (2019-05-04 16:19:34)

Rework the commitQueue and commitPipeline syncing
Replace the commitQueue implementation with one based on the
single-producer, multi-consumer queue used to implement `sync.Pool` in
go1.13. This implementation is similar to the previous one, but uses
fewer atomic operations per enqueue/dequeue and also takes care of
nil'ing out unused slots to avoid unnecessary retention of
batches. Removed the use of the cond var to signal when the queue had
space available. That approach was racy as it isn't safe to signal a
`sync.Cond` without holding a lock on the condition being examined
unless you jump through some hoops. Instead, we configure a maximum
commit concurrency which is enforced through a semaphore (implemented
via a `chan struct{}`).

Move most of the syncing logic into LogWriter. Added
`LogWriter.SyncRecord` which is similar to `WriteRecord` except that it
optionally queues a wait group and an asynchronous sync of the log. The
wait group is signalled via a call to `Done()` when the sync
completes. Pushing the handling of syncing into `LogWriter` allowed
removal of `commitQueue.syncLoop` which was somewhat redundant with
`LogWriter.flushLoop`. It also removed the need for `LogWriter.Sync()`
which allowed further simplification in `LogWriter.flushLoop`. Lastly,
this prepares for a future where there are different syncing styles in
`LogWriter`. For example, `O_DIRECT` or `O_DSYNC` will not require an
explicit sync call. As a final benefit, it removes a performance wart
where pipeline performance fell off a cliff at high parallelism (note
that the actual parallelism is `<N>*8` below). I'm not fully clear on
why that was happening. Profiles showed significant goroutine thrashing.

Added `syncQueue`, a lock-free single-producer, single-consumer
fixed-size queue. Added `flusherCond`, a specialized cond var that
allows signalling the addition to `syncQueue` without holding the
associated lock.

```
name                           old time/op    new time/op     delta
CommitPipeline/parallel=1-8       968ns ± 4%      859ns ± 1%   -11.28%  (p=0.000 n=10+10)
CommitPipeline/parallel=2-8       861ns ± 3%      824ns ± 2%    -4.36%  (p=0.000 n=10+10)
CommitPipeline/parallel=4-8       813ns ± 2%      812ns ± 3%      ~     (p=0.985 n=9+10)
CommitPipeline/parallel=8-8       797ns ± 3%      786ns ± 9%      ~     (p=0.984 n=10+10)
CommitPipeline/parallel=16-8      800ns ± 4%      824ns ± 2%      ~     (p=0.058 n=10+7)
CommitPipeline/parallel=32-8     1.12µs ±11%     0.83µs ± 9%   -25.90%  (p=0.000 n=8+10)
CommitPipeline/parallel=64-8     2.48µs ± 7%     0.85µs ±17%   -65.78%  (p=0.000 n=10+10)
CommitPipeline/parallel=128-8    2.42µs ± 6%     0.72µs ± 7%   -70.28%  (p=0.000 n=9+9)
```

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


Patch:
diff --git a/commit.go b/commit.go
index b1bcf2f..828c4c3 100644
--- a/commit.go
+++ b/commit.go
@@ -5,105 +5,119 @@
 package pebble
 
 import (
-	"math/bits"
 	"runtime"
 	"sync"
 	"sync/atomic"
 	"unsafe"
+
+	"github.com/petermattis/pebble/internal/record"
 )
 
-type commitQueueNode struct {
-	position uint64
-	value    unsafe.Pointer
-	_        [6]uint64
-}
+// The maximum concurrency allowed for commit operations. This limit is
+// enforced by commitPipeline.sem.
+const commitConcurrency = record.SyncConcurrency
 
-// commitQueue maintains a circular ring buffer for a fixed-size
-// single-producer, multiple-consumer queue.
+// commitQueue is a lock-free fixed-size single-producer, multi-consumer
+// queue. The single producer can enqueue (push) to the head, and consumers can
+// dequeue (pop) from the tail.
+//
+// It has the added feature that it nils out unused slots to avoid unnecessary
+// retention of objects.
 type commitQueue struct {
-	// The padding members are here to ensure each item is on a separate cache
-	// line. This prevents false sharing and improves performance.
-	_     [8]uint64
-	write uint64
-	_     [8]uint64
-	read  uint64
-	_     [8]uint64
-	mask  uint64
-	_     [8]uint64
-	nodes []commitQueueNode
+	// headTail packs together a 32-bit head index and a 32-bit tail index. Both
+	// are indexes into slots modulo len(slots)-1.
+	//
+	// tail = index of oldest data in queue
+	// head = index of next slot to fill
+	//
+	// Slots in the range [tail, head) are owned by consumers.  A consumer
+	// continues to own a slot outside this range until it nils the slot, at
+	// which point ownership passes to the producer.
+	//
+	// The head index is stored in the most-significant bits so that we can
+	// atomically add to it and the overflow is harmless.
+	headTail uint64
+
+	// slots is a ring buffer of values stored in this queue. The size must be a
+	// power of 2. A slot is in use until *both* the tail index has moved beyond
+	// it and the slot value has been set to nil. The slot value is set to nil
+	// atomically by the consumer and read atomically by the producer.
+	slots [commitConcurrency]unsafe.Pointer
 }
 
-func (q *commitQueue) init() {
-	// Size the commitQueue to 8x the number of CPUs. Note that this works around
-	// a limitation (bug?) when the size of the queue is 1 (enqueue does not
-	// properly block waiting for the sequence to advance).
-	n := runtime.NumCPU()
-	n = 1 << uint((64 - bits.LeadingZeros(uint(n-1))))
-	n *= 8
-
-	q.mask = uint64(n - 1)
-	q.nodes = make([]commitQueueNode, n)
-	for i := range q.nodes {
-		q.nodes[i].position = uint64(i)
-	}
+const dequeueBits = 32
+
+func (q *commitQueue) unpack(ptrs uint64) (head, tail uint32) {
+	const mask = 1<<dequeueBits - 1
+	head = uint32((ptrs >> dequeueBits) & mask)
+	tail = uint32(ptrs & mask)
+	return
+}
+
+func (q *commitQueue) pack(head, tail uint32) uint64 {
+	const mask = 1<<dequeueBits - 1
+	return (uint64(head) << dequeueBits) |
+		uint64(tail&mask)
 }
 
-// Enqueue a single batch. Wait on cond if the queue is full.
-func (q *commitQueue) enqueue(b *Batch, cond *sync.Cond) {
-	// Note that this is a single-producer, multi-consumer queue. The q.write
-	// field is protected by commitPipeline.write.mu.
+func (q *commitQueue) enqueue(b *Batch) {
 	for {
-		pos := q.write
-		n := &q.nodes[pos&q.mask]
-		seq := atomic.LoadUint64(&n.position)
-		if seq != pos {
-			cond.Wait()
-			continue
+		ptrs := atomic.LoadUint64(&q.headTail)
+		head, tail := q.unpack(ptrs)
+		if (tail+uint32(len(q.slots)))&(1<<dequeueBits-1) == head {
+			// Queue is full.
+			panic("not reached")
+		}
+		slot := &q.slots[head&uint32(len(q.slots)-1)]
+
+		// Check if the head slot has been released by dequeue.
+		for atomic.LoadPointer(slot) != nil {
+			// Another goroutine is still cleaning up the tail, so the queue is
+			// actually still full. We spin because this should resolve itself
+			// momentarily.
+			runtime.Gosched()
 		}
 
-		atomic.StoreUint64(&q.write, pos+1)
-		atomic.StorePointer(&n.value, unsafe.Pointer(b))
-		atomic.StoreUint64(&n.position, pos+1)
+		// The head slot is free, so we own it.
+		*slot = unsafe.Pointer(b)
+
+		// Increment head. This passes ownership of slot to dequeue and acts as a
+		// store barrier for writing the slot.
+		atomic.AddUint64(&q.headTail, 1<<dequeueBits)
 		return
 	}
 }
 
-// Dequeue removes and returns the batch at the head of the queue if that batch
-// has been applied (Batch.applied != 0). Returns nil if either the queue empty
-// or the head batch is not been applied.
-func (q *commitQueue) dequeue(cond *sync.Cond) *Batch {
+func (q *commitQueue) dequeue() *Batch {
+	var slot *unsafe.Pointer
 	for {
-		pos := atomic.LoadUint64(&q.read)
-		n := &q.nodes[pos&q.mask]
-		seq := atomic.LoadUint64(&n.position)
-		if seq != pos+1 {
-			if pos == atomic.LoadUint64(&q.write) {
-				// The queue is empty.
-				return nil
-			}
-			// The element at pos is being written.
-			runtime.Gosched()
-			continue
-		}
-
-		b := (*Batch)(atomic.LoadPointer(&n.value))
-		if b == nil || atomic.LoadUint32(&b.applied) == 0 {
-			// The batch at the read index has either already been published or has
-			// not been applied.
+		ptrs := atomic.LoadUint64(&q.headTail)
+		head, tail := q.unpack(ptrs)
+		if tail == head {
+			// Queue is empty.
 			return nil
 		}
 
-		// Dequeue the element by bumping the read pointer.
-		if atomic.CompareAndSwapUint64(&q.read, pos, pos+1) {
-			atomic.StorePointer(&n.value, nil)
-			atomic.StoreUint64(&n.position, pos+q.mask+1)
-			cond.Signal()
-			return b
+		// Confirm head and tail (for our speculative check above) and increment
+		// tail. If this succeeds, then we own the slot at tail.
+		ptrs2 := q.pack(head, tail+1)
+		if atomic.CompareAndSwapUint64(&q.headTail, ptrs, ptrs2) {
+			// Success.
+			slot = &q.slots[tail&uint32(len(q.slots)-1)]
+			break
 		}
-
-		// We failed to bump the read pointer. Loop and try again.
-		runtime.Gosched() // free up the cpu before the next iteration
 	}
+
+	// We now own slot.
+	b := (*Batch)(*slot)
+
+	// Tell enqueue that we're done with this slot. Zeroing the slot is also
+	// important so we don't leave behind references that could keep this object
+	// live longer than necessary.
+	atomic.StorePointer(slot, nil)
+	// At this point enqueue owns the slot.
+
+	return b
 }
 
 // commitEnv contains the environment that a commitPipeline interacts
@@ -119,12 +133,11 @@ type commitEnv struct {
 
 	// Apply the batch to the specified memtable. Called concurrently.
 	apply func(b *Batch, mem *memTable) error
-	// Sync the WAL. Called serially by the sync goroutine.
-	sync func() error
-	// Write the batch to the WAL. The data is not persisted until a call to
-	// sync() is performed. Returns the memtable the batch should be applied
-	// to. Serial execution enforced by commitPipeline.mu.
-	write func(b *Batch) (*memTable, error)
+	// Write the batch to the WAL. If wg!=nil, the data will be persisted
+	// asynchronously and done will be called on wait group upon
+	// completion. Returns the memtable the batch should be applied to. Serial
+	// execution enforced by commitPipeline.mu.
+	write func(b *Batch, wg *sync.WaitGroup) (*memTable, error)
 }
 
 // A commitPipeline manages the stages of committing a set of mutations
@@ -199,69 +212,23 @@ type commitEnv struct {
 // batch in the queue. See commitPipeline.publish for additional commentary.
 type commitPipeline struct {
 	env commitEnv
+	sem chan struct{}
 	// The mutex to use for synchronizing access to logSeqNum and serializing
 	// calls to commitEnv.write().
 	mu sync.Mutex
-	// Condition var to signal upon changes to the pending queue.
-	cond sync.Cond
 	// Queue of pending batches to commit.
 	pending commitQueue
-
-	syncer struct {
-		sync.Mutex
-		cond    sync.Cond
-		closed  bool
-		pending []*Batch
-	}
 }
 
 func newCommitPipeline(env commitEnv) *commitPipeline {
 	p := &commitPipeline{
 		env: env,
+		sem: make(chan struct{}, commitConcurrency),
 	}
-	p.cond.L = &p.mu
-	p.pending.init()
-	p.syncer.cond.L = &p.syncer.Mutex
-	go p.syncLoop()
 	return p
 }
 
-func (p *commitPipeline) syncLoop() {
-	s := &p.syncer
-	s.Lock()
-	defer s.Unlock()
-
-	for {
-		for len(s.pending) == 0 && !s.closed {
-			s.cond.Wait()
-		}
-		if s.closed {
-			return
-		}
-
-		pending := s.pending
-		s.pending = nil
-
-		s.Unlock()
-
-		if err := p.env.sync(); err != nil {
-			// TODO(peter): Handle error notification.
-			panic(err)
-		}
-
-		for _, b := range pending {
-			b.commit.Done()
-		}
-
-		s.Lock()
-	}
-}
-
 func (p *commitPipeline) Close() {
-	p.syncer.Lock()
-	p.syncer.closed = true
-	p.syncer.cond.Broadcast()
-	p.syncer.Unlock()
 }
 
 // Commit the specified batch, writing it to the WAL, optionally syncing the
@@ -272,6 +239,8 @@ func (p *commitPipeline) Commit(b *Batch, syncWAL bool) error {
 		return nil
 	}
 
+	p.sem <- struct{}{}
+
 	// Prepare the batch for committing: enqueuing the batch in the pending
 	// queue, determining the batch sequence number and writing the data to the
 	// WAL.
@@ -292,6 +261,7 @@ func (p *commitPipeline) Commit(b *Batch, syncWAL bool) error {
 	// Publish the batch sequence number.
 	p.publish(b)
 
+	<-p.sem
 	return nil
 }
 
@@ -316,12 +286,14 @@ func (p *commitPipeline) AllocateSeqNum(prepare func(), apply func(seqNum uint64
 	b.setCount(1)
 	b.commit.Add(1)
 
+	p.sem <- struct{}{}
+
 	p.mu.Lock()
 
 	// Enqueue the batch in the pending queue. Note that while the pending queue
 	// is lock-free, we want the order of batches to be the same as the sequence
 	// number order.
-	p.pending.enqueue(b, &p.cond)
+	p.pending.enqueue(b)
 
 	// Assign the batch a sequence number.
 	seqNum := *p.env.logSeqNum
@@ -345,6 +317,8 @@ func (p *commitPipeline) AllocateSeqNum(prepare func(), apply func(seqNum uint64
 
 	// Publish the sequence number.
 	p.publish(b)
+
+	<-p.sem
 }
 
 func (p *commitPipeline) prepare(b *Batch, syncWAL bool) (*memTable, error) {
@@ -358,30 +332,27 @@ func (p *commitPipeline) prepare(b *Batch, syncWAL bool) (*memTable, error) {
 	}
 	b.commit.Add(count)
 
+	var syncWG *sync.WaitGroup
+	if syncWAL {
+		syncWG = &b.commit
+	}
+
 	p.mu.Lock()
 
 	// Enqueue the batch in the pending queue. Note that while the pending queue
 	// is lock-free, we want the order of batches to be the same as the sequence
 	// number order.
-	p.pending.enqueue(b, &p.cond)
+	p.pending.enqueue(b)
 
 	// Assign the batch a sequence number.
 	b.setSeqNum(*p.env.logSeqNum)
 	*p.env.logSeqNum += n
 
 	// Write the data to the WAL.
-	mem, err := p.env.write(b)
+	mem, err := p.env.write(b, syncWG)
 
 	p.mu.Unlock()
 
-	if syncWAL {
-		s := &p.syncer
-		s.Lock()
-		s.pending = append(s.pending, b)
-		s.cond.Signal()
-		s.Unlock()
-	}
-
 	return mem, err
 }
 
@@ -397,7 +368,7 @@ func (p *commitPipeline) publish(b *Batch) {
 	// batch applies it will go through the same process and publish our batch
 	// for us.
 	for {
-		t := p.pending.dequeue(&p.cond)
+		t := p.pending.dequeue()
 		if t == nil {
 			// Wait for another goroutine to publish us. We might also be waiting for
 			// the WAL sync to finish.
diff --git a/db.go b/db.go
index f9b3757..33f7eb6 100644
--- a/db.go
+++ b/db.go
@@ -371,21 +371,7 @@ func (d *DB) commitApply(b *Batch, mem *memTable) error {
 	return nil
 }
 
-func (d *DB) commitSync() error {
-	if d.opts.DisableWAL {
-		return errors.New("pebble: WAL disabled")
-	}
-
-	d.mu.Lock()
-	log := d.mu.log.LogWriter
-	d.mu.Unlock()
-	// NB: The log might have been closed after we unlock d.mu. That's ok because
-	// it will have been synced and all we're guaranteeing is that the log that
-	// was open at the start of this call was synced by the end of it.
-	return log.Sync()
-}
-
-func (d *DB) commitWrite(b *Batch) (*memTable, error) {
+func (d *DB) commitWrite(b *Batch, wg *sync.WaitGroup) (*memTable, error) {
 	d.mu.Lock()
 
 	// Throttle writes if there are too many L0 tables.
@@ -411,7 +397,7 @@ func (d *DB) commitWrite(b *Batch) (*memTable, error) {
 		return d.mu.mem.mutable, nil
 	}
 
-	size, err := d.mu.log.WriteRecord(b.storage.data)
+	size, err := d.mu.log.SyncRecord(b.storage.data, wg)
 	if err != nil {
 		panic(err)
 	}
diff --git a/internal/record/log_writer.go b/internal/record/log_writer.go
index b8ba8a5..32fcd94 100644
--- a/internal/record/log_writer.go
+++ b/internal/record/log_writer.go
@@ -30,6 +30,164 @@ type syncer interface {
 	Sync() error
 }
 
+const (
+	syncConcurrencyBits = 9
+
+	// SyncConcurrency is the maximum number of concurrent sync operations that
+	// can be performed. Exported as this value also limits the commit
+	// concurrency in commitPipeline.
+	SyncConcurrency = 1 << syncConcurrencyBits
+)
+
+// syncQueue is a lock-free fixed-size single-producer, single-consumer
+// queue. The single-producer can push to the head, and the single-consumer can
+// pop multiple values from the tail. Popping calls Done() on each of the
+// available *sync.WaitGroup elements.
+type syncQueue struct {
+	// headTail packs together a 32-bit head index and a 32-bit tail index. Both
+	// are indexes into slots modulo len(slots)-1.
+	//
+	// tail = index of oldest data in queue
+	// head = index of next slot to fill
+	//
+	// Slots in the range [tail, head) are owned by consumers.  A consumer
+	// continues to own a slot outside this range until it nils the slot, at
+	// which point ownership passes to the producer.
+	//
+	// The head index is stored in the most-significant bits so that we can
+	// atomically add to it and the overflow is harmless.
+	headTail uint64
+
+	// slots is a ring buffer of values stored in this queue. The size must be a
+	// power of 2. A slot is in use until the tail index has moved beyond it.
+	slots [SyncConcurrency]*sync.WaitGroup
+}
+
+const dequeueBits = 32
+
+func (q *syncQueue) unpack(ptrs uint64) (head, tail uint32) {
+	const mask = 1<<dequeueBits - 1
+	head = uint32((ptrs >> dequeueBits) & mask)
+	tail = uint32(ptrs & mask)
+	return
+}
+
+func (q *syncQueue) push(wg *sync.WaitGroup) {
+	ptrs := atomic.LoadUint64(&q.headTail)
+	head, tail := q.unpack(ptrs)
+	if (tail+uint32(len(q.slots)))&(1<<dequeueBits-1) == head {
+		panic("queue is full")
+	}
+
+	slot := &q.slots[head&uint32(len(q.slots)-1)]
+	*slot = wg
+
+	// Increment head. This passes ownership of slot to dequeue and acts as a
+	// store barrier for writing the slot.
+	atomic.AddUint64(&q.headTail, 1<<dequeueBits)
+}
+
+func (q *syncQueue) empty() bool {
+	head, tail := q.load()
+	return head == tail
+}
+
+func (q *syncQueue) load() (head, tail uint32) {
+	ptrs := atomic.LoadUint64(&q.headTail)
+	head, tail = q.unpack(ptrs)
+	return head, tail
+}
+
+func (q *syncQueue) pop(head, tail uint32) {
+	if tail == head {
+		// Queue is empty.
+		return
+	}
+
+	for ; tail != head; tail++ {
+		slot := &q.slots[tail&uint32(len(q.slots)-1)]
+		wg := *slot
+		if wg == nil {
+			panic("nil waiter")
+		}
+		*slot = nil
+		// We need to bump the tail count before signalling the wait group as
+		// signalling the wait group can trigger release a blocked goroutine which
+		// will try to enqueue before we've "freed" space in the queue.
+		atomic.AddUint64(&q.headTail, 1)
+		wg.Done()
+	}
+}
+
+// flusherCond is a specialized condition variable that is safe to signal for
+// readiness without holding the associated mutex in some circumstances. In
+// particular, when waiter is added to syncQueue, this condition variable can
+// be signalled without holding flusher.Mutex.
+type flusherCond struct {
+	mu   *sync.Mutex
+	q    *syncQueue
+	cond sync.Cond
+}
+
+func (c *flusherCond) init(mu *sync.Mutex, q *syncQueue) {
+	c.mu = mu
+	c.q = q
+	// Yes, this is a bit circular, but that is intentional. flusherCond.cond.L
+	// points flusherCond so that when cond.L.Unlock is called flusherCond.Unlock
+	// will be called and we can check the !syncQueue.empty() condition.
+	c.cond.L = c
+}
+
+func (c *flusherCond) Signal() {
+	// Pass-through to the cond var.
+	c.cond.Signal()
+}
+
+func (c *flusherCond) Wait() {
+	// Pass-through to the cond var. Note that internally the cond var implements
+	// Wait as:
+	//
+	//   t := notifyListAdd()
+	//   L.Unlock()
+	//   notifyListWait(t)
+	//   L.Lock()
+	//
+	// We've configured the cond var to call flusherReady.Unlock() which allows
+	// us to check the !syncQueue.empty() condition without a danger of missing a
+	// notification. Any call to flusherReady.Signal() after notifyListAdd() is
+	// called will cause the subsequent notifyListWait() to return immediately.
+	c.cond.Wait()
+}
+
+func (c *flusherCond) Lock() {
+	c.mu.Lock()
+}
+
+func (c *flusherCond) Unlock() {
+	c.mu.Unlock()
+	if !c.q.empty() {
+		// If the current goroutine is about to block on sync.Cond.Wait, this call
+		// to Signal will prevent that. The comment in Wait above explains a bit
+		// about what is going on here, but it is worth reiterating:
+		//
+		//   flusherCond.Wait()
+		//     sync.Cond.Wait()
+		//       t := notifyListAdd()
+		//       flusherCond.Unlock()    <-- we are here
+		//       notifyListWait(t)
+		//       flusherCond.Lock()
+		//
+		// The call to Signal here results in:
+		//
+		//     sync.Cond.Signal()
+		//       notifyListNotifyOne()
+		//
+		// The call to notifyListNotifyOne() will prevent the call to
+		// notifyListWait(t) from blocking.
+		c.cond.Signal()
+	}
+}
+
 // LogWriter writes records to an underlying io.Writer. In order to support WAL
 // file reuse, a LogWriter's records are tagged with the WAL's file
 // number. When reading a log file a record from a previous incarnation of the
@@ -54,23 +212,19 @@ type LogWriter struct {
 	block *block
 	free  chan *block
 
-	// Protects against concurrent calls to Flush/Sync().
-	flushMu sync.Mutex
-
 	flusher struct {
 		sync.Mutex
-		// Cond var signalled when there are blocks to flush or the Writer has been
-		// closed.
-		ready sync.Cond
-		// Cond var signalled when flushing of pending blocks has been completed.
-		done sync.Cond
-		// Is flushing currently active?
-		flushing bool
+		// Flusher ready is a condition variable that is signalled when there are
+		// blocks to flush, syncing has been requested, or the LogWriter has been
+		// closed. For signalling of a sync, it is safe to call without holding
+		// flusher.Mutex.
+		ready flusherCond
 		// Has the writer been closed?
 		closed bool
 		// Accumulated flush error.
 		err     error
 		pending []*block
+		syncQ   syncQueue
 	}
 }
 
@@ -95,8 +249,7 @@ func NewLogWriter(w io.Writer, logNum uint64) *LogWriter {
 		r.free <- &block{}
 	}
 	r.block = <-r.free
-	r.flusher.ready.L = &r.flusher.Mutex
-	r.flusher.done.L = &r.flusher.Mutex
+	r.flusher.ready.init(&r.flusher.Mutex, &r.flusher.syncQ)
 	go r.flushLoop()
 	return r
 }
@@ -107,24 +260,27 @@ func (w *LogWriter) flushLoop() {
 	defer f.Unlock()
 
 	for {
+		var data []byte
 		for {
 			if f.closed {
 				return
 			}
-			if f.flushing {
-				f.done.Wait()
-				continue
-			}
-			if len(f.pending) == 0 {
-				f.ready.Wait()
-				continue
+			// Grab the portion of the current block that requires flushing. Note that
+			// the current block can be added to the pending blocks list after we release
+			// the flusher lock, but it won't be part of pending.
+			written := atomic.LoadInt32(&w.block.written)
+			data = w.block.buf[w.block.flushed:written]
+			w.block.flushed = written
+			if len(f.pending) > 0 || len(data) > 0 || !f.syncQ.empty() {
+				break
 			}
-			break
+			f.ready.Wait()
+			continue
 		}
 
 		pending := f.pending
-		f.pending = nil
-		f.flushing = true
+		f.pending = f.pending[len(f.pending):]
+		head, tail := f.syncQ.load()
 
 		f.Unlock()
 
@@ -134,14 +290,23 @@ func (w *LogWriter) flushLoop() {
 				break
 			}
 		}
+		if err == nil && len(data) > 0 {
+			_, err = w.w.Write(data)
+		}
+		if err == nil && head != tail {
+			if w.s != nil {
+				err = w.s.Sync()
+			}
+			if err == nil {
+				f.syncQ.pop(head, tail)
+			}
+		}
 
 		f.Lock()
 		f.err = err
 		if f.err != nil {
 			return
 		}
-		f.flushing = false
-		f.done.Signal()
 	}
 }
 
@@ -173,16 +338,22 @@ func (w *LogWriter) queueBlock() {
 	w.blockNum++
 }
 
-// Close flushes any unwritten data and closes the writer.
+// Close flushes and syncs any unwritten data and closes the writer.
 func (w *LogWriter) Close() error {
-	w.flusher.Lock()
-	w.flusher.closed = true
-	w.flusher.ready.Signal()
-	w.flusher.Unlock()
+	f := &w.flusher
+
+	// Force a sync of any unwritten data.
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	f.syncQ.push(wg)
+	f.ready.Signal()
+	wg.Wait()
+
+	f.Lock()
+	f.closed = true
+	f.ready.Signal()
+	f.Unlock()
 
-	if err := w.Sync(); err != nil {
-		return err
-	}
 	if w.c != nil {
 		if err := w.c.Close(); err != nil {
 			return err
@@ -192,93 +363,17 @@ func (w *LogWriter) Close() error {
 	return nil
 }
 
-func (w *LogWriter) closed() bool {
-	w.flusher.Lock()
-	closed := w.flusher.closed
-	w.flusher.Unlock()
-	return closed
-}
-
-func (w *LogWriter) flushLocked() error {
-	if w.err != nil {
-		if w.closed() {
-			return nil
-		}
-		return w.err
-	}
-
-	w.flusher.Lock()
-	// Wait for any existing flushing to complete.
-	for w.flusher.flushing {
-		w.flusher.done.Wait()
-	}
-	// Block any new flushing from starting.
-	w.flusher.flushing = true
-	// Grab the list of pending blocks to be flushed.
-	pending := w.flusher.pending
-	w.flusher.pending = nil
-	// Grab the portion of the current block that requires flushing. Note that
-	// the current block can be added to the pending blocks list after we release
-	// the flusher lock, but it won't be part of pending.
-	written := atomic.LoadInt32(&w.block.written)
-	data := w.block.buf[w.block.flushed:written]
-	w.block.flushed = written
-	w.flusher.Unlock()
-
-	// Flush any pending blocks.
-	var err error
-	for _, t := range pending {
-		if err = w.flushBlock(t); err != nil {
-			break
-		}
-	}
-	if err == nil && len(data) > 0 {
-		_, err = w.w.Write(data)
-	}
-
-	// Release the flush loop.
-	w.flusher.Lock()
-	w.err = err
-	w.flusher.err = err
-	w.flusher.flushing = false
-	w.flusher.done.Signal()
-	w.flusher.Unlock()
-
-	if w.f != nil {
-		w.err = w.f.Flush()
-		return w.err
-	}
-	return nil
-}
-
-// Sync flushes unwritten data and synchronizes the underlying file. May be
-// called concurrently with Write, Flush and itself.
-func (w *LogWriter) Sync() error {
-	w.flushMu.Lock()
-	defer w.flushMu.Unlock()
-
-	if err := w.flushLocked(); err != nil {
-		if w.closed() {
-			return nil
-		}
-		return err
-	}
-
-	if w.s != nil {
-		w.err = w.s.Sync()
-		if w.err != nil {
-			if w.closed() {
-				return nil
-			}
-		}
-		return w.err
-	}
-	return nil
-}
-
 // WriteRecord writes a complete record. Returns the offset just past the end
 // of the record.
 func (w *LogWriter) WriteRecord(p []byte) (int64, error) {
+	return w.SyncRecord(p, nil)
+}
+
+// SyncRecord writes a complete record. If wg!= nil the record will be
+// asynchronously persisted to the underlying writer and done will be called on
+// the wait group upon completion. Returns the offset just past the end of the
+// record.
+func (w *LogWriter) SyncRecord(p []byte, wg *sync.WaitGroup) (int64, error) {
 	if w.err != nil {
 		return -1, w.err
 	}
@@ -287,6 +382,12 @@ func (w *LogWriter) WriteRecord(p []byte) (int64, error) {
 		p = w.emitFragment(i, p)
 	}
 
+	if wg != nil {
+		f := &w.flusher
+		f.syncQ.push(wg)
+		f.ready.Signal()
+	}
+
 	offset := w.blockNum*blockSize + int64(w.block.written)
 	return offset, w.err
 }
diff --git a/open.go b/open.go
index fbc6ff8..feda0a0 100644
--- a/open.go
+++ b/open.go
@@ -79,7 +79,6 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		logSeqNum:     &d.mu.versions.logSeqNum,
 		visibleSeqNum: &d.mu.versions.visibleSeqNum,
 		apply:         d.commitApply,
-		sync:          d.commitSync,
 		write:         d.commitWrite,
 	})
 	d.mu.nextJobID = 1


Test patch:
diff --git a/commit_test.go b/commit_test.go
index f4c80b4..8500914 100644
--- a/commit_test.go
+++ b/commit_test.go
@@ -34,7 +34,6 @@ func (e *testCommitEnv) env() commitEnv {
 		logSeqNum:     &e.logSeqNum,
 		visibleSeqNum: &e.visibleSeqNum,
 		apply:         e.apply,
-		sync:          e.sync,
 		write:         e.write,
 	}
 }
@@ -46,11 +45,7 @@ func (e *testCommitEnv) apply(b *Batch, mem *memTable) error {
 	return nil
 }
 
-func (e *testCommitEnv) sync() error {
-	return nil
-}
-
-func (e *testCommitEnv) write(b *Batch) (*memTable, error) {
+func (e *testCommitEnv) write(b *Batch, _ *sync.WaitGroup) (*memTable, error) {
 	n := int64(len(b.storage.data))
 	atomic.AddInt64(&e.writePos, n)
 	atomic.AddUint64(&e.writeCount, 1)
@@ -144,10 +139,7 @@ func BenchmarkCommitPipeline(b *testing.B) {
 					mem.unref()
 					return nil
 				},
-				sync: func() error {
-					return wal.Sync()
-				},
-				write: func(b *Batch) (*memTable, error) {
+				write: func(b *Batch, wg *sync.WaitGroup) (*memTable, error) {
 					for {
 						err := mem.prepare(b)
 						if err == arenaskl.ErrArenaFull {
@@ -160,7 +152,7 @@ func BenchmarkCommitPipeline(b *testing.B) {
 						break
 					}
 
-					_, err := wal.WriteRecord(b.storage.data)
+					_, err := wal.SyncRecord(b.storage.data, wg)
 					return mem, err
 				},
 			}
diff --git a/internal/record/log_writer_test.go b/internal/record/log_writer_test.go
new file mode 100644
index 0000000..919ccbd
--- /dev/null
+++ b/internal/record/log_writer_test.go
@@ -0,0 +1,116 @@
+// 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 record
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+)
+
+func TestSyncQueue(t *testing.T) {
+	var q syncQueue
+	var closed int32
+
+	var flusherWG sync.WaitGroup
+	flusherWG.Add(1)
+	go func() {
+		defer flusherWG.Done()
+		for {
+			if atomic.LoadInt32(&closed) == 1 {
+				return
+			}
+			head, tail := q.load()
+			q.pop(head, tail)
+		}
+	}()
+
+	var commitMu sync.Mutex
+	var doneWG sync.WaitGroup
+	for i := 0; i < SyncConcurrency; i++ {
+		doneWG.Add(1)
+		go func(i int) {
+			defer doneWG.Done()
+			for j := 0; j < 1000; j++ {
+				wg := &sync.WaitGroup{}
+				wg.Add(1)
+				// syncQueue is a single-producer, single-consumer queue. We need to
+				// provide mutual exclusion on the producer side.
+				commitMu.Lock()
+				q.push(wg)
+				commitMu.Unlock()
+				wg.Wait()
+			}
+		}(i)
+	}
+	doneWG.Wait()
+
+	atomic.StoreInt32(&closed, 1)
+	flusherWG.Wait()
+}
+
+func TestFlusherCond(t *testing.T) {
+	var mu sync.Mutex
+	var q syncQueue
+	var c flusherCond
+	var closed bool
+
+	c.init(&mu, &q)
+
+	var flusherWG sync.WaitGroup
+	flusherWG.Add(1)
+	go func() {
+		defer flusherWG.Done()
+
+		mu.Lock()
+		defer mu.Unlock()
+
+		for {
+			for {
+				if closed {
+					return
+				}
+				if !q.empty() {
+					break
+				}
+				c.Wait()
+			}
+
+			head, tail := q.load()
+			q.pop(head, tail)
+		}
+	}()
+
+	var commitMu sync.Mutex
+	var doneWG sync.WaitGroup
+	// NB: we're testing with low concurrency here, because what we want to
+	// stress is that signalling of the flusherCond works
+	// correctly. Specifically, we want to make sure that a signal is "lost",
+	// causing the test to wedge.
+	for i := 0; i < 2; i++ {
+		doneWG.Add(1)
+		go func(i int) {
+			defer doneWG.Done()
+			for j := 0; j < 10000; j++ {
+				wg := &sync.WaitGroup{}
+				wg.Add(1)
+				// syncQueue is a single-producer, single-consumer queue. We need to
+				// provide mutual exclusion on the producer side.
+				commitMu.Lock()
+				q.push(wg)
+				commitMu.Unlock()
+				c.Signal()
+				wg.Wait()
+			}
+		}(i)
+	}
+	doneWG.Wait()
+
+	mu.Lock()
+	closed = true
+	c.Signal()
+	mu.Unlock()
+	flusherWG.Wait()
+}