diff --git a/extern/storage-sealing/fsm.go b/extern/storage-sealing/fsm.go index fd9ff2d29..ddfebad24 100644 --- a/extern/storage-sealing/fsm.go +++ b/extern/storage-sealing/fsm.go @@ -328,7 +328,7 @@ func (m *Sealing) restartSectors(ctx context.Context) error { log.Errorf("loading sector list: %+v", err) } - sd, err := m.getSealDelay() + cfg, err := m.getConfig() if err != nil { return xerrors.Errorf("getting the sealing delay: %w", err) } @@ -339,8 +339,8 @@ func (m *Sealing) restartSectors(ctx context.Context) error { } if sector.State == WaitDeals { - if sd > 0 { - timer := time.NewTimer(sd) + if cfg.WaitDealsDelay > 0 { + timer := time.NewTimer(cfg.WaitDealsDelay) go func() { <-timer.C m.StartPacking(sector.SectorNumber) diff --git a/extern/storage-sealing/sealing.go b/extern/storage-sealing/sealing.go index eca387da3..7ae2e5163 100644 --- a/extern/storage-sealing/sealing.go +++ b/extern/storage-sealing/sealing.go @@ -3,6 +3,7 @@ package sealing import ( "context" "io" + "math" "sync" "time" @@ -33,6 +34,16 @@ type SectorLocation struct { Partition uint64 } +type Config struct { + // 0 = no limit + MaxWaitDealsSectors uint64 + + // includes failed, 0 = no limit + MaxSealingSectors uint64 + + WaitDealsDelay time.Duration +} + type SealingAPI interface { StateWaitMsg(context.Context, cid.Cid) (MsgLookup, error) StateComputeDataCommitment(ctx context.Context, maddr address.Address, sectorType abi.RegisteredSealProof, deals []abi.DealID, tok TipSetToken) (cid.Cid, error) @@ -70,7 +81,7 @@ type Sealing struct { upgradeLk sync.Mutex toUpgrade map[abi.SectorNumber]struct{} - getSealDelay GetSealingDelayFunc + getConfig GetSealingConfigFunc } type FeeConfig struct { @@ -80,7 +91,7 @@ type FeeConfig struct { type UnsealedSectorMap struct { infos map[abi.SectorNumber]UnsealedSectorInfo - mux sync.Mutex + lk sync.Mutex } type UnsealedSectorInfo struct { @@ -90,7 +101,7 @@ type UnsealedSectorInfo struct { pieceSizes []abi.UnpaddedPieceSize } -func New(api SealingAPI, fc FeeConfig, events Events, maddr address.Address, ds datastore.Batching, sealer sectorstorage.SectorManager, sc SectorIDCounter, verif ffiwrapper.Verifier, pcp PreCommitPolicy, gsd GetSealingDelayFunc) *Sealing { +func New(api SealingAPI, fc FeeConfig, events Events, maddr address.Address, ds datastore.Batching, sealer sectorstorage.SectorManager, sc SectorIDCounter, verif ffiwrapper.Verifier, pcp PreCommitPolicy, gc GetSealingConfigFunc) *Sealing { s := &Sealing{ api: api, feeCfg: fc, @@ -103,11 +114,11 @@ func New(api SealingAPI, fc FeeConfig, events Events, maddr address.Address, ds pcp: pcp, unsealedInfoMap: UnsealedSectorMap{ infos: make(map[abi.SectorNumber]UnsealedSectorInfo), - mux: sync.Mutex{}, + lk: sync.Mutex{}, }, - toUpgrade: map[abi.SectorNumber]struct{}{}, - getSealDelay: gsd, + toUpgrade: map[abi.SectorNumber]struct{}{}, + getConfig: gc, } s.sectors = statemachine.New(namespace.Wrap(ds, datastore.NewKey(SectorStorePrefix)), s, SectorInfo{}) @@ -137,18 +148,18 @@ func (m *Sealing) AddPieceToAnySector(ctx context.Context, size abi.UnpaddedPiec return 0, 0, xerrors.Errorf("piece cannot fit into a sector") } - m.unsealedInfoMap.mux.Lock() + m.unsealedInfoMap.lk.Lock() sid, pads, err := m.getSectorAndPadding(size) if err != nil { - m.unsealedInfoMap.mux.Unlock() + m.unsealedInfoMap.lk.Unlock() return 0, 0, xerrors.Errorf("getting available sector: %w", err) } for _, p := range pads { err = m.addPiece(ctx, sid, p.Unpadded(), NewNullReader(p.Unpadded()), nil) if err != nil { - m.unsealedInfoMap.mux.Unlock() + m.unsealedInfoMap.lk.Unlock() return 0, 0, xerrors.Errorf("writing pads: %w", err) } } @@ -157,12 +168,15 @@ func (m *Sealing) AddPieceToAnySector(ctx context.Context, size abi.UnpaddedPiec err = m.addPiece(ctx, sid, size, r, &d) if err != nil { - m.unsealedInfoMap.mux.Unlock() + m.unsealedInfoMap.lk.Unlock() return 0, 0, xerrors.Errorf("adding piece to sector: %w", err) } - m.unsealedInfoMap.mux.Unlock() - if m.unsealedInfoMap.infos[sid].numDeals == getDealPerSectorLimit(m.sealer.SectorSize()) { + startPacking := m.unsealedInfoMap.infos[sid].numDeals >= getDealPerSectorLimit(m.sealer.SectorSize()) + + m.unsealedInfoMap.lk.Unlock() + + if startPacking { if err := m.StartPacking(sid); err != nil { return 0, 0, xerrors.Errorf("start packing: %w", err) } @@ -171,7 +185,7 @@ func (m *Sealing) AddPieceToAnySector(ctx context.Context, size abi.UnpaddedPiec return sid, offset, nil } -// Caller should hold m.unsealedInfoMap.mux +// Caller should hold m.unsealedInfoMap.lk func (m *Sealing) addPiece(ctx context.Context, sectorID abi.SectorNumber, size abi.UnpaddedPieceSize, r io.Reader, di *DealInfo) error { log.Infof("Adding piece to sector %d", sectorID) ppi, err := m.sealer.AddPiece(sectorstorage.WithPriority(ctx, DealSectorPriority), m.minerSector(sectorID), m.unsealedInfoMap.infos[sectorID].pieceSizes, size, r) @@ -206,7 +220,7 @@ func (m *Sealing) Remove(ctx context.Context, sid abi.SectorNumber) error { return m.sectors.Send(uint64(sid), SectorRemove{}) } -// Caller should NOT hold m.unsealedInfoMap.mux +// Caller should NOT hold m.unsealedInfoMap.lk func (m *Sealing) StartPacking(sectorID abi.SectorNumber) error { log.Infof("Starting packing sector %d", sectorID) err := m.sectors.Send(uint64(sectorID), SectorStartPacking{}) @@ -214,14 +228,14 @@ func (m *Sealing) StartPacking(sectorID abi.SectorNumber) error { return err } - m.unsealedInfoMap.mux.Lock() + m.unsealedInfoMap.lk.Lock() delete(m.unsealedInfoMap.infos, sectorID) - m.unsealedInfoMap.mux.Unlock() + m.unsealedInfoMap.lk.Unlock() return nil } -// Caller should hold m.unsealedInfoMap.mux +// Caller should hold m.unsealedInfoMap.lk func (m *Sealing) getSectorAndPadding(size abi.UnpaddedPieceSize) (abi.SectorNumber, []abi.PaddedPieceSize, error) { ss := abi.PaddedPieceSize(m.sealer.SectorSize()) for k, v := range m.unsealedInfoMap.infos { @@ -231,7 +245,7 @@ func (m *Sealing) getSectorAndPadding(size abi.UnpaddedPieceSize) (abi.SectorNum } } - ns, err := m.newSector() + ns, err := m.newDealSector() if err != nil { return 0, nil, err } @@ -245,8 +259,57 @@ func (m *Sealing) getSectorAndPadding(size abi.UnpaddedPieceSize) (abi.SectorNum return ns, nil, nil } -// newSector creates a new sector for deal storage -func (m *Sealing) newSector() (abi.SectorNumber, error) { +// newDealSector creates a new sector for deal storage +func (m *Sealing) newDealSector() (abi.SectorNumber, error) { + // First make sure we don't have too many 'open' sectors + + cfg, err := m.getConfig() + if err != nil { + return 0, xerrors.Errorf("getting config: %w", err) + } + + if cfg.MaxWaitDealsSectors > 0 { + // run in a loop because we have to drop the map lock here for a bit + tries := 0 + + for uint64(len(m.unsealedInfoMap.infos)) >= cfg.MaxWaitDealsSectors { + if tries > 10 { + // whatever... + break + } + + if tries > 0 { + m.unsealedInfoMap.lk.Unlock() + time.Sleep(time.Second) + m.unsealedInfoMap.lk.Lock() + } + + tries++ + var mostStored abi.PaddedPieceSize = math.MaxUint64 + var best abi.SectorNumber = math.MaxUint64 + + for sn, info := range m.unsealedInfoMap.infos { + if info.stored + 1 > mostStored + 1 { // 18446744073709551615 + 1 = 0 + best = sn + } + } + + if best == math.MaxUint64 { + // probably not possible, but who knows + break + } + + m.unsealedInfoMap.lk.Unlock() + if err := m.StartPacking(best); err != nil { + log.Error("newDealSector StartPacking error: %+v", err) + continue // let's pretend this is fine + } + m.unsealedInfoMap.lk.Lock() + } + } + + // Now actually create a new sector + sid, err := m.sc.Next() if err != nil { return 0, xerrors.Errorf("getting sector number: %w", err) @@ -272,13 +335,13 @@ func (m *Sealing) newSector() (abi.SectorNumber, error) { return 0, xerrors.Errorf("starting the sector fsm: %w", err) } - sd, err := m.getSealDelay() + cf, err := m.getConfig() if err != nil { return 0, xerrors.Errorf("getting the sealing delay: %w", err) } - if sd > 0 { - timer := time.NewTimer(sd) + if cf.WaitDealsDelay > 0 { + timer := time.NewTimer(cf.WaitDealsDelay) go func() { <-timer.C m.StartPacking(sid) diff --git a/extern/storage-sealing/types.go b/extern/storage-sealing/types.go index 45993bb82..1e7c9f76c 100644 --- a/extern/storage-sealing/types.go +++ b/extern/storage-sealing/types.go @@ -3,8 +3,6 @@ package sealing import ( "bytes" "context" - "time" - "github.com/ipfs/go-cid" "github.com/filecoin-project/specs-actors/actors/abi" @@ -188,7 +186,7 @@ type MessageReceipt struct { GasUsed int64 } -type GetSealingDelayFunc func() (time.Duration, error) +type GetSealingConfigFunc func() (Config, error) func (mr *MessageReceipt) Equals(o *MessageReceipt) bool { return mr.ExitCode == o.ExitCode && bytes.Equal(mr.Return, o.Return) && mr.GasUsed == o.GasUsed diff --git a/node/builder.go b/node/builder.go index 7f1247084..cf07a660a 100644 --- a/node/builder.go +++ b/node/builder.go @@ -328,8 +328,8 @@ func Online() Option { Override(new(dtypes.SetConsiderOfflineStorageDealsConfigFunc), modules.NewSetConsideringOfflineStorageDealsFunc), Override(new(dtypes.ConsiderOfflineRetrievalDealsConfigFunc), modules.NewConsiderOfflineRetrievalDealsConfigFunc), Override(new(dtypes.SetConsiderOfflineRetrievalDealsConfigFunc), modules.NewSetConsiderOfflineRetrievalDealsConfigFunc), - Override(new(dtypes.SetSealingDelayFunc), modules.NewSetSealDelayFunc), - Override(new(dtypes.GetSealingDelayFunc), modules.NewGetSealDelayFunc), + Override(new(dtypes.SetSealingConfigFunc), modules.NewSetSealConfigFunc), + Override(new(dtypes.GetSealingConfigFunc), modules.NewGetSealConfigFunc), Override(new(dtypes.SetExpectedSealDurationFunc), modules.NewSetExpectedSealDurationFunc), Override(new(dtypes.GetExpectedSealDurationFunc), modules.NewGetExpectedSealDurationFunc), ), diff --git a/node/config/def.go b/node/config/def.go index 436935916..fcddf73ba 100644 --- a/node/config/def.go +++ b/node/config/def.go @@ -31,10 +31,9 @@ type StorageMiner struct { Common Dealmaking DealmakingConfig + Sealing SealingConfig Storage sectorstorage.SealerConfig Fees MinerFeeConfig - - SealingDelay Duration } type DealmakingConfig struct { @@ -48,6 +47,16 @@ type DealmakingConfig struct { Filter string } +type SealingConfig struct { + // 0 = no limit + MaxWaitDealsSectors uint64 + + // includes failed, 0 = no limit + MaxSealingSectors uint64 + + WaitDealsDelay Duration +} + type MinerFeeConfig struct { MaxPreCommitGasFee types.FIL MaxCommitGasFee types.FIL @@ -130,6 +139,12 @@ func DefaultFullNode() *FullNode { func DefaultStorageMiner() *StorageMiner { cfg := &StorageMiner{ Common: defCommon(), + + Sealing: SealingConfig{ + MaxWaitDealsSectors: 2, // 64G with 32G sectors + MaxSealingSectors: 0, + WaitDealsDelay: Duration(time.Hour), + }, Storage: sectorstorage.SealerConfig{ AllowAddPiece: true, @@ -158,8 +173,6 @@ func DefaultStorageMiner() *StorageMiner { MaxCommitGasFee: types.FIL(types.BigDiv(types.FromFil(1), types.NewInt(20))), MaxWindowPoStGasFee: types.FIL(types.FromFil(50)), }, - - SealingDelay: Duration(time.Hour), } cfg.Common.API.ListenAddress = "/ip4/127.0.0.1/tcp/2345/http" cfg.Common.API.RemoteListenAddress = "127.0.0.1:2345" diff --git a/node/impl/storminer.go b/node/impl/storminer.go index e302f7051..77052164b 100644 --- a/node/impl/storminer.go +++ b/node/impl/storminer.go @@ -62,8 +62,8 @@ type StorageMinerAPI struct { SetConsiderOfflineStorageDealsConfigFunc dtypes.SetConsiderOfflineStorageDealsConfigFunc ConsiderOfflineRetrievalDealsConfigFunc dtypes.ConsiderOfflineRetrievalDealsConfigFunc SetConsiderOfflineRetrievalDealsConfigFunc dtypes.SetConsiderOfflineRetrievalDealsConfigFunc - SetSealingDelayFunc dtypes.SetSealingDelayFunc - GetSealingDelayFunc dtypes.GetSealingDelayFunc + SetSealingConfigFunc dtypes.SetSealingConfigFunc + GetSealingConfigFunc dtypes.GetSealingConfigFunc GetExpectedSealDurationFunc dtypes.GetExpectedSealDurationFunc SetExpectedSealDurationFunc dtypes.SetExpectedSealDurationFunc } @@ -232,11 +232,22 @@ func (sm *StorageMinerAPI) SectorStartSealing(ctx context.Context, number abi.Se } func (sm *StorageMinerAPI) SectorSetSealDelay(ctx context.Context, delay time.Duration) error { - return sm.SetSealingDelayFunc(delay) + cfg, err := sm.GetSealingConfigFunc() + if err != nil { + return xerrors.Errorf("get config: %w", err) + } + + cfg.WaitDealsDelay = delay + + return sm.SetSealingConfigFunc(cfg) } func (sm *StorageMinerAPI) SectorGetSealDelay(ctx context.Context) (time.Duration, error) { - return sm.GetSealingDelayFunc() + cfg, err := sm.GetSealingConfigFunc() + if err != nil { + return 0, err + } + return cfg.WaitDealsDelay, nil } func (sm *StorageMinerAPI) SectorSetExpectedSealDuration(ctx context.Context, delay time.Duration) error { diff --git a/node/modules/dtypes/miner.go b/node/modules/dtypes/miner.go index 34911df5e..600d19dc7 100644 --- a/node/modules/dtypes/miner.go +++ b/node/modules/dtypes/miner.go @@ -8,6 +8,7 @@ import ( "github.com/filecoin-project/go-address" "github.com/filecoin-project/go-fil-markets/storagemarket" + sealing "github.com/filecoin-project/lotus/extern/storage-sealing" "github.com/filecoin-project/specs-actors/actors/abi" ) @@ -56,10 +57,10 @@ type ConsiderOfflineRetrievalDealsConfigFunc func() (bool, error) type SetConsiderOfflineRetrievalDealsConfigFunc func(bool) error // SetSealingDelay sets how long a sector waits for more deals before sealing begins. -type SetSealingDelayFunc func(time.Duration) error +type SetSealingConfigFunc func(sealing.Config) error // GetSealingDelay returns how long a sector waits for more deals before sealing begins. -type GetSealingDelayFunc func() (time.Duration, error) +type GetSealingConfigFunc func() (sealing.Config, error) // SetExpectedSealDurationFunc is a function which is used to set how long sealing is expected to take. // Deals that would need to start earlier than this duration will be rejected. diff --git a/node/modules/storageminer.go b/node/modules/storageminer.go index 7aec09482..e0cbb434e 100644 --- a/node/modules/storageminer.go +++ b/node/modules/storageminer.go @@ -141,8 +141,8 @@ func SectorIDCounter(ds dtypes.MetadataDS) sealing.SectorIDCounter { return &sidsc{sc} } -func StorageMiner(fc config.MinerFeeConfig) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, api lapi.FullNode, h host.Host, ds dtypes.MetadataDS, sealer sectorstorage.SectorManager, sc sealing.SectorIDCounter, verif ffiwrapper.Verifier, gsd dtypes.GetSealingDelayFunc) (*storage.Miner, error) { - return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, api lapi.FullNode, h host.Host, ds dtypes.MetadataDS, sealer sectorstorage.SectorManager, sc sealing.SectorIDCounter, verif ffiwrapper.Verifier, gsd dtypes.GetSealingDelayFunc) (*storage.Miner, error) { +func StorageMiner(fc config.MinerFeeConfig) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, api lapi.FullNode, h host.Host, ds dtypes.MetadataDS, sealer sectorstorage.SectorManager, sc sealing.SectorIDCounter, verif ffiwrapper.Verifier, gsd dtypes.GetSealingConfigFunc) (*storage.Miner, error) { + return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, api lapi.FullNode, h host.Host, ds dtypes.MetadataDS, sealer sectorstorage.SectorManager, sc sealing.SectorIDCounter, verif ffiwrapper.Verifier, gsd dtypes.GetSealingConfigFunc) (*storage.Miner, error) { maddr, err := minerAddrFromDS(ds) if err != nil { return nil, err @@ -593,19 +593,27 @@ func NewSetConsiderOfflineRetrievalDealsConfigFunc(r repo.LockedRepo) (dtypes.Se }, nil } -func NewSetSealDelayFunc(r repo.LockedRepo) (dtypes.SetSealingDelayFunc, error) { - return func(delay time.Duration) (err error) { - err = mutateCfg(r, func(cfg *config.StorageMiner) { - cfg.SealingDelay = config.Duration(delay) +func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error) { + return func(cfg sealing.Config) (err error) { + err = mutateCfg(r, func(c *config.StorageMiner) { + c.Sealing = config.SealingConfig{ + MaxWaitDealsSectors: cfg.MaxWaitDealsSectors, + MaxSealingSectors: cfg.MaxSealingSectors, + WaitDealsDelay: config.Duration(cfg.WaitDealsDelay), + } }) return }, nil } -func NewGetSealDelayFunc(r repo.LockedRepo) (dtypes.GetSealingDelayFunc, error) { - return func() (out time.Duration, err error) { +func NewGetSealConfigFunc(r repo.LockedRepo) (dtypes.GetSealingConfigFunc, error) { + return func() (out sealing.Config, err error) { err = readCfg(r, func(cfg *config.StorageMiner) { - out = time.Duration(cfg.SealingDelay) + out = sealing.Config{ + MaxWaitDealsSectors: cfg.Sealing.MaxWaitDealsSectors, + MaxSealingSectors: cfg.Sealing.MaxSealingSectors, + WaitDealsDelay: time.Duration(cfg.Sealing.WaitDealsDelay), + } }) return }, nil diff --git a/storage/miner.go b/storage/miner.go index f8a6691f4..803556cdf 100644 --- a/storage/miner.go +++ b/storage/miner.go @@ -43,8 +43,8 @@ type Miner struct { maddr address.Address worker address.Address - getSealDelay dtypes.GetSealingDelayFunc - sealing *sealing.Sealing + getSealConfig dtypes.GetSealingConfigFunc + sealing *sealing.Sealing } type storageMinerApi interface { @@ -84,7 +84,7 @@ type storageMinerApi interface { WalletHas(context.Context, address.Address) (bool, error) } -func NewMiner(api storageMinerApi, maddr, worker address.Address, h host.Host, ds datastore.Batching, sealer sectorstorage.SectorManager, sc sealing.SectorIDCounter, verif ffiwrapper.Verifier, gsd dtypes.GetSealingDelayFunc, feeCfg config.MinerFeeConfig) (*Miner, error) { +func NewMiner(api storageMinerApi, maddr, worker address.Address, h host.Host, ds datastore.Batching, sealer sectorstorage.SectorManager, sc sealing.SectorIDCounter, verif ffiwrapper.Verifier, gsd dtypes.GetSealingConfigFunc, feeCfg config.MinerFeeConfig) (*Miner, error) { m := &Miner{ api: api, feeCfg: feeCfg, @@ -94,9 +94,9 @@ func NewMiner(api storageMinerApi, maddr, worker address.Address, h host.Host, d sc: sc, verif: verif, - maddr: maddr, - worker: worker, - getSealDelay: gsd, + maddr: maddr, + worker: worker, + getSealConfig: gsd, } return m, nil @@ -120,7 +120,7 @@ func (m *Miner) Run(ctx context.Context) error { evts := events.NewEvents(ctx, m.api) adaptedAPI := NewSealingAPIAdapter(m.api) pcp := sealing.NewBasicPreCommitPolicy(adaptedAPI, miner.MaxSectorExpirationExtension-(miner.WPoStProvingPeriod*2), md.PeriodStart%miner.WPoStProvingPeriod) - m.sealing = sealing.New(adaptedAPI, fc, NewEventsAdapter(evts), m.maddr, m.ds, m.sealer, m.sc, m.verif, &pcp, sealing.GetSealingDelayFunc(m.getSealDelay)) + m.sealing = sealing.New(adaptedAPI, fc, NewEventsAdapter(evts), m.maddr, m.ds, m.sealer, m.sc, m.verif, &pcp, sealing.GetSealingConfigFunc(m.getSealConfig)) go m.sealing.Run(ctx) //nolint:errcheck // logged intside the function