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

Merge banch feat/go-fil-markets-v1.13.0 into feat/datamodel-selector-retrieval

Browse Source
This commit is contained in:
Peter Rabbitson
2021-10-04 21:56:11 +02:00
parent 0444435589 f70ec02a6c
commit 2a1094fc17
90 changed files with 2886 additions and 843 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder_miner.go
+3 -2
View File
@@ -199,8 +199,9 @@ func ConfigStorageMiner(c interface{}) Option {
Override(new(dtypes.RetrievalDealFilter), modules.RetrievalDealFilter(dealfilter.CliRetrievalDealFilter(cfg.Dealmaking.RetrievalFilter))),
),
Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(&cfg.Fees, storageadapter.PublishMsgConfig{
Period: time.Duration(cfg.Dealmaking.PublishMsgPeriod),
MaxDealsPerMsg: cfg.Dealmaking.MaxDealsPerPublishMsg,
Period: time.Duration(cfg.Dealmaking.PublishMsgPeriod),
MaxDealsPerMsg: cfg.Dealmaking.MaxDealsPerPublishMsg,
StartEpochSealingBuffer: cfg.Dealmaking.StartEpochSealingBuffer,
})),
Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(&cfg.Fees, &cfg.Dealmaking)),
),
node/config/def.go
+3 -1
View File
@@ -101,7 +101,7 @@ func DefaultStorageMiner() *StorageMiner {
PreCommitBatchWait: Duration(24 * time.Hour), // this should be less than 31.5 hours, which is the expiration of a precommit ticket
PreCommitBatchSlack: Duration(3 * 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
CommittedCapacitySectorLifetime: Duration(builtin.EpochDurationSeconds * policy.GetMaxSectorExpirationExtension()),
CommittedCapacitySectorLifetime: Duration(builtin.EpochDurationSeconds * uint64(policy.GetMaxSectorExpirationExtension()) * uint64(time.Second)),
AggregateCommits: true,
MinCommitBatch: miner5.MinAggregatedSectors, // per FIP13, we must have at least four proofs to aggregate, where 4 is the cross over point where aggregation wins out on single provecommit gas costs
@@ -148,6 +148,8 @@ func DefaultStorageMiner() *StorageMiner {
SimultaneousTransfers: DefaultSimultaneousTransfers,
StartEpochSealingBuffer: 480, // 480 epochs buffer == 4 hours from adding deal to sector to sector being sealed
RetrievalPricing: &RetrievalPricing{
Strategy: RetrievalPricingDefaultMode,
Default: &RetrievalPricingDefault{
node/config/doc_gen.go
+6
View File
@@ -265,6 +265,12 @@ passed to the sealing node by the markets service. 0 is unlimited.`,
Comment: `The maximum number of parallel online data transfers (storage+retrieval)`,
},
{
Name: "StartEpochSealingBuffer",
Type: "uint64",
Comment: `Minimum start epoch buffer to give time for sealing of sector with deal.`,
},
{
Name: "Filter",
Type: "string",
node/config/load.go
+2
View File
@@ -125,6 +125,8 @@ func ConfigUpdate(cfgCur, cfgDef interface{}, comment bool) ([]byte, error) {
outLines = append(outLines, pad+"# type: "+doc.Type)
}
outLines = append(outLines, pad+"# env var: LOTUS_"+strings.ToUpper(strings.ReplaceAll(section, ".", "_"))+"_"+strings.ToUpper(lf[0]))
}
}
node/config/types.go
+2
View File
@@ -131,6 +131,8 @@ type DealmakingConfig struct {
MaxStagingDealsBytes int64
// The maximum number of parallel online data transfers (storage+retrieval)
SimultaneousTransfers uint64
// Minimum start epoch buffer to give time for sealing of sector with deal.
StartEpochSealingBuffer uint64
// A command used for fine-grained evaluation of storage deals
// see https://docs.filecoin.io/mine/lotus/miner-configuration/#using-filters-for-fine-grained-storage-and-retrieval-deal-acceptance for more details
node/impl/full/chain.go
-30
View File
@@ -10,7 +10,6 @@ import (
"strings"
"sync"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/stmgr"
"go.uber.org/fx"
@@ -30,7 +29,6 @@ import (
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-state-types/crypto"
"github.com/filecoin-project/specs-actors/actors/util/adt"
"github.com/filecoin-project/lotus/api"
@@ -99,34 +97,6 @@ func (m *ChainModule) ChainHead(context.Context) (*types.TipSet, error) {
return m.Chain.GetHeaviestTipSet(), nil
}
func (a *ChainAPI) ChainGetRandomnessFromTickets(ctx context.Context, tsk types.TipSetKey, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte) (abi.Randomness, error) {
pts, err := a.Chain.LoadTipSet(tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset key: %w", err)
}
// Doing this here is slightly nicer than doing it in the chainstore directly, but it's still bad for ChainAPI to reason about network upgrades
if randEpoch > build.UpgradeHyperdriveHeight {
return a.Chain.GetChainRandomnessLookingForward(ctx, pts.Cids(), personalization, randEpoch, entropy)
}
return a.Chain.GetChainRandomnessLookingBack(ctx, pts.Cids(), personalization, randEpoch, entropy)
}
func (a *ChainAPI) ChainGetRandomnessFromBeacon(ctx context.Context, tsk types.TipSetKey, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte) (abi.Randomness, error) {
pts, err := a.Chain.LoadTipSet(tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset key: %w", err)
}
// Doing this here is slightly nicer than doing it in the chainstore directly, but it's still bad for ChainAPI to reason about network upgrades
if randEpoch > build.UpgradeHyperdriveHeight {
return a.Chain.GetBeaconRandomnessLookingForward(ctx, pts.Cids(), personalization, randEpoch, entropy)
}
return a.Chain.GetBeaconRandomnessLookingBack(ctx, pts.Cids(), personalization, randEpoch, entropy)
}
func (a *ChainAPI) ChainGetBlock(ctx context.Context, msg cid.Cid) (*types.BlockHeader, error) {
return a.Chain.GetBlock(msg)
}
node/impl/full/state.go
+33
View File
@@ -6,6 +6,8 @@ import (
"encoding/json"
"strconv"
"github.com/filecoin-project/go-state-types/crypto"
"github.com/filecoin-project/go-state-types/cbor"
cid "github.com/ipfs/go-cid"
"go.uber.org/fx"
@@ -1418,3 +1420,34 @@ func (m *StateModule) StateNetworkVersion(ctx context.Context, tsk types.TipSetK
// But that's likely going to break a bunch of stuff.
return m.StateManager.GetNtwkVersion(ctx, ts.Height()), nil
}
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)
}
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)
}
node/modules/client.go
+1 -1
View File
@@ -112,7 +112,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
node/modules/graphsync.go
+1 -1
View File
@@ -20,7 +20,7 @@ func Graphsync(parallelTransfers uint64) func(mctx helpers.MetricsCtx, lc fx.Lif
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(parallelTransfers))
chainLinkSystem := storeutil.LinkSystemForBlockstore(chainBs)
err := gs.RegisterPersistenceOption("chainstore", chainLinkSystem)
if err != nil {
node/modules/lp2p/addrs.go
-13
View File
@@ -11,19 +11,6 @@ import (
mamask "github.com/whyrusleeping/multiaddr-filter"
)
func AddrFilters(filters []string) func() (opts Libp2pOpts, err error) {
return func() (opts Libp2pOpts, err error) {
for _, s := range filters {
f, err := mamask.NewMask(s)
if err != nil {
return opts, fmt.Errorf("incorrectly formatted address filter in config: %s", s)
}
opts.Opts = append(opts.Opts, libp2p.FilterAddresses(f)) //nolint:staticcheck
}
return opts, nil
}
}
func makeAddrsFactory(announce []string, noAnnounce []string) (p2pbhost.AddrsFactory, error) {
var annAddrs []ma.Multiaddr
for _, addr := range announce {
node/modules/lp2p/libp2p.go
+1 -1
View File
@@ -42,7 +42,7 @@ func PrivKey(ks types.KeyStore) (crypto.PrivKey, error) {
if err != nil {
return nil, err
}
kbytes, err := pk.Bytes()
kbytes, err := crypto.MarshalPrivateKey(pk)
if err != nil {
return nil, err
}
node/modules/storageminer.go
+25 -17
View File
@@ -70,7 +70,10 @@ import (
"github.com/filecoin-project/lotus/storage"
)
var StorageCounterDSPrefix = "/storage/nextid"
var (
StorageCounterDSPrefix = "/storage/nextid"
StagingAreaDirName = "deal-staging"
)
func minerAddrFromDS(ds dtypes.MetadataDS) (address.Address, error) {
maddrb, err := ds.Get(datastore.NewKey("miner-address"))
@@ -337,7 +340,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
@@ -395,7 +398,7 @@ func StagingGraphsync(parallelTransfers uint64) func(mctx helpers.MetricsCtx, lc
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(parallelTransfers))
return gs
}
@@ -536,7 +539,8 @@ func BasicDealFilter(cfg config.DealmakingConfig, user dtypes.StorageDealFilter)
return false, "miner error", err
}
diskUsageBytes, err := r.DiskUsage(r.Path() + "/deal-staging")
dir := filepath.Join(r.Path(), StagingAreaDirName)
diskUsageBytes, err := r.DiskUsage(dir)
if err != nil {
return false, "miner error", err
}
@@ -574,7 +578,7 @@ func StorageProvider(minerAddress dtypes.MinerAddress,
) (storagemarket.StorageProvider, error) {
net := smnet.NewFromLibp2pHost(h)
dir := filepath.Join(r.Path(), "deal-staging")
dir := filepath.Join(r.Path(), StagingAreaDirName)
// migrate temporary files that were created directly under the repo, by
// moving them to the new directory and symlinking them.
@@ -866,12 +870,13 @@ func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error
return func(cfg sealiface.Config) (err error) {
err = mutateCfg(r, func(c *config.StorageMiner) {
c.Sealing = config.SealingConfig{
MaxWaitDealsSectors: cfg.MaxWaitDealsSectors,
MaxSealingSectors: cfg.MaxSealingSectors,
MaxSealingSectorsForDeals: cfg.MaxSealingSectorsForDeals,
WaitDealsDelay: config.Duration(cfg.WaitDealsDelay),
AlwaysKeepUnsealedCopy: cfg.AlwaysKeepUnsealedCopy,
FinalizeEarly: cfg.FinalizeEarly,
MaxWaitDealsSectors: cfg.MaxWaitDealsSectors,
MaxSealingSectors: cfg.MaxSealingSectors,
MaxSealingSectorsForDeals: cfg.MaxSealingSectorsForDeals,
CommittedCapacitySectorLifetime: config.Duration(cfg.CommittedCapacitySectorLifetime),
WaitDealsDelay: config.Duration(cfg.WaitDealsDelay),
AlwaysKeepUnsealedCopy: cfg.AlwaysKeepUnsealedCopy,
FinalizeEarly: cfg.FinalizeEarly,
CollateralFromMinerBalance: cfg.CollateralFromMinerBalance,
AvailableBalanceBuffer: types.FIL(cfg.AvailableBalanceBuffer),
@@ -900,12 +905,13 @@ func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error
func ToSealingConfig(cfg *config.StorageMiner) sealiface.Config {
return sealiface.Config{
MaxWaitDealsSectors: cfg.Sealing.MaxWaitDealsSectors,
MaxSealingSectors: cfg.Sealing.MaxSealingSectors,
MaxSealingSectorsForDeals: cfg.Sealing.MaxSealingSectorsForDeals,
WaitDealsDelay: time.Duration(cfg.Sealing.WaitDealsDelay),
AlwaysKeepUnsealedCopy: cfg.Sealing.AlwaysKeepUnsealedCopy,
FinalizeEarly: cfg.Sealing.FinalizeEarly,
MaxWaitDealsSectors: cfg.Sealing.MaxWaitDealsSectors,
MaxSealingSectors: cfg.Sealing.MaxSealingSectors,
MaxSealingSectorsForDeals: cfg.Sealing.MaxSealingSectorsForDeals,
CommittedCapacitySectorLifetime: time.Duration(cfg.Sealing.CommittedCapacitySectorLifetime),
WaitDealsDelay: time.Duration(cfg.Sealing.WaitDealsDelay),
AlwaysKeepUnsealedCopy: cfg.Sealing.AlwaysKeepUnsealedCopy,
FinalizeEarly: cfg.Sealing.FinalizeEarly,
CollateralFromMinerBalance: cfg.Sealing.CollateralFromMinerBalance,
AvailableBalanceBuffer: types.BigInt(cfg.Sealing.AvailableBalanceBuffer),
@@ -926,6 +932,8 @@ func ToSealingConfig(cfg *config.StorageMiner) sealiface.Config {
TerminateBatchMax: cfg.Sealing.TerminateBatchMax,
TerminateBatchMin: cfg.Sealing.TerminateBatchMin,
TerminateBatchWait: time.Duration(cfg.Sealing.TerminateBatchWait),
StartEpochSealingBuffer: abi.ChainEpoch(cfg.Dealmaking.StartEpochSealingBuffer),
}
}
node/rpc.go
+3 -2
View File
@@ -25,6 +25,7 @@ import (
"github.com/filecoin-project/lotus/api/v1api"
"github.com/filecoin-project/lotus/lib/rpcenc"
"github.com/filecoin-project/lotus/metrics"
"github.com/filecoin-project/lotus/metrics/proxy"
"github.com/filecoin-project/lotus/node/impl"
)
@@ -78,7 +79,7 @@ func FullNodeHandler(a v1api.FullNode, permissioned bool, opts ...jsonrpc.Server
m.Handle(path, handler)
}
fnapi := metrics.MetricedFullAPI(a)
fnapi := proxy.MetricedFullAPI(a)
if permissioned {
fnapi = api.PermissionedFullAPI(fnapi)
}
@@ -113,7 +114,7 @@ func FullNodeHandler(a v1api.FullNode, permissioned bool, opts ...jsonrpc.Server
func MinerHandler(a api.StorageMiner, permissioned bool) (http.Handler, error) {
m := mux.NewRouter()
mapi := metrics.MetricedStorMinerAPI(a)
mapi := proxy.MetricedStorMinerAPI(a)
if permissioned {
mapi = api.PermissionedStorMinerAPI(mapi)
}