diff --git a/config/config.go b/config/config.go index b464c05..481dee9 100644 --- a/config/config.go +++ b/config/config.go @@ -4,14 +4,14 @@ package config import ( "fmt" "math" + "time" ) // WAL format constants derived from the binary layout specification. const ( - walFileHeaderSize uint64 = 32 - walBlockSize uint32 = 32 * 1024 // 32KB + walFileHeaderSize uint64 = 32 physicalRecordHeaderSize uint64 = 7 - walBatchHeaderSize uint64 = 18 // flags(2) + baseSequence(8) + entryCount(4) + entriesSize(4) + walBatchHeaderSize uint64 = 18 // flags(2) + baseSequence(8) + entryCount(4) + entriesSize(4) ) // WalConfig holds configuration for the Write-Ahead Log subsystem. @@ -38,6 +38,10 @@ type WalConfig struct { // 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 @@ -60,12 +64,13 @@ type WalConfig struct { func Defaults() WalConfig { return WalConfig{ MaxSegmentSize: 64 * 1024 * 1024, // 64MB - BlockSize: 32 * 1024, // 32KB + BlockSize: 32 * 1024, // 32KB SyncMode: "always", MaxBatchEntries: 10000, MaxBatchSize: 4 * 1024 * 1024, // 4MB - MaxKeyBytes: 4 * 1024, // 4KB - MaxInlineValue: 4 * 1024, // 4KB + GroupCommitDelay: 500 * time.Microsecond, + MaxKeyBytes: 4 * 1024, // 4KB + MaxInlineValue: 4 * 1024, // 4KB MemTableSize: 64 * 1024 * 1024, // 64MB MaxImmutableCount: 3, } @@ -89,6 +94,9 @@ func (c *WalConfig) applyDefaults() { if c.MaxBatchSize == 0 { c.MaxBatchSize = d.MaxBatchSize } + if c.GroupCommitDelay == 0 { + c.GroupCommitDelay = d.GroupCommitDelay + } if c.MaxKeyBytes == 0 { c.MaxKeyBytes = d.MaxKeyBytes } @@ -121,6 +129,10 @@ func (c *WalConfig) Validate() error { 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. diff --git a/config/config_test.go b/config/config_test.go index d1a3dfe..7c3e2b1 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -2,6 +2,7 @@ package config import ( "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -13,17 +14,42 @@ func TestValidateDefaults(t *testing.T) { require.NoError(t, err, "default WalConfig should pass validation") d := Defaults() + assert.Equal(t, 500*time.Microsecond, d.GroupCommitDelay) assert.Equal(t, d.MaxSegmentSize, cfg.MaxSegmentSize, "MaxSegmentSize should be defaulted") assert.Equal(t, d.BlockSize, cfg.BlockSize, "BlockSize should be defaulted") assert.Equal(t, d.SyncMode, cfg.SyncMode, "SyncMode should be defaulted") assert.Equal(t, d.MaxBatchEntries, cfg.MaxBatchEntries, "MaxBatchEntries should be defaulted") assert.Equal(t, d.MaxBatchSize, cfg.MaxBatchSize, "MaxBatchSize should be defaulted") + assert.Equal(t, d.GroupCommitDelay, cfg.GroupCommitDelay, "GroupCommitDelay should be defaulted") assert.Equal(t, d.MaxKeyBytes, cfg.MaxKeyBytes, "MaxKeyBytes should be defaulted") assert.Equal(t, d.MaxInlineValue, cfg.MaxInlineValue, "MaxInlineValue should be defaulted") assert.Equal(t, d.MemTableSize, cfg.MemTableSize, "MemTableSize should be defaulted") assert.Equal(t, d.MaxImmutableCount, cfg.MaxImmutableCount, "MaxImmutableCount should be defaulted") } +func TestValidateGroupCommitDelay(t *testing.T) { + tests := []struct { + name string + delay time.Duration + want string + }{ + {name: "negative delay", delay: -time.Microsecond, want: "GroupCommitDelay"}, + {name: "ten milliseconds rejected", delay: 10 * time.Millisecond, want: "GroupCommitDelay"}, + {name: "above ten milliseconds rejected", delay: 11 * time.Millisecond, want: "GroupCommitDelay"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := Defaults() + cfg.GroupCommitDelay = tt.delay + + err := cfg.Validate() + require.Error(t, err) + assert.Contains(t, err.Error(), tt.want) + }) + } +} + func TestValidateSegmentTooSmall(t *testing.T) { cfg := WalConfig{MaxSegmentSize: 1024} err := cfg.Validate() diff --git a/wal/writer.go b/wal/writer.go index aee496f..b214651 100644 --- a/wal/writer.go +++ b/wal/writer.go @@ -6,8 +6,8 @@ import ( "sync/atomic" "time" - "github.com/dailz/go-kv/errkit" "github.com/dailz/go-kv/config" + "github.com/dailz/go-kv/errkit" "github.com/dailz/go-kv/memtable" ) @@ -101,24 +101,67 @@ func NewWalWriter(cfg *config.WalConfig, dir string, startSegmentID uint64, star func (ww *WalWriter) runLoop() { defer ww.wg.Done() + const batchSizeThreshold = 32 * 1024 + for { select { case <-ww.done: ww.processRemaining() return - default: - } + case req, ok := <-ww.queue.ch: + if !ok { + ww.processRemaining() + return + } - requests := ww.queue.Collect() - if len(requests) == 0 { - time.Sleep(100 * time.Microsecond) - continue - } + requests := []*CommitRequest{req} + accumulatedSize := estimateRequestSize(req) + timer := time.NewTimer(ww.cfg.GroupCommitDelay) - ww.processBatch(requests) + collect: + for accumulatedSize < batchSizeThreshold { + select { + case <-timer.C: + break collect + case req, ok := <-ww.queue.ch: + if !ok { + break collect + } + requests = append(requests, req) + accumulatedSize += estimateRequestSize(req) + case <-ww.done: + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + ww.sendError(requests, errkit.ErrWriteStopped) + ww.processRemaining() + return + } + } + + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + + ww.processBatch(requests) + } } } +func estimateRequestSize(req *CommitRequest) int { + size := 0 + for _, entry := range req.Entries { + size += len(entry.Key) + len(entry.Value) + 4 + } + return size +} + func (ww *WalWriter) processRemaining() { for { requests := ww.queue.Collect()