From 73778f918729d902d780c65ee85202689f1ad2d1 Mon Sep 17 00:00:00 2001 From: SantanuKar43 Date: Mon, 27 Apr 2026 15:13:05 +0530 Subject: [PATCH 1/2] add batch commit at fixed time interval --- internal/handler/handler.go | 14 ++++++++-- internal/wal/wal.go | 51 ++++++++++++++++++++++--------------- 2 files changed, 42 insertions(+), 23 deletions(-) diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 7897350..0a30527 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -16,10 +16,15 @@ func Handle(input string, store *store.Store, wal *wal.Wal) string { return "ERR wrong number of arguments" } - _, err := wal.Append(cmd.Name, cmd.Args[0], cmd.Args[1]) + lsn, err := wal.Append(cmd.Name, cmd.Args[0], cmd.Args[1]) if err != nil { return fmt.Sprintf("ERR unable to write to WAL: %s", err.Error()) } + wal.CommitBroadcast.L.Lock() + for lsn > wal.CommittedLSN { + wal.CommitBroadcast.Wait() + } + wal.CommitBroadcast.L.Unlock() store.Set(cmd.Args[0], cmd.Args[1]) return "OK" @@ -37,10 +42,15 @@ func Handle(input string, store *store.Store, wal *wal.Wal) string { return "ERR wrong number of arguments" } - _, err := wal.Append(cmd.Name, cmd.Args[0]) + lsn, err := wal.Append(cmd.Name, cmd.Args[0]) if err != nil { return fmt.Sprintf("ERR unable to write to WAL: %s", err.Error()) } + wal.CommitBroadcast.L.Lock() + for lsn > wal.CommittedLSN { + wal.CommitBroadcast.Wait() + } + wal.CommitBroadcast.L.Unlock() store.Delete(cmd.Args[0]) return "OK" diff --git a/internal/wal/wal.go b/internal/wal/wal.go index 8cefcb2..7a4f071 100644 --- a/internal/wal/wal.go +++ b/internal/wal/wal.go @@ -33,12 +33,14 @@ type Wal struct { entryBuf []byte fatalErrChan chan error maxEntryLen uint32 + CommitBroadcast *sync.Cond + CommittedLSN uint64 } var byteOrder binary.ByteOrder = binary.LittleEndian func NewWal(filepath string, fsyncStrategy FsyncStrategy, maxWalEntryLength uint32, ctx context.Context, fatalErrChan chan error) (*Wal, error) { - file, err := os.OpenFile(filepath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644) + file, err := os.OpenFile(filepath, os.O_WRONLY|os.O_CREATE|os.O_APPEND|os.O_SYNC, 0644) if err != nil { return nil, err } @@ -50,17 +52,21 @@ func NewWal(filepath string, fsyncStrategy FsyncStrategy, maxWalEntryLength uint fatalErrChan: fatalErrChan, maxEntryLen: maxWalEntryLength, entryBuf: make([]byte, maxWalEntryLength), + asyncBuffer: new(bytes.Buffer), } - if fsyncStrategy != ALWAYS { - w.asyncBuffer = new(bytes.Buffer) - go scheduleFsyncEverySec(w, ctx) - } + locker := sync.Mutex{} + w.CommitBroadcast = sync.NewCond(&locker) + flushIntervalMillis := 1000 + if fsyncStrategy == ALWAYS { + flushIntervalMillis = 10 + } + go scheduleWalFlush(w, ctx, flushIntervalMillis) return w, nil } -func scheduleFsyncEverySec(w *Wal, ctx context.Context) { - ticker := time.NewTicker(1 * time.Second) +func scheduleWalFlush(w *Wal, ctx context.Context, flushIntervalMillis int) { + ticker := time.NewTicker(time.Duration(flushIntervalMillis) * time.Millisecond) defer ticker.Stop() for { select { @@ -75,8 +81,13 @@ func scheduleFsyncEverySec(w *Wal, ctx context.Context) { data := w.asyncBuffer.Bytes() for len(data) > 0 { n, err := w.file.Write(data) - if n > 0 { + if n > 0 && err == nil { data = data[n:] + if w.fsyncStrategy == ALWAYS { + w.CommitBroadcast.L.Lock() + w.CommittedLSN += uint64(n) + w.CommitBroadcast.L.Unlock() + } } if err != nil { writeErr = err @@ -91,22 +102,21 @@ func scheduleFsyncEverySec(w *Wal, ctx context.Context) { w.asyncBuffer.Write(data) w.mu.Unlock() - if writeErr == nil && w.fsyncStrategy != NEVER { - writeErr = w.file.Sync() - } - if writeErr != nil { log.Printf("wal write failure: %s, stopping the server", writeErr) w.fatalErrChan <- writeErr return } + if w.fsyncStrategy == ALWAYS { + w.CommitBroadcast.Broadcast() + } case <-ctx.Done(): return } } } -func (w *Wal) Append(cmd ... string) (int, error) { +func (w *Wal) Append(cmd ... string) (uint64, error) { w.mu.Lock() defer w.mu.Unlock() if !w.closed { @@ -115,21 +125,20 @@ func (w *Wal) Append(cmd ... string) (int, error) { if err != nil { return 0, err } - + + bytes, err := w.asyncBuffer.Write(entry) if w.fsyncStrategy != ALWAYS { - bytes, err := w.asyncBuffer.Write(entry) - return bytes, err + w.CommitBroadcast.L.Lock() + w.CommittedLSN += uint64(bytes) + w.CommitBroadcast.L.Unlock() + return w.CommittedLSN, err } - bytes, err := w.file.Write(entry) - if err == nil { - err = w.file.Sync() - } if err != nil { log.Printf("wal append failure: %s, stopping the server", err) w.fatalErrChan <- err } - return bytes, err + return w.CommittedLSN + uint64(bytes), err } return 0, fmt.Errorf("unable to append, wal already closed") } From 2bc77692b75c0a9efdc419ab5236099f97fec6b2 Mon Sep 17 00:00:00 2001 From: SantanuKar43 Date: Mon, 27 Apr 2026 18:19:59 +0530 Subject: [PATCH 2/2] signal flush instead of fixed time interval --- cmd/server/main.go | 2 +- internal/handler/handler.go | 36 +++--- internal/wal/wal.go | 222 ++++++++++++++++++++---------------- 3 files changed, 140 insertions(+), 120 deletions(-) diff --git a/cmd/server/main.go b/cmd/server/main.go index 8273cb4..de17466 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -13,7 +13,7 @@ import ( const port = ":9000" // TODO configs const walFilePath = "bin/wal.log" // TODO configs -const fsyncStrategy = wal.EVERY_SEC // TODO configs +const fsyncStrategy = wal.ALWAYS // TODO configs const maxWalEntryLength = 1 << 30 // TODO configs func main() { diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 0a30527..7f76a54 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -3,6 +3,7 @@ package handler import ( "fmt" "log" + "com.github.SantanuKar43/simple-kv/internal/protocol" "com.github.SantanuKar43/simple-kv/internal/store" "com.github.SantanuKar43/simple-kv/internal/wal" @@ -15,16 +16,9 @@ func Handle(input string, store *store.Store, wal *wal.Wal) string { if len(cmd.Args) != 2 { return "ERR wrong number of arguments" } - - lsn, err := wal.Append(cmd.Name, cmd.Args[0], cmd.Args[1]) - if err != nil { + if err := appendToWal(cmd, wal); err != nil { return fmt.Sprintf("ERR unable to write to WAL: %s", err.Error()) } - wal.CommitBroadcast.L.Lock() - for lsn > wal.CommittedLSN { - wal.CommitBroadcast.Wait() - } - wal.CommitBroadcast.L.Unlock() store.Set(cmd.Args[0], cmd.Args[1]) return "OK" @@ -41,16 +35,9 @@ func Handle(input string, store *store.Store, wal *wal.Wal) string { if len(cmd.Args) != 1 { return "ERR wrong number of arguments" } - - lsn, err := wal.Append(cmd.Name, cmd.Args[0]) - if err != nil { + if err := appendToWal(cmd, wal); err != nil { return fmt.Sprintf("ERR unable to write to WAL: %s", err.Error()) } - wal.CommitBroadcast.L.Lock() - for lsn > wal.CommittedLSN { - wal.CommitBroadcast.Wait() - } - wal.CommitBroadcast.L.Unlock() store.Delete(cmd.Args[0]) return "OK" @@ -59,6 +46,19 @@ func Handle(input string, store *store.Store, wal *wal.Wal) string { } } +func appendToWal(cmd protocol.Command, wal *wal.Wal) error { + lsn, err := wal.Append(cmd.Name, cmd.Args[0]) + if err != nil { + return err + } + wal.CommitSignal.L.Lock() + for lsn > wal.CommittedLSN { + wal.CommitSignal.Wait() + } + wal.CommitSignal.L.Unlock() + return nil +} + func Replay(input string, store *store.Store) { cmd := protocol.Parse(input) switch cmd.Name { @@ -67,6 +67,6 @@ func Replay(input string, store *store.Store) { case "DEL": store.Delete(cmd.Args[0]) default: - log.Printf("invalid command parsed %s\n", cmd.Name) + log.Printf("invalid command parsed %s\n", cmd.Name) } -} \ No newline at end of file +} diff --git a/internal/wal/wal.go b/internal/wal/wal.go index 7a4f071..5980ec9 100644 --- a/internal/wal/wal.go +++ b/internal/wal/wal.go @@ -2,6 +2,7 @@ package wal import ( "bufio" + "bytes" "context" "encoding/binary" "fmt" @@ -11,8 +12,8 @@ import ( "log" "os" "strings" - "bytes" "sync" + "sync/atomic" "time" ) @@ -25,16 +26,19 @@ const ( ) type Wal struct { - mu sync.Mutex - file *os.File - closed bool - fsyncStrategy FsyncStrategy - asyncBuffer *bytes.Buffer - entryBuf []byte - fatalErrChan chan error - maxEntryLen uint32 - CommitBroadcast *sync.Cond - CommittedLSN uint64 + mu *sync.Mutex + file *os.File + closed bool + fsyncStrategy FsyncStrategy + asyncBuffer *bytes.Buffer + entryBuf []byte + fatalErrChan chan error + maxEntryLen uint32 + CommitSignal *sync.Cond + CommittedLSN uint64 + UncommittedLSN uint64 + flushChan chan bool + isFlushing atomic.Bool } var byteOrder binary.ByteOrder = binary.LittleEndian @@ -45,22 +49,21 @@ func NewWal(filepath string, fsyncStrategy FsyncStrategy, maxWalEntryLength uint return nil, err } w := &Wal{ - mu: sync.Mutex{}, - file: file, - closed: false, + mu: &sync.Mutex{}, + file: file, + closed: false, fsyncStrategy: fsyncStrategy, - fatalErrChan: fatalErrChan, - maxEntryLen: maxWalEntryLength, - entryBuf: make([]byte, maxWalEntryLength), - asyncBuffer: new(bytes.Buffer), + fatalErrChan: fatalErrChan, + maxEntryLen: maxWalEntryLength, + entryBuf: make([]byte, maxWalEntryLength), + asyncBuffer: new(bytes.Buffer), + flushChan: make(chan bool, 1), + isFlushing: atomic.Bool{}, } - locker := sync.Mutex{} - w.CommitBroadcast = sync.NewCond(&locker) + locker := &sync.Mutex{} + w.CommitSignal = sync.NewCond(locker) flushIntervalMillis := 1000 - if fsyncStrategy == ALWAYS { - flushIntervalMillis = 10 - } go scheduleWalFlush(w, ctx, flushIntervalMillis) return w, nil } @@ -71,52 +74,65 @@ func scheduleWalFlush(w *Wal, ctx context.Context, flushIntervalMillis int) { for { select { case <-ticker.C: - w.mu.Lock() - if w.closed { - w.mu.Unlock() - return - } + w.signalFlush() + case <-w.flushChan: + w.flush() + case <-ctx.Done(): + return + } + } +} - var writeErr error - data := w.asyncBuffer.Bytes() - for len(data) > 0 { - n, err := w.file.Write(data) - if n > 0 && err == nil { - data = data[n:] - if w.fsyncStrategy == ALWAYS { - w.CommitBroadcast.L.Lock() - w.CommittedLSN += uint64(n) - w.CommitBroadcast.L.Unlock() - } - } - if err != nil { - writeErr = err - break - } - if n == 0 { - writeErr = io.ErrUnexpectedEOF - break - } - } - w.asyncBuffer.Reset() - w.asyncBuffer.Write(data) - w.mu.Unlock() - - if writeErr != nil { - log.Printf("wal write failure: %s, stopping the server", writeErr) - w.fatalErrChan <- writeErr - return - } +func (w *Wal) signalFlush() { + if w.isFlushing.CompareAndSwap(false, true) { + w.flushChan <- true + } +} + +func (w *Wal) flush() { + w.mu.Lock() + if w.closed { + w.mu.Unlock() + return + } + defer w.isFlushing.Store(false) + w.CommitSignal.L.Lock() + defer w.CommitSignal.L.Unlock() + + var writeErr error + data := w.asyncBuffer.Bytes() + for len(data) > 0 { + n, err := w.file.Write(data) + if n > 0 && err == nil { + data = data[n:] if w.fsyncStrategy == ALWAYS { - w.CommitBroadcast.Broadcast() + w.CommittedLSN += uint64(n) } - case <-ctx.Done(): - return } + if err != nil { + writeErr = err + break + } + if n == 0 { + writeErr = io.ErrUnexpectedEOF + break + } + } + w.asyncBuffer.Reset() + w.asyncBuffer.Write(data) + w.mu.Unlock() + + if writeErr != nil { + log.Printf("wal write failure: %s, stopping the server", writeErr) + w.fatalErrChan <- writeErr + return + } + if w.fsyncStrategy == ALWAYS { + w.CommitSignal.Broadcast() } } -func (w *Wal) Append(cmd ... string) (uint64, error) { +func (w *Wal) Append(cmd ...string) (uint64, error) { w.mu.Lock() defer w.mu.Unlock() if !w.closed { @@ -125,12 +141,13 @@ func (w *Wal) Append(cmd ... string) (uint64, error) { if err != nil { return 0, err } - + bytes, err := w.asyncBuffer.Write(entry) if w.fsyncStrategy != ALWAYS { - w.CommitBroadcast.L.Lock() + w.CommitSignal.L.Lock() + w.UncommittedLSN += uint64(bytes) w.CommittedLSN += uint64(bytes) - w.CommitBroadcast.L.Unlock() + w.CommitSignal.L.Unlock() return w.CommittedLSN, err } @@ -138,7 +155,10 @@ func (w *Wal) Append(cmd ... string) (uint64, error) { log.Printf("wal append failure: %s, stopping the server", err) w.fatalErrChan <- err } - return w.CommittedLSN + uint64(bytes), err + w.UncommittedLSN += uint64(bytes) + + w.signalFlush() + return w.UncommittedLSN, err } return 0, fmt.Errorf("unable to append, wal already closed") } @@ -152,10 +172,10 @@ func (w *Wal) getWalEntry(line string) ([]byte, error) { } byteOrder.PutUint32(w.entryBuf[:4], uint32(length)) - copy(w.entryBuf[4:(length + 4)], lineBytes) + copy(w.entryBuf[4:(length+4)], lineBytes) checksum := calcChecksum(w.entryBuf[:(length + 4)]) - byteOrder.PutUint32(w.entryBuf[(length + 4):(length + 8)], checksum) + byteOrder.PutUint32(w.entryBuf[(length+4):(length+8)], checksum) return w.entryBuf[:(length + 8)], nil } @@ -166,59 +186,59 @@ func calcChecksum(bytes []byte) uint32 { func (w *Wal) WALIterator(fromOffset int64) iter.Seq2[string, error] { // the iterator also validates the wal file and truncates till last read safe offset - return func(yield func(string, error) bool) { + return func(yield func(string, error) bool) { w.mu.Lock() defer w.mu.Unlock() - f, err := os.OpenFile(w.file.Name(), os.O_RDWR, 0644) // create a separate fd - if err != nil { - yield("", err) - return - } - defer f.Close() - - if _, err := f.Seek(fromOffset, io.SeekStart); err != nil { - yield("", err) - return - } + f, err := os.OpenFile(w.file.Name(), os.O_RDWR, 0644) // create a separate fd + if err != nil { + yield("", err) + return + } + defer f.Close() + + if _, err := f.Seek(fromOffset, io.SeekStart); err != nil { + yield("", err) + return + } var safeOffset int64 = 0 var currOffset int64 = 0 buf := make([]byte, w.maxEntryLen) - reader := bufio.NewReader(f) + reader := bufio.NewReader(f) - for { + for { // read length - _, err := io.ReadFull(reader, buf[:4]) - if err == io.EOF { - return - } - if err != nil { + _, err := io.ReadFull(reader, buf[:4]) + if err == io.EOF { + return + } + if err != nil { log.Printf("wal corrupted, err %s, truncating till last safe read offset %d\n", err, safeOffset) f.Truncate(safeOffset) f.Sync() yield("", err) - return - } - length := byteOrder.Uint32(buf[:4]) + return + } + length := byteOrder.Uint32(buf[:4]) currOffset += 4 if length > w.maxEntryLen { log.Printf("wal corrupted, length found too big, truncating till last safe read offset %d\n", safeOffset) f.Truncate(safeOffset) f.Sync() - return + return } - // read full wal entry ([length | data | checksum]) - _, err = io.ReadFull(reader, buf[4:(8 + length)]) - if err != nil { + // read full wal entry ([length | data | checksum]) + _, err = io.ReadFull(reader, buf[4:(8+length)]) + if err != nil { log.Printf("wal corrupted, err %s, truncating till last safe read offset %d\n", err, safeOffset) f.Truncate(safeOffset) f.Sync() yield("", err) - return - } - data := buf[4:(length + 4)] + return + } + data := buf[4:(length + 4)] currOffset += int64(length) + 4 // validate checksum @@ -234,10 +254,10 @@ func (w *Wal) WALIterator(fromOffset int64) iter.Seq2[string, error] { safeOffset = currOffset if !yield(string(data), nil) { - return - } - } - } + return + } + } + } } func (w *Wal) Close() error {