cerc-io/lotus
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

segregate chain and state blockstores.

Browse Source 
This paves the way for better object lifetime management.

Concretely, it makes it possible to:
- have different stores backing chain and state data.
- having the same datastore library, but using different parameters.
- attach different caching layers/policies to each class of data, e.g.
  sizing caches differently.
- specifying different retention policies for chain and state data.

This separation is important because:
- access patterns/frequency of chain and state data are different.
- state is derivable from chain, so one could never expunge the chain
  store, and only retain state objects reachable from the last finality
  in the state store.
This commit is contained in:
Raúl Kripalani
2021-02-28 22:49:44 +00:00
parent 853de3daf7
commit 3795cc2bd2
45 changed files with 630 additions and 370 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+7 -4
View File
@@ -145,7 +145,7 @@ const (
HeadMetricsKey
SettlePaymentChannelsKey
RunPeerTaggerKey
SetupFallbackBlockstoreKey
SetupFallbackBlockstoresKey
SetApiEndpointKey
@@ -590,12 +590,15 @@ func Repo(r repo.Repo) Option {
Override(new(repo.LockedRepo), modules.LockedRepo(lr)), // module handles closing
Override(new(dtypes.MetadataDS), modules.Datastore),
Override(new(dtypes.ChainRawBlockstore), modules.ChainRawBlockstore),
Override(new(dtypes.ChainBlockstore), From(new(dtypes.ChainRawBlockstore))),
Override(new(dtypes.UniversalBlockstore), modules.UniversalBlockstore),
Override(new(dtypes.ChainBlockstore), modules.ChainBlockstore),
Override(new(dtypes.StateBlockstore), modules.StateBlockstore),
Override(new(dtypes.ExposedBlockstore), From(new(dtypes.UniversalBlockstore))),
If(os.Getenv("LOTUS_ENABLE_CHAINSTORE_FALLBACK") == "1",
Override(new(dtypes.ChainBlockstore), modules.FallbackChainBlockstore),
Override(SetupFallbackBlockstoreKey, modules.SetupFallbackBlockstore),
Override(new(dtypes.StateBlockstore), modules.FallbackStateBlockstore),
Override(SetupFallbackBlockstoresKey, modules.InitFallbackBlockstores),
),
Override(new(dtypes.ClientImportMgr), modules.ClientImportMgr),
node/impl/full/chain.go
+16 -5
View File
@@ -35,6 +35,7 @@ import (
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/chain/vm"
"github.com/filecoin-project/lotus/node/modules/dtypes"
)
var log = logging.Logger("fullnode")
@@ -57,6 +58,11 @@ type ChainModule struct {
fx.In
Chain *store.ChainStore
// ExposedBlockstore is the global monolith blockstore that is safe to
// expose externally. In the future, this will be segregated into two
// blockstores.
ExposedBlockstore dtypes.ExposedBlockstore
}
var _ ChainModuleAPI = (*ChainModule)(nil)
@@ -68,6 +74,11 @@ type ChainAPI struct {
ChainModuleAPI
Chain *store.ChainStore
// ExposedBlockstore is the global monolith blockstore that is safe to
// expose externally. In the future, this will be segregated into two
// blockstores.
ExposedBlockstore dtypes.ExposedBlockstore
}
func (m *ChainModule) ChainNotify(ctx context.Context) (<-chan []*api.HeadChange, error) {
@@ -212,7 +223,7 @@ func (m *ChainModule) ChainGetTipSetByHeight(ctx context.Context, h abi.ChainEpo
}
func (m *ChainModule) ChainReadObj(ctx context.Context, obj cid.Cid) ([]byte, error) {
blk, err := m.Chain.Blockstore().Get(obj)
blk, err := m.ExposedBlockstore.Get(obj)
if err != nil {
return nil, xerrors.Errorf("blockstore get: %w", err)
}
@@ -221,15 +232,15 @@ func (m *ChainModule) ChainReadObj(ctx context.Context, obj cid.Cid) ([]byte, er
}
func (a *ChainAPI) ChainDeleteObj(ctx context.Context, obj cid.Cid) error {
return a.Chain.Blockstore().DeleteBlock(obj)
return a.ExposedBlockstore.DeleteBlock(obj)
}
func (m *ChainModule) ChainHasObj(ctx context.Context, obj cid.Cid) (bool, error) {
return m.Chain.Blockstore().Has(obj)
return m.ExposedBlockstore.Has(obj)
}
func (a *ChainAPI) ChainStatObj(ctx context.Context, obj cid.Cid, base cid.Cid) (api.ObjStat, error) {
bs := a.Chain.Blockstore()
bs := a.ExposedBlockstore
bsvc := blockservice.New(bs, offline.Exchange(bs))
dag := merkledag.NewDAGService(bsvc)
@@ -514,7 +525,7 @@ func (a *ChainAPI) ChainGetNode(ctx context.Context, p string) (*api.IpldObject,
return nil, xerrors.Errorf("parsing path: %w", err)
}
bs := a.Chain.Blockstore()
bs := a.ExposedBlockstore
bsvc := blockservice.New(bs, offline.Exchange(bs))
dag := merkledag.NewDAGService(bsvc)
node/impl/full/state.go
+26 -26
View File
@@ -97,7 +97,7 @@ func (a *StateAPI) StateMinerSectors(ctx context.Context, addr address.Address,
return nil, xerrors.Errorf("failed to load miner actor: %w", err)
}
mas, err := miner.Load(a.StateManager.ChainStore().Store(ctx), act)
mas, err := miner.Load(a.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return nil, xerrors.Errorf("failed to load miner actor state: %w", err)
}
@@ -111,7 +111,7 @@ func (a *StateAPI) StateMinerActiveSectors(ctx context.Context, maddr address.Ad
return nil, xerrors.Errorf("failed to load miner actor: %w", err)
}
mas, err := miner.Load(a.StateManager.ChainStore().Store(ctx), act)
mas, err := miner.Load(a.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return nil, xerrors.Errorf("failed to load miner actor state: %w", err)
}
@@ -135,7 +135,7 @@ func (m *StateModule) StateMinerInfo(ctx context.Context, actor address.Address,
return miner.MinerInfo{}, xerrors.Errorf("failed to load miner actor: %w", err)
}
mas, err := miner.Load(m.StateManager.ChainStore().Store(ctx), act)
mas, err := miner.Load(m.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return miner.MinerInfo{}, xerrors.Errorf("failed to load miner actor state: %w", err)
}
@@ -153,7 +153,7 @@ func (a *StateAPI) StateMinerDeadlines(ctx context.Context, m address.Address, t
return nil, xerrors.Errorf("failed to load miner actor: %w", err)
}
mas, err := miner.Load(a.StateManager.ChainStore().Store(ctx), act)
mas, err := miner.Load(a.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return nil, xerrors.Errorf("failed to load miner actor state: %w", err)
}
@@ -192,7 +192,7 @@ func (a *StateAPI) StateMinerPartitions(ctx context.Context, m address.Address,
return nil, xerrors.Errorf("failed to load miner actor: %w", err)
}
mas, err := miner.Load(a.StateManager.ChainStore().Store(ctx), act)
mas, err := miner.Load(a.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return nil, xerrors.Errorf("failed to load miner actor state: %w", err)
}
@@ -253,7 +253,7 @@ func (m *StateModule) StateMinerProvingDeadline(ctx context.Context, addr addres
return nil, xerrors.Errorf("failed to load miner actor: %w", err)
}
mas, err := miner.Load(m.StateManager.ChainStore().Store(ctx), act)
mas, err := miner.Load(m.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return nil, xerrors.Errorf("failed to load miner actor state: %w", err)
}
@@ -272,7 +272,7 @@ func (a *StateAPI) StateMinerFaults(ctx context.Context, addr address.Address, t
return bitfield.BitField{}, xerrors.Errorf("failed to load miner actor: %w", err)
}
mas, err := miner.Load(a.StateManager.ChainStore().Store(ctx), act)
mas, err := miner.Load(a.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return bitfield.BitField{}, xerrors.Errorf("failed to load miner actor state: %w", err)
}
@@ -329,7 +329,7 @@ func (a *StateAPI) StateMinerRecoveries(ctx context.Context, addr address.Addres
return bitfield.BitField{}, xerrors.Errorf("failed to load miner actor: %w", err)
}
mas, err := miner.Load(a.StateManager.ChainStore().Store(ctx), act)
mas, err := miner.Load(a.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return bitfield.BitField{}, xerrors.Errorf("failed to load miner actor state: %w", err)
}
@@ -461,7 +461,7 @@ func (a *StateAPI) StateReadState(ctx context.Context, actor address.Address, ts
return nil, xerrors.Errorf("getting actor: %w", err)
}
blk, err := a.Chain.Blockstore().Get(act.Head)
blk, err := a.Chain.StateBlockstore().Get(act.Head)
if err != nil {
return nil, xerrors.Errorf("getting actor head: %w", err)
}
@@ -707,7 +707,7 @@ func (m *StateModule) StateMarketStorageDeal(ctx context.Context, dealId abi.Dea
}
func (a *StateAPI) StateChangedActors(ctx context.Context, old cid.Cid, new cid.Cid) (map[string]types.Actor, error) {
store := a.Chain.Store(ctx)
store := a.Chain.ActorStore(ctx)
oldTree, err := state.LoadStateTree(store, old)
if err != nil {
@@ -727,7 +727,7 @@ func (a *StateAPI) StateMinerSectorCount(ctx context.Context, addr address.Addre
if err != nil {
return api.MinerSectors{}, err
}
mas, err := miner.Load(a.Chain.Store(ctx), act)
mas, err := miner.Load(a.Chain.ActorStore(ctx), act)
if err != nil {
return api.MinerSectors{}, err
}
@@ -792,7 +792,7 @@ func (a *StateAPI) StateSectorExpiration(ctx context.Context, maddr address.Addr
if err != nil {
return nil, err
}
mas, err := miner.Load(a.StateManager.ChainStore().Store(ctx), act)
mas, err := miner.Load(a.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return nil, err
}
@@ -804,7 +804,7 @@ func (a *StateAPI) StateSectorPartition(ctx context.Context, maddr address.Addre
if err != nil {
return nil, err
}
mas, err := miner.Load(a.StateManager.ChainStore().Store(ctx), act)
mas, err := miner.Load(a.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return nil, err
}
@@ -890,7 +890,7 @@ func (m *StateModule) MsigGetAvailableBalance(ctx context.Context, addr address.
if err != nil {
return types.EmptyInt, xerrors.Errorf("failed to load multisig actor: %w", err)
}
msas, err := multisig.Load(m.Chain.Store(ctx), act)
msas, err := multisig.Load(m.Chain.ActorStore(ctx), act)
if err != nil {
return types.EmptyInt, xerrors.Errorf("failed to load multisig actor state: %w", err)
}
@@ -912,7 +912,7 @@ func (a *StateAPI) MsigGetVestingSchedule(ctx context.Context, addr address.Addr
return api.EmptyVesting, xerrors.Errorf("failed to load multisig actor: %w", err)
}
msas, err := multisig.Load(a.Chain.Store(ctx), act)
msas, err := multisig.Load(a.Chain.ActorStore(ctx), act)
if err != nil {
return api.EmptyVesting, xerrors.Errorf("failed to load multisig actor state: %w", err)
}
@@ -961,7 +961,7 @@ func (m *StateModule) MsigGetVested(ctx context.Context, addr address.Address, s
return types.EmptyInt, xerrors.Errorf("failed to load multisig actor at end epoch: %w", err)
}
msas, err := multisig.Load(m.Chain.Store(ctx), act)
msas, err := multisig.Load(m.Chain.ActorStore(ctx), act)
if err != nil {
return types.EmptyInt, xerrors.Errorf("failed to load multisig actor state: %w", err)
}
@@ -989,7 +989,7 @@ func (m *StateModule) MsigGetPending(ctx context.Context, addr address.Address,
if err != nil {
return nil, xerrors.Errorf("failed to load multisig actor: %w", err)
}
msas, err := multisig.Load(m.Chain.Store(ctx), act)
msas, err := multisig.Load(m.Chain.ActorStore(ctx), act)
if err != nil {
return nil, xerrors.Errorf("failed to load multisig actor state: %w", err)
}
@@ -1032,7 +1032,7 @@ func (a *StateAPI) StateMinerPreCommitDepositForPower(ctx context.Context, maddr
return types.EmptyInt, xerrors.Errorf("failed to get resolve size: %w", err)
}
store := a.Chain.Store(ctx)
store := a.Chain.ActorStore(ctx)
var sectorWeight abi.StoragePower
if act, err := state.GetActor(market.Address); err != nil {
@@ -1093,7 +1093,7 @@ func (a *StateAPI) StateMinerInitialPledgeCollateral(ctx context.Context, maddr
return types.EmptyInt, xerrors.Errorf("failed to get resolve size: %w", err)
}
store := a.Chain.Store(ctx)
store := a.Chain.ActorStore(ctx)
var sectorWeight abi.StoragePower
if act, err := state.GetActor(market.Address); err != nil {
@@ -1164,7 +1164,7 @@ func (a *StateAPI) StateMinerAvailableBalance(ctx context.Context, maddr address
return types.EmptyInt, xerrors.Errorf("failed to load miner actor: %w", err)
}
mas, err := miner.Load(a.StateManager.ChainStore().Store(ctx), act)
mas, err := miner.Load(a.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return types.EmptyInt, xerrors.Errorf("failed to load miner actor state: %w", err)
}
@@ -1193,7 +1193,7 @@ func (a *StateAPI) StateMinerSectorAllocated(ctx context.Context, maddr address.
return false, xerrors.Errorf("failed to load miner actor: %w", err)
}
mas, err := miner.Load(a.StateManager.ChainStore().Store(ctx), act)
mas, err := miner.Load(a.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return false, xerrors.Errorf("failed to load miner actor state: %w", err)
}
@@ -1216,7 +1216,7 @@ func (a *StateAPI) StateVerifierStatus(ctx context.Context, addr address.Address
return nil, err
}
vrs, err := verifreg.Load(a.StateManager.ChainStore().Store(ctx), act)
vrs, err := verifreg.Load(a.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return nil, xerrors.Errorf("failed to load verified registry state: %w", err)
}
@@ -1247,7 +1247,7 @@ func (m *StateModule) StateVerifiedClientStatus(ctx context.Context, addr addres
return nil, err
}
vrs, err := verifreg.Load(m.StateManager.ChainStore().Store(ctx), act)
vrs, err := verifreg.Load(m.StateManager.ChainStore().ActorStore(ctx), act)
if err != nil {
return nil, xerrors.Errorf("failed to load verified registry state: %w", err)
}
@@ -1269,7 +1269,7 @@ func (a *StateAPI) StateVerifiedRegistryRootKey(ctx context.Context, tsk types.T
return address.Undef, err
}
vst, err := verifreg.Load(a.StateManager.ChainStore().Store(ctx), vact)
vst, err := verifreg.Load(a.StateManager.ChainStore().ActorStore(ctx), vact)
if err != nil {
return address.Undef, err
}
@@ -1298,12 +1298,12 @@ func (m *StateModule) StateDealProviderCollateralBounds(ctx context.Context, siz
return api.DealCollateralBounds{}, xerrors.Errorf("failed to load reward actor: %w", err)
}
pst, err := power.Load(m.StateManager.ChainStore().Store(ctx), pact)
pst, err := power.Load(m.StateManager.ChainStore().ActorStore(ctx), pact)
if err != nil {
return api.DealCollateralBounds{}, xerrors.Errorf("failed to load power actor state: %w", err)
}
rst, err := reward.Load(m.StateManager.ChainStore().Store(ctx), ract)
rst, err := reward.Load(m.StateManager.ChainStore().ActorStore(ctx), ract)
if err != nil {
return api.DealCollateralBounds{}, xerrors.Errorf("failed to load reward actor state: %w", err)
}
node/modules/blockstore.go
+65
View File
@@ -0,0 +1,65 @@
package modules
import (
"context"
"io"
bstore "github.com/ipfs/go-ipfs-blockstore"
"go.uber.org/fx"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/blockstore"
"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.
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 {
return nil, err
}
if c, ok := bs.(io.Closer); ok {
lc.Append(fx.Hook{
OnStop: func(_ context.Context) error {
return c.Close()
},
})
}
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) {
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) {
return bs, nil
}
func FallbackChainBlockstore(cbs dtypes.ChainBlockstore) dtypes.ChainBlockstore {
return &blockstore.FallbackStore{Blockstore: cbs}
}
func FallbackStateBlockstore(sbs dtypes.StateBlockstore) dtypes.StateBlockstore {
return &blockstore.FallbackStore{Blockstore: sbs}
}
func InitFallbackBlockstores(cbs dtypes.ChainBlockstore, sbs dtypes.StateBlockstore, rem dtypes.ChainBitswap) error {
for _, bs := range []bstore.Blockstore{cbs, sbs} {
if fbs, ok := bs.(*blockstore.FallbackStore); ok {
fbs.SetFallback(rem.GetBlock)
continue
}
return xerrors.Errorf("expected a FallbackStore")
}
return nil
}
node/modules/chain.go
+10 -106
View File
@@ -1,25 +1,18 @@
package modules
import (
"bytes"
"context"
"os"
"time"
"github.com/ipfs/go-bitswap"
"github.com/ipfs/go-bitswap/network"
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-datastore"
"github.com/ipld/go-car"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/routing"
pubsub "github.com/libp2p/go-libp2p-pubsub"
"go.uber.org/fx"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
"github.com/filecoin-project/lotus/journal"
"github.com/filecoin-project/lotus/blockstore"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain"
@@ -29,14 +22,15 @@ import (
"github.com/filecoin-project/lotus/chain/messagepool"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/chain/vm"
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
"github.com/filecoin-project/lotus/journal"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
)
func ChainBitswap(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt routing.Routing, bs dtypes.ChainBlockstore) dtypes.ChainBitswap {
// ChainBitswap uses a blockstore that bypasses all caches.
func ChainBitswap(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt routing.Routing, bs dtypes.ExposedBlockstore) dtypes.ChainBitswap {
// prefix protocol for chain bitswap
// (so bitswap uses /chain/ipfs/bitswap/1.0.0 internally for chain sync stuff)
bitswapNetwork := network.NewFromIpfsHost(host, rt, network.Prefix("/chain"))
@@ -60,6 +54,10 @@ func ChainBitswap(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt r
return exch
}
func ChainBlockService(bs dtypes.ExposedBlockstore, rem dtypes.ChainBitswap) dtypes.ChainBlockService {
return blockservice.New(bs, rem)
}
func MessagePool(lc fx.Lifecycle, sm *stmgr.StateManager, ps *pubsub.PubSub, ds dtypes.MetadataDS, nn dtypes.NetworkName, j journal.Journal) (*messagepool.MessagePool, error) {
mpp := messagepool.NewProvider(sm, ps)
mp, err := messagepool.New(mpp, ds, nn, j)
@@ -74,43 +72,8 @@ func MessagePool(lc fx.Lifecycle, sm *stmgr.StateManager, ps *pubsub.PubSub, ds
return mp, nil
}
func ChainRawBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.ChainRawBlockstore, error) {
bs, err := r.Blockstore(helpers.LifecycleCtx(mctx, lc), repo.BlockstoreChain)
if err != nil {
return nil, err
}
// TODO potentially replace this cached blockstore by a CBOR cache.
cbs, err := blockstore.CachedBlockstore(helpers.LifecycleCtx(mctx, lc), bs, blockstore.DefaultCacheOpts())
if err != nil {
return nil, err
}
return cbs, nil
}
func ChainBlockService(bs dtypes.ChainRawBlockstore, rem dtypes.ChainBitswap) dtypes.ChainBlockService {
return blockservice.New(bs, rem)
}
func FallbackChainBlockstore(rbs dtypes.ChainRawBlockstore) dtypes.ChainBlockstore {
return &blockstore.FallbackStore{
Blockstore: rbs,
}
}
func SetupFallbackBlockstore(cbs dtypes.ChainBlockstore, rem dtypes.ChainBitswap) error {
fbs, ok := cbs.(*blockstore.FallbackStore)
if !ok {
return xerrors.Errorf("expected a FallbackStore")
}
fbs.SetFallback(rem.GetBlock)
return nil
}
func ChainStore(lc fx.Lifecycle, bs dtypes.ChainBlockstore, lbs dtypes.ChainRawBlockstore, ds dtypes.MetadataDS, syscalls vm.SyscallBuilder, j journal.Journal) *store.ChainStore {
chain := store.NewChainStore(bs, lbs, ds, syscalls, j)
func ChainStore(lc fx.Lifecycle, cbs dtypes.ChainBlockstore, sbs dtypes.StateBlockstore, ds dtypes.MetadataDS, 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)
@@ -125,65 +88,6 @@ func ChainStore(lc fx.Lifecycle, bs dtypes.ChainBlockstore, lbs dtypes.ChainRawB
return chain
}
func ErrorGenesis() Genesis {
return func() (header *types.BlockHeader, e error) {
return nil, xerrors.New("No genesis block provided, provide the file with 'lotus daemon --genesis=[genesis file]'")
}
}
func LoadGenesis(genBytes []byte) func(dtypes.ChainBlockstore) Genesis {
return func(bs dtypes.ChainBlockstore) Genesis {
return func() (header *types.BlockHeader, e error) {
c, err := car.LoadCar(bs, bytes.NewReader(genBytes))
if err != nil {
return nil, xerrors.Errorf("loading genesis car file failed: %w", err)
}
if len(c.Roots) != 1 {
return nil, xerrors.New("expected genesis file to have one root")
}
root, err := bs.Get(c.Roots[0])
if err != nil {
return nil, err
}
h, err := types.DecodeBlock(root.RawData())
if err != nil {
return nil, xerrors.Errorf("decoding block failed: %w", err)
}
return h, nil
}
}
}
func DoSetGenesis(_ dtypes.AfterGenesisSet) {}
func SetGenesis(cs *store.ChainStore, g Genesis) (dtypes.AfterGenesisSet, error) {
genFromRepo, err := cs.GetGenesis()
if err == nil {
if os.Getenv("LOTUS_SKIP_GENESIS_CHECK") != "_yes_" {
expectedGenesis, err := g()
if err != nil {
return dtypes.AfterGenesisSet{}, xerrors.Errorf("getting expected genesis failed: %w", err)
}
if genFromRepo.Cid() != expectedGenesis.Cid() {
return dtypes.AfterGenesisSet{}, xerrors.Errorf("genesis in the repo is not the one expected by this version of Lotus!")
}
}
return dtypes.AfterGenesisSet{}, nil // already set, noop
}
if err != datastore.ErrNotFound {
return dtypes.AfterGenesisSet{}, xerrors.Errorf("getting genesis block failed: %w", err)
}
genesis, err := g()
if err != nil {
return dtypes.AfterGenesisSet{}, xerrors.Errorf("genesis func failed: %w", err)
}
return dtypes.AfterGenesisSet{}, cs.SetGenesis(genesis)
}
func NetworkName(mctx helpers.MetricsCtx, lc fx.Lifecycle, cs *store.ChainStore, us stmgr.UpgradeSchedule, _ dtypes.AfterGenesisSet) (dtypes.NetworkName, error) {
if !build.Devnet {
return "testnetnet", nil
node/modules/client.go
+1 -1
View File
@@ -83,7 +83,7 @@ func ClientMultiDatastore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.Locke
ctx := helpers.LifecycleCtx(mctx, lc)
ds, err := r.Datastore(ctx, "/client")
if err != nil {
return nil, xerrors.Errorf("getting datastore out of reop: %w", err)
return nil, xerrors.Errorf("getting datastore out of repo: %w", err)
}
mds, err := multistore.NewMultiDstore(ds)
node/modules/dtypes/storage.go
+28 -6
View File
@@ -19,19 +19,41 @@ import (
"github.com/filecoin-project/lotus/node/repo/retrievalstoremgr"
)
// MetadataDS stores metadata
// dy default it's namespaced under /metadata in main repo datastore
// MetadataDS stores metadata. By default it's namespaced under /metadata in
// main repo datastore.
type MetadataDS datastore.Batching
type ChainRawBlockstore blockstore.Blockstore
type ChainBlockstore blockstore.Blockstore // optionally bitswap backed
type (
// UniversalBlockstore is the cold blockstore.
UniversalBlockstore 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
// patterns.
ChainBlockstore blockstore.Blockstore
// StateBlockstore is a blockstore to store state data (state tree). It is
// physically backed by the BareMonolithBlockstore, but it has a cache on
// top that is specially tuned for state data access patterns.
StateBlockstore blockstore.Blockstore
// ExposedBlockstore is a blockstore that interfaces directly with the
// network or with users, from which queries are served, and where incoming
// data is deposited. For security reasons, this store is disconnected from
// any internal caches. If blocks are added to this store in a way that
// could render caches dirty (e.g. a block is added when an existence cache
// holds a 'false' for that block), the process should signal so by calling
// blockstore.AllCaches.Dirty(cid).
ExposedBlockstore blockstore.Blockstore
)
type ChainBitswap exchange.Interface
type ChainBlockService bserv.BlockService
type ClientMultiDstore *multistore.MultiStore
type ClientImportMgr *importmgr.Mgr
type ClientBlockstore blockstore.Blockstore
type ClientBlockstore blockstore.BasicBlockstore
type ClientDealStore *statestore.StateStore
type ClientRequestValidator *requestvalidation.UnifiedRequestValidator
type ClientDatastore datastore.Batching
@@ -50,6 +72,6 @@ type ProviderRequestValidator *requestvalidation.UnifiedRequestValidator
type ProviderDataTransfer datatransfer.Manager
type StagingDAG format.DAGService
type StagingBlockstore blockstore.Blockstore
type StagingBlockstore blockstore.BasicBlockstore
type StagingGraphsync graphsync.GraphExchange
type StagingMultiDstore *multistore.MultiStore
node/modules/genesis.go
+73
View File
@@ -0,0 +1,73 @@
package modules
import (
"bytes"
"os"
"github.com/ipfs/go-datastore"
"github.com/ipld/go-car"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/node/modules/dtypes"
)
func ErrorGenesis() Genesis {
return func() (header *types.BlockHeader, e error) {
return nil, xerrors.New("No genesis block provided, provide the file with 'lotus daemon --genesis=[genesis file]'")
}
}
func LoadGenesis(genBytes []byte) func(dtypes.ChainBlockstore) Genesis {
return func(bs dtypes.ChainBlockstore) Genesis {
return func() (header *types.BlockHeader, e error) {
c, err := car.LoadCar(bs, bytes.NewReader(genBytes))
if err != nil {
return nil, xerrors.Errorf("loading genesis car file failed: %w", err)
}
if len(c.Roots) != 1 {
return nil, xerrors.New("expected genesis file to have one root")
}
root, err := bs.Get(c.Roots[0])
if err != nil {
return nil, err
}
h, err := types.DecodeBlock(root.RawData())
if err != nil {
return nil, xerrors.Errorf("decoding block failed: %w", err)
}
return h, nil
}
}
}
func DoSetGenesis(_ dtypes.AfterGenesisSet) {}
func SetGenesis(cs *store.ChainStore, g Genesis) (dtypes.AfterGenesisSet, error) {
genFromRepo, err := cs.GetGenesis()
if err == nil {
if os.Getenv("LOTUS_SKIP_GENESIS_CHECK") != "_yes_" {
expectedGenesis, err := g()
if err != nil {
return dtypes.AfterGenesisSet{}, xerrors.Errorf("getting expected genesis failed: %w", err)
}
if genFromRepo.Cid() != expectedGenesis.Cid() {
return dtypes.AfterGenesisSet{}, xerrors.Errorf("genesis in the repo is not the one expected by this version of Lotus!")
}
}
return dtypes.AfterGenesisSet{}, nil // already set, noop
}
if err != datastore.ErrNotFound {
return dtypes.AfterGenesisSet{}, xerrors.Errorf("getting genesis block failed: %w", err)
}
genesis, err := g()
if err != nil {
return dtypes.AfterGenesisSet{}, xerrors.Errorf("genesis func failed: %w", err)
}
return dtypes.AfterGenesisSet{}, cs.SetGenesis(genesis)
}
node/modules/graphsync.go
+2 -2
View File
@@ -14,8 +14,8 @@ import (
)
// Graphsync creates a graphsync instance from the given loader and storer
func Graphsync(parallelTransfers uint64) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, clientBs dtypes.ClientBlockstore, chainBs dtypes.ChainBlockstore, h host.Host) (dtypes.Graphsync, error) {
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, clientBs dtypes.ClientBlockstore, chainBs dtypes.ChainBlockstore, h host.Host) (dtypes.Graphsync, error) {
func Graphsync(parallelTransfers uint64) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, clientBs dtypes.ClientBlockstore, chainBs dtypes.ExposedBlockstore, h host.Host) (dtypes.Graphsync, error) {
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, clientBs dtypes.ClientBlockstore, chainBs dtypes.ExposedBlockstore, h host.Host) (dtypes.Graphsync, error) {
graphsyncNetwork := gsnet.NewFromLibp2pHost(h)
loader := storeutil.LoaderForBlockstore(clientBs)
storer := storeutil.StorerForBlockstore(clientBs)
node/modules/ipfsclient.go
+1 -1
View File
@@ -18,7 +18,7 @@ import (
func IpfsClientBlockstore(ipfsMaddr string, onlineMode bool) func(helpers.MetricsCtx, fx.Lifecycle, dtypes.ClientImportMgr) (dtypes.ClientBlockstore, error) {
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, localStore dtypes.ClientImportMgr) (dtypes.ClientBlockstore, error) {
var err error
var ipfsbs blockstore.Blockstore
var ipfsbs blockstore.BasicBlockstore
if ipfsMaddr != "" {
var ma multiaddr.Multiaddr
ma, err = multiaddr.NewMultiaddr(ipfsMaddr)
node/repo/blockstore_opts.go
-4
View File
@@ -5,10 +5,6 @@ import badgerbs "github.com/filecoin-project/lotus/blockstore/badger"
// BadgerBlockstoreOptions returns the badger options to apply for the provided
// domain.
func BadgerBlockstoreOptions(domain BlockstoreDomain, path string, readonly bool) (badgerbs.Options, error) {
if domain != BlockstoreChain {
return badgerbs.Options{}, ErrInvalidBlockstoreDomain
}
opts := badgerbs.DefaultOptions(path)
// Due to legacy usage of blockstore.Blockstore, over a datastore, all
node/repo/fsrepo.go
+26 -4
View File
@@ -13,7 +13,7 @@ import (
"sync"
"github.com/BurntSushi/toml"
"github.com/filecoin-project/lotus/blockstore"
"github.com/ipfs/go-datastore"
fslock "github.com/ipfs/go-fs-lock"
logging "github.com/ipfs/go-log/v2"
@@ -22,7 +22,7 @@ import (
"github.com/multiformats/go-multiaddr"
"golang.org/x/xerrors"
lblockstore "github.com/filecoin-project/lotus/blockstore"
"github.com/filecoin-project/lotus/blockstore"
badgerbs "github.com/filecoin-project/lotus/blockstore/badger"
"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
"github.com/filecoin-project/lotus/extern/sector-storage/stores"
@@ -264,11 +264,18 @@ type fsLockedRepo struct {
bs blockstore.Blockstore
bsErr error
bsOnce sync.Once
ssPath string
ssErr error
ssOnce sync.Once
storageLk sync.Mutex
configLk sync.Mutex
}
func (fsr *fsLockedRepo) Readonly() bool {
return fsr.readonly
}
func (fsr *fsLockedRepo) Path() string {
return fsr.path
}
@@ -301,7 +308,7 @@ func (fsr *fsLockedRepo) Close() error {
// Blockstore returns a blockstore for the provided data domain.
func (fsr *fsLockedRepo) Blockstore(ctx context.Context, domain BlockstoreDomain) (blockstore.Blockstore, error) {
if domain != BlockstoreChain {
if domain != UniversalBlockstore {
return nil, ErrInvalidBlockstoreDomain
}
@@ -325,12 +332,27 @@ func (fsr *fsLockedRepo) Blockstore(ctx context.Context, domain BlockstoreDomain
fsr.bsErr = err
return
}
fsr.bs = lblockstore.WrapIDStore(bs)
fsr.bs = blockstore.WrapIDStore(bs)
})
return fsr.bs, fsr.bsErr
}
func (fsr *fsLockedRepo) SplitstorePath() (string, error) {
fsr.ssOnce.Do(func() {
path := fsr.join(filepath.Join(fsDatastore, "splitstore"))
if err := os.MkdirAll(path, 0755); err != nil {
fsr.ssErr = err
return
}
fsr.ssPath = path
})
return fsr.ssPath, fsr.ssErr
}
// join joins path elements with fsr.path
func (fsr *fsLockedRepo) join(paths ...string) string {
return filepath.Join(append([]string{fsr.path}, paths...)...)
node/repo/importmgr/mgr.go
+1 -1
View File
@@ -16,7 +16,7 @@ type Mgr struct {
mds *multistore.MultiStore
ds datastore.Batching
Blockstore blockstore.Blockstore
Blockstore blockstore.BasicBlockstore
}
type Label string
node/repo/interface.go
+9 -3
View File
@@ -4,10 +4,10 @@ import (
"context"
"errors"
"github.com/filecoin-project/lotus/blockstore"
"github.com/ipfs/go-datastore"
"github.com/multiformats/go-multiaddr"
"github.com/filecoin-project/lotus/blockstore"
"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
"github.com/filecoin-project/lotus/extern/sector-storage/stores"
@@ -18,11 +18,11 @@ import (
type BlockstoreDomain string
const (
// BlockstoreChain represents the blockstore domain for chain data.
// UniversalBlockstore represents the blockstore domain for all data.
// Right now, this includes chain objects (tipsets, blocks, messages), as
// well as state. In the future, they may get segregated into different
// domains.
BlockstoreChain = BlockstoreDomain("chain")
UniversalBlockstore = BlockstoreDomain("universal")
)
var (
@@ -63,6 +63,9 @@ type LockedRepo interface {
// the lifecycle.
Blockstore(ctx context.Context, domain BlockstoreDomain) (blockstore.Blockstore, error)
// SplitstorePath returns the path for the SplitStore
SplitstorePath() (string, error)
// Returns config in this repo
Config() (interface{}, error)
SetConfig(func(interface{})) error
@@ -84,4 +87,7 @@ type LockedRepo interface {
// Path returns absolute path of the repo
Path() string
// Readonly returns true if the repo is readonly
Readonly() bool
}
node/repo/memrepo.go
+9 -1
View File
@@ -201,6 +201,10 @@ func (mem *MemRepo) Lock(t RepoType) (LockedRepo, error) {
}, nil
}
func (lmem *lockedMemRepo) Readonly() bool {
return false
}
func (lmem *lockedMemRepo) checkToken() error {
lmem.RLock()
defer lmem.RUnlock()
@@ -246,12 +250,16 @@ func (lmem *lockedMemRepo) Datastore(_ context.Context, ns string) (datastore.Ba
}
func (lmem *lockedMemRepo) Blockstore(ctx context.Context, domain BlockstoreDomain) (blockstore.Blockstore, error) {
if domain != BlockstoreChain {
if domain != UniversalBlockstore {
return nil, ErrInvalidBlockstoreDomain
}
return lmem.mem.blockstore, nil
}
func (lmem *lockedMemRepo) SplitstorePath() (string, error) {
return ioutil.TempDir("", "splitstore.*")
}
func (lmem *lockedMemRepo) ListDatastores(ns string) ([]int64, error) {
return nil, nil
}
node/repo/retrievalstoremgr/retrievalstoremgr.go
+2 -2
View File
@@ -73,13 +73,13 @@ func (mrs *multiStoreRetrievalStore) DAGService() ipldformat.DAGService {
// BlockstoreRetrievalStoreManager manages a single blockstore as if it were multiple stores
type BlockstoreRetrievalStoreManager struct {
bs blockstore.Blockstore
bs blockstore.BasicBlockstore
}
var _ RetrievalStoreManager = &BlockstoreRetrievalStoreManager{}
// NewBlockstoreRetrievalStoreManager returns a new blockstore based RetrievalStoreManager
func NewBlockstoreRetrievalStoreManager(bs blockstore.Blockstore) RetrievalStoreManager {
func NewBlockstoreRetrievalStoreManager(bs blockstore.BasicBlockstore) RetrievalStoreManager {
return &BlockstoreRetrievalStoreManager{
bs: bs,
}