get rid of goroutine iteration in tracking store; long live ForEach

chain/store/splitstore/snoop.go

@@ -1,7 +1,6 @@
 package splitstore
 
 import (
-	"context"
 	"os"
 
 	"golang.org/x/xerrors"
@@ -20,7 +19,7 @@ type TrackingStore interface {
 	PutBatch([]cid.Cid, abi.ChainEpoch) error
 	Get(cid.Cid) (abi.ChainEpoch, error)
 	Delete(cid.Cid) error
-	Keys(context.Context) (<-chan cid.Cid, error)
+	ForEach(func(cid.Cid, abi.ChainEpoch) error) error
 	Close() error
 }
 
@@ -40,9 +39,6 @@ func NewTrackingStore(path string) (TrackingStore, error) {
 	if err = env.SetMaxDBs(1); err != nil {
 		return nil, xerrors.Errorf("failed to set LMDB max dbs: %w", err)
 	}
-	// if err = env.SetMaxReaders(2); err != nil {
-	// 	return nil, xerrors.Errorf("failed to set LMDB max readers: %w", err)
-	// }
 
 	if st, err := os.Stat(path); os.IsNotExist(err) {
 		if err := os.MkdirAll(path, 0777); err != nil {
@@ -129,15 +125,10 @@ func (s *trackingStore) Delete(cid cid.Cid) error {
 		})
 }
 
-func (s *trackingStore) Keys(ctx context.Context) (<-chan cid.Cid, error) {
+func (s *trackingStore) ForEach(f func(cid.Cid, abi.ChainEpoch) error) error {
-	ch := make(chan cid.Cid)
+	return withMaxReadersRetry(
-	go func() {
-		defer close(ch)
-
-		err := withMaxReadersRetry(
 		func() error {
 			return s.env.View(func(txn *lmdb.Txn) error {
-
 				txn.RawRead = true
 				cur, err := txn.OpenCursor(s.db)
 				if err != nil {
@@ -146,7 +137,7 @@ func (s *trackingStore) Keys(ctx context.Context) (<-chan cid.Cid, error) {
 				defer cur.Close()
 
 				for {
-						k, _, err := cur.Get(nil, nil, lmdb.Next)
+					k, v, err := cur.Get(nil, nil, lmdb.Next)
 					if err != nil {
 						if lmdb.IsNotFound(err) {
 							return nil
@@ -155,21 +146,16 @@ func (s *trackingStore) Keys(ctx context.Context) (<-chan cid.Cid, error) {
 						return err
 					}
 
-						select {
+					cid := cid.NewCidV1(cid.Raw, k)
-						case ch <- cid.NewCidV1(cid.Raw, k):
+					epoch := bytesToEpoch(v)
-						case <-ctx.Done():
-							return nil
-						}
-					}
-				})
-			})
 
+					err = f(cid, epoch)
 					if err != nil {
-			log.Errorf("error iterating over tracking store keys: %s", err)
+						return err
 					}
-	}()
+				}
-
+			})
-	return ch, nil
+		})
 }
 
 func (s *trackingStore) Close() error {

chain/store/splitstore/splitstore.go

@@ -9,6 +9,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"golang.org/x/xerrors"
+
 	"github.com/ledgerwatch/lmdb-go/lmdb"
 
 	blocks "github.com/ipfs/go-block-format"
@@ -369,51 +371,37 @@ func (s *SplitStore) compact() {
 	// Phase 2: sweep cold objects:
 	// - If a cold object is reachable in the hot range, it stays in the hotstore.
 	// - If a cold object is reachable in the cold range, it is moved to the coldstore.
-	// - If a cold object is unreachable, it is deleted.
+	// - If a cold object is unreachable, it is deleted if GC is enabled, otherwise moved to the coldstore.
-	ch, err := s.snoop.Keys(context.Background())
-	if err != nil {
-		// TODO do something better here
-		panic(err)
-	}
-
 	startSweep := time.Now()
 	log.Info("sweeping cold objects")
 
 	// some stats for logging
 	var stHot, stCold, stDead int
 
-	for cid := range ch {
+	err = s.snoop.ForEach(func(cid cid.Cid, wrEpoch abi.ChainEpoch) error {
-		wrEpoch, err := s.snoop.Get(cid)
-		if err != nil {
-			// TODO do something better here
-			panic(err)
-		}
-
 		// is the object stil hot?
 		if wrEpoch > coldEpoch {
 			// yes, stay in the hotstore
 			stHot++
-			continue
+			return nil
 		}
 
 		// the object is cold -- check whether it is reachable in the hot range
 		mark, err := hotSet.Has(cid)
 		if err != nil {
-			// TODO do something better here
+			return xerrors.Errorf("error checking live mark for %s: %w", cid, err)
-			panic(err)
 		}
 
 		if mark {
 			// the object is reachable in the hot range, stay in the hotstore
 			stHot++
-			continue
+			return nil
 		}
 
 		// check whether it is reachable in the cold range
 		mark, err = coldSet.Has(cid)
 		if err != nil {
-			// TODO do something better here
+			return xerrors.Errorf("error checkiing cold set for %s: %w", cid, err)
-			panic(err)
 		}
 
 		if s.enableGC {
@@ -421,14 +409,12 @@ func (s *SplitStore) compact() {
 				// the object is reachable in the cold range, move it to the cold store
 				blk, err := s.hot.Get(cid)
 				if err != nil {
-					// TODO do something better here
+					return xerrors.Errorf("error retrieving tracked block %s from hotstore: %w ", cid, err)
-					panic(err)
 				}
 
 				err = s.cold.Put(blk)
 				if err != nil {
-					// TODO do something better here
+					return xerrors.Errorf("error puting block %s to coldstore: %w", cid, err)
-					panic(err)
 				}
 
 				stCold++
@@ -440,14 +426,12 @@ func (s *SplitStore) compact() {
 			// if GC is disabled, we move both cold and dead objects to the coldstore
 			blk, err := s.hot.Get(cid)
 			if err != nil {
-				// TODO do something better here
+				return xerrors.Errorf("error retrieving tracked block %s from hotstore: %w ", cid, err)
-				panic(err)
 			}
 
 			err = s.cold.Put(blk)
 			if err != nil {
-				// TODO do something better here
+				return xerrors.Errorf("error puting block %s to coldstore: %w", cid, err)
-				panic(err)
 			}
 
 			if mark {
@@ -460,17 +444,22 @@ func (s *SplitStore) compact() {
 		// delete the object from the hotstore
 		err = s.hot.DeleteBlock(cid)
 		if err != nil {
-			// TODO do something better here
+			return xerrors.Errorf("error deleting block %s from hotstore: %w", cid, err)
-			panic(err)
 		}
 
 		// remove the snoop tracking
 		err = s.snoop.Delete(cid)
+		if err != nil {
+			return xerrors.Errorf("error deleting cid %s from tracking store: %w", cid, err)
+		}
+
+		return nil
+	})
+
 	if err != nil {
 		// TODO do something better here
 		panic(err)
 	}
-	}
 
 	log.Infow("sweeping done", "took", time.Since(startSweep))
 	log.Infow("compaction stats", "hot", stHot, "cold", stCold, "dead", stDead)