Merge pull request #4992 from filecoin-project/feat/splitstore

hot/cold blockstore segregation (aka. splitstore)

node/modules/blockstore.go

@@ -3,19 +3,25 @@ package modules
 import (
 	"context"
 	"io"
+	"os"
+	"path/filepath"
 
 	bstore "github.com/ipfs/go-ipfs-blockstore"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/lotus/blockstore"
+	badgerbs "github.com/filecoin-project/lotus/blockstore/badger"
+	"github.com/filecoin-project/lotus/blockstore/splitstore"
+	"github.com/filecoin-project/lotus/node/config"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	"github.com/filecoin-project/lotus/node/modules/helpers"
 	"github.com/filecoin-project/lotus/node/repo"
 )
 
 // UniversalBlockstore returns a single universal blockstore that stores both
-// chain data and state data.
+// chain data and state data. It can be backed by a blockstore directly
+// (e.g. Badger), or by a Splitstore.
 func UniversalBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.UniversalBlockstore, error) {
 	bs, err := r.Blockstore(helpers.LifecycleCtx(mctx, lc), repo.UniversalBlockstore)
 	if err != nil {
@@ -31,17 +37,76 @@ func UniversalBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.Locked
 	return bs, err
 }
 
-// StateBlockstore is a hook to overlay caches for state objects, or in the
-// future, to segregate the universal blockstore into different physical state
-// and chain stores.
-func StateBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, bs dtypes.UniversalBlockstore) (dtypes.StateBlockstore, error) {
+func BadgerHotBlockstore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.HotBlockstore, error) {
+	path, err := r.SplitstorePath()
+	if err != nil {
+		return nil, err
+	}
+
+	path = filepath.Join(path, "hot.badger")
+	if err := os.MkdirAll(path, 0755); err != nil {
+		return nil, err
+	}
+
+	opts, err := repo.BadgerBlockstoreOptions(repo.HotBlockstore, path, r.Readonly())
+	if err != nil {
+		return nil, err
+	}
+
+	bs, err := badgerbs.Open(opts)
+	if err != nil {
+		return nil, err
+	}
+
+	lc.Append(fx.Hook{
+		OnStop: func(_ context.Context) error {
+			return bs.Close()
+		}})
+
 	return bs, nil
 }
 
-// ChainBlockstore is a hook to overlay caches for state objects, or in the
-// future, to segregate the universal blockstore into different physical state
-// and chain stores.
-func ChainBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, bs dtypes.UniversalBlockstore) (dtypes.ChainBlockstore, error) {
+func SplitBlockstore(cfg *config.Chainstore) func(lc fx.Lifecycle, r repo.LockedRepo, ds dtypes.MetadataDS, cold dtypes.UniversalBlockstore, hot dtypes.HotBlockstore) (dtypes.SplitBlockstore, error) {
+	return func(lc fx.Lifecycle, r repo.LockedRepo, ds dtypes.MetadataDS, cold dtypes.UniversalBlockstore, hot dtypes.HotBlockstore) (dtypes.SplitBlockstore, error) {
+		path, err := r.SplitstorePath()
+		if err != nil {
+			return nil, err
+		}
+
+		cfg := &splitstore.Config{
+			TrackingStoreType:    cfg.Splitstore.TrackingStoreType,
+			MarkSetType:          cfg.Splitstore.MarkSetType,
+			EnableFullCompaction: cfg.Splitstore.EnableFullCompaction,
+			EnableGC:             cfg.Splitstore.EnableGC,
+			Archival:             cfg.Splitstore.Archival,
+		}
+		ss, err := splitstore.Open(path, ds, hot, cold, cfg)
+		if err != nil {
+			return nil, err
+		}
+		lc.Append(fx.Hook{
+			OnStop: func(context.Context) error {
+				return ss.Close()
+			},
+		})
+
+		return ss, err
+	}
+}
+
+func StateFlatBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.UniversalBlockstore) (dtypes.StateBlockstore, error) {
+	return bs, nil
+}
+
+func StateSplitBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.SplitBlockstore) (dtypes.StateBlockstore, error) {
+	return bs, nil
+}
+
+func ChainFlatBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.UniversalBlockstore) (dtypes.ChainBlockstore, error) {
+	return bs, nil
+}
+
+func ChainSplitBlockstore(_ fx.Lifecycle, _ helpers.MetricsCtx, bs dtypes.SplitBlockstore) (dtypes.ChainBlockstore, error) {
 	return bs, nil
 }
 

node/modules/chain.go

@@ -14,6 +14,7 @@ import (
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/lotus/blockstore"
+	"github.com/filecoin-project/lotus/blockstore/splitstore"
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain"
 	"github.com/filecoin-project/lotus/chain/beacon"
@@ -72,14 +73,26 @@ func MessagePool(lc fx.Lifecycle, sm *stmgr.StateManager, ps *pubsub.PubSub, ds
 	return mp, nil
 }
 
-func ChainStore(lc fx.Lifecycle, cbs dtypes.ChainBlockstore, sbs dtypes.StateBlockstore, ds dtypes.MetadataDS, syscalls vm.SyscallBuilder, j journal.Journal) *store.ChainStore {
+func ChainStore(lc fx.Lifecycle, cbs dtypes.ChainBlockstore, sbs dtypes.StateBlockstore, ds dtypes.MetadataDS, basebs dtypes.BaseBlockstore, syscalls vm.SyscallBuilder, j journal.Journal) *store.ChainStore {
 	chain := store.NewChainStore(cbs, sbs, ds, syscalls, j)
 
 	if err := chain.Load(); err != nil {
 		log.Warnf("loading chain state from disk: %s", err)
 	}
 
+	var startHook func(context.Context) error
+	if ss, ok := basebs.(*splitstore.SplitStore); ok {
+		startHook = func(_ context.Context) error {
+			err := ss.Start(chain)
+			if err != nil {
+				err = xerrors.Errorf("error starting splitstore: %w", err)
+			}
+			return err
+		}
+	}
+
 	lc.Append(fx.Hook{
+		OnStart: startHook,
 		OnStop: func(_ context.Context) error {
 			return chain.Close()
 		},

node/modules/dtypes/storage.go

@@ -27,6 +27,15 @@ type (
 	// UniversalBlockstore is the cold blockstore.
 	UniversalBlockstore blockstore.Blockstore
 
+	// HotBlockstore is the Hot blockstore abstraction for the splitstore
+	HotBlockstore blockstore.Blockstore
+
+	// SplitBlockstore is the hot/cold blockstore that sits on top of the ColdBlockstore.
+	SplitBlockstore blockstore.Blockstore
+
+	// BaseBlockstore is something, coz DI
+	BaseBlockstore blockstore.Blockstore
+
 	// ChainBlockstore is a blockstore to store chain data (tipsets, blocks,
 	// messages). It is physically backed by the BareMonolithBlockstore, but it
 	// has a cache on top that is specially tuned for chain data access