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

Merge pull request #5695 from filecoin-project/feat/segregate-blockstores

Browse Source 
segregate chain and state blockstores
This commit is contained in:
Łukasz Magiera
2021-03-03 10:53:46 +01:00
committed by GitHub
parent 61005e61e9 1ac0c9a926
commit 448813d2fe
48 changed files with 850 additions and 374 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,
}