sealing: More SnapDeals config knobs

documentation/en/default-lotus-miner-config.toml

@@ -325,18 +325,32 @@
   # env var: LOTUS_SEALING_MAXWAITDEALSSECTORS
   #MaxWaitDealsSectors = 2
 
-  # Upper bound on how many sectors can be sealing at the same time when creating new CC sectors (0 = unlimited)
+  # Upper bound on how many sectors can be sealing+upgrading at the same time when creating new CC sectors (0 = unlimited)
   #
   # type: uint64
   # env var: LOTUS_SEALING_MAXSEALINGSECTORS
   #MaxSealingSectors = 0
 
-  # Upper bound on how many sectors can be sealing at the same time when creating new sectors with deals (0 = unlimited)
+  # Upper bound on how many sectors can be sealing+upgrading at the same time when creating new sectors with deals (0 = unlimited)
   #
   # type: uint64
   # env var: LOTUS_SEALING_MAXSEALINGSECTORSFORDEALS
   #MaxSealingSectorsForDeals = 0
 
+  # This setting combined with MaxUpgradingSectors set to a value higher than MaxSealingSectorsForDeals makes it
+  # possible to use fast sector upgrades to handle high volumes of storage deals, while still using the simple sealing
+  # flow when the volume of storage deals is lower.
+  #
+  # type: bool
+  # env var: LOTUS_SEALING_PREFERNEWSECTORSFORDEALS
+  #PreferNewSectorsForDeals = false
+
+  # Upper bound on how many sectors can be sealing+upgrading at the same time when upgrading CC sectors with deals (0 = MaxSealingSectorsForDeals)
+  #
+  # type: uint64
+  # env var: LOTUS_SEALING_MAXUPGRADINGSECTORS
+  #MaxUpgradingSectors = 0
+
   # CommittedCapacitySectorLifetime is the duration a Committed Capacity (CC) sector will
   # live before it must be extended or converted into sector containing deals before it is
   # terminated. Value must be between 180-540 days inclusive

extern/sector-storage/piece_provider.go

@@ -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

extern/storage-sealing/input.go

@@ -550,7 +550,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 +619,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 +634,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 {

extern/storage-sealing/sealiface/config.go

@@ -18,6 +18,10 @@ type Config struct {
 	// includes failed, 0 = no limit
 	MaxSealingSectorsForDeals uint64
 
+	PreferNewSectorsForDeals bool
+
+	MaxUpgradingSectors uint64
+
 	MakeNewSectorForDeals bool
 
 	MakeCCSectorsAvailable bool

node/config/doc_gen.go

@@ -714,13 +714,27 @@ Note that setting this number too high in relation to deal ingestion rate may re
 			Name: "MaxSealingSectors",
 			Type: "uint64",
 
-			Comment: `Upper bound on how many sectors can be sealing at the same time when creating new CC sectors (0 = unlimited)`,
+			Comment: `Upper bound on how many sectors can be sealing+upgrading at the same time when creating new CC sectors (0 = unlimited)`,
 		},
 		{
 			Name: "MaxSealingSectorsForDeals",
 			Type: "uint64",
 
-			Comment: `Upper bound on how many sectors can be sealing at the same time when creating new sectors with deals (0 = unlimited)`,
+			Comment: `Upper bound on how many sectors can be sealing+upgrading at the same time when creating new sectors with deals (0 = unlimited)`,
+		},
+		{
+			Name: "PreferNewSectorsForDeals",
+			Type: "bool",
+
+			Comment: `This setting combined with MaxUpgradingSectors set to a value higher than MaxSealingSectorsForDeals makes it
+possible to use fast sector upgrades to handle high volumes of storage deals, while still using the simple sealing
+flow when the volume of storage deals is lower.`,
+		},
+		{
+			Name: "MaxUpgradingSectors",
+			Type: "uint64",
+
+			Comment: `Upper bound on how many sectors can be sealing+upgrading at the same time when upgrading CC sectors with deals (0 = MaxSealingSectorsForDeals)`,
 		},
 		{
 			Name: "CommittedCapacitySectorLifetime",

node/config/types.go

@@ -228,12 +228,22 @@ type SealingConfig struct {
 	// 0 = no limit
 	MaxWaitDealsSectors uint64
 
-	// Upper bound on how many sectors can be sealing at the same time when creating new CC sectors (0 = unlimited)
+	// Upper bound on how many sectors can be sealing+upgrading at the same time when creating new CC sectors (0 = unlimited)
 	MaxSealingSectors uint64
 
-	// Upper bound on how many sectors can be sealing at the same time when creating new sectors with deals (0 = unlimited)
+	// Upper bound on how many sectors can be sealing+upgrading at the same time when creating new sectors with deals (0 = unlimited)
 	MaxSealingSectorsForDeals uint64
 
+	// Prefer creating new sectors even if there are sectors Available for upgrading
+	//
+	// This setting combined with MaxUpgradingSectors set to a value higher than MaxSealingSectorsForDeals makes it
+	// possible to use fast sector upgrades to handle high volumes of storage deals, while still using the simple sealing
+	// flow when the volume of storage deals is lower.
+	PreferNewSectorsForDeals bool
+
+	// Upper bound on how many sectors can be sealing+upgrading at the same time when upgrading CC sectors with deals (0 = MaxSealingSectorsForDeals)
+	MaxUpgradingSectors uint64
+
 	// CommittedCapacitySectorLifetime is the duration a Committed Capacity (CC) sector will
 	// live before it must be extended or converted into sector containing deals before it is
 	// terminated. Value must be between 180-540 days inclusive

node/modules/storageminer.go

@@ -916,6 +916,8 @@ func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error
 				MaxWaitDealsSectors:             cfg.MaxWaitDealsSectors,
 				MaxSealingSectors:               cfg.MaxSealingSectors,
 				MaxSealingSectorsForDeals:       cfg.MaxSealingSectorsForDeals,
+				PreferNewSectorsForDeals:        cfg.PreferNewSectorsForDeals,
+				MaxUpgradingSectors:             cfg.MaxUpgradingSectors,
 				CommittedCapacitySectorLifetime: config.Duration(cfg.CommittedCapacitySectorLifetime),
 				WaitDealsDelay:                  config.Duration(cfg.WaitDealsDelay),
 				MakeCCSectorsAvailable:          cfg.MakeCCSectorsAvailable,
@@ -954,6 +956,8 @@ func ToSealingConfig(dealmakingCfg config.DealmakingConfig, sealingCfg config.Se
 		MaxWaitDealsSectors:             sealingCfg.MaxWaitDealsSectors,
 		MaxSealingSectors:               sealingCfg.MaxSealingSectors,
 		MaxSealingSectorsForDeals:       sealingCfg.MaxSealingSectorsForDeals,
+		PreferNewSectorsForDeals:        sealingCfg.PreferNewSectorsForDeals,
+		MaxUpgradingSectors:             sealingCfg.MaxUpgradingSectors,
 		StartEpochSealingBuffer:         abi.ChainEpoch(dealmakingCfg.StartEpochSealingBuffer),
 		MakeNewSectorForDeals:           dealmakingCfg.MakeNewSectorForDeals,
 		CommittedCapacitySectorLifetime: time.Duration(sealingCfg.CommittedCapacitySectorLifetime),