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

Merge pull request #6356 from filecoin-project/nonsense/split-market-miner-processes

Browse Source 
Support standalone miner-market process
This commit is contained in:
Łukasz Magiera
2021-07-13 17:16:07 +02:00
committed by GitHub
parent c37401a1a3 2af02af1c1
commit 837322ea59
84 changed files with 2596 additions and 1600 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/modules/dtypes/storage.go
+1
View File
@@ -86,6 +86,7 @@ type ClientDataTransfer datatransfer.Manager
type ProviderDealStore *statestore.StateStore
type ProviderPieceStore piecestore.PieceStore
type ProviderRequestValidator *requestvalidation.UnifiedRequestValidator
// ProviderDataTransfer is a data transfer manager for the provider
node/modules/storageminer.go
+25 -20
View File
@@ -8,6 +8,7 @@ import (
"net/http"
"os"
"path/filepath"
"strings"
"time"
"github.com/filecoin-project/lotus/markets/pricing"
@@ -44,7 +45,7 @@ import (
smnet "github.com/filecoin-project/go-fil-markets/storagemarket/network"
"github.com/filecoin-project/go-jsonrpc/auth"
"github.com/filecoin-project/go-multistore"
paramfetch "github.com/filecoin-project/go-paramfetch"
"github.com/filecoin-project/go-paramfetch"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-statestore"
"github.com/filecoin-project/go-storedcounter"
@@ -67,7 +68,6 @@ import (
"github.com/filecoin-project/lotus/journal"
"github.com/filecoin-project/lotus/markets"
marketevents "github.com/filecoin-project/lotus/markets/loggers"
"github.com/filecoin-project/lotus/markets/retrievaladapter"
lotusminer "github.com/filecoin-project/lotus/miner"
"github.com/filecoin-project/lotus/node/config"
"github.com/filecoin-project/lotus/node/modules/dtypes"
@@ -207,7 +207,6 @@ type StorageMinerParams struct {
Lifecycle fx.Lifecycle
MetricsCtx helpers.MetricsCtx
API v1api.FullNode
Host host.Host
MetadataDS dtypes.MetadataDS
Sealer sectorstorage.SectorManager
SectorIDCounter sealing.SectorIDCounter
@@ -226,7 +225,6 @@ func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*st
lc = params.Lifecycle
api = params.API
sealer = params.Sealer
h = params.Host
sc = params.SectorIDCounter
verif = params.Verifier
prover = params.Prover
@@ -247,7 +245,7 @@ func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*st
return nil, err
}
sm, err := storage.NewMiner(api, maddr, h, ds, sealer, sc, verif, prover, gsd, fc, j, as)
sm, err := storage.NewMiner(api, maddr, ds, sealer, sc, verif, prover, gsd, fc, j, as)
if err != nil {
return nil, err
}
@@ -652,6 +650,10 @@ func RetrievalDealFilter(userFilter dtypes.RetrievalDealFilter) func(onlineOk dt
}
}
func RetrievalNetwork(h host.Host) rmnet.RetrievalMarketNetwork {
return rmnet.NewFromLibp2pHost(h)
}
// RetrievalPricingFunc configures the pricing function to use for retrieval deals.
func RetrievalPricingFunc(cfg config.DealmakingConfig) func(_ dtypes.ConsiderOnlineRetrievalDealsConfigFunc,
_ dtypes.ConsiderOfflineRetrievalDealsConfigFunc) dtypes.RetrievalPricingFunc {
@@ -667,35 +669,26 @@ func RetrievalPricingFunc(cfg config.DealmakingConfig) func(_ dtypes.ConsiderOnl
}
// RetrievalProvider creates a new retrieval provider attached to the provider blockstore
func RetrievalProvider(h host.Host,
miner *storage.Miner,
full v1api.FullNode,
func RetrievalProvider(
maddr dtypes.MinerAddress,
adapter retrievalmarket.RetrievalProviderNode,
netwk rmnet.RetrievalMarketNetwork,
ds dtypes.MetadataDS,
pieceStore dtypes.ProviderPieceStore,
mds dtypes.StagingMultiDstore,
dt dtypes.ProviderDataTransfer,
pieceProvider sectorstorage.PieceProvider,
pricingFnc dtypes.RetrievalPricingFunc,
userFilter dtypes.RetrievalDealFilter,
) (retrievalmarket.RetrievalProvider, error) {
adapter := retrievaladapter.NewRetrievalProviderNode(miner, pieceProvider, full)
maddr, err := minerAddrFromDS(ds)
if err != nil {
return nil, err
}
netwk := rmnet.NewFromLibp2pHost(h)
opt := retrievalimpl.DealDeciderOpt(retrievalimpl.DealDecider(userFilter))
return retrievalimpl.NewProvider(maddr, adapter, netwk, pieceStore, mds, dt, namespace.Wrap(ds, datastore.NewKey("/retrievals/provider")),
return retrievalimpl.NewProvider(address.Address(maddr), adapter, netwk, pieceStore, mds, dt, namespace.Wrap(ds, datastore.NewKey("/retrievals/provider")),
retrievalimpl.RetrievalPricingFunc(pricingFnc), opt)
}
var WorkerCallsPrefix = datastore.NewKey("/worker/calls")
var ManagerWorkPrefix = datastore.NewKey("/stmgr/calls")
func LocalStorage(mctx helpers.MetricsCtx, lc fx.Lifecycle, ls stores.LocalStorage, si stores.SectorIndex, urls sectorstorage.URLs) (*stores.Local, error) {
func LocalStorage(mctx helpers.MetricsCtx, lc fx.Lifecycle, ls stores.LocalStorage, si stores.SectorIndex, urls stores.URLs) (*stores.Local, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
return stores.NewLocal(ctx, ls, si, urls)
}
@@ -733,6 +726,18 @@ func StorageAuth(ctx helpers.MetricsCtx, ca v0api.Common) (sectorstorage.Storage
return sectorstorage.StorageAuth(headers), nil
}
func StorageAuthWithURL(apiInfo string) func(ctx helpers.MetricsCtx, ca v0api.Common) (sectorstorage.StorageAuth, error) {
return func(ctx helpers.MetricsCtx, ca v0api.Common) (sectorstorage.StorageAuth, error) {
s := strings.Split(apiInfo, ":")
if len(s) != 2 {
return nil, errors.New("unexpected format of `apiInfo`")
}
headers := http.Header{}
headers.Add("Authorization", "Bearer "+s[0])
return sectorstorage.StorageAuth(headers), nil
}
}
func NewConsiderOnlineStorageDealsConfigFunc(r repo.LockedRepo) (dtypes.ConsiderOnlineStorageDealsConfigFunc, error) {
return func() (out bool, err error) {
err = readCfg(r, func(cfg *config.StorageMiner) {
node/modules/storageminer_svc.go
+71
View File
@@ -0,0 +1,71 @@
package modules
import (
"context"
"github.com/filecoin-project/lotus/storage/sectorblocks"
"go.uber.org/fx"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/api/client"
cliutil "github.com/filecoin-project/lotus/cli/util"
"github.com/filecoin-project/lotus/node/modules/helpers"
)
type MinerSealingService api.StorageMiner
type MinerStorageService api.StorageMiner
var _ sectorblocks.SectorBuilder = *new(MinerSealingService)
func connectMinerService(apiInfo string) func(mctx helpers.MetricsCtx, lc fx.Lifecycle) (api.StorageMiner, error) {
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle) (api.StorageMiner, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
info := cliutil.ParseApiInfo(apiInfo)
addr, err := info.DialArgs("v0")
if err != nil {
return nil, xerrors.Errorf("could not get DialArgs: %w", err)
}
log.Infof("Checking (svc) api version of %s", addr)
mapi, closer, err := client.NewStorageMinerRPCV0(ctx, addr, info.AuthHeader())
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
v, err := mapi.Version(ctx)
if err != nil {
return xerrors.Errorf("checking version: %w", err)
}
if !v.APIVersion.EqMajorMinor(api.MinerAPIVersion0) {
return xerrors.Errorf("remote service API version didn't match (expected %s, remote %s)", api.MinerAPIVersion0, v.APIVersion)
}
return nil
},
OnStop: func(context.Context) error {
closer()
return nil
}})
return mapi, nil
}
}
func ConnectSealingService(apiInfo string) func(mctx helpers.MetricsCtx, lc fx.Lifecycle) (MinerSealingService, error) {
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle) (MinerSealingService, error) {
log.Info("Connecting sealing service to miner")
return connectMinerService(apiInfo)(mctx, lc)
}
}
func ConnectStorageService(apiInfo string) func(mctx helpers.MetricsCtx, lc fx.Lifecycle) (MinerStorageService, error) {
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle) (MinerStorageService, error) {
log.Info("Connecting storage service to miner")
return connectMinerService(apiInfo)(mctx, lc)
}
}