Merge pull request #5778 from filecoin-project/feat/splitstore-compact-hotstore

splitstore: compact hotstore prior to garbage collection

blockstore/badger/blockstore.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"runtime"
 	"sync/atomic"
 
 	"github.com/dgraph-io/badger/v2"
@@ -150,6 +151,20 @@ func (b *Blockstore) CollectGarbage() error {
 	return err
 }
 
+// Compact runs a synchronous compaction
+func (b *Blockstore) Compact() error {
+	if atomic.LoadInt64(&b.state) != stateOpen {
+		return ErrBlockstoreClosed
+	}
+
+	nworkers := runtime.NumCPU() / 2
+	if nworkers < 2 {
+		nworkers = 2
+	}
+
+	return b.DB.Flatten(nworkers)
+}
+
 // View implements blockstore.Viewer, which leverages zero-copy read-only
 // access to values.
 func (b *Blockstore) View(cid cid.Cid, fn func([]byte) error) error {

blockstore/splitstore/splitstore.go

@@ -798,15 +798,26 @@ func (s *SplitStore) purgeTracking(cids []cid.Cid) error {
 }
 
 func (s *SplitStore) gcHotstore() {
+	if compact, ok := s.hot.(interface{ Compact() error }); ok {
+		log.Infof("compacting hotstore")
+		startCompact := time.Now()
+		err := compact.Compact()
+		if err != nil {
+			log.Warnf("error compacting hotstore: %s", err)
+			return
+		}
+		log.Infow("hotstore compaction done", "took", time.Since(startCompact))
+	}
+
 	if gc, ok := s.hot.(interface{ CollectGarbage() error }); ok {
 		log.Infof("garbage collecting hotstore")
 		startGC := time.Now()
 		err := gc.CollectGarbage()
 		if err != nil {
 			log.Warnf("error garbage collecting hotstore: %s", err)
-		} else {
-			log.Infow("garbage collection done", "took", time.Since(startGC))
+			return
 		}
+		log.Infow("hotstore garbage collection done", "took", time.Since(startGC))
 	}
 }