diff --git a/cli/pprof.go b/cli/pprof.go index dff089e7f..50a67ef86 100644 --- a/cli/pprof.go +++ b/cli/pprof.go @@ -13,7 +13,7 @@ import ( ) var pprofCmd = &cli.Command{ - Name: "pprof", + Name: "pprof", Hidden: true, Subcommands: []*cli.Command{ PprofGoroutines, @@ -42,7 +42,7 @@ var PprofGoroutines = &cli.Command{ return err } - addr = "http://" + addr + "/debug/pprof/goroutine?debug=2" + addr = "http://" + addr + "/debug/pprof/goroutine?debug=2" r, err := http.Get(addr) if err != nil { @@ -56,4 +56,3 @@ var PprofGoroutines = &cli.Command{ return r.Body.Close() }, } - diff --git a/cmd/lotus-storage-miner/config.go b/cmd/lotus-storage-miner/config.go index 0c843fe23..e5e4fc4c4 100644 --- a/cmd/lotus-storage-miner/config.go +++ b/cmd/lotus-storage-miner/config.go @@ -9,7 +9,7 @@ import ( ) var configCmd = &cli.Command{ - Name: "config", + Name: "config", Usage: "Output default configuration", Action: func(cctx *cli.Context) error { comm, err := config.ConfigComment(config.DefaultStorageMiner()) diff --git a/extern/storage-sealing/checks.go b/extern/storage-sealing/checks.go index 95ef101fa..af62b9548 100644 --- a/extern/storage-sealing/checks.go +++ b/extern/storage-sealing/checks.go @@ -26,10 +26,12 @@ type ErrBadCommD struct{ error } type ErrExpiredTicket struct{ error } type ErrBadTicket struct{ error } type ErrPrecommitOnChain struct{ error } +type ErrSectorNumberAllocated struct{ error } type ErrBadSeed struct{ error } type ErrInvalidProof struct{ error } type ErrNoPrecommit struct{ error } +type ErrCommitWaitFailed struct{ error } func checkPieces(ctx context.Context, si SectorInfo, api SealingAPI) error { tok, height, err := api.ChainHead(ctx) @@ -87,6 +89,9 @@ func checkPrecommit(ctx context.Context, maddr address.Address, si SectorInfo, t pci, err := api.StateSectorPreCommitInfo(ctx, maddr, si.SectorNumber, tok) if err != nil { + if err == ErrSectorAllocated { + return &ErrSectorNumberAllocated{err} + } return &ErrApi{xerrors.Errorf("getting precommit info: %w", err)} } @@ -106,6 +111,16 @@ func (m *Sealing) checkCommit(ctx context.Context, si SectorInfo, proof []byte, } pci, err := m.api.StateSectorPreCommitInfo(ctx, m.maddr, si.SectorNumber, tok) + if err == ErrSectorAllocated { + // not much more we can check here, basically try to wait for commit, + // and hope that this will work + + if si.CommitMessage != nil { + return &ErrCommitWaitFailed{err} + } + + return err + } if err != nil { return xerrors.Errorf("getting precommit info: %w", err) } diff --git a/extern/storage-sealing/fsm.go b/extern/storage-sealing/fsm.go index fd9ff2d29..d9648a99d 100644 --- a/extern/storage-sealing/fsm.go +++ b/extern/storage-sealing/fsm.go @@ -108,6 +108,7 @@ var fsmPlanners = map[SectorState]func(events []statemachine.Event, state *Secto on(SectorRetryPreCommitWait{}, PreCommitWait), on(SectorChainPreCommitFailed{}, PreCommitFailed), on(SectorRetryPreCommit{}, PreCommitting), + on(SectorRetryCommitWait{}, CommitWait), ), FinalizeFailed: planOne( on(SectorRetryFinalize{}, FinalizeSector), @@ -225,6 +226,8 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta */ + m.stats.updateSector(m.minerSector(state.SectorNumber), state.State) + switch state.State { // Happy path case Empty: @@ -315,6 +318,8 @@ func planCommitting(events []statemachine.Event, state *SectorInfo) error { state.State = SealPreCommit1Failed case SectorCommitFailed: state.State = CommitFailed + case SectorRetryCommitWait: + state.State = CommitWait default: return xerrors.Errorf("planCommitting got event of unknown type %T, events: %+v", event.User, events) } @@ -328,7 +333,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 +344,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/fsm_events.go b/extern/storage-sealing/fsm_events.go index c4278991e..f270b3668 100644 --- a/extern/storage-sealing/fsm_events.go +++ b/extern/storage-sealing/fsm_events.go @@ -252,6 +252,10 @@ func (evt SectorRetryInvalidProof) apply(state *SectorInfo) { state.InvalidProofs++ } +type SectorRetryCommitWait struct{} + +func (evt SectorRetryCommitWait) apply(state *SectorInfo) {} + // Faults type SectorFaulty struct{} diff --git a/extern/storage-sealing/fsm_test.go b/extern/storage-sealing/fsm_test.go index b1e53133c..f41d8c535 100644 --- a/extern/storage-sealing/fsm_test.go +++ b/extern/storage-sealing/fsm_test.go @@ -3,6 +3,8 @@ package sealing import ( "testing" + "github.com/filecoin-project/go-address" + "github.com/filecoin-project/specs-actors/actors/abi" logging "github.com/ipfs/go-log/v2" "github.com/stretchr/testify/require" @@ -25,8 +27,14 @@ type test struct { } func TestHappyPath(t *testing.T) { + ma, _ := address.NewIDAddress(55151) m := test{ - s: &Sealing{}, + s: &Sealing{ + maddr: ma, + stats: SectorStats{ + bySector: map[abi.SectorID]statSectorState{}, + }, + }, t: t, state: &SectorInfo{State: Packing}, } @@ -60,8 +68,14 @@ func TestHappyPath(t *testing.T) { } func TestSeedRevert(t *testing.T) { + ma, _ := address.NewIDAddress(55151) m := test{ - s: &Sealing{}, + s: &Sealing{ + maddr: ma, + stats: SectorStats{ + bySector: map[abi.SectorID]statSectorState{}, + }, + }, t: t, state: &SectorInfo{State: Packing}, } @@ -101,8 +115,14 @@ func TestSeedRevert(t *testing.T) { } func TestPlanCommittingHandlesSectorCommitFailed(t *testing.T) { + ma, _ := address.NewIDAddress(55151) m := test{ - s: &Sealing{}, + s: &Sealing{ + maddr: ma, + stats: SectorStats{ + bySector: map[abi.SectorID]statSectorState{}, + }, + }, t: t, state: &SectorInfo{State: Committing}, } diff --git a/extern/storage-sealing/garbage.go b/extern/storage-sealing/garbage.go index d8cdb4248..4b95c1b67 100644 --- a/extern/storage-sealing/garbage.go +++ b/extern/storage-sealing/garbage.go @@ -31,6 +31,17 @@ func (m *Sealing) pledgeSector(ctx context.Context, sectorID abi.SectorID, exist } func (m *Sealing) PledgeSector() error { + cfg, err := m.getConfig() + if err != nil { + return xerrors.Errorf("getting config: %w", err) + } + + if cfg.MaxSealingSectors > 0 { + if m.stats.curSealing() > cfg.MaxSealingSectors { + return xerrors.Errorf("too many sectors sealing (curSealing: %d, max: %d)", m.stats.curSealing(), cfg.MaxSealingSectors) + } + } + go func() { ctx := context.TODO() // we can't use the context from command which invokes // this, as we run everything here async, and it's cancelled when the diff --git a/extern/storage-sealing/sealiface/config.go b/extern/storage-sealing/sealiface/config.go new file mode 100644 index 000000000..945565562 --- /dev/null +++ b/extern/storage-sealing/sealiface/config.go @@ -0,0 +1,18 @@ +package sealiface + +import "time" + +// this has to be in a separate package to not make lotus API depend on filecoin-ffi + +type Config struct { + // 0 = no limit + MaxWaitDealsSectors uint64 + + // includes failed, 0 = no limit + MaxSealingSectors uint64 + + // includes failed, 0 = no limit + MaxSealingSectorsForDeals uint64 + + WaitDealsDelay time.Duration +} diff --git a/extern/storage-sealing/sealing.go b/extern/storage-sealing/sealing.go index eca387da3..062ade2a3 100644 --- a/extern/storage-sealing/sealing.go +++ b/extern/storage-sealing/sealing.go @@ -2,7 +2,9 @@ package sealing import ( "context" + "errors" "io" + "math" "sync" "time" @@ -33,9 +35,14 @@ type SectorLocation struct { Partition uint64 } +var ErrSectorAllocated = errors.New("sectorNumber is allocated, but PreCommit info wasn't found on chain") + type SealingAPI interface { StateWaitMsg(context.Context, cid.Cid) (MsgLookup, error) + StateSearchMsg(context.Context, cid.Cid) (*MsgLookup, error) StateComputeDataCommitment(ctx context.Context, maddr address.Address, sectorType abi.RegisteredSealProof, deals []abi.DealID, tok TipSetToken) (cid.Cid, error) + + // Can return ErrSectorAllocated in case precommit info wasn't found, but the sector number is marked as allocated StateSectorPreCommitInfo(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tok TipSetToken) (*miner.SectorPreCommitOnChainInfo, error) StateSectorGetInfo(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tok TipSetToken) (*miner.SectorOnChainInfo, error) StateSectorPartition(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tok TipSetToken) (*SectorLocation, error) @@ -70,7 +77,9 @@ type Sealing struct { upgradeLk sync.Mutex toUpgrade map[abi.SectorNumber]struct{} - getSealDelay GetSealingDelayFunc + stats SectorStats + + getConfig GetSealingConfigFunc } type FeeConfig struct { @@ -80,7 +89,7 @@ type FeeConfig struct { type UnsealedSectorMap struct { infos map[abi.SectorNumber]UnsealedSectorInfo - mux sync.Mutex + lk sync.Mutex } type UnsealedSectorInfo struct { @@ -90,7 +99,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 +112,15 @@ 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, + + stats: SectorStats{ + bySector: map[abi.SectorID]statSectorState{}, + }, } s.sectors = statemachine.New(namespace.Wrap(ds, datastore.NewKey(SectorStorePrefix)), s, SectorInfo{}) @@ -137,18 +150,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 +170,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 +187,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 +222,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 +230,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 +247,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 +261,66 @@ 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.MaxSealingSectorsForDeals > 0 { + if m.stats.curSealing() > cfg.MaxSealingSectorsForDeals { + return 0, xerrors.Errorf("too many sectors sealing") + } + } + + if cfg.MaxWaitDealsSectors > 0 { + // run in a loop because we have to drop the map lock here for a bit + tries := 0 + + // we have to run in a loop as we're dropping unsealedInfoMap.lk + // to actually call StartPacking. When we do that, another entry can + // get added to unsealedInfoMap. + 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 +346,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/sector_state.go b/extern/storage-sealing/sector_state.go index f2801c9fc..2f57d83e8 100644 --- a/extern/storage-sealing/sector_state.go +++ b/extern/storage-sealing/sector_state.go @@ -36,3 +36,14 @@ const ( RemoveFailed SectorState = "RemoveFailed" Removed SectorState = "Removed" ) + +func toStatState(st SectorState) statSectorState { + switch st { + case Empty, WaitDeals, Packing, PreCommit1, PreCommit2, PreCommitting, PreCommitWait, WaitSeed, Committing, CommitWait, FinalizeSector: + return sstSealing + case Proving, Removed, Removing: + return sstProving + } + + return sstFailed +} diff --git a/extern/storage-sealing/states_failed.go b/extern/storage-sealing/states_failed.go index e208a8cca..cf829f44f 100644 --- a/extern/storage-sealing/states_failed.go +++ b/extern/storage-sealing/states_failed.go @@ -85,6 +85,10 @@ func (m *Sealing) handlePreCommitFailed(ctx statemachine.Context, sector SectorI return ctx.Send(SectorRetryPreCommit{}) case *ErrPrecommitOnChain: // noop + case *ErrSectorNumberAllocated: + log.Errorf("handlePreCommitFailed: sector number already allocated, not proceeding: %+v", err) + // TODO: check if the sector is committed (not sure how we'd end up here) + return nil default: return xerrors.Errorf("checkPrecommit sanity check error: %w", err) } @@ -158,6 +162,8 @@ func (m *Sealing) handleCommitFailed(ctx statemachine.Context, sector SectorInfo return ctx.Send(SectorChainPreCommitFailed{xerrors.Errorf("no precommit: %w", err)}) case *ErrPrecommitOnChain: // noop, this is expected + case *ErrSectorNumberAllocated: + // noop, already committed? default: return xerrors.Errorf("checkPrecommit sanity check error (%T): %w", err, err) } @@ -186,6 +192,12 @@ func (m *Sealing) handleCommitFailed(ctx statemachine.Context, sector SectorInfo return ctx.Send(SectorRetryPreCommitWait{}) case *ErrNoPrecommit: return ctx.Send(SectorRetryPreCommit{}) + case *ErrCommitWaitFailed: + if err := failedCooldown(ctx, sector); err != nil { + return err + } + + return ctx.Send(SectorRetryCommitWait{}) default: return xerrors.Errorf("checkCommit sanity check error (%T): %w", err, err) } diff --git a/extern/storage-sealing/states_sealing.go b/extern/storage-sealing/states_sealing.go index 2178ce0b4..110a73ac0 100644 --- a/extern/storage-sealing/states_sealing.go +++ b/extern/storage-sealing/states_sealing.go @@ -149,6 +149,10 @@ func (m *Sealing) handlePreCommitting(ctx statemachine.Context, sector SectorInf return ctx.Send(SectorSealPreCommit1Failed{xerrors.Errorf("bad ticket: %w", err)}) case *ErrPrecommitOnChain: return ctx.Send(SectorPreCommitLanded{TipSet: tok}) // we re-did precommit + case *ErrSectorNumberAllocated: + log.Errorf("handlePreCommitFailed: sector number already allocated, not proceeding: %+v", err) + // TODO: check if the sector is committed (not sure how we'd end up here) + return nil default: return xerrors.Errorf("checkPrecommit sanity check error: %w", err) } @@ -275,6 +279,20 @@ func (m *Sealing) handleWaitSeed(ctx statemachine.Context, sector SectorInfo) er } func (m *Sealing) handleCommitting(ctx statemachine.Context, sector SectorInfo) error { + if sector.CommitMessage != nil { + log.Warnf("sector %d entered committing state with a commit message cid", sector.SectorNumber) + + ml, err := m.api.StateSearchMsg(ctx.Context(), *sector.CommitMessage) + if err != nil { + log.Warnf("sector %d searching existing commit message %s: %+v", sector.SectorNumber, *sector.CommitMessage, err) + } + + if ml != nil { + // some weird retry paths can lead here + return ctx.Send(SectorRetryCommitWait{}) + } + } + log.Info("scheduling seal proof computation...") log.Infof("KOMIT %d %x(%d); %x(%d); %v; r:%x; d:%x", sector.SectorNumber, sector.TicketValue, sector.TicketEpoch, sector.SeedValue, sector.SeedEpoch, sector.pieceInfos(), sector.CommR, sector.CommD) diff --git a/extern/storage-sealing/stats.go b/extern/storage-sealing/stats.go new file mode 100644 index 000000000..871c962c1 --- /dev/null +++ b/extern/storage-sealing/stats.go @@ -0,0 +1,45 @@ +package sealing + +import ( + "sync" + + "github.com/filecoin-project/specs-actors/actors/abi" +) + +type statSectorState int + +const ( + sstSealing statSectorState = iota + sstFailed + sstProving + nsst +) + +type SectorStats struct { + lk sync.Mutex + + bySector map[abi.SectorID]statSectorState + totals [nsst]uint64 +} + +func (ss *SectorStats) updateSector(id abi.SectorID, st SectorState) { + ss.lk.Lock() + defer ss.lk.Unlock() + + oldst, found := ss.bySector[id] + if found { + ss.totals[oldst]-- + } + + sst := toStatState(st) + ss.bySector[id] = sst + ss.totals[sst]++ +} + +// return the number of sectors currently in the sealing pipeline +func (ss *SectorStats) curSealing() uint64 { + ss.lk.Lock() + defer ss.lk.Unlock() + + return ss.totals[sstSealing] + ss.totals[sstFailed] +} diff --git a/extern/storage-sealing/types.go b/extern/storage-sealing/types.go index 45993bb82..a9c2a7203 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" @@ -14,6 +12,7 @@ import ( "github.com/filecoin-project/specs-storage/storage" sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage" + "github.com/filecoin-project/lotus/extern/storage-sealing/sealiface" ) // Piece is a tuple of piece and deal info @@ -188,7 +187,7 @@ type MessageReceipt struct { GasUsed int64 } -type GetSealingDelayFunc func() (time.Duration, error) +type GetSealingConfigFunc func() (sealiface.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..d10810998 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,19 @@ type DealmakingConfig struct { Filter string } +type SealingConfig struct { + // 0 = no limit + MaxWaitDealsSectors uint64 + + // includes failed, 0 = no limit + MaxSealingSectors uint64 + + // includes failed, 0 = no limit + MaxSealingSectorsForDeals uint64 + + WaitDealsDelay Duration +} + type MinerFeeConfig struct { MaxPreCommitGasFee types.FIL MaxCommitGasFee types.FIL @@ -131,6 +143,13 @@ func DefaultStorageMiner() *StorageMiner { cfg := &StorageMiner{ Common: defCommon(), + Sealing: SealingConfig{ + MaxWaitDealsSectors: 2, // 64G with 32G sectors + MaxSealingSectors: 0, + MaxSealingSectorsForDeals: 0, + WaitDealsDelay: Duration(time.Hour), + }, + Storage: sectorstorage.SealerConfig{ AllowAddPiece: true, AllowPreCommit1: true, @@ -158,8 +177,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..d559a2de1 100644 --- a/node/modules/dtypes/miner.go +++ b/node/modules/dtypes/miner.go @@ -9,6 +9,8 @@ import ( "github.com/filecoin-project/go-address" "github.com/filecoin-project/go-fil-markets/storagemarket" "github.com/filecoin-project/specs-actors/actors/abi" + + "github.com/filecoin-project/lotus/extern/storage-sealing/sealiface" ) type MinerAddress address.Address @@ -56,10 +58,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(sealiface.Config) error // GetSealingDelay returns how long a sector waits for more deals before sealing begins. -type GetSealingDelayFunc func() (time.Duration, error) +type GetSealingConfigFunc func() (sealiface.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..ce2427f2c 100644 --- a/node/modules/storageminer.go +++ b/node/modules/storageminer.go @@ -48,6 +48,7 @@ import ( "github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper" "github.com/filecoin-project/lotus/extern/sector-storage/stores" sealing "github.com/filecoin-project/lotus/extern/storage-sealing" + "github.com/filecoin-project/lotus/extern/storage-sealing/sealiface" lapi "github.com/filecoin-project/lotus/api" "github.com/filecoin-project/lotus/build" @@ -141,8 +142,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 +594,28 @@ 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 sealiface.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 sealiface.Config, err error) { err = readCfg(r, func(cfg *config.StorageMiner) { - out = time.Duration(cfg.SealingDelay) + out = sealiface.Config{ + MaxWaitDealsSectors: cfg.Sealing.MaxWaitDealsSectors, + MaxSealingSectors: cfg.Sealing.MaxSealingSectors, + MaxSealingSectorsForDeals: cfg.Sealing.MaxSealingSectorsForDeals, + WaitDealsDelay: time.Duration(cfg.Sealing.WaitDealsDelay), + } }) return }, nil diff --git a/storage/adapter_storage_miner.go b/storage/adapter_storage_miner.go index 8881e599e..1890a369f 100644 --- a/storage/adapter_storage_miner.go +++ b/storage/adapter_storage_miner.go @@ -108,6 +108,27 @@ func (s SealingAPIAdapter) StateWaitMsg(ctx context.Context, mcid cid.Cid) (seal }, nil } +func (s SealingAPIAdapter) StateSearchMsg(ctx context.Context, c cid.Cid) (*sealing.MsgLookup, error) { + wmsg, err := s.delegate.StateSearchMsg(ctx, c) + if err != nil { + return nil, err + } + + if wmsg == nil { + return nil, nil + } + + return &sealing.MsgLookup{ + Receipt: sealing.MessageReceipt{ + ExitCode: wmsg.Receipt.ExitCode, + Return: wmsg.Receipt.Return, + GasUsed: wmsg.Receipt.GasUsed, + }, + TipSetTok: wmsg.TipSet.Bytes(), + Height: wmsg.Height, + }, nil +} + func (s SealingAPIAdapter) StateComputeDataCommitment(ctx context.Context, maddr address.Address, sectorType abi.RegisteredSealProof, deals []abi.DealID, tok sealing.TipSetToken) (cid.Cid, error) { tsk, err := types.TipSetKeyFromBytes(tok) if err != nil { @@ -186,7 +207,7 @@ func (s SealingAPIAdapter) StateSectorPreCommitInfo(ctx context.Context, maddr a return nil, xerrors.Errorf("checking if sector is allocated: %w", err) } if set { - return nil, xerrors.Errorf("sectorNumber is allocated") + return nil, sealing.ErrSectorAllocated } return nil, nil diff --git a/storage/miner.go b/storage/miner.go index f8a6691f4..7baffee30 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 { @@ -60,6 +60,7 @@ type storageMinerApi interface { StateMinerProvingDeadline(context.Context, address.Address, types.TipSetKey) (*miner.DeadlineInfo, error) StateMinerPreCommitDepositForPower(context.Context, address.Address, miner.SectorPreCommitInfo, types.TipSetKey) (types.BigInt, error) StateMinerInitialPledgeCollateral(context.Context, address.Address, miner.SectorPreCommitInfo, types.TipSetKey) (types.BigInt, error) + StateSearchMsg(context.Context, cid.Cid) (*api.MsgLookup, error) StateWaitMsg(ctx context.Context, cid cid.Cid, confidence uint64) (*api.MsgLookup, error) // TODO: removeme eventually StateGetActor(ctx context.Context, actor address.Address, ts types.TipSetKey) (*types.Actor, error) StateGetReceipt(context.Context, cid.Cid, types.TipSetKey) (*types.MessageReceipt, error) @@ -84,7 +85,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 +95,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 +121,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