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

wip: storage miner index provider

Browse Source
This commit is contained in:
Dirk McCormick
2021-11-16 17:45:42 +01:00
parent a633f93707
commit 627d4fbf74
10 changed files with 292 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder_miner.go
+2
View File
@@ -12,6 +12,7 @@ import (
"github.com/filecoin-project/go-fil-markets/storagemarket"
"github.com/filecoin-project/go-fil-markets/storagemarket/impl/storedask"
"github.com/filecoin-project/go-state-types/abi"
provider "github.com/filecoin-project/index-provider"
storage2 "github.com/filecoin-project/specs-storage/storage"
"github.com/filecoin-project/lotus/api"
@@ -163,6 +164,7 @@ func ConfigStorageMiner(c interface{}) Option {
Override(HandleRetrievalKey, modules.HandleRetrieval),
// Markets (storage)
Override(new(provider.Interface), modules.IndexerProvider(cfg.IndexerProvider)),
Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDAGServiceDataTransfer),
Override(new(*storedask.StoredAsk), modules.NewStorageAsk),
Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(cfg.Dealmaking, nil)),
node/config/def.go
+9
View File
@@ -176,6 +176,15 @@ func DefaultStorageMiner() *StorageMiner {
},
},
IndexerProvider: IndexerProviderConfig{
ListenAddresses: []string{
"/ip4/0.0.0.0/tcp/0",
"/ip6/::/tcp/0",
},
MaxSimultaneousTransfers: DefaultSimultaneousTransfers,
},
Subsystems: MinerSubsystemConfig{
EnableMining: true,
EnableSealing: true,
node/config/types.go
+22 -7
View File
@@ -3,6 +3,8 @@ package config
import (
"github.com/ipfs/go-cid"
"github.com/filecoin-project/index-provider/config"
"github.com/filecoin-project/lotus/chain/types"
sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
)
@@ -43,13 +45,14 @@ type Backup struct {
type StorageMiner struct {
Common
Subsystems MinerSubsystemConfig
Dealmaking DealmakingConfig
Sealing SealingConfig
Storage sectorstorage.SealerConfig
Fees MinerFeeConfig
Addresses MinerAddressConfig
DAGStore DAGStoreConfig
Subsystems MinerSubsystemConfig
Dealmaking DealmakingConfig
IndexerProvider IndexerProviderConfig
Sealing SealingConfig
Storage sectorstorage.SealerConfig
Fees MinerFeeConfig
Addresses MinerAddressConfig
DAGStore DAGStoreConfig
}
type DAGStoreConfig struct {
@@ -146,6 +149,18 @@ type DealmakingConfig struct {
RetrievalPricing *RetrievalPricing
}
type IndexerProviderConfig struct {
config.Ingest
// Binding address for the libp2p host - 0 means random port.
// Format: multiaddress; see https://multiformats.io/multiaddr/
ListenAddresses []string
// The maximum number of simultaneous data transfers between the indexers
// and the indexer provider
MaxSimultaneousTransfers uint64
}
type RetrievalPricing struct {
Strategy string // possible values: "default", "external"
node/modules/storageminer.go
+3
View File
@@ -34,6 +34,7 @@ import (
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-statestore"
"github.com/filecoin-project/go-storedcounter"
provider "github.com/filecoin-project/index-provider"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
@@ -581,6 +582,7 @@ func StorageProvider(minerAddress dtypes.MinerAddress,
h host.Host, ds dtypes.MetadataDS,
r repo.LockedRepo,
pieceStore dtypes.ProviderPieceStore,
indexer provider.Interface,
dataTransfer dtypes.ProviderDataTransfer,
spn storagemarket.StorageProviderNode,
df dtypes.StorageDealFilter,
@@ -609,6 +611,7 @@ func StorageProvider(minerAddress dtypes.MinerAddress,
namespace.Wrap(ds, datastore.NewKey("/deals/provider")),
store,
dsw,
indexer,
pieceStore,
dataTransfer,
spn,
node/modules/storageminer_idxprov.go
+158
View File
@@ -0,0 +1,158 @@
package modules
import (
"context"
"fmt"
"os"
"path/filepath"
"golang.org/x/xerrors"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-graphsync"
graphsyncimpl "github.com/ipfs/go-graphsync/impl"
gsnet "github.com/ipfs/go-graphsync/network"
cidlink "github.com/ipld/go-ipld-prime/linking/cid"
"github.com/libp2p/go-libp2p"
ci "github.com/libp2p/go-libp2p-core/crypto"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/libp2p/go-libp2p-core/peerstore"
"go.uber.org/fx"
datatransfer "github.com/filecoin-project/go-data-transfer"
dtimpl "github.com/filecoin-project/go-data-transfer/impl"
dtnet "github.com/filecoin-project/go-data-transfer/network"
dtgstransport "github.com/filecoin-project/go-data-transfer/transport/graphsync"
provider "github.com/filecoin-project/index-provider"
"github.com/filecoin-project/index-provider/engine"
p2pserver "github.com/filecoin-project/index-provider/server/provider/libp2p"
"github.com/ipfs/go-datastore/namespace"
"github.com/libp2p/go-libp2p-core/host"
"github.com/filecoin-project/lotus/build"
marketevents "github.com/filecoin-project/lotus/markets/loggers"
"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"
)
type IdxProv struct {
fx.In
helpers.MetricsCtx
fx.Lifecycle
Repo repo.LockedRepo
Datastore dtypes.MetadataDS
PeerID peer.ID
peerstore.Peerstore
}
func IndexerProvider(cfg config.IndexerProviderConfig) func(params IdxProv) (provider.Interface, error) {
return func(args IdxProv) (provider.Interface, error) {
ipds := namespace.Wrap(args.Datastore, datastore.NewKey("/indexer-provider"))
pkey := args.Peerstore.PrivKey(args.PeerID)
if pkey == nil {
return nil, fmt.Errorf("missing private key for node ID: %s", args.PeerID.Pretty())
}
h, err := createIndexerProviderHost(args.MetricsCtx, args.Lifecycle, pkey, args.Peerstore, cfg.ListenAddresses)
if err != nil {
return nil, xerrors.Errorf("creating indexer provider host: %w", err)
}
dt, err := newIndexerProviderDataTransfer(cfg, args.MetricsCtx, args.Lifecycle, args.Repo, h, ipds)
if err != nil {
return nil, err
}
e, err := engine.New(cfg.Ingest, pkey, dt, h, ipds, nil)
if err != nil {
return nil, xerrors.Errorf("creating indexer provider engine: %w", err)
}
args.Lifecycle.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
// Start the engine
err := e.Start(ctx)
if err != nil {
return xerrors.Errorf("starting indexer provider engine: %s", err)
}
// Add a handler to libp2p that listens for incoming index requests
p2pserver.New(ctx, h, e)
return nil
},
OnStop: func(ctx context.Context) error {
return e.Shutdown(ctx)
},
})
return e, nil
}
}
func createIndexerProviderHost(mctx helpers.MetricsCtx, lc fx.Lifecycle, pkey ci.PrivKey, pstore peerstore.Peerstore, listenAddrs []string) (host.Host, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
opts := []libp2p.Option{
libp2p.Identity(pkey),
libp2p.Peerstore(pstore),
libp2p.ListenAddrStrings(listenAddrs...),
libp2p.Ping(true),
libp2p.UserAgent("lotus-indexer-provider-" + build.UserVersion()),
}
h, err := libp2p.New(ctx, opts...)
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStop: func(ctx context.Context) error {
return h.Close()
},
})
return h, nil
}
func newIndexerProviderDataTransfer(cfg config.IndexerProviderConfig, mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, h host.Host, ds datastore.Batching) (datatransfer.Manager, error) {
net := dtnet.NewFromLibp2pHost(h)
// Set up graphsync
gs := newIndexerProviderGraphsync(cfg, mctx, lc, h)
// Set up data transfer
dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/transfers"))
transport := dtgstransport.NewTransport(h.ID(), gs, net)
dtPath := filepath.Join(r.Path(), "indexer-provider", "data-transfer")
err := os.MkdirAll(dtPath, 0755) //nolint: gosec
if err != nil && !os.IsExist(err) {
return nil, xerrors.Errorf("creating indexer provider data transfer dir %s: %w", dtPath, err)
}
dt, err := dtimpl.NewDataTransfer(dtDs, dtPath, net, transport)
if err != nil {
return nil, xerrors.Errorf("creating indexer provider data transfer module: %w", err)
}
dt.OnReady(marketevents.ReadyLogger("indexer-provider data transfer"))
lc.Append(fx.Hook{
OnStart: dt.Start,
OnStop: dt.Stop,
})
return dt, nil
}
func newIndexerProviderGraphsync(cfg config.IndexerProviderConfig, mctx helpers.MetricsCtx, lc fx.Lifecycle, h host.Host) graphsync.GraphExchange {
graphsyncNetwork := gsnet.NewFromLibp2pHost(h)
return graphsyncimpl.New(helpers.LifecycleCtx(mctx, lc),
graphsyncNetwork,
cidlink.DefaultLinkSystem(),
graphsyncimpl.MaxInProgressIncomingRequests(cfg.MaxSimultaneousTransfers),
graphsyncimpl.MaxLinksPerIncomingRequests(config.MaxTraversalLinks),
graphsyncimpl.MaxLinksPerOutgoingRequests(config.MaxTraversalLinks))
}