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

Merge pull request #7405 from filecoin-project/feat/update-graphsync-0.10.0

Browse Source 
Update to graphsync to v0.10.0, enable seperate storage and retrieval transfer limits
This commit is contained in:
Jiaying Wang
2021-10-05 11:50:59 -04:00
committed by GitHub
parent 1d2f84322d f60692c7ef
commit cbb147de23
15 changed files with 115 additions and 48 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
+16 -2
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,
@@ -147,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
+17 -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",
node/config/types.go
+9 -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
@@ -364,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
@@ -54,6 +54,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"
@@ -981,6 +982,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)
@@ -1232,7 +1234,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/modules/client.go
+3 -2
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"
@@ -112,7 +113,7 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
net := dtnet.NewFromLibp2pHost(h, dtRetryParams)
dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/client/transfers"))
transport := dtgstransport.NewTransport(h.ID(), gs)
transport := dtgstransport.NewTransport(h.ID(), gs, net)
err := os.MkdirAll(filepath.Join(r.Path(), "data-transfer"), 0755) //nolint: gosec
if err != nil && !os.IsExist(err) {
return nil, err
@@ -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.MaxInProgressRequests(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/storageminer.go
+11 -3
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"
@@ -340,7 +341,7 @@ func NewProviderDAGServiceDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.S
net := dtnet.NewFromLibp2pHost(h)
dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/provider/transfers"))
transport := dtgstransport.NewTransport(h.ID(), gs)
transport := dtgstransport.NewTransport(h.ID(), gs, net)
err := os.MkdirAll(filepath.Join(r.Path(), "data-transfer"), 0755) //nolint: gosec
if err != nil && !os.IsExist(err) {
return nil, err
@@ -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.MaxInProgressRequests(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
}