Merge remote-tracking branch 'origin/master' into feat/post-worker

2022-03-25 16:49:46 -04:00
parent fcaad20ecc d437f19aae
commit 82343a06e7
40 changed files with 885 additions and 235 deletions
@@ -166,8 +166,6 @@ func (p *pieceProvider) ReadPiece(ctx context.Context, sector storage.SectorRef,

 	r, err := p.tryReadUnsealedPiece(ctx, unsealed, sector, pieceOffset, size)

-	log.Debugf("result of first tryReadUnsealedPiece: r=%s, err=%s", r, err)
-
 	if xerrors.Is(err, storiface.ErrSectorNotFound) {
 		log.Debugf("no unsealed sector file with unsealed piece, sector=%+v, pieceOffset=%d, size=%d", sector, pieceOffset, size)
 		err = nil
@@ -473,6 +473,8 @@ func (m *Sealing) updateInput(ctx context.Context, sp abi.RegisteredSealProof) e
 		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))
+
 	var assigned int
 	for _, mt := range matches {
 		if m.pendingPieces[mt.deal].assigned {
@@ -506,6 +508,8 @@ func (m *Sealing) updateInput(ctx context.Context, sp abi.RegisteredSealProof) e
 		}
 	}

+	log.Debugw("updateInput matching done", "matches", len(matches), "toAssign", len(toAssign), "assigned", assigned, "openSectors", len(m.openSectors), "pieces", len(m.pendingPieces))
+
 	if len(toAssign) > 0 {
 		log.Errorf("we are trying to create a new sector with open sectors %v", m.openSectors)
 		if err := m.tryGetDealSector(ctx, sp, expF); err != nil {
@@ -550,7 +554,7 @@ func (m *Sealing) calcTargetExpiration(ctx context.Context, ssize abi.SectorSize
 	return curEpoch + minDur, curEpoch + maxDur, nil
 }

-func (m *Sealing) tryGetUpgradeSector(ctx context.Context, sp abi.RegisteredSealProof, ef expFn) (bool, error) {
+func (m *Sealing) maybeUpgradeSector(ctx context.Context, sp abi.RegisteredSealProof, ef expFn) (bool, error) {
 	if len(m.available) == 0 {
 		return false, nil
 	}
@@ -619,56 +623,8 @@ func (m *Sealing) tryGetUpgradeSector(ctx context.Context, sp abi.RegisteredSeal
 	return true, m.sectors.Send(uint64(candidate.Number), SectorStartCCUpdate{})
 }

-func (m *Sealing) tryGetDealSector(ctx context.Context, sp abi.RegisteredSealProof, ef expFn) error {
-	m.startupWait.Wait()
-
-	if m.nextDealSector != nil {
-		return nil // new sector is being created right now
-	}
-
-	cfg, err := m.getConfig()
-	if err != nil {
-		return xerrors.Errorf("getting storage config: %w", err)
-	}
-
-	if cfg.MaxSealingSectorsForDeals > 0 && m.stats.curSealing() >= cfg.MaxSealingSectorsForDeals {
-		return nil
-	}
-
-	if cfg.MaxWaitDealsSectors > 0 && m.stats.curStaging() >= cfg.MaxWaitDealsSectors {
-		return nil
-	}
-
-	got, err := m.tryGetUpgradeSector(ctx, sp, ef)
-	if err != nil {
-		return err
-	}
-	if got {
-		return nil
-	}
-
-	if !cfg.MakeNewSectorForDeals {
-		return nil
-	}
-
-	sid, err := m.createSector(ctx, cfg, sp)
-	if err != nil {
-		return err
-	}
-
-	m.nextDealSector = &sid
-
-	log.Infow("Creating sector", "number", sid, "type", "deal", "proofType", sp)
-	return m.sectors.Send(uint64(sid), SectorStart{
-		ID:         sid,
-		SectorType: sp,
-	})
-}
-
 // call with m.inputLk
 func (m *Sealing) createSector(ctx context.Context, cfg sealiface.Config, sp abi.RegisteredSealProof) (abi.SectorNumber, error) {
-	// Now actually create a new sector
-
 	sid, err := m.sc.Next()
 	if err != nil {
 		return 0, xerrors.Errorf("getting sector number: %w", err)
@@ -682,7 +638,74 @@ func (m *Sealing) createSector(ctx context.Context, cfg sealiface.Config, sp abi
 	// update stats early, fsm planner would do that async
 	m.stats.updateSector(ctx, cfg, m.minerSectorID(sid), UndefinedSectorState)

-	return sid, nil
+	return sid, err
+}
+
+func (m *Sealing) tryGetDealSector(ctx context.Context, sp abi.RegisteredSealProof, ef expFn) error {
+	m.startupWait.Wait()
+
+	if m.nextDealSector != nil {
+		return nil // new sector is being created right now
+	}
+
+	cfg, err := m.getConfig()
+	if err != nil {
+		return xerrors.Errorf("getting storage config: %w", err)
+	}
+
+	// if we're above WaitDeals limit, we don't want to add more staging sectors
+	if cfg.MaxWaitDealsSectors > 0 && m.stats.curStaging() >= cfg.MaxWaitDealsSectors {
+		return nil
+	}
+
+	maxUpgrading := cfg.MaxSealingSectorsForDeals
+	if cfg.MaxUpgradingSectors > 0 {
+		maxUpgrading = cfg.MaxUpgradingSectors
+	}
+
+	canCreate := cfg.MakeNewSectorForDeals && !(cfg.MaxSealingSectorsForDeals > 0 && m.stats.curSealing() >= cfg.MaxSealingSectorsForDeals)
+	canUpgrade := !(maxUpgrading > 0 && m.stats.curSealing() >= maxUpgrading)
+
+	// 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)
+
+	log.Infow("new deal sector decision",
+		"sealing", m.stats.curSealing(),
+		"maxSeal", cfg.MaxSealingSectorsForDeals,
+		"maxUpgrade", maxUpgrading,
+		"preferNew", cfg.PreferNewSectorsForDeals,
+		"canCreate", canCreate,
+		"canUpgrade", canUpgrade,
+		"shouldUpgrade", shouldUpgrade)
+
+	if shouldUpgrade {
+		got, err := m.maybeUpgradeSector(ctx, sp, ef)
+		if err != nil {
+			return err
+		}
+		if got {
+			return nil
+		}
+	}
+
+	if canCreate {
+		sid, err := m.createSector(ctx, cfg, sp)
+		if err != nil {
+			return err
+		}
+		m.nextDealSector = &sid
+
+		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
 }

 func (m *Sealing) StartPacking(sid abi.SectorNumber) error {
@@ -18,6 +18,10 @@ type Config struct {
 	// includes failed, 0 = no limit
 	MaxSealingSectorsForDeals uint64

+	PreferNewSectorsForDeals bool
+
+	MaxUpgradingSectors uint64
+
 	MakeNewSectorForDeals bool

 	MakeCCSectorsAvailable bool
@@ -184,14 +184,14 @@ func (m *Sealing) handleComputeProofFailed(ctx statemachine.Context, sector Sect
 }

 func (m *Sealing) handleSubmitReplicaUpdateFailed(ctx statemachine.Context, sector SectorInfo) error {
+	if err := failedCooldown(ctx, sector); err != nil {
+		return err
+	}
+
 	if sector.ReplicaUpdateMessage != nil {
 		mw, err := m.Api.StateSearchMsg(ctx.Context(), *sector.ReplicaUpdateMessage)
 		if err != nil {
 			// API error
-			if err := failedCooldown(ctx, sector); err != nil {
-				return err
-			}
-
 			return ctx.Send(SectorRetrySubmitReplicaUpdateWait{})
 		}

@@ -248,10 +248,6 @@ func (m *Sealing) handleSubmitReplicaUpdateFailed(ctx statemachine.Context, sect
 		return ctx.Send(SectorAbortUpgrade{})
 	}

-	if err := failedCooldown(ctx, sector); err != nil {
-		return err
-	}
-
 	return ctx.Send(SectorRetrySubmitReplicaUpdate{})
 }

@@ -16,7 +16,7 @@ func (m *Sealing) MarkForSnapUpgrade(ctx context.Context, id abi.SectorNumber) e
 	}

 	if si.State != Proving {
-		return xerrors.Errorf("can't mark sectors not in the 'Proving' state for upgrade")
+		return xerrors.Errorf("unable to snap-up sectors not in the 'Proving' state")
 	}

 	if si.hasDeals() {