aleesa-bot/aleesa-jabber-go
0
Fork 0
Code Activity

Initial commit

Browse Source
This commit is contained in:
eleksir
2023-09-17 12:18:09 +03:00
commit 843dd418b0
1514 changed files with 643531 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

769
vendor/github.com/cockroachdb/pebble/record/log_writer.go generated vendored Normal file
View File

@@ -0,0 +1,769 @@
// 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 record
import (
"context"
"encoding/binary"
"io"
"runtime/pprof"
"sync"
"sync/atomic"
"time"
"github.com/cockroachdb/errors"
"github.com/cockroachdb/pebble/internal/base"
"github.com/cockroachdb/pebble/internal/crc"
"github.com/prometheus/client_golang/prometheus"
)
var walSyncLabels = pprof.Labels("pebble", "wal-sync")
var errClosedWriter = errors.New("pebble/record: closed LogWriter")
type block struct {
// buf[:written] has already been filled with fragments. Updated atomically.
written atomic.Int32
// buf[:flushed] has already been flushed to w.
flushed int32
buf [blockSize]byte
}
type flusher interface {
Flush() error
}
type syncer interface {
Sync() error
}
const (
syncConcurrencyBits = 12
// SyncConcurrency is the maximum number of concurrent sync operations that
// can be performed. Note that a sync operation is initiated either by a call
// to SyncRecord or by a call to Close. Exported as this value also limits
// the commit concurrency in commitPipeline.
SyncConcurrency = 1 << syncConcurrencyBits
)
type syncSlot struct {
wg *sync.WaitGroup
err *error
}
// 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 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 the tail index has moved beyond it.
slots [SyncConcurrency]syncSlot
// blocked is an atomic boolean which indicates whether syncing is currently
// blocked or can proceed. It is used by the implementation of
// min-sync-interval to block syncing until the min interval has passed.
blocked atomic.Bool
}
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, err *error) {
ptrs := q.headTail.Load()
head, tail := q.unpack(ptrs)
if (tail+uint32(len(q.slots)))&(1<<dequeueBits-1) == head {
panic("pebble: queue is full")
}
slot := &q.slots[head&uint32(len(q.slots)-1)]
slot.wg = wg
slot.err = err
// Increment head. This passes ownership of slot to dequeue and acts as a
// store barrier for writing the slot.
q.headTail.Add(1 << dequeueBits)
}
func (q *syncQueue) setBlocked() {
q.blocked.Store(true)
}
func (q *syncQueue) clearBlocked() {
q.blocked.Store(false)
}
func (q *syncQueue) empty() bool {
head, tail, _ := q.load()
return head == tail
}
// load returns the head, tail of the queue for what should be synced to the
// caller. It can return a head, tail of zero if syncing is blocked due to
// min-sync-interval. It additionally returns the real length of this queue,
// regardless of whether syncing is blocked.
func (q *syncQueue) load() (head, tail, realLength uint32) {
ptrs := q.headTail.Load()
head, tail = q.unpack(ptrs)
realLength = head - tail
if q.blocked.Load() {
return 0, 0, realLength
}
return head, tail, realLength
}
// REQUIRES: queueSemChan is non-nil.
func (q *syncQueue) pop(head, tail uint32, err error, queueSemChan chan struct{}) error {
if tail == head {
// Queue is empty.
return nil
}
for ; tail != head; tail++ {
slot := &q.slots[tail&uint32(len(q.slots)-1)]
wg := slot.wg
if wg == nil {
return errors.Errorf("nil waiter at %d", errors.Safe(tail&uint32(len(q.slots)-1)))
}
*slot.err = err
slot.wg = nil
slot.err = 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.
q.headTail.Add(1)
wg.Done()
// Is always non-nil in production.
if queueSemChan != nil {
<-queueSemChan
}
}
return nil
}
// flusherCond is a specialized condition variable that allows its condition to
// change and readiness be signalled without holding its associated mutex. In
// particular, when a waiter is added to syncQueue atomically, 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()
}
}
type durationFunc func() time.Duration
// syncTimer is an interface for timers, modeled on the closure callback mode
// of time.Timer. See time.AfterFunc and LogWriter.afterFunc. syncTimer is used
// by tests to mock out the timer functionality used to implement
// min-sync-interval.
type syncTimer interface {
Reset(time.Duration) bool
Stop() bool
}
// 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
// file will return the error ErrInvalidLogNum.
type LogWriter struct {
// w is the underlying writer.
w io.Writer
// c is w as a closer.
c io.Closer
// s is w as a syncer.
s syncer
// logNum is the low 32-bits of the log's file number.
logNum uint32
// blockNum is the zero based block number for the current block.
blockNum int64
// err is any accumulated error. TODO(peter): This needs to be protected in
// some fashion. Perhaps using atomic.Value.
err error
// block is the current block being written. Protected by flusher.Mutex.
block *block
free struct {
sync.Mutex
blocks []*block
}
flusher struct {
sync.Mutex
// 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
// Set to true when the flush loop should be closed.
close bool
// Closed when the flush loop has terminated.
closed chan struct{}
// Accumulated flush error.
err error
// minSyncInterval is the minimum duration between syncs.
minSyncInterval durationFunc
fsyncLatency prometheus.Histogram
pending []*block
syncQ syncQueue
metrics *LogWriterMetrics
}
// afterFunc is a hook to allow tests to mock out the timer functionality
// used for min-sync-interval. In normal operation this points to
// time.AfterFunc.
afterFunc func(d time.Duration, f func()) syncTimer
// See the comment for LogWriterConfig.QueueSemChan.
queueSemChan chan struct{}
}
// LogWriterConfig is a struct used for configuring new LogWriters
type LogWriterConfig struct {
WALMinSyncInterval durationFunc
WALFsyncLatency prometheus.Histogram
// QueueSemChan is an optional channel to pop from when popping from
// LogWriter.flusher.syncQueue. It functions as a semaphore that prevents
// the syncQueue from overflowing (which will cause a panic). All production
// code ensures this is non-nil.
QueueSemChan chan struct{}
}
// initialAllocatedBlocksCap is the initial capacity of the various slices
// intended to hold LogWriter blocks. The LogWriter may allocate more blocks
// than this threshold allows.
const initialAllocatedBlocksCap = 32
// blockPool pools *blocks to avoid allocations. Blocks are only added to the
// Pool when a LogWriter is closed. Before that, free blocks are maintained
// within a LogWriter's own internal free list `w.free.blocks`.
var blockPool = sync.Pool{
New: func() any { return &block{} },
}
// NewLogWriter returns a new LogWriter.
func NewLogWriter(w io.Writer, logNum base.FileNum, logWriterConfig LogWriterConfig) *LogWriter {
c, _ := w.(io.Closer)
s, _ := w.(syncer)
r := &LogWriter{
w: w,
c: c,
s: s,
// NB: we truncate the 64-bit log number to 32-bits. This is ok because a)
// we are very unlikely to reach a file number of 4 billion and b) the log
// number is used as a validation check and using only the low 32-bits is
// sufficient for that purpose.
logNum: uint32(logNum),
afterFunc: func(d time.Duration, f func()) syncTimer {
return time.AfterFunc(d, f)
},
queueSemChan: logWriterConfig.QueueSemChan,
}
r.free.blocks = make([]*block, 0, initialAllocatedBlocksCap)
r.block = blockPool.Get().(*block)
r.flusher.ready.init(&r.flusher.Mutex, &r.flusher.syncQ)
r.flusher.closed = make(chan struct{})
r.flusher.pending = make([]*block, 0, cap(r.free.blocks))
r.flusher.metrics = &LogWriterMetrics{}
f := &r.flusher
f.minSyncInterval = logWriterConfig.WALMinSyncInterval
f.fsyncLatency = logWriterConfig.WALFsyncLatency
go func() {
pprof.Do(context.Background(), walSyncLabels, r.flushLoop)
}()
return r
}
func (w *LogWriter) flushLoop(context.Context) {
f := &w.flusher
f.Lock()
// Initialize idleStartTime to when the loop starts.
idleStartTime := time.Now()
var syncTimer syncTimer
defer func() {
// Capture the idle duration between the last piece of work and when the
// loop terminated.
f.metrics.WriteThroughput.IdleDuration += time.Since(idleStartTime)
if syncTimer != nil {
syncTimer.Stop()
}
close(f.closed)
f.Unlock()
}()
// The flush loop performs flushing of full and partial data blocks to the
// underlying writer (LogWriter.w), syncing of the writer, and notification
// to sync requests that they have completed.
//
// - flusher.ready is a condition variable that is signalled when there is
// work to do. Full blocks are contained in flusher.pending. The current
// partial block is in LogWriter.block. And sync operations are held in
// flusher.syncQ.
//
// - The decision to sync is determined by whether there are any sync
// requests present in flusher.syncQ and whether enough time has elapsed
// since the last sync. If not enough time has elapsed since the last sync,
// flusher.syncQ.blocked will be set to 1. If syncing is blocked,
// syncQueue.empty() will return true and syncQueue.load() will return 0,0
// (i.e. an empty list).
//
// - flusher.syncQ.blocked is cleared by a timer that is initialized when
// blocked is set to 1. When blocked is 1, no syncing will take place, but
// flushing will continue to be performed. The on/off toggle for syncing
// does not need to be carefully synchronized with the rest of processing
// -- all we need to ensure is that after any transition to blocked=1 there
// is eventually a transition to blocked=0. syncTimer performs this
// transition. Note that any change to min-sync-interval will not take
// effect until the previous timer elapses.
//
// - Picking up the syncing work to perform requires coordination with
// picking up the flushing work. Specifically, flushing work is queued
// before syncing work. The guarantee of this code is that when a sync is
// requested, any previously queued flush work will be synced. This
// motivates reading the syncing work (f.syncQ.load()) before picking up
// the flush work (w.block.written.Load()).
// The list of full blocks that need to be written. This is copied from
// f.pending on every loop iteration, though the number of elements is
// usually small (most frequently 1). In the case of the WAL LogWriter, the
// number of blocks is bounded by the size of the WAL's corresponding
// memtable (MemtableSize/BlockSize). With the default 64 MiB memtables,
// this works out to at most 2048 elements if the entirety of the memtable's
// contents are queued.
pending := make([]*block, 0, cap(f.pending))
for {
for {
// 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 := w.block.written.Load()
if len(f.pending) > 0 || written > w.block.flushed || !f.syncQ.empty() {
break
}
if f.close {
// If the writer is closed, pretend the sync timer fired immediately so
// that we can process any queued sync requests.
f.syncQ.clearBlocked()
if !f.syncQ.empty() {
break
}
return
}
f.ready.Wait()
continue
}
// Found work to do, so no longer idle.
workStartTime := time.Now()
idleDuration := workStartTime.Sub(idleStartTime)
pending = append(pending[:0], f.pending...)
f.pending = f.pending[:0]
f.metrics.PendingBufferLen.AddSample(int64(len(pending)))
// Grab the list of sync waiters. Note that syncQueue.load() will return
// 0,0 while we're waiting for the min-sync-interval to expire. This
// allows flushing to proceed even if we're not ready to sync.
head, tail, realSyncQLen := f.syncQ.load()
f.metrics.SyncQueueLen.AddSample(int64(realSyncQLen))
// 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. This has to
// be ordered after we get the list of sync waiters from syncQ in order to
// prevent a race where a waiter adds itself to syncQ, but this thread
// picks up the entry in syncQ and not the buffered data.
written := w.block.written.Load()
data := w.block.buf[w.block.flushed:written]
w.block.flushed = written
// If flusher has an error, we propagate it to waiters. Note in spite of
// error we consume the pending list above to free blocks for writers.
if f.err != nil {
f.syncQ.pop(head, tail, f.err, w.queueSemChan)
// Update the idleStartTime if work could not be done, so that we don't
// include the duration we tried to do work as idle. We don't bother
// with the rest of the accounting, which means we will undercount.
idleStartTime = time.Now()
continue
}
f.Unlock()
synced, syncLatency, bytesWritten, err := w.flushPending(data, pending, head, tail)
f.Lock()
if synced && f.fsyncLatency != nil {
f.fsyncLatency.Observe(float64(syncLatency))
}
f.err = err
if f.err != nil {
f.syncQ.clearBlocked()
// Update the idleStartTime if work could not be done, so that we don't
// include the duration we tried to do work as idle. We don't bother
// with the rest of the accounting, which means we will undercount.
idleStartTime = time.Now()
continue
}
if synced && f.minSyncInterval != nil {
// A sync was performed. Make sure we've waited for the min sync
// interval before syncing again.
if min := f.minSyncInterval(); min > 0 {
f.syncQ.setBlocked()
if syncTimer == nil {
syncTimer = w.afterFunc(min, func() {
f.syncQ.clearBlocked()
f.ready.Signal()
})
} else {
syncTimer.Reset(min)
}
}
}
// Finished work, and started idling.
idleStartTime = time.Now()
workDuration := idleStartTime.Sub(workStartTime)
f.metrics.WriteThroughput.Bytes += bytesWritten
f.metrics.WriteThroughput.WorkDuration += workDuration
f.metrics.WriteThroughput.IdleDuration += idleDuration
}
}
func (w *LogWriter) flushPending(
data []byte, pending []*block, head, tail uint32,
) (synced bool, syncLatency time.Duration, bytesWritten int64, err error) {
defer func() {
// Translate panics into errors. The errors will cause flushLoop to shut
// down, but allows us to do so in a controlled way and avoid swallowing
// the stack that created the panic if panic'ing itself hits a panic
// (e.g. unlock of unlocked mutex).
if r := recover(); r != nil {
err = errors.Newf("%v", r)
}
}()
for _, b := range pending {
bytesWritten += blockSize - int64(b.flushed)
if err = w.flushBlock(b); err != nil {
break
}
}
if n := len(data); err == nil && n > 0 {
bytesWritten += int64(n)
_, err = w.w.Write(data)
}
synced = head != tail
if synced {
if err == nil && w.s != nil {
syncLatency, err = w.syncWithLatency()
}
f := &w.flusher
if popErr := f.syncQ.pop(head, tail, err, w.queueSemChan); popErr != nil {
return synced, syncLatency, bytesWritten, popErr
}
}
return synced, syncLatency, bytesWritten, err
}
func (w *LogWriter) syncWithLatency() (time.Duration, error) {
start := time.Now()
err := w.s.Sync()
syncLatency := time.Since(start)
return syncLatency, err
}
func (w *LogWriter) flushBlock(b *block) error {
if _, err := w.w.Write(b.buf[b.flushed:]); err != nil {
return err
}
b.written.Store(0)
b.flushed = 0
w.free.Lock()
w.free.blocks = append(w.free.blocks, b)
w.free.Unlock()
return nil
}
// queueBlock queues the current block for writing to the underlying writer,
// allocates a new block and reserves space for the next header.
func (w *LogWriter) queueBlock() {
// Allocate a new block, blocking until one is available. We do this first
// because w.block is protected by w.flusher.Mutex.
w.free.Lock()
if len(w.free.blocks) == 0 {
w.free.blocks = append(w.free.blocks, blockPool.Get().(*block))
}
nextBlock := w.free.blocks[len(w.free.blocks)-1]
w.free.blocks = w.free.blocks[:len(w.free.blocks)-1]
w.free.Unlock()
f := &w.flusher
f.Lock()
f.pending = append(f.pending, w.block)
w.block = nextBlock
f.ready.Signal()
w.err = w.flusher.err
f.Unlock()
w.blockNum++
}
// Close flushes and syncs any unwritten data and closes the writer.
// Where required, external synchronisation is provided by commitPipeline.mu.
func (w *LogWriter) Close() error {
f := &w.flusher
// Emit an EOF trailer signifying the end of this log. This helps readers
// differentiate between a corrupted entry in the middle of a log from
// garbage at the tail from a recycled log file.
w.emitEOFTrailer()
// Signal the flush loop to close.
f.Lock()
f.close = true
f.ready.Signal()
f.Unlock()
// Wait for the flush loop to close. The flush loop will not close until all
// pending data has been written or an error occurs.
<-f.closed
// Sync any flushed data to disk. NB: flushLoop will sync after flushing the
// last buffered data only if it was requested via syncQ, so we need to sync
// here to ensure that all the data is synced.
err := w.flusher.err
var syncLatency time.Duration
if err == nil && w.s != nil {
syncLatency, err = w.syncWithLatency()
}
f.Lock()
if f.fsyncLatency != nil {
f.fsyncLatency.Observe(float64(syncLatency))
}
free := w.free.blocks
f.Unlock()
if w.c != nil {
cerr := w.c.Close()
w.c = nil
if cerr != nil {
return cerr
}
}
for _, b := range free {
b.flushed = 0
b.written.Store(0)
blockPool.Put(b)
}
w.err = errClosedWriter
return err
}
// WriteRecord writes a complete record. Returns the offset just past the end
// of the record.
// External synchronisation provided by commitPipeline.mu.
func (w *LogWriter) WriteRecord(p []byte) (int64, error) {
logSize, err := w.SyncRecord(p, nil, nil)
return logSize, err
}
// 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.
// External synchronisation provided by commitPipeline.mu.
func (w *LogWriter) SyncRecord(
p []byte, wg *sync.WaitGroup, err *error,
) (logSize int64, err2 error) {
if w.err != nil {
return -1, w.err
}
// The `i == 0` condition ensures we handle empty records. Such records can
// possibly be generated for VersionEdits stored in the MANIFEST. While the
// MANIFEST is currently written using Writer, it is good to support the same
// semantics with LogWriter.
for i := 0; i == 0 || len(p) > 0; i++ {
p = w.emitFragment(i, p)
}
if wg != nil {
// If we've been asked to persist the record, add the WaitGroup to the sync
// queue and signal the flushLoop. Note that flushLoop will write partial
// blocks to the file if syncing has been requested. The contract is that
// any record written to the LogWriter to this point will be flushed to the
// OS and synced to disk.
f := &w.flusher
f.syncQ.push(wg, err)
f.ready.Signal()
}
offset := w.blockNum*blockSize + int64(w.block.written.Load())
// Note that we don't return w.err here as a concurrent call to Close would
// race with our read. That's ok because the only error we could be seeing is
// one to syncing for which the caller can receive notification of by passing
// in a non-nil err argument.
return offset, nil
}
// Size returns the current size of the file.
// External synchronisation provided by commitPipeline.mu.
func (w *LogWriter) Size() int64 {
return w.blockNum*blockSize + int64(w.block.written.Load())
}
func (w *LogWriter) emitEOFTrailer() {
// Write a recyclable chunk header with a different log number. Readers
// will treat the header as EOF when the log number does not match.
b := w.block
i := b.written.Load()
binary.LittleEndian.PutUint32(b.buf[i+0:i+4], 0) // CRC
binary.LittleEndian.PutUint16(b.buf[i+4:i+6], 0) // Size
b.buf[i+6] = recyclableFullChunkType
binary.LittleEndian.PutUint32(b.buf[i+7:i+11], w.logNum+1) // Log number
b.written.Store(i + int32(recyclableHeaderSize))
}
func (w *LogWriter) emitFragment(n int, p []byte) (remainingP []byte) {
b := w.block
i := b.written.Load()
first := n == 0
last := blockSize-i-recyclableHeaderSize >= int32(len(p))
if last {
if first {
b.buf[i+6] = recyclableFullChunkType
} else {
b.buf[i+6] = recyclableLastChunkType
}
} else {
if first {
b.buf[i+6] = recyclableFirstChunkType
} else {
b.buf[i+6] = recyclableMiddleChunkType
}
}
binary.LittleEndian.PutUint32(b.buf[i+7:i+11], w.logNum)
r := copy(b.buf[i+recyclableHeaderSize:], p)
j := i + int32(recyclableHeaderSize+r)
binary.LittleEndian.PutUint32(b.buf[i+0:i+4], crc.New(b.buf[i+6:j]).Value())
binary.LittleEndian.PutUint16(b.buf[i+4:i+6], uint16(r))
b.written.Store(j)
if blockSize-b.written.Load() < recyclableHeaderSize {
// There is no room for another fragment in the block, so fill the
// remaining bytes with zeros and queue the block for flushing.
for i := b.written.Load(); i < blockSize; i++ {
b.buf[i] = 0
}
w.queueBlock()
}
return p[r:]
}
// Metrics must be called after Close. The callee will no longer modify the
// returned LogWriterMetrics.
func (w *LogWriter) Metrics() *LogWriterMetrics {
return w.flusher.metrics
}
// LogWriterMetrics contains misc metrics for the log writer.
type LogWriterMetrics struct {
WriteThroughput base.ThroughputMetric
PendingBufferLen base.GaugeSampleMetric
SyncQueueLen base.GaugeSampleMetric
}
// Merge merges metrics from x. Requires that x is non-nil.
func (m *LogWriterMetrics) Merge(x *LogWriterMetrics) error {
m.WriteThroughput.Merge(x.WriteThroughput)
m.PendingBufferLen.Merge(x.PendingBufferLen)
m.SyncQueueLen.Merge(x.SyncQueueLen)
return nil
}

644
vendor/github.com/cockroachdb/pebble/record/record.go generated vendored Normal file
View File

@@ -0,0 +1,644 @@
// Copyright 2011 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 reads and writes sequences of records. Each record is a stream
// of bytes that completes before the next record starts.
//
// When reading, call Next to obtain an io.Reader for the next record. Next will
// return io.EOF when there are no more records. It is valid to call Next
// without reading the current record to exhaustion.
//
// When writing, call Next to obtain an io.Writer for the next record. Calling
// Next finishes the current record. Call Close to finish the final record.
//
// Optionally, call Flush to finish the current record and flush the underlying
// writer without starting a new record. To start a new record after flushing,
// call Next.
//
// Neither Readers or Writers are safe to use concurrently.
//
// Example code:
//
// func read(r io.Reader) ([]string, error) {
// var ss []string
// records := record.NewReader(r)
// for {
// rec, err := records.Next()
// if err == io.EOF {
// break
// }
// if err != nil {
// log.Printf("recovering from %v", err)
// r.Recover()
// continue
// }
// s, err := io.ReadAll(rec)
// if err != nil {
// log.Printf("recovering from %v", err)
// r.Recover()
// continue
// }
// ss = append(ss, string(s))
// }
// return ss, nil
// }
//
// func write(w io.Writer, ss []string) error {
// records := record.NewWriter(w)
// for _, s := range ss {
// rec, err := records.Next()
// if err != nil {
// return err
// }
// if _, err := rec.Write([]byte(s)), err != nil {
// return err
// }
// }
// return records.Close()
// }
//
// The wire format is that the stream is divided into 32KiB blocks, and each
// block contains a number of tightly packed chunks. Chunks cannot cross block
// boundaries. The last block may be shorter than 32 KiB. Any unused bytes in a
// block must be zero.
//
// A record maps to one or more chunks. There are two chunk formats: legacy and
// recyclable. The legacy chunk format:
//
// +----------+-----------+-----------+--- ... ---+
// | CRC (4B) | Size (2B) | Type (1B) | Payload |
// +----------+-----------+-----------+--- ... ---+
//
// CRC is computed over the type and payload
// Size is the length of the payload in bytes
// Type is the chunk type
//
// There are four chunk types: whether the chunk is the full record, or the
// first, middle or last chunk of a multi-chunk record. A multi-chunk record
// has one first chunk, zero or more middle chunks, and one last chunk.
//
// The recyclyable chunk format is similar to the legacy format, but extends
// the chunk header with an additional log number field. This allows reuse
// (recycling) of log files which can provide significantly better performance
// when syncing frequently as it avoids needing to update the file
// metadata. Additionally, recycling log files is a prequisite for using direct
// IO with log writing. The recyclyable format is:
//
// +----------+-----------+-----------+----------------+--- ... ---+
// | CRC (4B) | Size (2B) | Type (1B) | Log number (4B)| Payload |
// +----------+-----------+-----------+----------------+--- ... ---+
//
// Recyclable chunks are distinguished from legacy chunks by the addition of 4
// extra "recyclable" chunk types that map directly to the legacy chunk types
// (i.e. full, first, middle, last). The CRC is computed over the type, log
// number, and payload.
//
// The wire format allows for limited recovery in the face of data corruption:
// on a format error (such as a checksum mismatch), the reader moves to the
// next block and looks for the next full or first chunk.
package record
// The C++ Level-DB code calls this the log, but it has been renamed to record
// to avoid clashing with the standard log package, and because it is generally
// useful outside of logging. The C++ code also uses the term "physical record"
// instead of "chunk", but "chunk" is shorter and less confusing.
import (
"encoding/binary"
"io"
"github.com/cockroachdb/errors"
"github.com/cockroachdb/pebble/internal/base"
"github.com/cockroachdb/pebble/internal/crc"
)
// These constants are part of the wire format and should not be changed.
const (
fullChunkType = 1
firstChunkType = 2
middleChunkType = 3
lastChunkType = 4
recyclableFullChunkType = 5
recyclableFirstChunkType = 6
recyclableMiddleChunkType = 7
recyclableLastChunkType = 8
)
const (
blockSize = 32 * 1024
blockSizeMask = blockSize - 1
legacyHeaderSize = 7
recyclableHeaderSize = legacyHeaderSize + 4
)
var (
// ErrNotAnIOSeeker is returned if the io.Reader underlying a Reader does not implement io.Seeker.
ErrNotAnIOSeeker = errors.New("pebble/record: reader does not implement io.Seeker")
// ErrNoLastRecord is returned if LastRecordOffset is called and there is no previous record.
ErrNoLastRecord = errors.New("pebble/record: no last record exists")
// ErrZeroedChunk is returned if a chunk is encountered that is zeroed. This
// usually occurs due to log file preallocation.
ErrZeroedChunk = base.CorruptionErrorf("pebble/record: zeroed chunk")
// ErrInvalidChunk is returned if a chunk is encountered with an invalid
// header, length, or checksum. This usually occurs when a log is recycled,
// but can also occur due to corruption.
ErrInvalidChunk = base.CorruptionErrorf("pebble/record: invalid chunk")
)
// IsInvalidRecord returns true if the error matches one of the error types
// returned for invalid records. These are treated in a way similar to io.EOF
// in recovery code.
func IsInvalidRecord(err error) bool {
return err == ErrZeroedChunk || err == ErrInvalidChunk || err == io.ErrUnexpectedEOF
}
// Reader reads records from an underlying io.Reader.
type Reader struct {
// r is the underlying reader.
r io.Reader
// logNum is the low 32-bits of the log's file number. May be zero when used
// with log files that do not have a file number (e.g. the MANIFEST).
logNum uint32
// blockNum is the zero based block number currently held in buf.
blockNum int64
// seq is the sequence number of the current record.
seq int
// buf[begin:end] is the unread portion of the current chunk's payload. The
// low bound, begin, excludes the chunk header.
begin, end int
// n is the number of bytes of buf that are valid. Once reading has started,
// only the final block can have n < blockSize.
n int
// recovering is true when recovering from corruption.
recovering bool
// last is whether the current chunk is the last chunk of the record.
last bool
// err is any accumulated error.
err error
// buf is the buffer.
buf [blockSize]byte
}
// NewReader returns a new reader. If the file contains records encoded using
// the recyclable record format, then the log number in those records must
// match the specified logNum.
func NewReader(r io.Reader, logNum base.FileNum) *Reader {
return &Reader{
r: r,
logNum: uint32(logNum),
blockNum: -1,
}
}
// nextChunk sets r.buf[r.i:r.j] to hold the next chunk's payload, reading the
// next block into the buffer if necessary.
func (r *Reader) nextChunk(wantFirst bool) error {
for {
if r.end+legacyHeaderSize <= r.n {
checksum := binary.LittleEndian.Uint32(r.buf[r.end+0 : r.end+4])
length := binary.LittleEndian.Uint16(r.buf[r.end+4 : r.end+6])
chunkType := r.buf[r.end+6]
if checksum == 0 && length == 0 && chunkType == 0 {
if r.end+recyclableHeaderSize > r.n {
// Skip the rest of the block if the recyclable header size does not
// fit within it.
r.end = r.n
continue
}
if r.recovering {
// Skip the rest of the block, if it looks like it is all
// zeroes. This is common with WAL preallocation.
//
// Set r.err to be an error so r.recover actually recovers.
r.err = ErrZeroedChunk
r.recover()
continue
}
return ErrZeroedChunk
}
headerSize := legacyHeaderSize
if chunkType >= recyclableFullChunkType && chunkType <= recyclableLastChunkType {
headerSize = recyclableHeaderSize
if r.end+headerSize > r.n {
return ErrInvalidChunk
}
logNum := binary.LittleEndian.Uint32(r.buf[r.end+7 : r.end+11])
if logNum != r.logNum {
if wantFirst {
// If we're looking for the first chunk of a record, we can treat a
// previous instance of the log as EOF.
return io.EOF
}
// Otherwise, treat this chunk as invalid in order to prevent reading
// of a partial record.
return ErrInvalidChunk
}
chunkType -= (recyclableFullChunkType - 1)
}
r.begin = r.end + headerSize
r.end = r.begin + int(length)
if r.end > r.n {
// The chunk straddles a 32KB boundary (or the end of file).
if r.recovering {
r.recover()
continue
}
return ErrInvalidChunk
}
if checksum != crc.New(r.buf[r.begin-headerSize+6:r.end]).Value() {
if r.recovering {
r.recover()
continue
}
return ErrInvalidChunk
}
if wantFirst {
if chunkType != fullChunkType && chunkType != firstChunkType {
continue
}
}
r.last = chunkType == fullChunkType || chunkType == lastChunkType
r.recovering = false
return nil
}
if r.n < blockSize && r.blockNum >= 0 {
if !wantFirst || r.end != r.n {
// This can happen if the previous instance of the log ended with a
// partial block at the same blockNum as the new log but extended
// beyond the partial block of the new log.
return ErrInvalidChunk
}
return io.EOF
}
n, err := io.ReadFull(r.r, r.buf[:])
if err != nil && err != io.ErrUnexpectedEOF {
if err == io.EOF && !wantFirst {
return io.ErrUnexpectedEOF
}
return err
}
r.begin, r.end, r.n = 0, 0, n
r.blockNum++
}
}
// Next returns a reader for the next record. It returns io.EOF if there are no
// more records. The reader returned becomes stale after the next Next call,
// and should no longer be used.
func (r *Reader) Next() (io.Reader, error) {
r.seq++
if r.err != nil {
return nil, r.err
}
r.begin = r.end
r.err = r.nextChunk(true)
if r.err != nil {
return nil, r.err
}
return singleReader{r, r.seq}, nil
}
// Offset returns the current offset within the file. If called immediately
// before a call to Next(), Offset() will return the record offset.
func (r *Reader) Offset() int64 {
if r.blockNum < 0 {
return 0
}
return int64(r.blockNum)*blockSize + int64(r.end)
}
// recover clears any errors read so far, so that calling Next will start
// reading from the next good 32KiB block. If there are no such blocks, Next
// will return io.EOF. recover also marks the current reader, the one most
// recently returned by Next, as stale. If recover is called without any
// prior error, then recover is a no-op.
func (r *Reader) recover() {
if r.err == nil {
return
}
r.recovering = true
r.err = nil
// Discard the rest of the current block.
r.begin, r.end, r.last = r.n, r.n, false
// Invalidate any outstanding singleReader.
r.seq++
}
// seekRecord seeks in the underlying io.Reader such that calling r.Next
// returns the record whose first chunk header starts at the provided offset.
// Its behavior is undefined if the argument given is not such an offset, as
// the bytes at that offset may coincidentally appear to be a valid header.
//
// It returns ErrNotAnIOSeeker if the underlying io.Reader does not implement
// io.Seeker.
//
// seekRecord will fail and return an error if the Reader previously
// encountered an error, including io.EOF. Such errors can be cleared by
// calling Recover. Calling seekRecord after Recover will make calling Next
// return the record at the given offset, instead of the record at the next
// good 32KiB block as Recover normally would. Calling seekRecord before
// Recover has no effect on Recover's semantics other than changing the
// starting point for determining the next good 32KiB block.
//
// The offset is always relative to the start of the underlying io.Reader, so
// negative values will result in an error as per io.Seeker.
func (r *Reader) seekRecord(offset int64) error {
r.seq++
if r.err != nil {
return r.err
}
s, ok := r.r.(io.Seeker)
if !ok {
return ErrNotAnIOSeeker
}
// Only seek to an exact block offset.
c := int(offset & blockSizeMask)
if _, r.err = s.Seek(offset&^blockSizeMask, io.SeekStart); r.err != nil {
return r.err
}
// Clear the state of the internal reader.
r.begin, r.end, r.n = 0, 0, 0
r.blockNum, r.recovering, r.last = -1, false, false
if r.err = r.nextChunk(false); r.err != nil {
return r.err
}
// Now skip to the offset requested within the block. A subsequent
// call to Next will return the block at the requested offset.
r.begin, r.end = c, c
return nil
}
type singleReader struct {
r *Reader
seq int
}
func (x singleReader) Read(p []byte) (int, error) {
r := x.r
if r.seq != x.seq {
return 0, errors.New("pebble/record: stale reader")
}
if r.err != nil {
return 0, r.err
}
for r.begin == r.end {
if r.last {
return 0, io.EOF
}
if r.err = r.nextChunk(false); r.err != nil {
return 0, r.err
}
}
n := copy(p, r.buf[r.begin:r.end])
r.begin += n
return n, nil
}
// Writer writes records to an underlying io.Writer.
type Writer struct {
// w is the underlying writer.
w io.Writer
// seq is the sequence number of the current record.
seq int
// f is w as a flusher.
f flusher
// buf[i:j] is the bytes that will become the current chunk.
// The low bound, i, includes the chunk header.
i, j int
// buf[:written] has already been written to w.
// written is zero unless Flush has been called.
written int
// baseOffset is the base offset in w at which writing started. If
// w implements io.Seeker, it's relative to the start of w, 0 otherwise.
baseOffset int64
// blockNumber is the zero based block number currently held in buf.
blockNumber int64
// lastRecordOffset is the offset in w where the last record was
// written (including the chunk header). It is a relative offset to
// baseOffset, thus the absolute offset of the last record is
// baseOffset + lastRecordOffset.
lastRecordOffset int64
// first is whether the current chunk is the first chunk of the record.
first bool
// pending is whether a chunk is buffered but not yet written.
pending bool
// err is any accumulated error.
err error
// buf is the buffer.
buf [blockSize]byte
}
// NewWriter returns a new Writer.
func NewWriter(w io.Writer) *Writer {
f, _ := w.(flusher)
var o int64
if s, ok := w.(io.Seeker); ok {
var err error
if o, err = s.Seek(0, io.SeekCurrent); err != nil {
o = 0
}
}
return &Writer{
w: w,
f: f,
baseOffset: o,
lastRecordOffset: -1,
}
}
// fillHeader fills in the header for the pending chunk.
func (w *Writer) fillHeader(last bool) {
if w.i+legacyHeaderSize > w.j || w.j > blockSize {
panic("pebble/record: bad writer state")
}
if last {
if w.first {
w.buf[w.i+6] = fullChunkType
} else {
w.buf[w.i+6] = lastChunkType
}
} else {
if w.first {
w.buf[w.i+6] = firstChunkType
} else {
w.buf[w.i+6] = middleChunkType
}
}
binary.LittleEndian.PutUint32(w.buf[w.i+0:w.i+4], crc.New(w.buf[w.i+6:w.j]).Value())
binary.LittleEndian.PutUint16(w.buf[w.i+4:w.i+6], uint16(w.j-w.i-legacyHeaderSize))
}
// writeBlock writes the buffered block to the underlying writer, and reserves
// space for the next chunk's header.
func (w *Writer) writeBlock() {
_, w.err = w.w.Write(w.buf[w.written:])
w.i = 0
w.j = legacyHeaderSize
w.written = 0
w.blockNumber++
}
// writePending finishes the current record and writes the buffer to the
// underlying writer.
func (w *Writer) writePending() {
if w.err != nil {
return
}
if w.pending {
w.fillHeader(true)
w.pending = false
}
_, w.err = w.w.Write(w.buf[w.written:w.j])
w.written = w.j
}
// Close finishes the current record and closes the writer.
func (w *Writer) Close() error {
w.seq++
w.writePending()
if w.err != nil {
return w.err
}
w.err = errors.New("pebble/record: closed Writer")
return nil
}
// Flush finishes the current record, writes to the underlying writer, and
// flushes it if that writer implements interface{ Flush() error }.
func (w *Writer) Flush() error {
w.seq++
w.writePending()
if w.err != nil {
return w.err
}
if w.f != nil {
w.err = w.f.Flush()
return w.err
}
return nil
}
// Next returns a writer for the next record. The writer returned becomes stale
// after the next Close, Flush or Next call, and should no longer be used.
func (w *Writer) Next() (io.Writer, error) {
w.seq++
if w.err != nil {
return nil, w.err
}
if w.pending {
w.fillHeader(true)
}
w.i = w.j
w.j = w.j + legacyHeaderSize
// Check if there is room in the block for the header.
if w.j > blockSize {
// Fill in the rest of the block with zeroes.
for k := w.i; k < blockSize; k++ {
w.buf[k] = 0
}
w.writeBlock()
if w.err != nil {
return nil, w.err
}
}
w.lastRecordOffset = w.baseOffset + w.blockNumber*blockSize + int64(w.i)
w.first = true
w.pending = true
return singleWriter{w, w.seq}, nil
}
// WriteRecord writes a complete record. Returns the offset just past the end
// of the record.
func (w *Writer) WriteRecord(p []byte) (int64, error) {
if w.err != nil {
return -1, w.err
}
t, err := w.Next()
if err != nil {
return -1, err
}
if _, err := t.Write(p); err != nil {
return -1, err
}
w.writePending()
offset := w.blockNumber*blockSize + int64(w.j)
return offset, w.err
}
// Size returns the current size of the file.
func (w *Writer) Size() int64 {
if w == nil {
return 0
}
return w.blockNumber*blockSize + int64(w.j)
}
// LastRecordOffset returns the offset in the underlying io.Writer of the last
// record so far - the one created by the most recent Next call. It is the
// offset of the first chunk header, suitable to pass to Reader.SeekRecord.
//
// If that io.Writer also implements io.Seeker, the return value is an absolute
// offset, in the sense of io.SeekStart, regardless of whether the io.Writer
// was initially at the zero position when passed to NewWriter. Otherwise, the
// return value is a relative offset, being the number of bytes written between
// the NewWriter call and any records written prior to the last record.
//
// If there is no last record, i.e. nothing was written, LastRecordOffset will
// return ErrNoLastRecord.
func (w *Writer) LastRecordOffset() (int64, error) {
if w.err != nil {
return 0, w.err
}
if w.lastRecordOffset < 0 {
return 0, ErrNoLastRecord
}
return w.lastRecordOffset, nil
}
type singleWriter struct {
w *Writer
seq int
}
func (x singleWriter) Write(p []byte) (int, error) {
w := x.w
if w.seq != x.seq {
return 0, errors.New("pebble/record: stale writer")
}
if w.err != nil {
return 0, w.err
}
n0 := len(p)
for len(p) > 0 {
// Write a block, if it is full.
if w.j == blockSize {
w.fillHeader(false)
w.writeBlock()
if w.err != nil {
return 0, w.err
}
w.first = false
}
// Copy bytes into the buffer.
n := copy(w.buf[w.j:], p)
w.j += n
p = p[n:]
}
return n0, nil
}

82
vendor/github.com/cockroachdb/pebble/record/rotation.go generated vendored Normal file
View File

@@ -0,0 +1,82 @@
// Copyright 2023 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
// RotationHelper is a type used to inform the decision of rotating a record log
// file.
//
// The assumption is that multiple records can be coalesced into a single record
// (called a snapshot). Starting a new file, where the first record is a
// snapshot of the current state is referred to as "rotating" the log.
//
// Normally we rotate files when a certain file size is reached. But in certain
// cases (e.g. contents become very large), this can result in too frequent
// rotation. This helper contains logic to impose extra conditions on the
// rotation.
//
// The rotation helper uses "size" as a unit-less estimation that is correlated
// with the on-disk size of a record or snapshot.
type RotationHelper struct {
// lastSnapshotSize is the size of the last snapshot.
lastSnapshotSize int64
// sizeSinceLastSnapshot is the sum of sizes of records applied since the last
// snapshot.
sizeSinceLastSnapshot int64
lastRecordSize int64
}
// AddRecord makes the rotation helper aware of a new record.
func (rh *RotationHelper) AddRecord(recordSize int64) {
rh.sizeSinceLastSnapshot += recordSize
rh.lastRecordSize = recordSize
}
// ShouldRotate returns whether we should start a new log file (with a snapshot).
// Does not need to be called if other rotation factors (log file size) are not
// satisfied.
func (rh *RotationHelper) ShouldRotate(nextSnapshotSize int64) bool {
// The primary goal is to ensure that when reopening a log file, the number of
// edits that need to be replayed on top of the snapshot is "sane" while
// keeping the rotation frequency as low as possible.
//
// For the purposes of this description, we assume that the log is mainly
// storing a collection of "entries", with edits adding or removing entries.
// Consider the following cases:
//
// - The number of live entries is roughly stable: after writing the snapshot
// (with S entries), we require that there be enough edits such that the
// cumulative number of entries in those edits, E, be greater than S. This
// will ensure that at most 50% of data written out is due to rotation.
//
// - The number of live entries K in the DB is shrinking drastically, say from
// S to S/10: After this shrinking, E = 0.9S, and so if we used the previous
// snapshot entry count, S, as the threshold that needs to be exceeded, we
// will further delay the snapshot writing. Which means on reopen we will
// need to replay 0.9S edits to get to a version with 0.1S entries. It would
// be better to create a new snapshot when E exceeds the number of entries in
// the current version.
//
// - The number of live entries L in the DB is growing; say the last snapshot
// had S entries, and now we have 10S entries, so E = 9S. If we required
// that E is at least the current number of entries, we would further delay
// writing a new snapshot (which is not desirable).
//
// The logic below uses the min of the last snapshot size count and the size
// count in the current version.
return rh.sizeSinceLastSnapshot > rh.lastSnapshotSize || rh.sizeSinceLastSnapshot > nextSnapshotSize
}
// Rotate makes the rotation helper aware that we are rotating to a new snapshot
// (to which we will apply the latest edit).
func (rh *RotationHelper) Rotate(snapshotSize int64) {
rh.lastSnapshotSize = snapshotSize
rh.sizeSinceLastSnapshot = rh.lastRecordSize
}
// DebugInfo returns the last snapshot size and size of the edits since the last
// snapshot; used for testing and debugging.
func (rh *RotationHelper) DebugInfo() (lastSnapshotSize int64, sizeSinceLastSnapshot int64) {
return rh.lastSnapshotSize, rh.sizeSinceLastSnapshot
}