From 333a979529128394c61954b690d17a0577c64c39 Mon Sep 17 00:00:00 2001
From: sawka <mike.sawka@gmail.com>
Date: Tue, 28 May 2024 18:27:38 -0700
Subject: [PATCH] hook up blockstore flusher

---
 pkg/blockstore/blockstore.go                  | 49 +++++++++++++++++--
 .../{dbsetup.go => blockstore_dbsetup.go}     |  3 ++
 pkg/blockstore/blockstore_test.go             |  2 +-
 3 files changed, 48 insertions(+), 6 deletions(-)
 rename pkg/blockstore/{dbsetup.go => blockstore_dbsetup.go} (97%)

diff --git a/pkg/blockstore/blockstore.go b/pkg/blockstore/blockstore.go
index f770b51d..fcf19f9f 100644
--- a/pkg/blockstore/blockstore.go
+++ b/pkg/blockstore/blockstore.go
@@ -11,6 +11,8 @@ import (
 	"context"
 	"fmt"
 	"io/fs"
+	"log"
+	"runtime/debug"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -265,28 +267,40 @@ func (s *BlockStore) ReadFile(ctx context.Context, blockId string, name string)
 	return
 }
 
-func (s *BlockStore) FlushCache(ctx context.Context) error {
+type FlushStats struct {
+	FlushDuration   time.Duration
+	NumDirtyEntries int
+	NumCommitted    int
+}
+
+func (s *BlockStore) FlushCache(ctx context.Context) (stats FlushStats, rtnErr error) {
 	wasFlushing := s.setUnlessFlushing()
 	if wasFlushing {
-		return fmt.Errorf("flush already in progress")
+		return stats, fmt.Errorf("flush already in progress")
 	}
 	defer s.setIsFlushing(false)
+	startTime := time.Now()
+	defer func() {
+		stats.FlushDuration = time.Since(startTime)
+	}()
 
 	// get a copy of dirty keys so we can iterate without the lock
 	dirtyCacheKeys := s.getDirtyCacheKeys()
+	stats.NumDirtyEntries = len(dirtyCacheKeys)
 	for _, key := range dirtyCacheKeys {
 		err := withLock(s, key.BlockId, key.Name, func(entry *CacheEntry) error {
 			return entry.flushToDB(ctx, false)
 		})
 		if ctx.Err() != nil {
 			// transient error (also must stop the loop)
-			return ctx.Err()
+			return stats, ctx.Err()
 		}
 		if err != nil {
-			return fmt.Errorf("error flushing cache entry[%v]: %v", key, err)
+			return stats, fmt.Errorf("error flushing cache entry[%v]: %v", key, err)
 		}
+		stats.NumCommitted++
 	}
-	return nil
+	return stats, nil
 }
 
 ///////////////////////////////////
@@ -367,7 +381,32 @@ func (s *BlockStore) setUnlessFlushing() bool {
 	}
 	s.IsFlushing = true
 	return false
+}
 
+func (s *BlockStore) runFlushWithNewContext() (FlushStats, error) {
+	ctx, cancelFn := context.WithTimeout(context.Background(), DefaultFlushTime)
+	defer cancelFn()
+	return s.FlushCache(ctx)
+}
+
+func (s *BlockStore) runFlusher() {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Printf("panic in blockstore flusher: %v\n", r)
+			debug.PrintStack()
+		}
+	}()
+	for {
+		stats, err := s.runFlushWithNewContext()
+		if err != nil || stats.NumDirtyEntries > 0 {
+			log.Printf("blockstore flush: %d/%d entries flushed, err:%v\n", stats.NumCommitted, stats.NumDirtyEntries, err)
+		}
+		if stopFlush.Load() {
+			log.Printf("blockstore flusher stopping\n")
+			return
+		}
+		time.Sleep(DefaultFlushTime)
+	}
 }
 
 func minInt64(a, b int64) int64 {
diff --git a/pkg/blockstore/dbsetup.go b/pkg/blockstore/blockstore_dbsetup.go
similarity index 97%
rename from pkg/blockstore/dbsetup.go
rename to pkg/blockstore/blockstore_dbsetup.go
index a4937ab9..51d66911 100644
--- a/pkg/blockstore/dbsetup.go
+++ b/pkg/blockstore/blockstore_dbsetup.go
@@ -42,6 +42,9 @@ func InitBlockstore() error {
 	if err != nil {
 		return err
 	}
+	if !stopFlush.Load() {
+		go GBS.runFlusher()
+	}
 	log.Printf("blockstore initialized\n")
 	return nil
 }
diff --git a/pkg/blockstore/blockstore_test.go b/pkg/blockstore/blockstore_test.go
index 170f5897..f2fc475e 100644
--- a/pkg/blockstore/blockstore_test.go
+++ b/pkg/blockstore/blockstore_test.go
@@ -565,7 +565,7 @@ func TestSimpleDBFlush(t *testing.T) {
 		t.Fatalf("error writing data: %v", err)
 	}
 	checkFileData(t, ctx, blockId, fileName, "hello world!")
-	err = GBS.FlushCache(ctx)
+	_, err = GBS.FlushCache(ctx)
 	if err != nil {
 		t.Fatalf("error flushing cache: %v", err)
 	}