lotus/storage/pipeline/input.go

814 lines
24 KiB
Go
Raw Normal View History

2021-01-18 13:26:03 +00:00
package sealing
import (
"context"
"sort"
2021-01-18 20:59:34 +00:00
"time"
2021-01-18 13:26:03 +00:00
2022-04-12 00:45:13 +00:00
"github.com/ipfs/go-cid"
"golang.org/x/xerrors"
2021-01-18 13:26:03 +00:00
"github.com/filecoin-project/go-commp-utils/zerocomm"
2022-06-14 15:00:51 +00:00
"github.com/filecoin-project/go-padreader"
2021-01-18 13:26:03 +00:00
"github.com/filecoin-project/go-state-types/abi"
2022-03-16 16:33:05 +00:00
"github.com/filecoin-project/go-state-types/big"
2022-06-14 15:00:51 +00:00
"github.com/filecoin-project/go-statemachine"
2021-01-18 13:26:03 +00:00
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
2022-03-16 16:33:05 +00:00
"github.com/filecoin-project/lotus/chain/actors/policy"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/storage/pipeline/lib/nullreader"
"github.com/filecoin-project/lotus/storage/pipeline/sealiface"
"github.com/filecoin-project/lotus/storage/sealer"
"github.com/filecoin-project/lotus/storage/sealer/ffiwrapper"
"github.com/filecoin-project/lotus/storage/sealer/storiface"
"github.com/filecoin-project/lotus/storage/sectorblocks"
2021-01-18 13:26:03 +00:00
)
func (m *Sealing) handleWaitDeals(ctx statemachine.Context, sector SectorInfo) error {
var used abi.UnpaddedPieceSize
for _, piece := range sector.Pieces {
used += piece.Piece.Size.Unpadded()
}
2021-01-18 13:26:03 +00:00
m.inputLk.Lock()
2021-01-18 20:59:34 +00:00
if m.nextDealSector != nil && *m.nextDealSector == sector.SectorNumber {
m.nextDealSector = nil
}
sid := m.minerSectorID(sector.SectorNumber)
if len(m.assignedPieces[sid]) > 0 {
m.inputLk.Unlock()
// got assigned more pieces in the AddPiece state
return ctx.Send(SectorAddPiece{})
}
started, err := m.maybeStartSealing(ctx, sector, used)
if err != nil || started {
delete(m.openSectors, m.minerSectorID(sector.SectorNumber))
m.inputLk.Unlock()
return err
}
if _, has := m.openSectors[sid]; !has {
m.openSectors[sid] = &openSector{
used: used,
maybeAccept: func(cid cid.Cid) error {
// todo check deal start deadline (configurable)
m.assignedPieces[sid] = append(m.assignedPieces[sid], cid)
return ctx.Send(SectorAddPiece{})
},
number: sector.SectorNumber,
ccUpdate: sector.CCUpdate,
}
} else {
// make sure we're only accounting for pieces which were correctly added
// (note that m.assignedPieces[sid] will always be empty here)
m.openSectors[sid].used = used
}
go func() {
defer m.inputLk.Unlock()
if err := m.updateInput(ctx.Context(), sector.SectorType); err != nil {
log.Errorf("%+v", err)
}
}()
return nil
}
func (m *Sealing) maybeStartSealing(ctx statemachine.Context, sector SectorInfo, used abi.UnpaddedPieceSize) (bool, error) {
2021-01-18 20:59:34 +00:00
now := time.Now()
st := m.sectorTimers[m.minerSectorID(sector.SectorNumber)]
if st != nil {
if !st.Stop() { // timer expired, SectorStartPacking was/is being sent
// we send another SectorStartPacking in case one was sent in the handleAddPiece state
log.Infow("starting to seal deal sector", "sector", sector.SectorNumber, "trigger", "wait-timeout")
return true, ctx.Send(SectorStartPacking{})
2021-01-18 20:59:34 +00:00
}
}
ssize, err := sector.SectorType.SectorSize()
if err != nil {
return false, xerrors.Errorf("getting sector size")
}
maxDeals, err := getDealPerSectorLimit(ssize)
if err != nil {
return false, xerrors.Errorf("getting per-sector deal limit: %w", err)
}
if len(sector.dealIDs()) >= maxDeals {
// can't accept more deals
log.Infow("starting to seal deal sector", "sector", sector.SectorNumber, "trigger", "maxdeals")
return true, ctx.Send(SectorStartPacking{})
}
if used.Padded() == abi.PaddedPieceSize(ssize) {
// sector full
log.Infow("starting to seal deal sector", "sector", sector.SectorNumber, "trigger", "filled")
return true, ctx.Send(SectorStartPacking{})
}
if sector.CreationTime != 0 {
2021-01-18 20:59:34 +00:00
cfg, err := m.getConfig()
if err != nil {
return false, xerrors.Errorf("getting storage config: %w", err)
2021-01-18 20:59:34 +00:00
}
sealTime := time.Unix(sector.CreationTime, 0).Add(cfg.WaitDealsDelay)
// check deal age, start sealing when the deal closest to starting is within slack time
ts, err := m.Api.ChainHead(ctx.Context())
blockTime := time.Second * time.Duration(build.BlockDelaySecs)
if err != nil {
return false, xerrors.Errorf("API error getting head: %w", err)
}
for _, piece := range sector.Pieces {
2022-02-08 17:39:18 +00:00
if piece.DealInfo == nil {
continue
}
dealSafeSealEpoch := piece.DealInfo.DealProposal.StartEpoch - cfg.StartEpochSealingBuffer
dealSafeSealTime := time.Now().Add(time.Duration(dealSafeSealEpoch-ts.Height()) * blockTime)
2022-02-08 17:39:18 +00:00
if dealSafeSealTime.Before(sealTime) {
sealTime = dealSafeSealTime
}
}
2021-01-18 20:59:34 +00:00
if now.After(sealTime) {
log.Infow("starting to seal deal sector", "sector", sector.SectorNumber, "trigger", "wait-timeout")
return true, ctx.Send(SectorStartPacking{})
2021-01-18 20:59:34 +00:00
}
2021-01-19 18:04:05 +00:00
m.sectorTimers[m.minerSectorID(sector.SectorNumber)] = time.AfterFunc(sealTime.Sub(now), func() {
log.Infow("starting to seal deal sector", "sector", sector.SectorNumber, "trigger", "wait-timer")
2021-01-19 18:04:05 +00:00
if err := ctx.Send(SectorStartPacking{}); err != nil {
log.Errorw("sending SectorStartPacking event failed", "sector", sector.SectorNumber, "error", err)
}
})
2021-01-18 20:59:34 +00:00
}
return false, nil
2021-01-18 13:26:03 +00:00
}
func (m *Sealing) handleAddPiece(ctx statemachine.Context, sector SectorInfo) error {
ssize, err := sector.SectorType.SectorSize()
if err != nil {
return err
}
res := SectorPieceAdded{}
2021-01-18 13:26:03 +00:00
m.inputLk.Lock()
pending, ok := m.assignedPieces[m.minerSectorID(sector.SectorNumber)]
if ok {
delete(m.assignedPieces, m.minerSectorID(sector.SectorNumber))
}
m.inputLk.Unlock()
if !ok {
// nothing to do here (might happen after a restart in AddPiece)
return ctx.Send(res)
}
2021-01-18 13:26:03 +00:00
2021-01-18 20:59:34 +00:00
var offset abi.UnpaddedPieceSize
pieceSizes := make([]abi.UnpaddedPieceSize, len(sector.Pieces))
for i, p := range sector.Pieces {
pieceSizes[i] = p.Piece.Size.Unpadded()
offset += p.Piece.Size.Unpadded()
}
maxDeals, err := getDealPerSectorLimit(ssize)
if err != nil {
return xerrors.Errorf("getting per-sector deal limit: %w", err)
}
for i, piece := range pending {
2021-01-18 13:26:03 +00:00
m.inputLk.Lock()
deal, ok := m.pendingPieces[piece]
m.inputLk.Unlock()
if !ok {
return xerrors.Errorf("piece %s assigned to sector %d not found", piece, sector.SectorNumber)
}
if len(sector.dealIDs())+(i+1) > maxDeals {
// todo: this is rather unlikely to happen, but in case it does, return the deal to waiting queue instead of failing it
deal.accepted(sector.SectorNumber, offset, xerrors.Errorf("too many deals assigned to sector %d, dropping deal", sector.SectorNumber))
continue
}
2021-01-18 20:59:34 +00:00
pads, padLength := ffiwrapper.GetRequiredPadding(offset.Padded(), deal.size.Padded())
2021-01-18 13:26:03 +00:00
2021-01-18 20:59:34 +00:00
if offset.Padded()+padLength+deal.size.Padded() > abi.PaddedPieceSize(ssize) {
// todo: this is rather unlikely to happen, but in case it does, return the deal to waiting queue instead of failing it
deal.accepted(sector.SectorNumber, offset, xerrors.Errorf("piece %s assigned to sector %d with not enough space", piece, sector.SectorNumber))
continue
2021-01-18 13:26:03 +00:00
}
2021-01-18 20:59:34 +00:00
offset += padLength.Unpadded()
2021-01-18 13:26:03 +00:00
for _, p := range pads {
expectCid := zerocomm.ZeroPieceCommitment(p.Unpadded())
ppi, err := m.sealer.AddPiece(sealer.WithPriority(ctx.Context(), DealSectorPriority),
2021-01-18 13:26:03 +00:00
m.minerSector(sector.SectorType, sector.SectorNumber),
pieceSizes,
p.Unpadded(),
2022-04-12 15:45:34 +00:00
nullreader.NewNullReader(p.Unpadded()))
2021-01-18 13:26:03 +00:00
if err != nil {
err = xerrors.Errorf("writing padding piece: %w", err)
deal.accepted(sector.SectorNumber, offset, err)
return ctx.Send(SectorAddPieceFailed{err})
2021-01-18 13:26:03 +00:00
}
if !ppi.PieceCID.Equals(expectCid) {
err = xerrors.Errorf("got unexpected padding piece CID: expected:%s, got:%s", expectCid, ppi.PieceCID)
deal.accepted(sector.SectorNumber, offset, err)
return ctx.Send(SectorAddPieceFailed{err})
}
2021-01-18 13:26:03 +00:00
pieceSizes = append(pieceSizes, p.Unpadded())
res.NewPieces = append(res.NewPieces, Piece{
Piece: ppi,
})
}
ppi, err := m.sealer.AddPiece(sealer.WithPriority(ctx.Context(), DealSectorPriority),
2021-01-18 13:26:03 +00:00
m.minerSector(sector.SectorType, sector.SectorNumber),
pieceSizes,
deal.size,
deal.data)
if err != nil {
err = xerrors.Errorf("writing piece: %w", err)
deal.accepted(sector.SectorNumber, offset, err)
return ctx.Send(SectorAddPieceFailed{err})
2021-01-18 13:26:03 +00:00
}
if !ppi.PieceCID.Equals(deal.deal.DealProposal.PieceCID) {
err = xerrors.Errorf("got unexpected piece CID: expected:%s, got:%s", deal.deal.DealProposal.PieceCID, ppi.PieceCID)
deal.accepted(sector.SectorNumber, offset, err)
return ctx.Send(SectorAddPieceFailed{err})
}
2021-01-18 13:26:03 +00:00
log.Infow("deal added to a sector", "deal", deal.deal.DealID, "sector", sector.SectorNumber, "piece", ppi.PieceCID)
2021-01-18 20:59:34 +00:00
deal.accepted(sector.SectorNumber, offset, nil)
offset += deal.size
2021-01-18 13:26:03 +00:00
pieceSizes = append(pieceSizes, deal.size)
2021-01-18 20:59:34 +00:00
2021-01-18 13:26:03 +00:00
res.NewPieces = append(res.NewPieces, Piece{
2021-01-18 20:59:34 +00:00
Piece: ppi,
2021-01-18 13:26:03 +00:00
DealInfo: &deal.deal,
})
}
return ctx.Send(res)
}
func (m *Sealing) handleAddPieceFailed(ctx statemachine.Context, sector SectorInfo) error {
return ctx.Send(SectorRetryWaitDeals{})
}
func (m *Sealing) SectorAddPieceToAny(ctx context.Context, size abi.UnpaddedPieceSize, data storiface.Data, deal api.PieceDealInfo) (api.SectorOffset, error) {
2021-01-18 13:26:03 +00:00
log.Infof("Adding piece for deal %d (publish msg: %s)", deal.DealID, deal.PublishCid)
if (padreader.PaddedSize(uint64(size))) != size {
return api.SectorOffset{}, xerrors.Errorf("cannot allocate unpadded piece")
2021-01-18 13:26:03 +00:00
}
sp, err := m.currentSealProof(ctx)
if err != nil {
return api.SectorOffset{}, xerrors.Errorf("getting current seal proof type: %w", err)
2021-01-18 13:26:03 +00:00
}
ssize, err := sp.SectorSize()
if err != nil {
return api.SectorOffset{}, err
2021-01-18 13:26:03 +00:00
}
if size > abi.PaddedPieceSize(ssize).Unpadded() {
return api.SectorOffset{}, xerrors.Errorf("piece cannot fit into a sector")
2021-01-18 13:26:03 +00:00
}
if _, err := deal.DealProposal.Cid(); err != nil {
return api.SectorOffset{}, xerrors.Errorf("getting proposal CID: %w", err)
2021-01-18 13:26:03 +00:00
}
cfg, err := m.getConfig()
if err != nil {
return api.SectorOffset{}, xerrors.Errorf("getting config: %w", err)
}
ts, err := m.Api.ChainHead(ctx)
if err != nil {
return api.SectorOffset{}, xerrors.Errorf("couldnt get chain head: %w", err)
}
if ts.Height()+cfg.StartEpochSealingBuffer > deal.DealProposal.StartEpoch {
return api.SectorOffset{}, xerrors.Errorf(
"cannot add piece for deal with piece CID %s: current epoch %d has passed deal proposal start epoch %d",
deal.DealProposal.PieceCID, ts.Height(), deal.DealProposal.StartEpoch)
}
2021-01-18 13:26:03 +00:00
m.inputLk.Lock()
2022-02-21 07:51:25 +00:00
if pp, exist := m.pendingPieces[proposalCID(deal)]; exist {
2021-01-18 13:26:03 +00:00
m.inputLk.Unlock()
2022-02-22 10:22:51 +00:00
// we already have a pre-existing add piece call for this deal, let's wait for it to finish and see if it's successful
res, err := waitAddPieceResp(ctx, pp)
if err != nil {
return api.SectorOffset{}, err
}
if res.err == nil {
2022-02-22 10:22:51 +00:00
// all good, return the response
2022-02-21 07:51:25 +00:00
return api.SectorOffset{Sector: res.sn, Offset: res.offset.Padded()}, res.err
}
// if there was an error waiting for a pre-existing add piece call, let's retry
m.inputLk.Lock()
2021-01-18 13:26:03 +00:00
}
// addPendingPiece takes over m.inputLk
2022-02-22 10:22:51 +00:00
pp := m.addPendingPiece(ctx, size, data, deal, sp)
2022-02-22 10:22:51 +00:00
res, err := waitAddPieceResp(ctx, pp)
if err != nil {
return api.SectorOffset{}, err
}
return api.SectorOffset{Sector: res.sn, Offset: res.offset.Padded()}, res.err
}
// called with m.inputLk; transfers the lock to another goroutine!
func (m *Sealing) addPendingPiece(ctx context.Context, size abi.UnpaddedPieceSize, data storiface.Data, deal api.PieceDealInfo, sp abi.RegisteredSealProof) *pendingPiece {
2022-02-22 09:48:39 +00:00
doneCh := make(chan struct{})
2022-02-21 09:51:43 +00:00
pp := &pendingPiece{
doneCh: doneCh,
2021-01-18 13:26:03 +00:00
size: size,
deal: deal,
data: data,
assigned: false,
2022-02-21 09:51:43 +00:00
}
pp.accepted = func(sn abi.SectorNumber, offset abi.UnpaddedPieceSize, err error) {
2022-02-21 14:27:51 +00:00
pp.resp = &pieceAcceptResp{sn, offset, err}
2022-02-21 09:51:43 +00:00
close(pp.doneCh)
2021-01-18 13:26:03 +00:00
}
2022-02-21 09:51:43 +00:00
m.pendingPieces[proposalCID(deal)] = pp
2021-01-18 13:26:03 +00:00
go func() {
defer m.inputLk.Unlock()
2021-01-18 13:26:03 +00:00
if err := m.updateInput(ctx, sp); err != nil {
log.Errorf("%+v", err)
}
}()
2022-02-22 10:22:51 +00:00
return pp
}
func waitAddPieceResp(ctx context.Context, pp *pendingPiece) (*pieceAcceptResp, error) {
2022-02-21 07:56:49 +00:00
select {
2022-02-22 10:22:51 +00:00
case <-pp.doneCh:
2022-02-21 14:27:51 +00:00
res := pp.resp
2022-02-22 10:22:51 +00:00
return res, nil
2022-02-21 07:56:49 +00:00
case <-ctx.Done():
2022-02-22 10:22:51 +00:00
return nil, ctx.Err()
2022-02-21 07:56:49 +00:00
}
2021-01-18 13:26:03 +00:00
}
func (m *Sealing) SectorMatchPendingPiecesToOpenSectors(ctx context.Context) error {
sp, err := m.currentSealProof(ctx)
if err != nil {
return xerrors.Errorf("failed to get current seal proof: %w", err)
}
log.Debug("pieces to sector matching waiting for lock")
m.inputLk.Lock()
defer m.inputLk.Unlock()
return m.updateInput(ctx, sp)
}
2022-03-16 16:33:05 +00:00
type expFn func(sn abi.SectorNumber) (abi.ChainEpoch, abi.TokenAmount, error)
2021-01-18 13:26:03 +00:00
// called with m.inputLk
func (m *Sealing) updateInput(ctx context.Context, sp abi.RegisteredSealProof) error {
2022-03-16 16:33:05 +00:00
memo := make(map[abi.SectorNumber]struct {
e abi.ChainEpoch
p abi.TokenAmount
})
expF := func(sn abi.SectorNumber) (abi.ChainEpoch, abi.TokenAmount, error) {
if e, ok := memo[sn]; ok {
return e.e, e.p, nil
}
onChainInfo, err := m.Api.StateSectorGetInfo(ctx, m.maddr, sn, types.TipSetKey{})
2022-03-16 16:33:05 +00:00
if err != nil {
return 0, big.Zero(), err
}
memo[sn] = struct {
e abi.ChainEpoch
p abi.TokenAmount
}{e: onChainInfo.Expiration, p: onChainInfo.InitialPledge}
return onChainInfo.Expiration, onChainInfo.InitialPledge, nil
}
2021-01-18 13:26:03 +00:00
ssize, err := sp.SectorSize()
if err != nil {
return err
}
type match struct {
sector abi.SectorID
deal cid.Cid
size abi.UnpaddedPieceSize
padding abi.UnpaddedPieceSize
}
var matches []match
toAssign := map[cid.Cid]struct{}{} // used to maybe create new sectors
// todo: this is distinctly O(n^2), may need to be optimized for tiny deals and large scale miners
// (unlikely to be a problem now)
for proposalCid, piece := range m.pendingPieces {
2021-01-18 20:59:34 +00:00
if piece.assigned {
continue // already assigned to a sector, skip
}
2021-01-18 13:26:03 +00:00
toAssign[proposalCid] = struct{}{}
2021-01-18 13:26:03 +00:00
2021-01-18 20:59:34 +00:00
for id, sector := range m.openSectors {
avail := abi.PaddedPieceSize(ssize).Unpadded() - sector.used
// check that sector lifetime is long enough to fit deal using latest expiration from on chain
ok, err := sector.dealFitsInLifetime(piece.deal.DealProposal.EndEpoch, expF)
if err != nil {
log.Errorf("failed to check expiration for cc Update sector %d", sector.number)
continue
}
if !ok {
2022-03-16 16:33:05 +00:00
exp, _, _ := expF(sector.number)
log.Debugf("CC update sector %d cannot fit deal, expiration %d before deal end epoch %d", id, exp, piece.deal.DealProposal.EndEpoch)
continue
}
2021-01-18 13:26:03 +00:00
if piece.size <= avail { // (note: if we have enough space for the piece, we also have enough space for inter-piece padding)
matches = append(matches, match{
sector: id,
deal: proposalCid,
2021-01-18 13:26:03 +00:00
size: piece.size,
padding: avail % piece.size,
})
}
}
}
sort.Slice(matches, func(i, j int) bool {
if matches[i].padding != matches[j].padding { // less padding is better
return matches[i].padding < matches[j].padding
}
if matches[i].size != matches[j].size { // larger pieces are better
return matches[i].size < matches[j].size
}
return matches[i].sector.Number < matches[j].sector.Number // prefer older sectors
})
log.Debugw("updateInput matching", "matches", len(matches), "toAssign", len(toAssign), "openSectors", len(m.openSectors), "pieces", len(m.pendingPieces))
2021-01-18 13:26:03 +00:00
var assigned int
for _, mt := range matches {
if m.pendingPieces[mt.deal].assigned {
assigned++
continue
}
if _, found := m.openSectors[mt.sector]; !found {
continue
}
avail := abi.PaddedPieceSize(ssize).Unpadded() - m.openSectors[mt.sector].used
if mt.size > avail {
continue
}
2021-01-18 13:26:03 +00:00
err := m.openSectors[mt.sector].maybeAccept(mt.deal)
if err != nil {
m.pendingPieces[mt.deal].accepted(mt.sector.Number, 0, err) // non-error case in handleAddPiece
}
m.openSectors[mt.sector].used += mt.padding + mt.size
2021-01-18 13:26:03 +00:00
m.pendingPieces[mt.deal].assigned = true
delete(toAssign, mt.deal)
if err != nil {
log.Errorf("sector %d rejected deal %s: %+v", mt.sector, mt.deal, err)
continue
}
}
log.Debugw("updateInput matching done", "matches", len(matches), "toAssign", len(toAssign), "assigned", assigned, "openSectors", len(m.openSectors), "pieces", len(m.pendingPieces))
2021-01-18 13:26:03 +00:00
if len(toAssign) > 0 {
log.Errorf("we are trying to create a new sector with open sectors %v", m.openSectors)
2022-03-16 16:33:05 +00:00
if err := m.tryGetDealSector(ctx, sp, expF); err != nil {
2021-01-18 20:59:34 +00:00
log.Errorw("Failed to create a new sector for deals", "error", err)
}
2021-01-18 13:26:03 +00:00
}
return nil
}
func (m *Sealing) calcTargetExpiration(ctx context.Context, ssize abi.SectorSize) (minTarget, target abi.ChainEpoch, err error) {
2022-03-16 16:33:05 +00:00
var candidates []*pendingPiece
for _, piece := range m.pendingPieces {
if piece.assigned {
continue // already assigned to a sector, skip
}
candidates = append(candidates, piece)
}
// earliest expiration first
sort.Slice(candidates, func(i, j int) bool {
return candidates[i].deal.DealProposal.EndEpoch < candidates[j].deal.DealProposal.EndEpoch
2022-03-16 16:33:05 +00:00
})
var totalBytes uint64
for _, candidate := range candidates {
totalBytes += uint64(candidate.size)
if totalBytes >= uint64(abi.PaddedPieceSize(ssize).Unpadded()) {
return candidates[0].deal.DealProposal.EndEpoch, candidate.deal.DealProposal.EndEpoch, nil
2022-03-16 16:33:05 +00:00
}
}
ts, err := m.Api.ChainHead(ctx)
2022-03-16 16:33:05 +00:00
if err != nil {
return 0, 0, xerrors.Errorf("getting current epoch: %w", err)
2022-03-16 16:33:05 +00:00
}
minDur, maxDur := policy.DealDurationBounds(0)
2022-03-16 16:33:05 +00:00
return ts.Height() + minDur, ts.Height() + maxDur, nil
2022-03-16 16:33:05 +00:00
}
2022-03-17 20:12:42 +00:00
func (m *Sealing) maybeUpgradeSector(ctx context.Context, sp abi.RegisteredSealProof, ef expFn) (bool, error) {
2022-03-16 16:33:05 +00:00
if len(m.available) == 0 {
return false, nil
}
2022-03-16 19:04:48 +00:00
ssize, err := sp.SectorSize()
if err != nil {
return false, xerrors.Errorf("getting sector size: %w", err)
}
minExpiration, targetExpiration, err := m.calcTargetExpiration(ctx, ssize)
if err != nil {
return false, xerrors.Errorf("calculating min target expiration: %w", err)
}
2022-03-16 16:33:05 +00:00
var candidate abi.SectorID
var bestExpiration abi.ChainEpoch
bestPledge := types.TotalFilecoinInt
for s := range m.available {
expiration, pledge, err := ef(s.Number)
if err != nil {
log.Errorw("checking sector expiration", "error", err)
continue
}
2022-03-16 19:28:15 +00:00
slowChecks := func(sid abi.SectorNumber) bool {
active, err := m.sectorActive(ctx, types.TipSetKey{}, sid)
2022-03-16 19:09:18 +00:00
if err != nil {
log.Errorw("checking sector active", "error", err)
return false
}
if !active {
log.Debugw("skipping available sector", "reason", "not active")
return false
}
return true
}
2022-03-16 16:33:05 +00:00
// if best is below target, we want larger expirations
// if best is above target, we want lower pledge, but only if still above target
if bestExpiration < targetExpiration {
2022-03-16 19:28:15 +00:00
if expiration > bestExpiration && slowChecks(s.Number) {
2022-03-16 16:33:05 +00:00
bestExpiration = expiration
bestPledge = pledge
candidate = s
}
continue
}
2022-03-16 19:28:15 +00:00
if expiration >= targetExpiration && pledge.LessThan(bestPledge) && slowChecks(s.Number) {
2022-03-16 16:33:05 +00:00
bestExpiration = expiration
bestPledge = pledge
candidate = s
}
}
if bestExpiration < minExpiration {
log.Infow("Not upgrading any sectors", "available", len(m.available), "pieces", len(m.pendingPieces), "bestExp", bestExpiration, "target", targetExpiration, "min", minExpiration, "candidate", candidate)
2022-03-16 19:04:48 +00:00
// didn't find a good sector / no sectors were available
2022-03-16 16:33:05 +00:00
return false, nil
}
log.Infow("Upgrading sector", "number", candidate.Number, "type", "deal", "proofType", sp, "expiration", bestExpiration, "pledge", types.FIL(bestPledge))
delete(m.available, candidate)
m.nextDealSector = &candidate.Number
2022-03-16 16:33:05 +00:00
return true, m.sectors.Send(uint64(candidate.Number), SectorStartCCUpdate{})
}
2022-03-17 20:12:42 +00:00
// call with m.inputLk
func (m *Sealing) createSector(ctx context.Context, cfg sealiface.Config, sp abi.RegisteredSealProof) (abi.SectorNumber, error) {
sid, err := m.sc.Next()
if err != nil {
return 0, xerrors.Errorf("getting sector number: %w", err)
}
err = m.sealer.NewSector(ctx, m.minerSector(sp, sid))
if err != nil {
return 0, xerrors.Errorf("initializing sector: %w", err)
}
// update stats early, fsm planner would do that async
m.stats.updateSector(ctx, cfg, m.minerSectorID(sid), UndefinedSectorState)
return sid, err
}
2022-03-16 16:33:05 +00:00
func (m *Sealing) tryGetDealSector(ctx context.Context, sp abi.RegisteredSealProof, ef expFn) error {
2021-06-15 19:04:11 +00:00
m.startupWait.Wait()
2021-06-15 19:04:11 +00:00
if m.nextDealSector != nil {
return nil // new sector is being created right now
}
2021-01-18 13:26:03 +00:00
cfg, err := m.getConfig()
if err != nil {
return xerrors.Errorf("getting storage config: %w", err)
}
2022-03-17 20:12:42 +00:00
// if we're above WaitDeals limit, we don't want to add more staging sectors
2022-03-16 16:33:05 +00:00
if cfg.MaxWaitDealsSectors > 0 && m.stats.curStaging() >= cfg.MaxWaitDealsSectors {
2021-01-18 20:59:34 +00:00
return nil
}
2022-03-17 20:12:42 +00:00
maxUpgrading := cfg.MaxSealingSectorsForDeals
if cfg.MaxUpgradingSectors > 0 {
maxUpgrading = cfg.MaxUpgradingSectors
2022-03-16 16:33:05 +00:00
}
2022-03-17 20:12:42 +00:00
canCreate := cfg.MakeNewSectorForDeals && !(cfg.MaxSealingSectorsForDeals > 0 && m.stats.curSealing() >= cfg.MaxSealingSectorsForDeals)
canUpgrade := !(maxUpgrading > 0 && m.stats.curSealing() >= maxUpgrading)
2021-01-18 13:26:03 +00:00
2022-03-17 20:12:42 +00:00
// we want to try to upgrade when:
// - we can upgrade and prefer upgrades
// - we don't prefer upgrades, but can't create a new sector
shouldUpgrade := canUpgrade && (!cfg.PreferNewSectorsForDeals || !canCreate)
2021-03-12 17:42:17 +00:00
2022-03-17 20:12:42 +00:00
log.Infow("new deal sector decision",
"sealing", m.stats.curSealing(),
"maxSeal", cfg.MaxSealingSectorsForDeals,
"maxUpgrade", maxUpgrading,
"preferNew", cfg.PreferNewSectorsForDeals,
"canCreate", canCreate,
"canUpgrade", canUpgrade,
"shouldUpgrade", shouldUpgrade)
2021-01-18 13:26:03 +00:00
2022-03-17 20:12:42 +00:00
if shouldUpgrade {
got, err := m.maybeUpgradeSector(ctx, sp, ef)
if err != nil {
return err
}
if got {
return nil
}
2021-01-18 13:26:03 +00:00
}
2022-03-17 20:12:42 +00:00
if canCreate {
sid, err := m.createSector(ctx, cfg, sp)
if err != nil {
return err
}
m.nextDealSector = &sid
2022-03-17 20:12:42 +00:00
log.Infow("Creating sector", "number", sid, "type", "deal", "proofType", sp)
if err := m.sectors.Send(uint64(sid), SectorStart{
ID: sid,
SectorType: sp,
}); err != nil {
return err
}
}
return nil
2021-01-18 13:26:03 +00:00
}
2021-01-18 20:59:34 +00:00
func (m *Sealing) StartPackingSector(sid abi.SectorNumber) error {
2021-06-15 19:04:11 +00:00
m.startupWait.Wait()
2021-05-30 13:13:38 +00:00
log.Infow("starting to seal deal sector", "sector", sid, "trigger", "user")
2021-01-18 20:59:34 +00:00
return m.sectors.Send(uint64(sid), SectorStartPacking{})
}
func (m *Sealing) SectorAbortUpgrade(sid abi.SectorNumber) error {
m.startupWait.Wait()
2022-03-16 19:04:48 +00:00
m.inputLk.Lock()
// always do this early
delete(m.available, m.minerSectorID(sid))
m.inputLk.Unlock()
log.Infow("aborting upgrade of sector", "sector", sid, "trigger", "user")
return m.sectors.Send(uint64(sid), SectorAbortUpgrade{xerrors.New("triggered by user")})
}
func (m *Sealing) SectorsStatus(ctx context.Context, sid abi.SectorNumber, showOnChainInfo bool) (api.SectorInfo, error) {
if showOnChainInfo {
return api.SectorInfo{}, xerrors.Errorf("on-chain info not supported")
}
info, err := m.GetSectorInfo(sid)
if err != nil {
return api.SectorInfo{}, err
}
deals := make([]abi.DealID, len(info.Pieces))
pieces := make([]api.SectorPiece, len(info.Pieces))
for i, piece := range info.Pieces {
pieces[i].Piece = piece.Piece
if piece.DealInfo == nil {
continue
}
pdi := *piece.DealInfo // copy
pieces[i].DealInfo = &pdi
deals[i] = piece.DealInfo.DealID
}
log := make([]api.SectorLog, len(info.Log))
for i, l := range info.Log {
log[i] = api.SectorLog{
Kind: l.Kind,
Timestamp: l.Timestamp,
Trace: l.Trace,
Message: l.Message,
}
}
sInfo := api.SectorInfo{
SectorID: sid,
State: api.SectorState(info.State),
CommD: info.CommD,
CommR: info.CommR,
Proof: info.Proof,
Deals: deals,
Pieces: pieces,
Ticket: api.SealTicket{
Value: info.TicketValue,
Epoch: info.TicketEpoch,
},
Seed: api.SealSeed{
Value: info.SeedValue,
Epoch: info.SeedEpoch,
},
PreCommitMsg: info.PreCommitMessage,
CommitMsg: info.CommitMessage,
Retries: info.InvalidProofs,
ToUpgrade: false,
ReplicaUpdateMessage: info.ReplicaUpdateMessage,
LastErr: info.LastErr,
Log: log,
// on chain info
SealProof: info.SectorType,
Activation: 0,
Expiration: 0,
DealWeight: big.Zero(),
VerifiedDealWeight: big.Zero(),
InitialPledge: big.Zero(),
OnTime: 0,
Early: 0,
}
return sInfo, nil
}
func proposalCID(deal api.PieceDealInfo) cid.Cid {
pc, err := deal.DealProposal.Cid()
if err != nil {
log.Errorf("DealProposal.Cid error: %+v", err)
return cid.Undef
}
return pc
}
var _ sectorblocks.SectorBuilder = &Sealing{}