// Copyright 2018 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 ( "runtime" "sync" "sync/atomic" "time" "github.com/cockroachdb/pebble/record" ) // 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 { // 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 atomic.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 [record.SyncConcurrency]atomic.Pointer[Batch] } const dequeueBits = 32 func (q *commitQueue) unpack(ptrs uint64) (head, tail uint32) { const mask = 1<> dequeueBits) & mask) tail = uint32(ptrs & mask) return } func (q *commitQueue) pack(head, tail uint32) uint64 { const mask = 1< syncWAL func (p *commitPipeline) Commit(b *Batch, syncWAL bool, noSyncWait bool) error { if b.Empty() { return nil } commitStartTime := time.Now() // Acquire semaphores. p.commitQueueSem <- struct{}{} if syncWAL { p.logSyncQSem <- struct{}{} } b.commitStats.SemaphoreWaitDuration = time.Since(commitStartTime) // Prepare the batch for committing: enqueuing the batch in the pending // queue, determining the batch sequence number and writing the data to the // WAL. // // NB: We set Batch.commitErr on error so that the batch won't be a candidate // for reuse. See Batch.release(). mem, err := p.prepare(b, syncWAL, noSyncWait) if err != nil { b.db = nil // prevent batch reuse on error // NB: we are not doing <-p.commitQueueSem since the batch is still // sitting in the pending queue. We should consider fixing this by also // removing the batch from the pending queue. return err } // Apply the batch to the memtable. if err := p.env.apply(b, mem); err != nil { b.db = nil // prevent batch reuse on error // NB: we are not doing <-p.commitQueueSem since the batch is still // sitting in the pending queue. We should consider fixing this by also // removing the batch from the pending queue. return err } // Publish the batch sequence number. p.publish(b) <-p.commitQueueSem if !noSyncWait { // Already waited for commit, so look at the error. if b.commitErr != nil { b.db = nil // prevent batch reuse on error err = b.commitErr } } // Else noSyncWait. The LogWriter can be concurrently writing to // b.commitErr. We will read b.commitErr in Batch.SyncWait after the // LogWriter is done writing. b.commitStats.TotalDuration = time.Since(commitStartTime) return err } // AllocateSeqNum allocates count sequence numbers, invokes the prepare // callback, then the apply callback, and then publishes the sequence // numbers. AllocateSeqNum does not write to the WAL or add entries to the // memtable. AllocateSeqNum can be used to sequence an operation such as // sstable ingestion within the commit pipeline. The prepare callback is // invoked with commitPipeline.mu held, but note that DB.mu is not held and // must be locked if necessary. func (p *commitPipeline) AllocateSeqNum( count int, prepare func(seqNum uint64), apply func(seqNum uint64), ) { // This method is similar to Commit and prepare. Be careful about trying to // share additional code with those methods because Commit and prepare are // performance critical code paths. b := newBatch(nil) defer b.release() // Give the batch a count of 1 so that the log and visible sequence number // are incremented correctly. b.data = make([]byte, batchHeaderLen) b.setCount(uint32(count)) b.commit.Add(1) p.commitQueueSem <- 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) // Assign the batch a sequence number. Note that we use atomic operations // here to handle concurrent reads of logSeqNum. commitPipeline.mu provides // mutual exclusion for other goroutines writing to logSeqNum. logSeqNum := p.env.logSeqNum.Add(uint64(count)) - uint64(count) seqNum := logSeqNum if seqNum == 0 { // We can't use the value 0 for the global seqnum during ingestion, because // 0 indicates no global seqnum. So allocate one more seqnum. p.env.logSeqNum.Add(1) seqNum++ } b.setSeqNum(seqNum) // Wait for any outstanding writes to the memtable to complete. This is // necessary for ingestion so that the check for memtable overlap can see any // writes that were sequenced before the ingestion. The spin loop is // unfortunate, but obviates the need for additional synchronization. for { visibleSeqNum := p.env.visibleSeqNum.Load() if visibleSeqNum == logSeqNum { break } runtime.Gosched() } // Invoke the prepare callback. Note the lack of error reporting. Even if the // callback internally fails, the sequence number needs to be published in // order to allow the commit pipeline to proceed. prepare(b.SeqNum()) p.mu.Unlock() // Invoke the apply callback. apply(b.SeqNum()) // Publish the sequence number. p.publish(b) <-p.commitQueueSem } func (p *commitPipeline) prepare(b *Batch, syncWAL bool, noSyncWait bool) (*memTable, error) { n := uint64(b.Count()) if n == invalidBatchCount { return nil, ErrInvalidBatch } var syncWG *sync.WaitGroup var syncErr *error switch { case !syncWAL: // Only need to wait for the publish. b.commit.Add(1) // Remaining cases represent syncWAL=true. case noSyncWait: syncErr = &b.commitErr syncWG = &b.fsyncWait // Only need to wait synchronously for the publish. The user will // (asynchronously) wait on the batch's fsyncWait. b.commit.Add(1) b.fsyncWait.Add(1) case !noSyncWait: syncErr = &b.commitErr syncWG = &b.commit // Must wait for both the publish and the WAL fsync. b.commit.Add(2) } 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) // Assign the batch a sequence number. Note that we use atomic operations // here to handle concurrent reads of logSeqNum. commitPipeline.mu provides // mutual exclusion for other goroutines writing to logSeqNum. b.setSeqNum(p.env.logSeqNum.Add(n) - n) // Write the data to the WAL. mem, err := p.env.write(b, syncWG, syncErr) p.mu.Unlock() return mem, err } func (p *commitPipeline) publish(b *Batch) { // Mark the batch as applied. b.applied.Store(true) // Loop dequeuing applied batches from the pending queue. If our batch was // the head of the pending queue we are guaranteed that either we'll publish // it or someone else will dequeueApplied and publish it. If our batch is not the // head of the queue then either we'll dequeueApplied applied batches and reach our // batch or there is an unapplied batch blocking us. When that unapplied // batch applies it will go through the same process and publish our batch // for us. for { t := p.pending.dequeueApplied() if t == nil { // Wait for another goroutine to publish us. We might also be waiting for // the WAL sync to finish. now := time.Now() b.commit.Wait() b.commitStats.CommitWaitDuration += time.Since(now) break } if !t.applied.Load() { panic("not reached") } // We're responsible for publishing the sequence number for batch t, but // another concurrent goroutine might sneak in and publish the sequence // number for a subsequent batch. That's ok as all we're guaranteeing is // that the sequence number ratchets up. for { curSeqNum := p.env.visibleSeqNum.Load() newSeqNum := t.SeqNum() + uint64(t.Count()) if newSeqNum <= curSeqNum { // t's sequence number has already been published. break } if p.env.visibleSeqNum.CompareAndSwap(curSeqNum, newSeqNum) { // We successfully published t's sequence number. break } } t.commit.Done() } }