fix: move saving decision into Save and add jitter

Author: razmser

cmd/statshouse-agg/statshouse-agg.go

@@ -174,8 +174,13 @@ main_loop:
 	log.Printf("4. Waiting RPC clients to receive responses and disconnect...")
 	agg.WaitRPCServer(10 * time.Second)
 	shutdownInfo.StopRPCServer = agent.ShutdownInfoDuration(&now).Nanoseconds()
-	log.Printf("5. Saving mappings...")
-	_ = mappingsCache.Save()
+	log.Printf("5. Saving mappings cache...")
+	ok, err := mappingsCache.Save()
+	if ok {
+		log.Printf("5. Mappings cache saved")
+	} else if err != nil {
+		log.Printf("5. Failed to save mappings cache: %v", err)
+	}
 	shutdownInfo.SaveMappings = agent.ShutdownInfoDuration(&now).Nanoseconds()
 	log.Printf("6. Saving journals...")
 	agg.SaveJournals()

cmd/statshouse/statshouse.go

@@ -420,11 +420,21 @@ loop:
 	logOk.Printf("7. Waiting preprocessor to save %d buckets of historic data...", nonEmpty)
 	main.agent.WaitPreprocessor()
 	shutdownInfo.StopPreprocessor = agent.ShutdownInfoDuration(&now).Nanoseconds()
-	logOk.Printf("8. Saving mappings...")
-	_ = mappingsCache.Save()
+	logOk.Printf("8. Saving mappings cache...")
+	ok, err := mappingsCache.Save()
+	if ok {
+		logOk.Printf("8. Mappings cache saved")
+	} else if err != nil {
+		logErr.Printf("8. Failed to save mappings cache: %v", err)
+	}
 	shutdownInfo.SaveMappings = agent.ShutdownInfoDuration(&now).Nanoseconds()
 	logOk.Printf("9. Saving journal...")
-	_ = journalFast.Save()
+	ok, version, err := journalFast.Save()
+	if ok {
+		logOk.Printf("9. Journal saved, new version %d", version)
+	} else if err != nil {
+		logErr.Printf("9. Failed to save journal: %v", err)
+	}
 	shutdownInfo.SaveJournal = agent.ShutdownInfoDuration(&now).Nanoseconds()
 	shutdownInfo.FinishShutdownTime = now.UnixNano()
 	agent.ShutdownInfoSave(argv.cacheDir, shutdownInfo)

internal/aggregator/aggregator.go

@@ -366,7 +366,7 @@ func MakeAggregator(dc pcache.DiskCache, fj *os.File, fjCompact *os.File, mappin
 	go func() {
 		for {
 			time.Sleep(time.Hour) // arbitrary
-			_ = mappingsCache.Save()
+			_, _ = mappingsCache.Save()
 			a.SaveJournals()
 		}
 	}()
@@ -375,8 +375,8 @@ func MakeAggregator(dc pcache.DiskCache, fj *os.File, fjCompact *os.File, mappin
 }
 
 func (a *Aggregator) SaveJournals() {
-	_ = a.journalFast.Save()
-	_ = a.journalCompact.Save()
+	_, _, _ = a.journalFast.Save()
+	_, _, _ = a.journalCompact.Save()
 }
 
 func (a *Aggregator) Agent() *agent.Agent {

internal/metajournal/journal_fast.go

@@ -21,6 +21,7 @@ import (
 	"github.com/google/btree"
 	"github.com/mailru/easyjson"
 	"github.com/zeebo/xxh3"
+	"pgregory.net/rand"
 
 	"github.com/VKCOM/statshouse/internal/agent"
 	"github.com/VKCOM/statshouse/internal/data_model"
@@ -71,7 +72,7 @@ type JournalFast struct {
 
 	// periodic saving
 	lastSavedVersion     int64
-	periodicSaveInterval time.Duration
+	periodicSaveInterval time.Duration // we run between one and two times this interval
 
 	BuiltinLongPollImmediateOK    data_model.ItemValue
 	BuiltinLongPollImmediateError data_model.ItemValue
@@ -209,7 +210,7 @@ func (ms *JournalFast) loadImpl(fileSize int64, readAt func(b []byte, offset int
 	return src, nil
 }
 
-func (ms *JournalFast) Save() error {
+func (ms *JournalFast) Save() (bool, int64, error) {
 	// We exclude writers so that they do not block on Lock() while code below runs in RLock().
 	// If we allow this, all new readers (GetValue) block on RLock(), effectively waiting for Save to finish.
 	ms.modifyMu.Lock()
@@ -218,11 +219,20 @@ func (ms *JournalFast) Save() error {
 	ms.mu.RLock()
 	defer ms.mu.RUnlock()
 
+	// If the journal is not changed, do not save it
+	if ms.lastSavedVersion == ms.currentVersion {
+		return false, ms.currentVersion, nil
+	}
 	saver := data_model.ChunkedStorageSaver{
 		WriteAt:  ms.writeAt,
 		Truncate: ms.truncate,
 	}
-	return ms.save(&saver, 0)
+	err := ms.save(&saver, 0)
+	if err != nil {
+		return false, ms.currentVersion, err
+	}
+	ms.lastSavedVersion = ms.currentVersion
+	return true, ms.currentVersion, nil
 }
 
 func (ms *JournalFast) save(saver *data_model.ChunkedStorageSaver, maxChunkSize int) error {
@@ -561,29 +571,15 @@ func (ms *JournalFast) StartPeriodicSaving() {
 }
 
 func (ms *JournalFast) goPeriodicSaving() {
-	// Fixed 10-minute delay to prevent simultaneous saves with mappings cache
-	time.Sleep(10 * time.Minute)
-
-	ticker := time.NewTicker(ms.periodicSaveInterval)
-	defer ticker.Stop()
-
-	for range ticker.C {
-		ms.mu.RLock()
-		currentVersion := ms.currentVersion
-		lastSaved := ms.lastSavedVersion
-		ms.mu.RUnlock()
-
-		// Only save if there are changes
-		if currentVersion > lastSaved {
-			err := ms.Save()
-			if err != nil {
-				log.Printf("Periodic journal save failed: %v", err)
-			} else {
-				ms.mu.Lock()
-				ms.lastSavedVersion = currentVersion
-				ms.mu.Unlock()
-				log.Printf("Periodic journal save completed, version %d", currentVersion)
-			}
+	for {
+		sleepDuration := ms.periodicSaveInterval + time.Duration(rand.Intn(int(ms.periodicSaveInterval)))*time.Second
+		time.Sleep(sleepDuration)
+
+		ok, version, err := ms.Save()
+		if err != nil {
+			log.Printf("Periodic journal save failed: %v", err)
+		} else if ok {
+			log.Printf("Periodic journal save completed, new version %d", version)
 		}
 	}
 }

internal/metajournal/journal_test.go

@@ -98,7 +98,7 @@ func Benchmark_SaveJournal(b *testing.B) {
 	b.ResetTimer()
 	b.ReportAllocs()
 	for i := 0; i < b.N; i++ {
-		err := journal.Save()
+		_, _, err := journal.Save()
 		require.NoError(b, err)
 	}
 }
@@ -139,7 +139,8 @@ func Benchmark_LoadFast(b *testing.B) {
 			break
 		}
 	}
-	require.NoError(b, j.Save())
+	_, _, err = j.Save()
+	require.NoError(b, err)
 	_ = fp.Close()
 	fp, err = os.OpenFile(testFile, os.O_CREATE|os.O_RDWR, 0666)
 	require.NoError(b, err)

internal/pcache/mappings_cache.go

@@ -78,7 +78,8 @@ type MappingsCache struct {
 	adds                 atomic.Int64
 
 	// periodic saving
-	hasChanges           atomic.Bool
+	version              uint64 // accessed under modifyMu
+	lastSavedVersion     uint64 // accessed under modifyMu
 	periodicSaveInterval time.Duration
 
 	// custom FS
@@ -391,6 +392,7 @@ func (c *MappingsCache) AddValues(nowUnix uint32, pairs []MappingPair) {
 		c.addItem(p.Str, p.Value, nowUnix)
 		// ins++
 	}
+	c.version += 1
 	// d4 := time.Since(a)
 	// fmt.Printf("addPairsLocked len(pairs) len(items) dels ins: %d %d %d %d %v %v %v %v %v\n", len(pairs), len(items), dels, ins, d0, d1-d0, d2-d1, d3-d2, d4-d3)
 	// 1024 2048 1023 1024 104.589µs 54.722µs 459.022µs 74.654µs 93.05µs
@@ -408,7 +410,6 @@ func (c *MappingsCache) addItem(k string, v int32, accessTS uint32) {
 	c.addSumTSLocked(int64(accessTS))
 	c.cache[k] = cacheValue{value: v, accessTS: accessTS}
 	c.adds.Add(1)
-	c.hasChanges.Store(true) // Mark that cache has changes
 }
 
 func (c *MappingsCache) removeItem(k string, v int32, accessTS uint32) {
@@ -426,14 +427,13 @@ func (c *MappingsCache) removeItem(k string, v int32, accessTS uint32) {
 	c.addSumTSLocked(-int64(accessTS))
 	delete(c.cache, k)
 	c.evicts.Add(1)
-	c.hasChanges.Store(true) // Mark that cache has changes
 }
 
 func elementSizeDisk(k string) int64 { // max, can be less if short string, requires no padding, etc.
 	return 4 + int64(len(k)) + 3 + 4 + 4 // strlen, string, padding, value accessTS
 }
 
-func (c *MappingsCache) Save() error {
+func (c *MappingsCache) Save() (bool, error) {
 	// We exclude writers so that they do not block on Lock() while code below runs in RLock().
 	// If we allow this, all new readers (GetValue) block on RLock(), effectively waiting for Save to finish.
 	c.modifyMu.Lock()
@@ -442,6 +442,10 @@ func (c *MappingsCache) Save() error {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 
+	if c.version == c.lastSavedVersion {
+		return false, nil
+	}
+
 	saver := data_model.ChunkedStorageSaver{
 		WriteAt:  c.writeAt,
 		Truncate: c.truncate,
@@ -471,17 +475,22 @@ func (c *MappingsCache) Save() error {
 		})
 		for _, p := range items {
 			if err := appendItem(p.str, p.val.value, p.val.accessTS); err != nil {
-				return err
+				return false, err
 			}
 		}
 	} else {
 		for k, p := range c.cache {
 			if err := appendItem(k, p.value, p.accessTS); err != nil {
-				return err
+				return false, err
 			}
 		}
 	}
-	return saver.FinishWrite(chunk)
+	err := saver.FinishWrite(chunk)
+	if err != nil {
+		return false, err
+	}
+	c.lastSavedVersion = c.version
+	return true, nil
 }
 
 // callers are expected to ignore error from this method
@@ -532,19 +541,15 @@ func (c *MappingsCache) StartPeriodicSaving() {
 }
 
 func (c *MappingsCache) goPeriodicSaving() {
-	ticker := time.NewTicker(c.periodicSaveInterval)
-	defer ticker.Stop()
-
-	for range ticker.C {
-		// Only save if there are changes
-		if c.hasChanges.Load() {
-			err := c.Save()
-			if err != nil {
-				log.Printf("Periodic mappings cache save failed: %v", err)
-			} else {
-				c.hasChanges.Store(false) // Reset the flag after successful save
-				log.Printf("Periodic mappings cache save completed")
-			}
+	for {
+		sleepDuration := c.periodicSaveInterval + time.Duration(rand.Intn(int(c.periodicSaveInterval)))*time.Second
+		time.Sleep(sleepDuration)
+
+		ok, err := c.Save()
+		if err != nil {
+			log.Printf("Periodic mappings cache save failed: %v", err)
+		} else if ok {
+			log.Printf("Periodic mappings cache save completed")
 		}
 	}
 }

internal/pcache/mappings_cache_test.go

@@ -75,13 +75,13 @@ func TestAddValues(t *testing.T) {
 		t.Fail()
 	}
 	cache.debugPrint(now, os.Stdout)
-	if err := cache.Save(); err != nil {
+	if _, err := cache.Save(); err != nil {
 		t.Error(err)
 	}
 	fp2 := []byte(string(fp))
 	cache = newTestMappingsCache(&fp2)
 	cache.debugPrint(now, os.Stdout)
-	if err := cache.Save(); err != nil {
+	if _, err := cache.Save(); err != nil {
 		t.Error(err)
 	}
 	if string(fp) != string(fp2) {