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

Merge remote-tracking branch 'origin/master' into steb/refactor-consistent-tipset-methods

Browse Source
This commit is contained in:
Łukasz Magiera
2021-01-30 12:05:21 +01:00
parent b78b9a441b 9053efd344
commit 56a9d05491
214 changed files with 9021 additions and 1634 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/modules/chain.go
+1 -1
View File
@@ -77,7 +77,7 @@ func MessagePool(lc fx.Lifecycle, sm *stmgr.StateManager, ps *pubsub.PubSub, ds
}
func ChainRawBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.ChainRawBlockstore, error) {
bs, err := r.Blockstore(repo.BlockstoreChain)
bs, err := r.Blockstore(helpers.LifecycleCtx(mctx, lc), repo.BlockstoreChain)
if err != nil {
return nil, err
}
node/modules/client.go
+17 -5
View File
@@ -39,6 +39,7 @@ import (
"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"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
"github.com/filecoin-project/lotus/node/repo/importmgr"
"github.com/filecoin-project/lotus/node/repo/retrievalstoremgr"
@@ -78,8 +79,9 @@ func HandleMigrateClientFunds(lc fx.Lifecycle, ds dtypes.MetadataDS, wallet full
})
}
func ClientMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.ClientMultiDstore, error) {
ds, err := r.Datastore("/client")
func ClientMultiDatastore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.ClientMultiDstore, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
ds, err := r.Datastore(ctx, "/client")
if err != nil {
return nil, xerrors.Errorf("getting datastore out of reop: %w", err)
}
@@ -120,7 +122,11 @@ func RegisterClientValidator(crv dtypes.ClientRequestValidator, dtm dtypes.Clien
// uses the clients's Client DAG service for transfers
func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Graphsync, ds dtypes.MetadataDS, r repo.LockedRepo) (dtypes.ClientDataTransfer, error) {
sc := storedcounter.New(ds, datastore.NewKey("/datatransfer/client/counter"))
net := dtnet.NewFromLibp2pHost(h)
// go-data-transfer protocol retries:
// 1s, 5s, 25s, 2m5s, 5m x 11 ~= 1 hour
dtRetryParams := dtnet.RetryParameters(time.Second, 5*time.Minute, 15, 5)
net := dtnet.NewFromLibp2pHost(h, dtRetryParams)
dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/client/transfers"))
transport := dtgstransport.NewTransport(h.ID(), gs)
@@ -129,7 +135,9 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
return nil, err
}
dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, sc)
// data-transfer push channel restart configuration
dtRestartConfig := dtimpl.PushChannelRestartConfig(time.Minute, 10, 1024, 10*time.Minute, 3)
dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, sc, dtRestartConfig)
if err != nil {
return nil, err
}
@@ -153,7 +161,11 @@ func NewClientDatastore(ds dtypes.MetadataDS) dtypes.ClientDatastore {
}
func StorageClient(lc fx.Lifecycle, h host.Host, ibs dtypes.ClientBlockstore, mds dtypes.ClientMultiDstore, r repo.LockedRepo, dataTransfer dtypes.ClientDataTransfer, discovery *discoveryimpl.Local, deals dtypes.ClientDatastore, scn storagemarket.StorageClientNode, j journal.Journal) (storagemarket.StorageClient, error) {
net := smnet.NewFromLibp2pHost(h)
// go-fil-markets protocol retries:
// 1s, 5s, 25s, 2m5s, 5m x 11 ~= 1 hour
marketsRetryParams := smnet.RetryParameters(time.Second, 5*time.Minute, 15, 5)
net := smnet.NewFromLibp2pHost(h, marketsRetryParams)
c, err := storageimpl.NewClient(net, ibs, mds, dataTransfer, discovery, deals, scn, storageimpl.DealPollingInterval(time.Second))
if err != nil {
return nil, err
node/modules/core.go
+55 -28
View File
@@ -7,6 +7,7 @@ import (
"io"
"io/ioutil"
"os"
"path/filepath"
"time"
"github.com/gbrlsnchs/jwt/v3"
@@ -14,6 +15,7 @@ import (
"github.com/libp2p/go-libp2p-core/peer"
"github.com/libp2p/go-libp2p-core/peerstore"
record "github.com/libp2p/go-libp2p-record"
"github.com/raulk/go-watchdog"
"go.uber.org/fx"
"golang.org/x/xerrors"
@@ -28,7 +30,6 @@ import (
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/repo"
"github.com/filecoin-project/lotus/system"
"github.com/raulk/go-watchdog"
)
const (
@@ -69,45 +70,71 @@ func MemoryConstraints() system.MemoryConstraints {
// MemoryWatchdog starts the memory watchdog, applying the computed resource
// constraints.
func MemoryWatchdog(lc fx.Lifecycle, constraints system.MemoryConstraints) {
func MemoryWatchdog(lr repo.LockedRepo, lc fx.Lifecycle, constraints system.MemoryConstraints) {
if os.Getenv(EnvWatchdogDisabled) == "1" {
log.Infof("memory watchdog is disabled via %s", EnvWatchdogDisabled)
return
}
cfg := watchdog.MemConfig{
Resolution: 5 * time.Second,
Policy: &watchdog.WatermarkPolicy{
Watermarks: []float64{0.50, 0.60, 0.70, 0.85, 0.90, 0.925, 0.95},
EmergencyWatermark: 0.95,
},
Logger: logWatchdog,
// configure heap profile capture so that one is captured per episode where
// utilization climbs over 90% of the limit. A maximum of 10 heapdumps
// will be captured during life of this process.
watchdog.HeapProfileDir = filepath.Join(lr.Path(), "heapprof")
watchdog.HeapProfileMaxCaptures = 10
watchdog.HeapProfileThreshold = 0.9
watchdog.Logger = logWatchdog
policy := watchdog.NewWatermarkPolicy(0.50, 0.60, 0.70, 0.85, 0.90, 0.925, 0.95)
// Try to initialize a watchdog in the following order of precedence:
// 1. If a max heap limit has been provided, initialize a heap-driven watchdog.
// 2. Else, try to initialize a cgroup-driven watchdog.
// 3. Else, try to initialize a system-driven watchdog.
// 4. Else, log a warning that the system is flying solo, and return.
addStopHook := func(stopFn func()) {
lc.Append(fx.Hook{
OnStop: func(ctx context.Context) error {
stopFn()
return nil
},
})
}
// if user has set max heap limit, apply it. Otherwise, fall back to total
// system memory constraint.
// 1. If user has set max heap limit, apply it.
if maxHeap := constraints.MaxHeapMem; maxHeap != 0 {
log.Infof("memory watchdog will apply max heap constraint: %d bytes", maxHeap)
cfg.Limit = maxHeap
cfg.Scope = watchdog.ScopeHeap
} else {
log.Infof("max heap size not provided; memory watchdog will apply total system memory constraint: %d bytes", constraints.TotalSystemMem)
cfg.Limit = constraints.TotalSystemMem
cfg.Scope = watchdog.ScopeSystem
const minGOGC = 10
err, stopFn := watchdog.HeapDriven(maxHeap, minGOGC, policy)
if err == nil {
log.Infof("initialized heap-driven watchdog; max heap: %d bytes", maxHeap)
addStopHook(stopFn)
return
}
log.Warnf("failed to initialize heap-driven watchdog; err: %s", err)
log.Warnf("trying a cgroup-driven watchdog")
}
err, stop := watchdog.Memory(cfg)
if err != nil {
log.Warnf("failed to instantiate memory watchdog: %s", err)
// 2. cgroup-driven watchdog.
err, stopFn := watchdog.CgroupDriven(5*time.Second, policy)
if err == nil {
log.Infof("initialized cgroup-driven watchdog")
addStopHook(stopFn)
return
}
log.Warnf("failed to initialize cgroup-driven watchdog; err: %s", err)
log.Warnf("trying a system-driven watchdog")
// 3. system-driven watchdog.
err, stopFn = watchdog.SystemDriven(0, 5*time.Second, policy) // 0 calculates the limit automatically.
if err == nil {
log.Infof("initialized system-driven watchdog")
addStopHook(stopFn)
return
}
lc.Append(fx.Hook{
OnStop: func(ctx context.Context) error {
stop()
return nil
},
})
// 4. log the failure
log.Warnf("failed to initialize system-driven watchdog; err: %s", err)
log.Warnf("system running without a memory watchdog")
}
type JwtPayload struct {
@@ -166,7 +193,7 @@ func BuiltinBootstrap() (dtypes.BootstrapPeers, error) {
func DrandBootstrap(ds dtypes.DrandSchedule) (dtypes.DrandBootstrap, error) {
// TODO: retry resolving, don't fail if at least one resolve succeeds
res := []peer.AddrInfo{}
var res []peer.AddrInfo
for _, d := range ds {
addrs, err := addrutil.ParseAddresses(context.TODO(), d.Config.Relays)
if err != nil {
node/modules/dtypes/miner.go
+16
View File
@@ -58,6 +58,22 @@ type ConsiderOfflineRetrievalDealsConfigFunc func() (bool, error)
// disable or enable retrieval deal acceptance.
type SetConsiderOfflineRetrievalDealsConfigFunc func(bool) error
// ConsiderVerifiedStorageDealsConfigFunc is a function which reads from miner
// config to determine if the user has disabled verified storage deals (or not).
type ConsiderVerifiedStorageDealsConfigFunc func() (bool, error)
// SetConsiderVerifiedStorageDealsConfigFunc is a function which is used to
// disable or enable verified storage deal acceptance.
type SetConsiderVerifiedStorageDealsConfigFunc func(bool) error
// ConsiderUnverifiedStorageDealsConfigFunc is a function which reads from miner
// config to determine if the user has disabled unverified storage deals (or not).
type ConsiderUnverifiedStorageDealsConfigFunc func() (bool, error)
// SetConsiderUnverifiedStorageDealsConfigFunc is a function which is used to
// disable or enable unverified storage deal acceptance.
type SetConsiderUnverifiedStorageDealsConfigFunc func(bool) error
// SetSealingDelay sets how long a sector waits for more deals before sealing begins.
type SetSealingConfigFunc func(sealiface.Config) error
node/modules/stmgr.go
+20
View File
@@ -0,0 +1,20 @@
package modules
import (
"go.uber.org/fx"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
)
func StateManager(lc fx.Lifecycle, cs *store.ChainStore, us stmgr.UpgradeSchedule) (*stmgr.StateManager, error) {
sm, err := stmgr.NewStateManagerWithUpgradeSchedule(cs, us)
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStart: sm.Start,
OnStop: sm.Stop,
})
return sm, nil
}
node/modules/storage.go
+4 -2
View File
@@ -8,6 +8,7 @@ import (
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/lib/backupds"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
)
@@ -27,8 +28,9 @@ func KeyStore(lr repo.LockedRepo) (types.KeyStore, error) {
return lr.KeyStore()
}
func Datastore(r repo.LockedRepo) (dtypes.MetadataDS, error) {
mds, err := r.Datastore("/metadata")
func Datastore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.MetadataDS, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
mds, err := r.Datastore(ctx, "/metadata")
if err != nil {
return nil, err
}
node/modules/storageminer.go
+75 -8
View File
@@ -58,6 +58,7 @@ import (
lapi "github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/actors/builtin"
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
"github.com/filecoin-project/lotus/chain/gen"
"github.com/filecoin-project/lotus/chain/gen/slashfilter"
"github.com/filecoin-project/lotus/chain/types"
@@ -66,7 +67,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/miner"
lotusminer "github.com/filecoin-project/lotus/miner"
"github.com/filecoin-project/lotus/node/config"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
@@ -127,8 +128,12 @@ func SealProofType(maddr dtypes.MinerAddress, fnapi lapi.FullNode) (abi.Register
if err != nil {
return 0, err
}
networkVersion, err := fnapi.StateNetworkVersion(context.TODO(), types.EmptyTSK)
if err != nil {
return 0, err
}
return mi.SealProofType, nil
return miner.PreferredSealProofTypeFromWindowPoStType(networkVersion, mi.WindowPoStProofType)
}
type sidsc struct {
@@ -357,8 +362,9 @@ func NewProviderPieceStore(lc fx.Lifecycle, ds dtypes.MetadataDS) (dtypes.Provid
return ps, nil
}
func StagingMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.StagingMultiDstore, error) {
ds, err := r.Datastore("/staging")
func StagingMultiDatastore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.StagingMultiDstore, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
ds, err := r.Datastore(ctx, "/staging")
if err != nil {
return nil, xerrors.Errorf("getting datastore out of reop: %w", err)
}
@@ -379,8 +385,9 @@ func StagingMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.StagingMu
// StagingBlockstore creates a blockstore for staging blocks for a miner
// in a storage deal, prior to sealing
func StagingBlockstore(r repo.LockedRepo) (dtypes.StagingBlockstore, error) {
stagingds, err := r.Datastore("/staging")
func StagingBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.StagingBlockstore, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
stagingds, err := r.Datastore(ctx, "/staging")
if err != nil {
return nil, err
}
@@ -419,13 +426,13 @@ func StagingGraphsync(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.Stagi
return gs
}
func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api lapi.FullNode, epp gen.WinningPoStProver, sf *slashfilter.SlashFilter, j journal.Journal) (*miner.Miner, error) {
func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api lapi.FullNode, epp gen.WinningPoStProver, sf *slashfilter.SlashFilter, j journal.Journal) (*lotusminer.Miner, error) {
minerAddr, err := minerAddrFromDS(ds)
if err != nil {
return nil, err
}
m := miner.NewMiner(api, epp, minerAddr, sf, j)
m := lotusminer.NewMiner(api, epp, minerAddr, sf, j)
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
@@ -461,11 +468,15 @@ func NewStorageAsk(ctx helpers.MetricsCtx, fapi lapi.FullNode, ds dtypes.Metadat
func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
verifiedOk dtypes.ConsiderVerifiedStorageDealsConfigFunc,
unverifiedOk dtypes.ConsiderUnverifiedStorageDealsConfigFunc,
blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc,
spn storagemarket.StorageProviderNode) dtypes.StorageDealFilter {
return func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
verifiedOk dtypes.ConsiderVerifiedStorageDealsConfigFunc,
unverifiedOk dtypes.ConsiderUnverifiedStorageDealsConfigFunc,
blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc,
spn storagemarket.StorageProviderNode) dtypes.StorageDealFilter {
@@ -491,6 +502,26 @@ func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.Conside
return false, "miner is not accepting offline storage deals", nil
}
b, err = verifiedOk()
if err != nil {
return false, "miner error", err
}
if deal.Proposal.VerifiedDeal && !b {
log.Warnf("verified storage deal consideration disabled; rejecting storage deal proposal from client: %s", deal.Client.String())
return false, "miner is not accepting verified storage deals", nil
}
b, err = unverifiedOk()
if err != nil {
return false, "miner error", err
}
if !deal.Proposal.VerifiedDeal && !b {
log.Warnf("unverified storage deal consideration disabled; rejecting storage deal proposal from client: %s", deal.Client.String())
return false, "miner is not accepting unverified storage deals", nil
}
blocklist, err := blocklistFunc()
if err != nil {
return false, "miner error", err
@@ -737,6 +768,42 @@ func NewSetConsiderOfflineRetrievalDealsConfigFunc(r repo.LockedRepo) (dtypes.Se
}, nil
}
func NewConsiderVerifiedStorageDealsConfigFunc(r repo.LockedRepo) (dtypes.ConsiderVerifiedStorageDealsConfigFunc, error) {
return func() (out bool, err error) {
err = readCfg(r, func(cfg *config.StorageMiner) {
out = cfg.Dealmaking.ConsiderVerifiedStorageDeals
})
return
}, nil
}
func NewSetConsideringVerifiedStorageDealsFunc(r repo.LockedRepo) (dtypes.SetConsiderVerifiedStorageDealsConfigFunc, error) {
return func(b bool) (err error) {
err = mutateCfg(r, func(cfg *config.StorageMiner) {
cfg.Dealmaking.ConsiderVerifiedStorageDeals = b
})
return
}, nil
}
func NewConsiderUnverifiedStorageDealsConfigFunc(r repo.LockedRepo) (dtypes.ConsiderUnverifiedStorageDealsConfigFunc, error) {
return func() (out bool, err error) {
err = readCfg(r, func(cfg *config.StorageMiner) {
out = cfg.Dealmaking.ConsiderUnverifiedStorageDeals
})
return
}, nil
}
func NewSetConsideringUnverifiedStorageDealsFunc(r repo.LockedRepo) (dtypes.SetConsiderUnverifiedStorageDealsConfigFunc, error) {
return func(b bool) (err error) {
err = mutateCfg(r, func(cfg *config.StorageMiner) {
cfg.Dealmaking.ConsiderUnverifiedStorageDeals = b
})
return
}, nil
}
func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error) {
return func(cfg sealiface.Config) (err error) {
err = mutateCfg(r, func(c *config.StorageMiner) {