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).
This commit is contained in:
dailz
2026-06-12 16:15:08 +08:00
parent 56bec62a6e
commit e34de4acc9
3 changed files with 96 additions and 15 deletions
+18 -6
View File
@@ -4,14 +4,14 @@ package config
import ( import (
"fmt" "fmt"
"math" "math"
"time"
) )
// WAL format constants derived from the binary layout specification. // WAL format constants derived from the binary layout specification.
const ( const (
walFileHeaderSize uint64 = 32 walFileHeaderSize uint64 = 32
walBlockSize uint32 = 32 * 1024 // 32KB
physicalRecordHeaderSize uint64 = 7 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. // WalConfig holds configuration for the Write-Ahead Log subsystem.
@@ -38,6 +38,10 @@ type WalConfig struct {
// Default: 4MB. // Default: 4MB.
MaxBatchSize uint32 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. // MaxKeyBytes is the maximum size of a single key in bytes.
// Default: 4KB. // Default: 4KB.
MaxKeyBytes uint32 MaxKeyBytes uint32
@@ -60,12 +64,13 @@ type WalConfig struct {
func Defaults() WalConfig { func Defaults() WalConfig {
return WalConfig{ return WalConfig{
MaxSegmentSize: 64 * 1024 * 1024, // 64MB MaxSegmentSize: 64 * 1024 * 1024, // 64MB
BlockSize: 32 * 1024, // 32KB BlockSize: 32 * 1024, // 32KB
SyncMode: "always", SyncMode: "always",
MaxBatchEntries: 10000, MaxBatchEntries: 10000,
MaxBatchSize: 4 * 1024 * 1024, // 4MB MaxBatchSize: 4 * 1024 * 1024, // 4MB
MaxKeyBytes: 4 * 1024, // 4KB GroupCommitDelay: 500 * time.Microsecond,
MaxInlineValue: 4 * 1024, // 4KB MaxKeyBytes: 4 * 1024, // 4KB
MaxInlineValue: 4 * 1024, // 4KB
MemTableSize: 64 * 1024 * 1024, // 64MB MemTableSize: 64 * 1024 * 1024, // 64MB
MaxImmutableCount: 3, MaxImmutableCount: 3,
} }
@@ -89,6 +94,9 @@ func (c *WalConfig) applyDefaults() {
if c.MaxBatchSize == 0 { if c.MaxBatchSize == 0 {
c.MaxBatchSize = d.MaxBatchSize c.MaxBatchSize = d.MaxBatchSize
} }
if c.GroupCommitDelay == 0 {
c.GroupCommitDelay = d.GroupCommitDelay
}
if c.MaxKeyBytes == 0 { if c.MaxKeyBytes == 0 {
c.MaxKeyBytes = d.MaxKeyBytes 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) 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 --- // --- Checked arithmetic invariant validation ---
// Mirrors the derivation in docs/design.md § WAL Segment Rotation. // Mirrors the derivation in docs/design.md § WAL Segment Rotation.
+26
View File
@@ -2,6 +2,7 @@ package config
import ( import (
"testing" "testing"
"time"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
@@ -13,17 +14,42 @@ func TestValidateDefaults(t *testing.T) {
require.NoError(t, err, "default WalConfig should pass validation") require.NoError(t, err, "default WalConfig should pass validation")
d := Defaults() 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.MaxSegmentSize, cfg.MaxSegmentSize, "MaxSegmentSize should be defaulted")
assert.Equal(t, d.BlockSize, cfg.BlockSize, "BlockSize 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.SyncMode, cfg.SyncMode, "SyncMode should be defaulted")
assert.Equal(t, d.MaxBatchEntries, cfg.MaxBatchEntries, "MaxBatchEntries 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.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.MaxKeyBytes, cfg.MaxKeyBytes, "MaxKeyBytes should be defaulted")
assert.Equal(t, d.MaxInlineValue, cfg.MaxInlineValue, "MaxInlineValue 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.MemTableSize, cfg.MemTableSize, "MemTableSize should be defaulted")
assert.Equal(t, d.MaxImmutableCount, cfg.MaxImmutableCount, "MaxImmutableCount 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) { func TestValidateSegmentTooSmall(t *testing.T) {
cfg := WalConfig{MaxSegmentSize: 1024} cfg := WalConfig{MaxSegmentSize: 1024}
err := cfg.Validate() err := cfg.Validate()
+52 -9
View File
@@ -6,8 +6,8 @@ import (
"sync/atomic" "sync/atomic"
"time" "time"
"github.com/dailz/go-kv/errkit"
"github.com/dailz/go-kv/config" "github.com/dailz/go-kv/config"
"github.com/dailz/go-kv/errkit"
"github.com/dailz/go-kv/memtable" "github.com/dailz/go-kv/memtable"
) )
@@ -101,24 +101,67 @@ func NewWalWriter(cfg *config.WalConfig, dir string, startSegmentID uint64, star
func (ww *WalWriter) runLoop() { func (ww *WalWriter) runLoop() {
defer ww.wg.Done() defer ww.wg.Done()
const batchSizeThreshold = 32 * 1024
for { for {
select { select {
case <-ww.done: case <-ww.done:
ww.processRemaining() ww.processRemaining()
return return
default: case req, ok := <-ww.queue.ch:
} if !ok {
ww.processRemaining()
return
}
requests := ww.queue.Collect() requests := []*CommitRequest{req}
if len(requests) == 0 { accumulatedSize := estimateRequestSize(req)
time.Sleep(100 * time.Microsecond) timer := time.NewTimer(ww.cfg.GroupCommitDelay)
continue
}
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() { func (ww *WalWriter) processRemaining() {
for { for {
requests := ww.queue.Collect() requests := ww.queue.Collect()