Files
dailz e34de4acc9 fix: implement group commit collection window (500µs or 32KB)
Replace polling loop (Collect + Sleep) with proper batch collection:
- Block on first request via channel receive
- Start 500µs timer for collection window
- Accumulate requests until timer fires or batch reaches 32KB
- Shutdown handling at all blocking points

Design doc §3.2.5: '等待组提交触发(500µs 或 32KB,先到者触发)'

Add GroupCommitDelay config field (default 500µs, must be > 0 and < 10ms).
2026-06-12 16:15:08 +08:00

218 lines
7.0 KiB
Go

// Package config defines configuration types and validation for the go-kv storage engine.
package config
import (
"fmt"
"math"
"time"
)
// WAL format constants derived from the binary layout specification.
const (
walFileHeaderSize uint64 = 32
physicalRecordHeaderSize uint64 = 7
walBatchHeaderSize uint64 = 18 // flags(2) + baseSequence(8) + entryCount(4) + entriesSize(4)
)
// WalConfig holds configuration for the Write-Ahead Log subsystem.
// Zero-value WalConfig is valid and uses defaults; call Validate() to apply
// defaults and verify invariants.
type WalConfig struct {
// MaxSegmentSize is the maximum size of a single WAL segment file in bytes.
// Default: 64MB. Must be large enough to hold the largest possible WAL Batch.
MaxSegmentSize uint64
// BlockSize is the WAL block size in bytes.
// Default: 32KB.
BlockSize uint32
// SyncMode controls when WAL writes are flushed to disk.
// Phase 1 only supports "always".
SyncMode string
// MaxBatchEntries is the maximum number of entries in a single WAL Batch.
// Default: 10000.
MaxBatchEntries uint32
// MaxBatchSize is the maximum total size of WAL Batch entries in bytes.
// Default: 4MB.
MaxBatchSize uint32
// GroupCommitDelay is the maximum time to collect requests before writing a
// WAL group commit batch. Default: 500µs. Must be > 0 and < 10ms.
GroupCommitDelay time.Duration
// MaxKeyBytes is the maximum size of a single key in bytes.
// Default: 4KB.
MaxKeyBytes uint32
// MaxInlineValue is the maximum size of an inline value in bytes.
// Values larger than this must use ValueLogPointer.
// Default: 4KB.
MaxInlineValue uint32
// MemTableSize is the target MemTable size in bytes before triggering flush.
// Default: 64MB.
MemTableSize uint32
// MaxImmutableCount is the maximum number of immutable MemTables allowed
// before writes are stalled. Default: 3.
MaxImmutableCount int
}
// Defaults returns a WalConfig populated with production defaults.
func Defaults() WalConfig {
return WalConfig{
MaxSegmentSize: 64 * 1024 * 1024, // 64MB
BlockSize: 32 * 1024, // 32KB
SyncMode: "always",
MaxBatchEntries: 10000,
MaxBatchSize: 4 * 1024 * 1024, // 4MB
GroupCommitDelay: 500 * time.Microsecond,
MaxKeyBytes: 4 * 1024, // 4KB
MaxInlineValue: 4 * 1024, // 4KB
MemTableSize: 64 * 1024 * 1024, // 64MB
MaxImmutableCount: 3,
}
}
// applyDefaults fills zero-valued fields with production defaults.
func (c *WalConfig) applyDefaults() {
d := Defaults()
if c.MaxSegmentSize == 0 {
c.MaxSegmentSize = d.MaxSegmentSize
}
if c.BlockSize == 0 {
c.BlockSize = d.BlockSize
}
if c.SyncMode == "" {
c.SyncMode = d.SyncMode
}
if c.MaxBatchEntries == 0 {
c.MaxBatchEntries = d.MaxBatchEntries
}
if c.MaxBatchSize == 0 {
c.MaxBatchSize = d.MaxBatchSize
}
if c.GroupCommitDelay == 0 {
c.GroupCommitDelay = d.GroupCommitDelay
}
if c.MaxKeyBytes == 0 {
c.MaxKeyBytes = d.MaxKeyBytes
}
if c.MaxInlineValue == 0 {
c.MaxInlineValue = d.MaxInlineValue
}
if c.MemTableSize == 0 {
c.MemTableSize = d.MemTableSize
}
if c.MaxImmutableCount == 0 {
c.MaxImmutableCount = d.MaxImmutableCount
}
}
// Validate applies defaults and verifies that all configuration invariants hold.
// The key invariant ensures that the largest possible WAL Batch can fit into
// an empty WAL segment:
//
// maxWalSegmentPayload >= maxEncodedWalBatchSize + worstCasePhysicalRecordOverhead + worstCaseBlockPadding
//
// All arithmetic is checked for overflow.
func (c *WalConfig) Validate() error {
c.applyDefaults()
if c.SyncMode != "always" {
return fmt.Errorf("config: SyncMode %q not supported (Phase 1: only \"always\")", c.SyncMode)
}
if c.MaxImmutableCount < 1 {
return fmt.Errorf("config: MaxImmutableCount must be >= 1, got %d", c.MaxImmutableCount)
}
if c.GroupCommitDelay <= 0 || c.GroupCommitDelay >= 10*time.Millisecond {
return fmt.Errorf("config: GroupCommitDelay must be > 0 and < 10ms, got %s", c.GroupCommitDelay)
}
// --- Checked arithmetic invariant validation ---
// Mirrors the derivation in docs/design.md § WAL Segment Rotation.
blockSize := uint64(c.BlockSize)
prHeaderSize := physicalRecordHeaderSize
batchHeaderSize := walBatchHeaderSize
maxBatchEntriesSize := uint64(c.MaxBatchSize)
// maxEncodedWalBatchSize = batchHeaderSize + maxBatchEntriesSize
maxEncodedWalBatchSize, err := safeAdd(batchHeaderSize, maxBatchEntriesSize)
if err != nil {
return fmt.Errorf("config: WAL batch size overflow: %w", err)
}
// maxPhysicalRecordPayload = blockSize - prHeaderSize
if blockSize <= prHeaderSize {
return fmt.Errorf("config: BlockSize %d must be > physical record header size %d", blockSize, prHeaderSize)
}
maxPhysicalRecordPayload := blockSize - prHeaderSize
// maxPhysicalRecordCount = ceil(maxEncodedWalBatchSize / maxPhysicalRecordPayload)
maxPhysicalRecordCount := divCeil(maxEncodedWalBatchSize, maxPhysicalRecordPayload)
// worstCasePhysicalRecordOverhead = maxPhysicalRecordCount * prHeaderSize
worstCasePhysicalRecordOverhead, err := safeMul(maxPhysicalRecordCount, prHeaderSize)
if err != nil {
return fmt.Errorf("config: physical record overhead overflow: %w", err)
}
// worstCaseBlockPadding = blockSize - 1 (at most one partial block of padding)
// From design doc: worstCaseBlockPadding = 7 bytes with default block size.
// Generalized: blockSize - maxPhysicalRecordPayload = prHeaderSize
worstCaseBlockPadding := prHeaderSize
// minWalSegmentPayload = maxEncodedWalBatchSize + worstCasePhysicalRecordOverhead + worstCaseBlockPadding
partial, err := safeAdd(maxEncodedWalBatchSize, worstCasePhysicalRecordOverhead)
if err != nil {
return fmt.Errorf("config: segment payload calculation overflow: %w", err)
}
minWalSegmentPayload, err := safeAdd(partial, worstCaseBlockPadding)
if err != nil {
return fmt.Errorf("config: segment payload calculation overflow: %w", err)
}
// maxWalSegmentPayload = MaxSegmentSize - walFileHeaderSize
if c.MaxSegmentSize <= walFileHeaderSize {
return fmt.Errorf("config: MaxSegmentSize %d must be > WAL file header size %d",
c.MaxSegmentSize, walFileHeaderSize)
}
maxWalSegmentPayload := c.MaxSegmentSize - walFileHeaderSize
if maxWalSegmentPayload < minWalSegmentPayload {
return fmt.Errorf("config: MaxSegmentSize %d too small: "+
"segment payload (%d) < minimum required (%d); "+
"need MaxSegmentSize >= %d",
c.MaxSegmentSize, maxWalSegmentPayload, minWalSegmentPayload,
minWalSegmentPayload+walFileHeaderSize)
}
return nil
}
// safeAdd returns a + b or an error if the result overflows uint64.
func safeAdd(a, b uint64) (uint64, error) {
if a > math.MaxUint64-b {
return 0, fmt.Errorf("uint64 overflow: %d + %d", a, b)
}
return a + b, nil
}
// safeMul returns a * b or an error if the result overflows uint64.
func safeMul(a, b uint64) (uint64, error) {
if a != 0 && b > math.MaxUint64/a {
return 0, fmt.Errorf("uint64 overflow: %d * %d", a, b)
}
return a * b, nil
}
// divCeil returns ceil(a / b) for b > 0.
func divCeil(a, b uint64) uint64 {
return (a + b - 1) / b
}