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

Merge remote-tracking branch 'origin/master' into feat/datamodel-selector-retrieval

Browse Source
This commit is contained in:
Peter Rabbitson
2021-10-05 18:15:51 +02:00
parent af0d9b650d cbb147de23
commit 5e6aceeb0e
120 changed files with 3978 additions and 645 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder_chain.go
+2 -2
View File
@@ -100,7 +100,7 @@ var ChainNode = Options(
Override(new(*dtypes.MpoolLocker), new(dtypes.MpoolLocker)),
// Shared graphsync (markets, serving chain)
Override(new(dtypes.Graphsync), modules.Graphsync(config.DefaultFullNode().Client.SimultaneousTransfers)),
Override(new(dtypes.Graphsync), modules.Graphsync(config.DefaultFullNode().Client.SimultaneousTransfersForStorage, config.DefaultFullNode().Client.SimultaneousTransfersForRetrieval)),
// Service: Wallet
Override(new(*messagesigner.MessageSigner), messagesigner.NewMessageSigner),
@@ -219,7 +219,7 @@ func ConfigFullNode(c interface{}) Option {
Override(new(retrievalmarket.BlockstoreAccessor), modules.IpfsRetrievalBlockstoreAccessor),
),
),
Override(new(dtypes.Graphsync), modules.Graphsync(cfg.Client.SimultaneousTransfers)),
Override(new(dtypes.Graphsync), modules.Graphsync(cfg.Client.SimultaneousTransfersForStorage, cfg.Client.SimultaneousTransfersForRetrieval)),
If(cfg.Wallet.RemoteBackend != "",
Override(new(*remotewallet.RemoteWallet), remotewallet.SetupRemoteWallet(cfg.Wallet.RemoteBackend)),
node/builder_miner.go
+1 -1
View File
@@ -136,7 +136,7 @@ func ConfigStorageMiner(c interface{}) Option {
If(cfg.Subsystems.EnableMarkets,
// Markets
Override(new(dtypes.StagingBlockstore), modules.StagingBlockstore),
Override(new(dtypes.StagingGraphsync), modules.StagingGraphsync(cfg.Dealmaking.SimultaneousTransfers)),
Override(new(dtypes.StagingGraphsync), modules.StagingGraphsync(cfg.Dealmaking.SimultaneousTransfersForStorage, cfg.Dealmaking.SimultaneousTransfersForRetrieval)),
Override(new(dtypes.ProviderPieceStore), modules.NewProviderPieceStore),
Override(new(*sectorblocks.SectorBlocks), sectorblocks.NewSectorBlocks),
node/config/def.go
+18 -3
View File
@@ -2,6 +2,8 @@ package config
import (
"encoding"
"os"
"strconv"
"time"
"github.com/ipfs/go-cid"
@@ -24,6 +26,16 @@ const (
RetrievalPricingExternalMode = "external"
)
// MaxTraversalLinks configures the maximum number of links to traverse in a DAG while calculating
// CommP and traversing a DAG with graphsync; invokes a budget on DAG depth and density.
var MaxTraversalLinks uint64 = 32 * (1 << 20)
func init() {
if envMaxTraversal, err := strconv.ParseUint(os.Getenv("LOTUS_MAX_TRAVERSAL_LINKS"), 10, 64); err == nil {
MaxTraversalLinks = envMaxTraversal
}
}
func (b *BatchFeeConfig) FeeForSectors(nSectors int) abi.TokenAmount {
return big.Add(big.Int(b.Base), big.Mul(big.NewInt(int64(nSectors)), big.Int(b.PerSector)))
}
@@ -65,7 +77,8 @@ func DefaultFullNode() *FullNode {
DefaultMaxFee: DefaultDefaultMaxFee,
},
Client: Client{
SimultaneousTransfers: DefaultSimultaneousTransfers,
SimultaneousTransfersForStorage: DefaultSimultaneousTransfers,
SimultaneousTransfersForRetrieval: DefaultSimultaneousTransfers,
},
Chainstore: Chainstore{
EnableSplitstore: false,
@@ -109,7 +122,8 @@ func DefaultStorageMiner() *StorageMiner {
CommitBatchWait: Duration(24 * time.Hour), // this can be up to 30 days
CommitBatchSlack: Duration(1 * time.Hour), // time buffer for forceful batch submission before sectors/deals in batch would start expiring, higher value will lower the chances for message fail due to expiration
AggregateAboveBaseFee: types.FIL(types.BigMul(types.PicoFil, types.NewInt(150))), // 0.15 nFIL
BatchPreCommitAboveBaseFee: types.FIL(types.BigMul(types.PicoFil, types.NewInt(320))), // 0.32 nFIL
AggregateAboveBaseFee: types.FIL(types.BigMul(types.PicoFil, types.NewInt(320))), // 0.32 nFIL
TerminateBatchMin: 1,
TerminateBatchMax: 100,
@@ -146,7 +160,8 @@ func DefaultStorageMiner() *StorageMiner {
MaxDealsPerPublishMsg: 8,
MaxProviderCollateralMultiplier: 2,
SimultaneousTransfers: DefaultSimultaneousTransfers,
SimultaneousTransfersForStorage: DefaultSimultaneousTransfers,
SimultaneousTransfersForRetrieval: DefaultSimultaneousTransfers,
StartEpochSealingBuffer: 480, // 480 epochs buffer == 4 hours from adding deal to sector to sector being sealed
node/config/doc_gen.go
+24 -4
View File
@@ -92,11 +92,18 @@ your node if metadata log is disabled`,
Comment: ``,
},
{
Name: "SimultaneousTransfers",
Name: "SimultaneousTransfersForStorage",
Type: "uint64",
Comment: `The maximum number of simultaneous data transfers between the client
and storage providers`,
and storage providers for storage deals`,
},
{
Name: "SimultaneousTransfersForRetrieval",
Type: "uint64",
Comment: `The maximum number of simultaneous data transfers between the client
and storage providers for retrieval deals`,
},
},
"Common": []DocField{
@@ -260,10 +267,16 @@ as a multiplier of the minimum collateral bound`,
passed to the sealing node by the markets service. 0 is unlimited.`,
},
{
Name: "SimultaneousTransfers",
Name: "SimultaneousTransfersForStorage",
Type: "uint64",
Comment: `The maximum number of parallel online data transfers (storage+retrieval)`,
Comment: `The maximum number of parallel online data transfers for storage deals`,
},
{
Name: "SimultaneousTransfersForRetrieval",
Type: "uint64",
Comment: `The maximum number of parallel online data transfers for retrieval deals`,
},
{
Name: "StartEpochSealingBuffer",
@@ -717,6 +730,13 @@ avoid the relatively high cost of unsealing the data later, at the cost of more
Comment: `time buffer for forceful batch submission before sectors/deals in batch would start expiring`,
},
{
Name: "BatchPreCommitAboveBaseFee",
Type: "types.FIL",
Comment: `network BaseFee below which to stop doing precommit batching, instead
sending precommit messages to the chain individually`,
},
{
Name: "AggregateAboveBaseFee",
Type: "types.FIL",
node/config/types.go
+13 -4
View File
@@ -129,8 +129,10 @@ type DealmakingConfig struct {
// The maximum allowed disk usage size in bytes of staging deals not yet
// passed to the sealing node by the markets service. 0 is unlimited.
MaxStagingDealsBytes int64
// The maximum number of parallel online data transfers (storage+retrieval)
SimultaneousTransfers uint64
// The maximum number of parallel online data transfers for storage deals
SimultaneousTransfersForStorage uint64
// The maximum number of parallel online data transfers for retrieval deals
SimultaneousTransfersForRetrieval uint64
// Minimum start epoch buffer to give time for sealing of sector with deal.
StartEpochSealingBuffer uint64
@@ -221,6 +223,10 @@ type SealingConfig struct {
// time buffer for forceful batch submission before sectors/deals in batch would start expiring
CommitBatchSlack Duration
// network BaseFee below which to stop doing precommit batching, instead
// sending precommit messages to the chain individually
BatchPreCommitAboveBaseFee types.FIL
// network BaseFee below which to stop doing commit aggregation, instead
// submitting proofs to the chain individually
AggregateAboveBaseFee types.FIL
@@ -360,8 +366,11 @@ type Client struct {
IpfsMAddr string
IpfsUseForRetrieval bool
// The maximum number of simultaneous data transfers between the client
// and storage providers
SimultaneousTransfers uint64
// and storage providers for storage deals
SimultaneousTransfersForStorage uint64
// The maximum number of simultaneous data transfers between the client
// and storage providers for retrieval deals
SimultaneousTransfersForRetrieval uint64
}
type Wallet struct {
node/impl/client/client.go
+7 -1
View File
@@ -59,6 +59,7 @@ import (
"github.com/filecoin-project/specs-actors/v3/actors/builtin/market"
marketevents "github.com/filecoin-project/lotus/markets/loggers"
"github.com/filecoin-project/lotus/node/config"
"github.com/filecoin-project/lotus/node/repo/imports"
"github.com/filecoin-project/lotus/api"
@@ -1011,6 +1012,7 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
Root: order.Root,
Selector: sel,
}},
car.MaxTraversalLinks(config.MaxTraversalLinks),
).Write(f)
if err != nil {
finish(err)
@@ -1302,7 +1304,11 @@ func (a *API) ClientGenCar(ctx context.Context, ref api.FileRef, outputPath stri
allSelector := ssb.ExploreRecursive(
selector.RecursionLimitNone(),
ssb.ExploreAll(ssb.ExploreRecursiveEdge())).Node()
sc := car.NewSelectiveCar(ctx, fs, []car.Dag{{Root: root, Selector: allSelector}})
sc := car.NewSelectiveCar(ctx,
fs,
[]car.Dag{{Root: root, Selector: allSelector}},
car.MaxTraversalLinks(config.MaxTraversalLinks),
)
f, err := os.Create(outputPath)
if err != nil {
return err
node/impl/full/state.go
+2 -24
View File
@@ -1422,32 +1422,10 @@ func (m *StateModule) StateNetworkVersion(ctx context.Context, tsk types.TipSetK
}
func (a *StateAPI) StateGetRandomnessFromTickets(ctx context.Context, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte, tsk types.TipSetKey) (abi.Randomness, error) {
pts, err := a.Chain.LoadTipSet(tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset key: %w", err)
}
rnv := a.StateManager.GetNtwkVersion(ctx, randEpoch)
if rnv >= network.Version13 {
return a.Chain.GetChainRandomnessLookingForward(ctx, pts.Cids(), personalization, randEpoch, entropy)
}
return a.Chain.GetChainRandomnessLookingBack(ctx, pts.Cids(), personalization, randEpoch, entropy)
return a.StateManager.GetRandomnessFromTickets(ctx, personalization, randEpoch, entropy, tsk)
}
func (a *StateAPI) StateGetRandomnessFromBeacon(ctx context.Context, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte, tsk types.TipSetKey) (abi.Randomness, error) {
pts, err := a.Chain.GetTipSetFromKey(tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
rnv := a.StateManager.GetNtwkVersion(ctx, randEpoch)
if rnv >= network.Version13 {
return a.Chain.GetBeaconRandomnessLookingForward(ctx, pts.Cids(), personalization, randEpoch, entropy)
}
return a.Chain.GetBeaconRandomnessLookingBack(ctx, pts.Cids(), personalization, randEpoch, entropy)
return a.StateManager.GetRandomnessFromBeacon(ctx, personalization, randEpoch, entropy, tsk)
}
node/modules/chain.go
+1 -1
View File
@@ -120,7 +120,7 @@ func NetworkName(mctx helpers.MetricsCtx,
ctx := helpers.LifecycleCtx(mctx, lc)
sm, err := stmgr.NewStateManager(cs, tsexec, syscalls, us)
sm, err := stmgr.NewStateManager(cs, tsexec, syscalls, us, nil)
if err != nil {
return "", err
}
node/modules/client.go
+2 -1
View File
@@ -36,6 +36,7 @@ import (
"github.com/filecoin-project/lotus/markets"
marketevents "github.com/filecoin-project/lotus/markets/loggers"
"github.com/filecoin-project/lotus/markets/retrievaladapter"
"github.com/filecoin-project/lotus/node/config"
"github.com/filecoin-project/lotus/node/impl/full"
payapi "github.com/filecoin-project/lotus/node/impl/paych"
"github.com/filecoin-project/lotus/node/modules/dtypes"
@@ -182,7 +183,7 @@ func StorageClient(lc fx.Lifecycle, h host.Host, dataTransfer dtypes.ClientDataT
marketsRetryParams := smnet.RetryParameters(time.Second, 5*time.Minute, 15, 5)
net := smnet.NewFromLibp2pHost(h, marketsRetryParams)
c, err := storageimpl.NewClient(net, dataTransfer, discovery, deals, scn, accessor, storageimpl.DealPollingInterval(time.Second))
c, err := storageimpl.NewClient(net, dataTransfer, discovery, deals, scn, accessor, storageimpl.DealPollingInterval(time.Second), storageimpl.MaxTraversalLinks(config.MaxTraversalLinks))
if err != nil {
return nil, err
}
node/modules/graphsync.go
+10 -2
View File
@@ -9,18 +9,26 @@ import (
"github.com/libp2p/go-libp2p-core/peer"
"go.uber.org/fx"
"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"
)
// 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.ExposedBlockstore, h host.Host) (dtypes.Graphsync, error) {
func Graphsync(parallelTransfersForStorage uint64, parallelTransfersForRetrieval 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)
lsys := storeutil.LinkSystemForBlockstore(clientBs)
gs := graphsyncimpl.New(helpers.LifecycleCtx(mctx, lc), graphsyncNetwork, lsys, graphsyncimpl.RejectAllRequestsByDefault(), graphsyncimpl.MaxInProgressIncomingRequests(parallelTransfers))
gs := graphsyncimpl.New(helpers.LifecycleCtx(mctx, lc),
graphsyncNetwork,
lsys,
graphsyncimpl.RejectAllRequestsByDefault(),
graphsyncimpl.MaxInProgressIncomingRequests(parallelTransfersForStorage),
graphsyncimpl.MaxInProgressOutgoingRequests(parallelTransfersForRetrieval),
graphsyncimpl.MaxLinksPerIncomingRequests(config.MaxTraversalLinks),
graphsyncimpl.MaxLinksPerOutgoingRequests(config.MaxTraversalLinks))
chainLinkSystem := storeutil.LinkSystemForBlockstore(chainBs)
err := gs.RegisterPersistenceOption("chainstore", chainLinkSystem)
if err != nil {
node/modules/stmgr.go
+3 -2
View File
@@ -1,6 +1,7 @@
package modules
import (
"github.com/filecoin-project/lotus/chain/beacon"
"github.com/filecoin-project/lotus/chain/vm"
"go.uber.org/fx"
@@ -8,8 +9,8 @@ import (
"github.com/filecoin-project/lotus/chain/store"
)
func StateManager(lc fx.Lifecycle, cs *store.ChainStore, exec stmgr.Executor, sys vm.SyscallBuilder, us stmgr.UpgradeSchedule) (*stmgr.StateManager, error) {
sm, err := stmgr.NewStateManager(cs, exec, sys, us)
func StateManager(lc fx.Lifecycle, cs *store.ChainStore, exec stmgr.Executor, sys vm.SyscallBuilder, us stmgr.UpgradeSchedule, b beacon.Schedule) (*stmgr.StateManager, error) {
sm, err := stmgr.NewStateManager(cs, exec, sys, us, b)
if err != nil {
return nil, err
}
node/modules/storageminer.go
+24 -14
View File
@@ -38,6 +38,7 @@ import (
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
graphsync "github.com/ipfs/go-graphsync/impl"
graphsyncimpl "github.com/ipfs/go-graphsync/impl"
gsnet "github.com/ipfs/go-graphsync/network"
"github.com/ipfs/go-graphsync/storeutil"
"github.com/libp2p/go-libp2p-core/host"
@@ -394,11 +395,18 @@ func StagingBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRe
// StagingGraphsync creates a graphsync instance which reads and writes blocks
// to the StagingBlockstore
func StagingGraphsync(parallelTransfers uint64) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.StagingBlockstore, h host.Host) dtypes.StagingGraphsync {
func StagingGraphsync(parallelTransfersForStorage uint64, parallelTransfersForRetrieval uint64) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.StagingBlockstore, h host.Host) dtypes.StagingGraphsync {
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.StagingBlockstore, h host.Host) dtypes.StagingGraphsync {
graphsyncNetwork := gsnet.NewFromLibp2pHost(h)
lsys := storeutil.LinkSystemForBlockstore(ibs)
gs := graphsync.New(helpers.LifecycleCtx(mctx, lc), graphsyncNetwork, lsys, graphsync.RejectAllRequestsByDefault(), graphsync.MaxInProgressIncomingRequests(parallelTransfers))
gs := graphsync.New(helpers.LifecycleCtx(mctx, lc),
graphsyncNetwork,
lsys,
graphsync.RejectAllRequestsByDefault(),
graphsync.MaxInProgressIncomingRequests(parallelTransfersForRetrieval),
graphsync.MaxInProgressOutgoingRequests(parallelTransfersForStorage),
graphsyncimpl.MaxLinksPerIncomingRequests(config.MaxTraversalLinks),
graphsyncimpl.MaxLinksPerOutgoingRequests(config.MaxTraversalLinks))
return gs
}
@@ -887,12 +895,13 @@ func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error
PreCommitBatchWait: config.Duration(cfg.PreCommitBatchWait),
PreCommitBatchSlack: config.Duration(cfg.PreCommitBatchSlack),
AggregateCommits: cfg.AggregateCommits,
MinCommitBatch: cfg.MinCommitBatch,
MaxCommitBatch: cfg.MaxCommitBatch,
CommitBatchWait: config.Duration(cfg.CommitBatchWait),
CommitBatchSlack: config.Duration(cfg.CommitBatchSlack),
AggregateAboveBaseFee: types.FIL(cfg.AggregateAboveBaseFee),
AggregateCommits: cfg.AggregateCommits,
MinCommitBatch: cfg.MinCommitBatch,
MaxCommitBatch: cfg.MaxCommitBatch,
CommitBatchWait: config.Duration(cfg.CommitBatchWait),
CommitBatchSlack: config.Duration(cfg.CommitBatchSlack),
AggregateAboveBaseFee: types.FIL(cfg.AggregateAboveBaseFee),
BatchPreCommitAboveBaseFee: types.FIL(cfg.BatchPreCommitAboveBaseFee),
TerminateBatchMax: cfg.TerminateBatchMax,
TerminateBatchMin: cfg.TerminateBatchMin,
@@ -922,12 +931,13 @@ func ToSealingConfig(cfg *config.StorageMiner) sealiface.Config {
PreCommitBatchWait: time.Duration(cfg.Sealing.PreCommitBatchWait),
PreCommitBatchSlack: time.Duration(cfg.Sealing.PreCommitBatchSlack),
AggregateCommits: cfg.Sealing.AggregateCommits,
MinCommitBatch: cfg.Sealing.MinCommitBatch,
MaxCommitBatch: cfg.Sealing.MaxCommitBatch,
CommitBatchWait: time.Duration(cfg.Sealing.CommitBatchWait),
CommitBatchSlack: time.Duration(cfg.Sealing.CommitBatchSlack),
AggregateAboveBaseFee: types.BigInt(cfg.Sealing.AggregateAboveBaseFee),
AggregateCommits: cfg.Sealing.AggregateCommits,
MinCommitBatch: cfg.Sealing.MinCommitBatch,
MaxCommitBatch: cfg.Sealing.MaxCommitBatch,
CommitBatchWait: time.Duration(cfg.Sealing.CommitBatchWait),
CommitBatchSlack: time.Duration(cfg.Sealing.CommitBatchSlack),
AggregateAboveBaseFee: types.BigInt(cfg.Sealing.AggregateAboveBaseFee),
BatchPreCommitAboveBaseFee: types.BigInt(cfg.Sealing.BatchPreCommitAboveBaseFee),
TerminateBatchMax: cfg.Sealing.TerminateBatchMax,
TerminateBatchMin: cfg.Sealing.TerminateBatchMin,