segregate chain and state blockstores

blockstore/badger/blockstore.go

@@ -110,10 +110,7 @@ func Open(opts Options) (*Blockstore, error) {
 		return nil, fmt.Errorf("failed to open badger blockstore: %w", err)
 	}
 
-	bs := &Blockstore{
-		DB: db,
-	}
-
+	bs := &Blockstore{DB: db}
 	if p := opts.Prefix; p != "" {
 		bs.prefixing = true
 		bs.prefix = []byte(p)

blockstore/badger/blockstore_test.go

@@ -61,8 +61,8 @@ func TestStorageKey(t *testing.T) {
 	require.Equal(t, k3, k2)
 }
 
-func newBlockstore(optsSupplier func(path string) Options) func(tb testing.TB) (bs blockstore.Blockstore, path string) {
-	return func(tb testing.TB) (bs blockstore.Blockstore, path string) {
+func newBlockstore(optsSupplier func(path string) Options) func(tb testing.TB) (bs blockstore.BasicBlockstore, path string) {
+	return func(tb testing.TB) (bs blockstore.BasicBlockstore, path string) {
 		tb.Helper()
 
 		path, err := ioutil.TempDir("", "")
@@ -83,8 +83,8 @@ func newBlockstore(optsSupplier func(path string) Options) func(tb testing.TB) (
 	}
 }
 
-func openBlockstore(optsSupplier func(path string) Options) func(tb testing.TB, path string) (bs blockstore.Blockstore, err error) {
-	return func(tb testing.TB, path string) (bs blockstore.Blockstore, err error) {
+func openBlockstore(optsSupplier func(path string) Options) func(tb testing.TB, path string) (bs blockstore.BasicBlockstore, err error) {
+	return func(tb testing.TB, path string) (bs blockstore.BasicBlockstore, err error) {
 		tb.Helper()
 		return Open(optsSupplier(path))
 	}

blockstore/badger/blockstore_test_suite.go

@@ -8,18 +8,19 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/filecoin-project/lotus/blockstore"
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
 	u "github.com/ipfs/go-ipfs-util"
 
+	"github.com/filecoin-project/lotus/blockstore"
+
 	"github.com/stretchr/testify/require"
 )
 
 // TODO: move this to go-ipfs-blockstore.
 type Suite struct {
-	NewBlockstore  func(tb testing.TB) (bs blockstore.Blockstore, path string)
-	OpenBlockstore func(tb testing.TB, path string) (bs blockstore.Blockstore, err error)
+	NewBlockstore  func(tb testing.TB) (bs blockstore.BasicBlockstore, path string)
+	OpenBlockstore func(tb testing.TB, path string) (bs blockstore.BasicBlockstore, err error)
 }
 
 func (s *Suite) RunTests(t *testing.T, prefix string) {
@@ -290,7 +291,7 @@ func (s *Suite) TestDelete(t *testing.T) {
 
 }
 
-func insertBlocks(t *testing.T, bs blockstore.Blockstore, count int) []cid.Cid {
+func insertBlocks(t *testing.T, bs blockstore.BasicBlockstore, count int) []cid.Cid {
 	keys := make([]cid.Cid, count)
 	for i := 0; i < count; i++ {
 		block := blocks.NewBlock([]byte(fmt.Sprintf("some data %d", i)))

blockstore/fallback.go

@@ -11,6 +11,19 @@ import (
 	"github.com/ipfs/go-cid"
 )
 
+// UnwrapFallbackStore takes a blockstore, and returns the underlying blockstore
+// if it was a FallbackStore. Otherwise, it just returns the supplied store
+// unmodified.
+func UnwrapFallbackStore(bs Blockstore) (Blockstore, bool) {
+	if fbs, ok := bs.(*FallbackStore); ok {
+		return fbs.Blockstore, true
+	}
+	return bs, false
+}
+
+// FallbackStore is a read-through store that queries another (potentially
+// remote) source if the block is not found locally. If the block is found
+// during the fallback, it stores it in the local store.
 type FallbackStore struct {
 	Blockstore
 
@@ -30,7 +43,7 @@ func (fbs *FallbackStore) SetFallback(missFn func(context.Context, cid.Cid) (blo
 }
 
 func (fbs *FallbackStore) getFallback(c cid.Cid) (blocks.Block, error) {
-	log.Errorw("fallbackstore: Block not found locally, fetching from the network", "cid", c)
+	log.Warnf("fallbackstore: block not found locally, fetching from the network; cid: %s", c)
 	fbs.lk.RLock()
 	defer fbs.lk.RUnlock()
 

blockstore/metrics.go

@@ -0,0 +1,154 @@
+package blockstore
+
+import (
+	"time"
+
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+)
+
+//
+// Currently unused, but kept in repo in case we introduce one of the candidate
+// cache implementations (Freecache, Ristretto), both of which report these
+// metrics.
+//
+
+// CacheMetricsEmitInterval is the interval at which metrics are emitted onto
+// OpenCensus.
+var CacheMetricsEmitInterval = 5 * time.Second
+
+var (
+	CacheName, _ = tag.NewKey("cache_name")
+)
+
+// CacheMeasures groups all metrics emitted by the blockstore caches.
+var CacheMeasures = struct {
+	HitRatio       *stats.Float64Measure
+	Hits           *stats.Int64Measure
+	Misses         *stats.Int64Measure
+	Entries        *stats.Int64Measure
+	QueriesServed  *stats.Int64Measure
+	Adds           *stats.Int64Measure
+	Updates        *stats.Int64Measure
+	Evictions      *stats.Int64Measure
+	CostAdded      *stats.Int64Measure
+	CostEvicted    *stats.Int64Measure
+	SetsDropped    *stats.Int64Measure
+	SetsRejected   *stats.Int64Measure
+	QueriesDropped *stats.Int64Measure
+}{
+	HitRatio:       stats.Float64("blockstore/cache/hit_ratio", "Hit ratio of blockstore cache", stats.UnitDimensionless),
+	Hits:           stats.Int64("blockstore/cache/hits", "Total number of hits at blockstore cache", stats.UnitDimensionless),
+	Misses:         stats.Int64("blockstore/cache/misses", "Total number of misses at blockstore cache", stats.UnitDimensionless),
+	Entries:        stats.Int64("blockstore/cache/entry_count", "Total number of entries currently in the blockstore cache", stats.UnitDimensionless),
+	QueriesServed:  stats.Int64("blockstore/cache/queries_served", "Total number of queries served by the blockstore cache", stats.UnitDimensionless),
+	Adds:           stats.Int64("blockstore/cache/adds", "Total number of adds to blockstore cache", stats.UnitDimensionless),
+	Updates:        stats.Int64("blockstore/cache/updates", "Total number of updates in blockstore cache", stats.UnitDimensionless),
+	Evictions:      stats.Int64("blockstore/cache/evictions", "Total number of evictions from blockstore cache", stats.UnitDimensionless),
+	CostAdded:      stats.Int64("blockstore/cache/cost_added", "Total cost (byte size) of entries added into blockstore cache", stats.UnitBytes),
+	CostEvicted:    stats.Int64("blockstore/cache/cost_evicted", "Total cost (byte size) of entries evicted by blockstore cache", stats.UnitBytes),
+	SetsDropped:    stats.Int64("blockstore/cache/sets_dropped", "Total number of sets dropped by blockstore cache", stats.UnitDimensionless),
+	SetsRejected:   stats.Int64("blockstore/cache/sets_rejected", "Total number of sets rejected by blockstore cache", stats.UnitDimensionless),
+	QueriesDropped: stats.Int64("blockstore/cache/queries_dropped", "Total number of queries dropped by blockstore cache", stats.UnitDimensionless),
+}
+
+// CacheViews groups all cache-related default views.
+var CacheViews = struct {
+	HitRatio       *view.View
+	Hits           *view.View
+	Misses         *view.View
+	Entries        *view.View
+	QueriesServed  *view.View
+	Adds           *view.View
+	Updates        *view.View
+	Evictions      *view.View
+	CostAdded      *view.View
+	CostEvicted    *view.View
+	SetsDropped    *view.View
+	SetsRejected   *view.View
+	QueriesDropped *view.View
+}{
+	HitRatio: &view.View{
+		Measure:     CacheMeasures.HitRatio,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+	Hits: &view.View{
+		Measure:     CacheMeasures.Hits,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+	Misses: &view.View{
+		Measure:     CacheMeasures.Misses,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+	Entries: &view.View{
+		Measure:     CacheMeasures.Entries,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+	QueriesServed: &view.View{
+		Measure:     CacheMeasures.QueriesServed,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+	Adds: &view.View{
+		Measure:     CacheMeasures.Adds,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+	Updates: &view.View{
+		Measure:     CacheMeasures.Updates,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+	Evictions: &view.View{
+		Measure:     CacheMeasures.Evictions,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+	CostAdded: &view.View{
+		Measure:     CacheMeasures.CostAdded,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+	CostEvicted: &view.View{
+		Measure:     CacheMeasures.CostEvicted,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+	SetsDropped: &view.View{
+		Measure:     CacheMeasures.SetsDropped,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+	SetsRejected: &view.View{
+		Measure:     CacheMeasures.SetsRejected,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+	QueriesDropped: &view.View{
+		Measure:     CacheMeasures.QueriesDropped,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{CacheName},
+	},
+}
+
+// DefaultViews exports all default views for this package.
+var DefaultViews = []*view.View{
+	CacheViews.HitRatio,
+	CacheViews.Hits,
+	CacheViews.Misses,
+	CacheViews.Entries,
+	CacheViews.QueriesServed,
+	CacheViews.Adds,
+	CacheViews.Updates,
+	CacheViews.Evictions,
+	CacheViews.CostAdded,
+	CacheViews.CostEvicted,
+	CacheViews.SetsDropped,
+	CacheViews.SetsRejected,
+	CacheViews.QueriesDropped,
+}

chain/gen/gen.go

@@ -125,7 +125,7 @@ func NewGeneratorWithSectors(numSectors int) (*ChainGen, error) {
 		return nil, xerrors.Errorf("failed to get metadata datastore: %w", err)
 	}
 
-	bs, err := lr.Blockstore(context.TODO(), repo.BlockstoreChain)
+	bs, err := lr.Blockstore(context.TODO(), repo.UniversalBlockstore)
 	if err != nil {
 		return nil, err
 	}

chain/gen/genesis/genesis.go

@@ -406,7 +406,7 @@ func VerifyPreSealedData(ctx context.Context, cs *store.ChainStore, stateroot ci
 		StateBase:      stateroot,
 		Epoch:          0,
 		Rand:           &fakeRand{},
-		Bstore:         cs.Blockstore(),
+		Bstore:         cs.StateBlockstore(),
 		Syscalls:       mkFakedSigSyscalls(cs.VMSys()),
 		CircSupplyCalc: nil,
 		NtwkVersion:    genesisNetworkVersion,

chain/gen/genesis/miners.go

@@ -70,7 +70,7 @@ func SetupStorageMiners(ctx context.Context, cs *store.ChainStore, sroot cid.Cid
 		StateBase:      sroot,
 		Epoch:          0,
 		Rand:           &fakeRand{},
-		Bstore:         cs.Blockstore(),
+		Bstore:         cs.StateBlockstore(),
 		Syscalls:       mkFakedSigSyscalls(cs.VMSys()),
 		CircSupplyCalc: csc,
 		NtwkVersion:    genesisNetworkVersion,

chain/gen/mining.go

@@ -79,7 +79,7 @@ func MinerCreateBlock(ctx context.Context, sm *stmgr.StateManager, w api.WalletA
 		}
 	}
 
-	store := sm.ChainStore().Store(ctx)
+	store := sm.ChainStore().ActorStore(ctx)
 	blsmsgroot, err := toArray(store, blsMsgCids)
 	if err != nil {
 		return nil, xerrors.Errorf("building bls amt: %w", err)

chain/stmgr/call.go

@@ -59,7 +59,7 @@ func (sm *StateManager) Call(ctx context.Context, msg *types.Message, ts *types.
 		StateBase:      bstate,
 		Epoch:          bheight,
 		Rand:           store.NewChainRand(sm.cs, ts.Cids()),
-		Bstore:         sm.cs.Blockstore(),
+		Bstore:         sm.cs.StateBlockstore(),
 		Syscalls:       sm.cs.VMSys(),
 		CircSupplyCalc: sm.GetVMCirculatingSupply,
 		NtwkVersion:    sm.GetNtwkVersion,
@@ -174,7 +174,7 @@ func (sm *StateManager) CallWithGas(ctx context.Context, msg *types.Message, pri
 		StateBase:      state,
 		Epoch:          ts.Height() + 1,
 		Rand:           r,
-		Bstore:         sm.cs.Blockstore(),
+		Bstore:         sm.cs.StateBlockstore(),
 		Syscalls:       sm.cs.VMSys(),
 		CircSupplyCalc: sm.GetVMCirculatingSupply,
 		NtwkVersion:    sm.GetNtwkVersion,