fix lotus-sim build

api/api_storage.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/filecoin-project/lotus/chain/actors/builtin"
+	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 
 	"github.com/google/uuid"
 	"github.com/ipfs/go-cid"
@@ -51,6 +52,8 @@ type StorageMiner interface {
 
 	MiningBase(context.Context) (*types.TipSet, error) //perm:read
 
+	ComputeWindowPoSt(ctx context.Context, dlIdx uint64, tsk types.TipSetKey) ([]miner.SubmitWindowedPoStParams, error) //perm:admin
+
 	// Temp api for testing
 	PledgeSector(context.Context) (abi.SectorID, error) //perm:write
 

api/proxy_gen.go

@@ -641,6 +641,8 @@ type StorageMinerStruct struct {
 
 		ComputeProof func(p0 context.Context, p1 []builtin.ExtendedSectorInfo, p2 abi.PoStRandomness, p3 abi.ChainEpoch, p4 abinetwork.Version) ([]builtin.PoStProof, error) `perm:"read"`
 
+		ComputeWindowPoSt func(p0 context.Context, p1 uint64, p2 types.TipSetKey) ([]miner.SubmitWindowedPoStParams, error) `perm:"admin"`
+
 		CreateBackup func(p0 context.Context, p1 string) error `perm:"admin"`
 
 		DagstoreGC func(p0 context.Context) ([]DagstoreShardResult, error) `perm:"admin"`
@@ -3857,6 +3859,17 @@ func (s *StorageMinerStub) ComputeProof(p0 context.Context, p1 []builtin.Extende
 	return *new([]builtin.PoStProof), ErrNotSupported
 }
 
+func (s *StorageMinerStruct) ComputeWindowPoSt(p0 context.Context, p1 uint64, p2 types.TipSetKey) ([]miner.SubmitWindowedPoStParams, error) {
+	if s.Internal.ComputeWindowPoSt == nil {
+		return *new([]miner.SubmitWindowedPoStParams), ErrNotSupported
+	}
+	return s.Internal.ComputeWindowPoSt(p0, p1, p2)
+}
+
+func (s *StorageMinerStub) ComputeWindowPoSt(p0 context.Context, p1 uint64, p2 types.TipSetKey) ([]miner.SubmitWindowedPoStParams, error) {
+	return *new([]miner.SubmitWindowedPoStParams), ErrNotSupported
+}
+
 func (s *StorageMinerStruct) CreateBackup(p0 context.Context, p1 string) error {
 	if s.Internal.CreateBackup == nil {
 		return ErrNotSupported

cmd/lotus-miner/init.go

@@ -467,7 +467,7 @@ func storageMinerInit(ctx context.Context, cctx *cli.Context, api v1api.FullNode
 			}
 			stor := stores.NewRemote(lstor, si, http.Header(sa), 10, &stores.DefaultPartialFileHandler{})
 
-			smgr, err := sectorstorage.New(ctx, lstor, stor, lr, si, sectorstorage.SealerConfig{
+			smgr, err := sectorstorage.New(ctx, lstor, stor, lr, si, sectorstorage.Config{
 				ParallelFetchLimit:       10,
 				AllowAddPiece:            true,
 				AllowPreCommit1:          true,

cmd/lotus-miner/proving.go

@@ -1,10 +1,12 @@
 package main
 
 import (
+	"encoding/json"
 	"fmt"
 	"os"
 	"strconv"
 	"text/tabwriter"
+	"time"
 
 	"github.com/fatih/color"
 	"github.com/urfave/cli/v2"
@@ -31,6 +33,7 @@ var provingCmd = &cli.Command{
 		provingFaultsCmd,
 		provingCheckProvableCmd,
 		workersCmd(false),
+		provingComputeCmd,
 	},
 }
 
@@ -510,3 +513,50 @@ var provingCheckProvableCmd = &cli.Command{
 		return tw.Flush()
 	},
 }
+
+var provingComputeCmd = &cli.Command{
+	Name: "compute",
+	Subcommands: []*cli.Command{
+		provingComputeWindowPoStCmd,
+	},
+}
+
+var provingComputeWindowPoStCmd = &cli.Command{
+	Name:  "window-post",
+	Usage: "Compute WindowPoSt for a specific deadline",
+	Description: `Note: This command is intended to be used to verify PoSt compute performance.
+It will not send any messages to the chain.`,
+	ArgsUsage: "[deadline index]",
+	Action: func(cctx *cli.Context) error {
+		if cctx.Args().Len() != 1 {
+			return xerrors.Errorf("must pass deadline index")
+		}
+
+		dlIdx, err := strconv.ParseUint(cctx.Args().Get(0), 10, 64)
+		if err != nil {
+			return xerrors.Errorf("could not parse deadline index: %w", err)
+		}
+
+		sapi, scloser, err := lcli.GetStorageMinerAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer scloser()
+
+		ctx := lcli.ReqContext(cctx)
+
+		start := time.Now()
+		res, err := sapi.ComputeWindowPoSt(ctx, dlIdx, types.EmptyTSK)
+		fmt.Printf("Took %s\n", time.Now().Sub(start))
+		if err != nil {
+			return err
+		}
+		jr, err := json.Marshal(res)
+		if err != nil {
+			return err
+		}
+		fmt.Println(string(jr))
+
+		return nil
+	},
+}

cmd/lotus-sim/info_commit.go

@@ -6,12 +6,11 @@ import (
 	"os"
 	"syscall"
 
-	"github.com/streadway/quantile"
+	"github.com/ipfs/go-cid"
+	"github.com/koalacxr/quantile"
 	"github.com/urfave/cli/v2"
 
 	"github.com/filecoin-project/go-state-types/exitcode"
-	"github.com/ipfs/go-cid"
-
 	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 	"github.com/filecoin-project/lotus/chain/stmgr"
 	"github.com/filecoin-project/lotus/chain/types"

cmd/lotus-sim/info_message.go

@@ -4,13 +4,14 @@ import (
 	"fmt"
 	"syscall"
 
+	"github.com/ipfs/go-cid"
+	"github.com/koalacxr/quantile"
+	"github.com/urfave/cli/v2"
+
 	"github.com/filecoin-project/lotus/chain/stmgr"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/cmd/lotus-sim/simulation"
 	"github.com/filecoin-project/lotus/lib/stati"
-	"github.com/ipfs/go-cid"
-	"github.com/streadway/quantile"
-	"github.com/urfave/cli/v2"
 )
 
 var infoMessageSizeSimCommand = &cli.Command{

cmd/lotus-worker/main.go

@@ -35,6 +35,7 @@ import (
 	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
 	"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
 	"github.com/filecoin-project/lotus/lib/lotuslog"
+	"github.com/filecoin-project/lotus/lib/ulimit"
 	"github.com/filecoin-project/lotus/metrics"
 	"github.com/filecoin-project/lotus/node/modules"
 	"github.com/filecoin-project/lotus/node/repo"
@@ -195,7 +196,7 @@ var runCmd = &cli.Command{
 		&cli.IntFlag{
 			Name:  "post-parallel-reads",
 			Usage: "maximum number of parallel challenge reads (0 = no limit)",
-			Value: 0,
+			Value: 128,
 		},
 		&cli.DurationFlag{
 			Name:  "post-read-timeout",
@@ -227,12 +228,23 @@ var runCmd = &cli.Command{
 			}
 		}
 
+		limit, _, err := ulimit.GetLimit()
+		switch {
+		case err == ulimit.ErrUnsupported:
+			log.Errorw("checking file descriptor limit failed", "error", err)
+		case err != nil:
+			return xerrors.Errorf("checking fd limit: %w", err)
+		default:
+			if limit < build.MinerFDLimit {
+				return xerrors.Errorf("soft file descriptor limit (ulimit -n) too low, want %d, current %d", build.MinerFDLimit, limit)
+			}
+		}
+
 		// Connect to storage-miner
 		ctx := lcli.ReqContext(cctx)
 
 		var nodeApi api.StorageMiner
 		var closer func()
-		var err error
 		for {
 			nodeApi, closer, err = lcli.GetStorageMinerAPI(cctx, cliutil.StorageMinerUseHttp)
 			if err == nil {

documentation/en/api-v0-methods-miner.md

@@ -16,6 +16,7 @@
   * [CheckProvable](#CheckProvable)
 * [Compute](#Compute)
   * [ComputeProof](#ComputeProof)
+  * [ComputeWindowPoSt](#ComputeWindowPoSt)
 * [Create](#Create)
   * [CreateBackup](#CreateBackup)
 * [Dagstore](#Dagstore)
@@ -394,6 +395,52 @@ Response:
 ]
 ```
 
+### ComputeWindowPoSt
+
+
+Perms: admin
+
+Inputs:
+```json
+[
+  42,
+  [
+    {
+      "/": "bafy2bzacea3wsdh6y3a36tb3skempjoxqpuyompjbmfeyf34fi3uy6uue42v4"
+    },
+    {
+      "/": "bafy2bzacebp3shtrn43k7g3unredz7fxn4gj533d3o43tqn2p2ipxxhrvchve"
+    }
+  ]
+]
+```
+
+Response:
+```json
+[
+  {
+    "Deadline": 42,
+    "Partitions": [
+      {
+        "Index": 42,
+        "Skipped": [
+          5,
+          1
+        ]
+      }
+    ],
+    "Proofs": [
+      {
+        "PoStProof": 8,
+        "ProofBytes": "Ynl0ZSBhcnJheQ=="
+      }
+    ],
+    "ChainCommitEpoch": 10101,
+    "ChainCommitRand": "Bw=="
+  }
+]
+```
+
 ## Create
 
 

documentation/en/cli-lotus-miner.md

@@ -2041,6 +2041,7 @@ COMMANDS:
    faults     View the currently known proving faulty sectors information
    check      Check sectors provable
    workers    list workers
+   compute    
    help, h    Shows a list of commands or help for one command
 
 OPTIONS:
@@ -2131,6 +2132,40 @@ OPTIONS:
    
 ```
 
+### lotus-miner proving compute
+```
+NAME:
+   lotus-miner proving compute - A new cli application
+
+USAGE:
+   lotus-miner proving compute command [command options] [arguments...]
+
+COMMANDS:
+   window-post  Compute WindowPoSt for a specific deadline
+   help, h      Shows a list of commands or help for one command
+
+OPTIONS:
+   --help, -h  show help (default: false)
+   
+```
+
+#### lotus-miner proving compute window-post
+```
+NAME:
+   lotus-miner proving compute window-post - Compute WindowPoSt for a specific deadline
+
+USAGE:
+   lotus-miner proving compute window-post [command options] [deadline index]
+
+DESCRIPTION:
+   Note: This command is intended to be used to verify PoSt compute performance.
+   It will not send any messages to the chain.
+
+OPTIONS:
+   --help, -h  show help (default: false)
+   
+```
+
 ## lotus-miner storage
 ```
 NAME:

documentation/en/cli-lotus-worker.md

@@ -50,7 +50,7 @@ OPTIONS:
    --windowpost                  enable window post (default: false)
    --winningpost                 enable winning post (default: false)
    --parallel-fetch-limit value  maximum fetch operations to run in parallel (default: 5)
-   --post-parallel-reads value   maximum number of parallel challenge reads (0 = no limit) (default: 0)
+   --post-parallel-reads value   maximum number of parallel challenge reads (0 = no limit) (default: 128)
    --post-read-timeout value     time limit for reading PoSt challenges (0 = no limit) (default: 0s)
    --timeout value               used when 'listen' is unspecified. must be a valid duration recognized by golang's time.ParseDuration function (default: "30m")
    --help, -h                    show help (default: false)

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

@@ -306,6 +306,14 @@
   #PurgeCacheOnStart = false
 
 
+[Proving]
+  # Maximum number of sector checks to run in parallel. (0 = unlimited)
+  #
+  # type: int
+  # env var: LOTUS_PROVING_PARALLELCHECKLIMIT
+  #ParallelCheckLimit = 128
+
+
 [Sealing]
   # Upper bound on how many sectors can be waiting for more deals to be packed in it before it begins sealing at any given time.
   # If the miner is accepting multiple deals in parallel, up to MaxWaitDealsSectors of new sectors will be created.
@@ -484,33 +492,49 @@
 
 
 [Storage]
+  # type: int
   # env var: LOTUS_STORAGE_PARALLELFETCHLIMIT
   #ParallelFetchLimit = 10
 
+  # Local worker config
+  #
+  # type: bool
   # env var: LOTUS_STORAGE_ALLOWADDPIECE
   #AllowAddPiece = true
 
+  # type: bool
   # env var: LOTUS_STORAGE_ALLOWPRECOMMIT1
   #AllowPreCommit1 = true
 
+  # type: bool
   # env var: LOTUS_STORAGE_ALLOWPRECOMMIT2
   #AllowPreCommit2 = true
 
+  # type: bool
   # env var: LOTUS_STORAGE_ALLOWCOMMIT
   #AllowCommit = true
 
+  # type: bool
   # env var: LOTUS_STORAGE_ALLOWUNSEAL
   #AllowUnseal = true
 
+  # type: bool
   # env var: LOTUS_STORAGE_ALLOWREPLICAUPDATE
   #AllowReplicaUpdate = true
 
+  # type: bool
   # env var: LOTUS_STORAGE_ALLOWPROVEREPLICAUPDATE2
   #AllowProveReplicaUpdate2 = true
 
+  # type: bool
   # env var: LOTUS_STORAGE_ALLOWREGENSECTORKEY
   #AllowRegenSectorKey = true
 
+  # ResourceFiltering instructs the system which resource filtering strategy
+  # to use when evaluating tasks against this worker. An empty value defaults
+  # to "hardware".
+  #
+  # type: sectorstorage.ResourceFilteringStrategy
   # env var: LOTUS_STORAGE_RESOURCEFILTERING
   #ResourceFiltering = "hardware"
 

extern/sector-storage/faults.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/rand"
 	"fmt"
+	"sync"
 	"time"
 
 	"golang.org/x/xerrors"
@@ -24,22 +25,55 @@ type FaultTracker interface {
 
 // CheckProvable returns unprovable sectors
 func (m *Manager) CheckProvable(ctx context.Context, pp abi.RegisteredPoStProof, sectors []storage.SectorRef, rg storiface.RGetter) (map[abi.SectorID]string, error) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
 	if rg == nil {
 		return nil, xerrors.Errorf("rg is nil")
 	}
 
 	var bad = make(map[abi.SectorID]string)
+	var badLk sync.Mutex
+
+	var postRand abi.PoStRandomness = make([]byte, abi.RandomnessLength)
+	_, _ = rand.Read(postRand)
+	postRand[31] &= 0x3f
+
+	limit := m.parallelCheckLimit
+	if limit <= 0 {
+		limit = len(sectors)
+	}
+	throttle := make(chan struct{}, limit)
+
+	addBad := func(s abi.SectorID, reason string) {
+		badLk.Lock()
+		bad[s] = reason
+		badLk.Unlock()
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(len(sectors))
 
 	for _, sector := range sectors {
-		err := func() error {
+		select {
+		case throttle <- struct{}{}:
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+
+		go func(sector storage.SectorRef) {
+			defer wg.Done()
+			defer func() {
+				<-throttle
+			}()
 			ctx, cancel := context.WithCancel(ctx)
 			defer cancel()
 
 			commr, update, err := rg(ctx, sector.ID)
 			if err != nil {
 				log.Warnw("CheckProvable Sector FAULT: getting commR", "sector", sector, "sealed", "err", err)
-				bad[sector.ID] = fmt.Sprintf("getting commR: %s", err)
-				return nil
+				addBad(sector.ID, fmt.Sprintf("getting commR: %s", err))
+				return
 			}
 
 			toLock := storiface.FTSealed | storiface.FTCache
@@ -49,31 +83,29 @@ func (m *Manager) CheckProvable(ctx context.Context, pp abi.RegisteredPoStProof,
 
 			locked, err := m.index.StorageTryLock(ctx, sector.ID, toLock, storiface.FTNone)
 			if err != nil {
-				return xerrors.Errorf("acquiring sector lock: %w", err)
+				addBad(sector.ID, fmt.Sprintf("tryLock error: %s", err))
+				return
 			}
 
 			if !locked {
 				log.Warnw("CheckProvable Sector FAULT: can't acquire read lock", "sector", sector)
-				bad[sector.ID] = fmt.Sprint("can't acquire read lock")
-				return nil
+				addBad(sector.ID, fmt.Sprint("can't acquire read lock"))
+				return
 			}
 
 			wpp, err := sector.ProofType.RegisteredWindowPoStProof()
 			if err != nil {
-				return err
+				addBad(sector.ID, fmt.Sprint("can't get proof type"))
+				return
 			}
 
-			var pr abi.PoStRandomness = make([]byte, abi.RandomnessLength)
-			_, _ = rand.Read(pr)
-			pr[31] &= 0x3f
-
-			ch, err := ffi.GeneratePoStFallbackSectorChallenges(wpp, sector.ID.Miner, pr, []abi.SectorNumber{
+			ch, err := ffi.GeneratePoStFallbackSectorChallenges(wpp, sector.ID.Miner, postRand, []abi.SectorNumber{
 				sector.ID.Number,
 			})
 			if err != nil {
 				log.Warnw("CheckProvable Sector FAULT: generating challenges", "sector", sector, "err", err)
-				bad[sector.ID] = fmt.Sprintf("generating fallback challenges: %s", err)
-				return nil
+				addBad(sector.ID, fmt.Sprintf("generating fallback challenges: %s", err))
+				return
 			}
 
 			vctx, cancel2 := context.WithTimeout(ctx, PostCheckTimeout)
@@ -88,17 +120,14 @@ func (m *Manager) CheckProvable(ctx context.Context, pp abi.RegisteredPoStProof,
 			}, wpp)
 			if err != nil {
 				log.Warnw("CheckProvable Sector FAULT: generating vanilla proof", "sector", sector, "err", err)
-				bad[sector.ID] = fmt.Sprintf("generating vanilla proof: %s", err)
-				return nil
+				addBad(sector.ID, fmt.Sprintf("generating vanilla proof: %s", err))
+				return
 			}
-
-			return nil
-		}()
-		if err != nil {
-			return nil, err
-		}
+		}(sector)
 	}
 
+	wg.Wait()
+
 	return bad, nil
 }
 

extern/sector-storage/manager.go

@@ -70,6 +70,8 @@ type Manager struct {
 	workLk sync.Mutex
 	work   *statestore.StateStore
 
+	parallelCheckLimit int
+
 	callToWork map[storiface.CallID]WorkID
 	// used when we get an early return and there's no callToWork mapping
 	callRes map[storiface.CallID]chan result
@@ -99,7 +101,7 @@ const (
 	ResourceFilteringDisabled = ResourceFilteringStrategy("disabled")
 )
 
-type SealerConfig struct {
+type Config struct {
 	ParallelFetchLimit int
 
 	// Local worker config
@@ -116,6 +118,9 @@ type SealerConfig struct {
 	// to use when evaluating tasks against this worker. An empty value defaults
 	// to "hardware".
 	ResourceFiltering ResourceFilteringStrategy
+
+	// PoSt config
+	ParallelCheckLimit int
 }
 
 type StorageAuth http.Header
@@ -123,7 +128,7 @@ type StorageAuth http.Header
 type WorkerStateStore *statestore.StateStore
 type ManagerStateStore *statestore.StateStore
 
-func New(ctx context.Context, lstor *stores.Local, stor stores.Store, ls stores.LocalStorage, si stores.SectorIndex, sc SealerConfig, wss WorkerStateStore, mss ManagerStateStore) (*Manager, error) {
+func New(ctx context.Context, lstor *stores.Local, stor stores.Store, ls stores.LocalStorage, si stores.SectorIndex, sc Config, wss WorkerStateStore, mss ManagerStateStore) (*Manager, error) {
 	prover, err := ffiwrapper.New(&readonlyProvider{stor: lstor, index: si})
 	if err != nil {
 		return nil, xerrors.Errorf("creating prover instance: %w", err)
@@ -142,6 +147,8 @@ func New(ctx context.Context, lstor *stores.Local, stor stores.Store, ls stores.
 
 		localProver: prover,
 
+		parallelCheckLimit: sc.ParallelCheckLimit,
+
 		work:       mss,
 		callToWork: map[storiface.CallID]WorkID{},
 		callRes:    map[storiface.CallID]chan result{},

extern/sector-storage/piece_provider_test.go

@@ -30,7 +30,7 @@ import (
 // only uses miner and does NOT use any remote worker.
 func TestPieceProviderSimpleNoRemoteWorker(t *testing.T) {
 	// Set up sector storage manager
-	sealerCfg := SealerConfig{
+	sealerCfg := Config{
 		ParallelFetchLimit: 10,
 		AllowAddPiece:      true,
 		AllowPreCommit1:    true,
@@ -89,7 +89,7 @@ func TestReadPieceRemoteWorkers(t *testing.T) {
 	logging.SetAllLoggers(logging.LevelDebug)
 
 	// miner's worker can only add pieces to an unsealed sector.
-	sealerCfg := SealerConfig{
+	sealerCfg := Config{
 		ParallelFetchLimit: 10,
 		AllowAddPiece:      true,
 		AllowPreCommit1:    false,
@@ -198,7 +198,7 @@ func generatePieceData(size uint64) []byte {
 	return bz
 }
 
-func newPieceProviderTestHarness(t *testing.T, mgrConfig SealerConfig, sectorProofType abi.RegisteredSealProof) *pieceProviderTestHarness {
+func newPieceProviderTestHarness(t *testing.T, mgrConfig Config, sectorProofType abi.RegisteredSealProof) *pieceProviderTestHarness {
 	ctx := context.Background()
 	// listen on tcp socket to create an http server later
 	address := "0.0.0.0:0"

go.mod

@@ -110,6 +110,7 @@ require (
 	github.com/ipld/go-ipld-selector-text-lite v0.0.1
 	github.com/jonboulle/clockwork v0.2.2 // indirect
 	github.com/kelseyhightower/envconfig v1.4.0
+	github.com/koalacxr/quantile v0.0.1
 	github.com/libp2p/go-buffer-pool v0.0.2
 	github.com/libp2p/go-eventbus v0.2.1
 	github.com/libp2p/go-libp2p v0.18.0
@@ -143,7 +144,6 @@ require (
 	github.com/prometheus/client_golang v1.11.0
 	github.com/raulk/clock v1.1.0
 	github.com/raulk/go-watchdog v1.2.0
-	github.com/streadway/quantile v0.0.0-20150917103942-b0c588724d25
 	github.com/stretchr/testify v1.7.0
 	github.com/syndtr/goleveldb v1.0.0
 	github.com/uber/jaeger-client-go v2.25.0+incompatible // indirect

go.sum

@@ -1006,6 +1006,8 @@ github.com/klauspost/cpuid/v2 v2.0.4/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa02
 github.com/klauspost/cpuid/v2 v2.0.6/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
+github.com/koalacxr/quantile v0.0.1 h1:wAW+SQ286Erny9wOjVww96t8ws+x5Zj6AKHDULUK+o0=
+github.com/koalacxr/quantile v0.0.1/go.mod h1:bGN/mCZLZ4lrSDHRQ6Lglj9chowGux8sGUIND+DQeD0=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/koron/go-ssdp v0.0.0-20180514024734-4a0ed625a78b/go.mod h1:5Ky9EC2xfoUKUor0Hjgi2BJhCSXJfMOFlmyYrVKGQMk=
@@ -1861,8 +1863,6 @@ github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271/go.mod h1:AZpEONHx3
 github.com/streadway/amqp v1.0.0/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/streadway/handy v0.0.0-20190108123426-d5acb3125c2a/go.mod h1:qNTQ5P5JnDBl6z3cMAg/SywNDC5ABu5ApDIw6lUbRmI=
 github.com/streadway/handy v0.0.0-20200128134331-0f66f006fb2e/go.mod h1:qNTQ5P5JnDBl6z3cMAg/SywNDC5ABu5ApDIw6lUbRmI=
-github.com/streadway/quantile v0.0.0-20150917103942-b0c588724d25 h1:7z3LSn867ex6VSaahyKadf4WtSsJIgne6A1WLOAGM8A=
-github.com/streadway/quantile v0.0.0-20150917103942-b0c588724d25/go.mod h1:lbP8tGiBjZ5YWIc2fzuRpTaz0b/53vT6PEs3QuAWzuU=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1 h1:2vfRuCMp5sSVIDSqO8oNnWJq7mPa6KVP3iPIwFBuy8A=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=

itests/kit/ensemble.go

@@ -585,7 +585,7 @@ func (n *Ensemble) Start() *Ensemble {
 
 			// disable resource filtering so that local worker gets assigned tasks
 			// regardless of system pressure.
-			node.Override(new(sectorstorage.SealerConfig), func() sectorstorage.SealerConfig {
+			node.Override(new(sectorstorage.Config), func() sectorstorage.Config {
 				scfg := config.DefaultStorageMiner()
 
 				if noLocal {
@@ -596,7 +596,7 @@ func (n *Ensemble) Start() *Ensemble {
 				}
 
 				scfg.Storage.ResourceFiltering = sectorstorage.ResourceFilteringDisabled
-				return scfg.Storage
+				return scfg.StorageManager()
 			}),
 
 			// upgrades

itests/worker_test.go

@@ -296,3 +296,49 @@ func TestWindowPostWorkerSkipBadSector(t *testing.T) {
 	require.Equal(t, p.MinerPower, p.TotalPower)
 	require.Equal(t, p.MinerPower.RawBytePower, types.NewInt(uint64(ssz)*uint64(sectors-1)))
 }
+
+func TestWindowPostWorkerManualPoSt(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	_ = logging.SetLogLevel("storageminer", "INFO")
+
+	sectors := 2 * 48 * 2
+
+	client, miner, _, ens := kit.EnsembleWorker(t,
+		kit.PresealSectors(sectors), // 2 sectors per partition, 2 partitions in all 48 deadlines
+		kit.LatestActorsAt(-1),
+		kit.ThroughRPC(),
+		kit.WithTaskTypes([]sealtasks.TaskType{sealtasks.TTGenerateWindowPoSt}))
+
+	maddr, err := miner.ActorAddress(ctx)
+	require.NoError(t, err)
+
+	di, err := client.StateMinerProvingDeadline(ctx, maddr, types.EmptyTSK)
+	require.NoError(t, err)
+
+	bm := ens.InterconnectAll().BeginMiningMustPost(2 * time.Millisecond)[0]
+
+	di = di.NextNotElapsed()
+
+	t.Log("Running one proving period")
+	waitUntil := di.Open + di.WPoStChallengeWindow*2 - 2
+	client.WaitTillChain(ctx, kit.HeightAtLeast(waitUntil))
+
+	t.Log("Waiting for post message")
+	bm.Stop()
+
+	tryDl := func(dl uint64) {
+		p, err := miner.ComputeWindowPoSt(ctx, dl, types.EmptyTSK)
+		require.NoError(t, err)
+		require.Len(t, p, 1)
+		require.Equal(t, dl, p[0].Deadline)
+	}
+	tryDl(0)
+	tryDl(40)
+	tryDl(di.Index + 4)
+
+	lastPending, err := client.MpoolPending(ctx, types.EmptyTSK)
+	require.NoError(t, err)
+	require.Len(t, lastPending, 0)
+}

node/builder_miner.go

@@ -108,10 +108,10 @@ func ConfigStorageMiner(c interface{}) Option {
 
 			// Mining / proving
 			Override(new(*slashfilter.SlashFilter), modules.NewSlashFilter),
-			Override(new(*storage.Miner), modules.StorageMiner(config.DefaultStorageMiner().Fees)),
 			Override(new(*miner.Miner), modules.SetupBlockProducer),
 			Override(new(gen.WinningPoStProver), storage.NewWinningPoStProver),
 			Override(new(*storage.Miner), modules.StorageMiner(cfg.Fees)),
+			Override(new(*storage.WindowPoStScheduler), modules.WindowPostScheduler(cfg.Fees)),
 			Override(new(sectorblocks.SectorBuilder), From(new(*storage.Miner))),
 		),
 
@@ -214,7 +214,7 @@ func ConfigStorageMiner(c interface{}) Option {
 			Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(&cfg.Fees, &cfg.Dealmaking)),
 		),
 
-		Override(new(sectorstorage.SealerConfig), cfg.Storage),
+		Override(new(sectorstorage.Config), cfg.StorageManager()),
 		Override(new(*storage.AddressSelector), modules.AddressSelector(&cfg.Addresses)),
 	)
 }

node/config/def.go

@@ -138,7 +138,11 @@ func DefaultStorageMiner() *StorageMiner {
 			TerminateBatchWait: Duration(5 * time.Minute),
 		},
 
-		Storage: sectorstorage.SealerConfig{
+		Proving: ProvingConfig{
+			ParallelCheckLimit: 128,
+		},
+
+		Storage: SealerConfig{
 			AllowAddPiece:            true,
 			AllowPreCommit1:          true,
 			AllowPreCommit2:          true,

node/config/doc_gen.go

@@ -620,6 +620,14 @@ over the worker address if this flag is set.`,
 			Comment: ``,
 		},
 	},
+	"ProvingConfig": []DocField{
+		{
+			Name: "ParallelCheckLimit",
+			Type: "int",
+
+			Comment: `Maximum number of sector checks to run in parallel. (0 = unlimited)`,
+		},
+	},
 	"Pubsub": []DocField{
 		{
 			Name: "Bootstrapper",
@@ -691,6 +699,70 @@ default value is true`,
 This parameter is ONLY applicable if the retrieval pricing policy strategy has been configured to "external".`,
 		},
 	},
+	"SealerConfig": []DocField{
+		{
+			Name: "ParallelFetchLimit",
+			Type: "int",
+
+			Comment: ``,
+		},
+		{
+			Name: "AllowAddPiece",
+			Type: "bool",
+
+			Comment: `Local worker config`,
+		},
+		{
+			Name: "AllowPreCommit1",
+			Type: "bool",
+
+			Comment: ``,
+		},
+		{
+			Name: "AllowPreCommit2",
+			Type: "bool",
+
+			Comment: ``,
+		},
+		{
+			Name: "AllowCommit",
+			Type: "bool",
+
+			Comment: ``,
+		},
+		{
+			Name: "AllowUnseal",
+			Type: "bool",
+
+			Comment: ``,
+		},
+		{
+			Name: "AllowReplicaUpdate",
+			Type: "bool",
+
+			Comment: ``,
+		},
+		{
+			Name: "AllowProveReplicaUpdate2",
+			Type: "bool",
+
+			Comment: ``,
+		},
+		{
+			Name: "AllowRegenSectorKey",
+			Type: "bool",
+
+			Comment: ``,
+		},
+		{
+			Name: "ResourceFiltering",
+			Type: "sectorstorage.ResourceFilteringStrategy",
+
+			Comment: `ResourceFiltering instructs the system which resource filtering strategy
+to use when evaluating tasks against this worker. An empty value defaults
+to "hardware".`,
+		},
+	},
 	"SealingConfig": []DocField{
 		{
 			Name: "MaxWaitDealsSectors",
@@ -933,6 +1005,12 @@ Default is 20 (about once a week).`,
 
 			Comment: ``,
 		},
+		{
+			Name: "Proving",
+			Type: "ProvingConfig",
+
+			Comment: ``,
+		},
 		{
 			Name: "Sealing",
 			Type: "SealingConfig",
@@ -941,7 +1019,7 @@ Default is 20 (about once a week).`,
 		},
 		{
 			Name: "Storage",
-			Type: "sectorstorage.SealerConfig",
+			Type: "SealerConfig",
 
 			Comment: ``,
 		},

node/config/storage.go

@@ -8,6 +8,7 @@ import (
 
 	"golang.org/x/xerrors"
 
+	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
 	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
 )
 
@@ -49,3 +50,20 @@ func WriteStorageFile(path string, config stores.StorageConfig) error {
 
 	return nil
 }
+
+func (c *StorageMiner) StorageManager() sectorstorage.Config {
+	return sectorstorage.Config{
+		ParallelFetchLimit:       c.Storage.ParallelFetchLimit,
+		AllowAddPiece:            c.Storage.AllowAddPiece,
+		AllowPreCommit1:          c.Storage.AllowPreCommit1,
+		AllowPreCommit2:          c.Storage.AllowPreCommit2,
+		AllowCommit:              c.Storage.AllowCommit,
+		AllowUnseal:              c.Storage.AllowUnseal,
+		AllowReplicaUpdate:       c.Storage.AllowReplicaUpdate,
+		AllowProveReplicaUpdate2: c.Storage.AllowProveReplicaUpdate2,
+		AllowRegenSectorKey:      c.Storage.AllowRegenSectorKey,
+		ResourceFiltering:        c.Storage.ResourceFiltering,
+
+		ParallelCheckLimit: c.Proving.ParallelCheckLimit,
+	}
+}

node/config/types.go

@@ -1,10 +1,9 @@
 package config
 
 import (
-	"github.com/ipfs/go-cid"
-
 	"github.com/filecoin-project/lotus/chain/types"
 	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
+	"github.com/ipfs/go-cid"
 )
 
 // // NOTE: ONLY PUT STRUCT DEFINITIONS IN THIS FILE
@@ -53,8 +52,9 @@ type StorageMiner struct {
 	Subsystems    MinerSubsystemConfig
 	Dealmaking    DealmakingConfig
 	IndexProvider IndexProviderConfig
+	Proving       ProvingConfig
 	Sealing       SealingConfig
-	Storage       sectorstorage.SealerConfig
+	Storage       SealerConfig
 	Fees          MinerFeeConfig
 	Addresses     MinerAddressConfig
 	DAGStore      DAGStoreConfig
@@ -216,6 +216,13 @@ type RetrievalPricingDefault struct {
 	VerifiedDealsFreeTransfer bool
 }
 
+type ProvingConfig struct {
+	// Maximum number of sector checks to run in parallel. (0 = unlimited)
+	ParallelCheckLimit int
+
+	// todo disable builtin post
+}
+
 type SealingConfig struct {
 	// Upper bound on how many sectors can be waiting for more deals to be packed in it before it begins sealing at any given time.
 	// If the miner is accepting multiple deals in parallel, up to MaxWaitDealsSectors of new sectors will be created.
@@ -307,6 +314,25 @@ type SealingConfig struct {
 	// todo TargetSectors - stop auto-pleding new sectors after this many sectors are sealed, default CC upgrade for deals sectors if above
 }
 
+type SealerConfig struct {
+	ParallelFetchLimit int
+
+	// Local worker config
+	AllowAddPiece            bool
+	AllowPreCommit1          bool
+	AllowPreCommit2          bool
+	AllowCommit              bool
+	AllowUnseal              bool
+	AllowReplicaUpdate       bool
+	AllowProveReplicaUpdate2 bool
+	AllowRegenSectorKey      bool
+
+	// ResourceFiltering instructs the system which resource filtering strategy
+	// to use when evaluating tasks against this worker. An empty value defaults
+	// to "hardware".
+	ResourceFiltering sectorstorage.ResourceFilteringStrategy
+}
+
 type BatchFeeConfig struct {
 	Base      types.FIL
 	PerSector types.FIL

node/impl/storminer.go

@@ -16,6 +16,7 @@ import (
 	"github.com/filecoin-project/go-jsonrpc/auth"
 
 	"github.com/filecoin-project/lotus/chain/actors/builtin"
+	lminer "github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 	"github.com/filecoin-project/lotus/chain/gen"
 
 	"github.com/google/uuid"
@@ -92,6 +93,8 @@ type StorageMinerAPI struct {
 	storiface.WorkerReturn `optional:"true"`
 	AddrSel                *storage.AddressSelector
 
+	WdPoSt *storage.WindowPoStScheduler
+
 	Epp gen.WinningPoStProver `optional:"true"`
 	DS  dtypes.MetadataDS
 
@@ -407,6 +410,21 @@ func (sm *StorageMinerAPI) SectorMatchPendingPiecesToOpenSectors(ctx context.Con
 	return sm.Miner.SectorMatchPendingPiecesToOpenSectors(ctx)
 }
 
+func (sm *StorageMinerAPI) ComputeWindowPoSt(ctx context.Context, dlIdx uint64, tsk types.TipSetKey) ([]lminer.SubmitWindowedPoStParams, error) {
+	var ts *types.TipSet
+	var err error
+	if tsk == types.EmptyTSK {
+		ts, err = sm.Full.ChainHead(ctx)
+	} else {
+		ts, err = sm.Full.ChainGetTipSet(ctx, tsk)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return sm.WdPoSt.ComputePoSt(ctx, dlIdx, ts)
+}
+
 func (sm *StorageMinerAPI) WorkerConnect(ctx context.Context, url string) error {
 	w, err := connectRemoteWorker(ctx, sm, url)
 	if err != nil {

node/modules/storageminer.go

@@ -215,6 +215,7 @@ type StorageMinerParams struct {
 	GetSealingConfigFn dtypes.GetSealingConfigFunc
 	Journal            journal.Journal
 	AddrSel            *storage.AddressSelector
+	Maddr              dtypes.MinerAddress
 }
 
 func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*storage.Miner, error) {
@@ -231,20 +232,11 @@ func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*st
 			gsd    = params.GetSealingConfigFn
 			j      = params.Journal
 			as     = params.AddrSel
+			maddr  = address.Address(params.Maddr)
 		)
 
-		maddr, err := minerAddrFromDS(ds)
-		if err != nil {
-			return nil, err
-		}
-
 		ctx := helpers.LifecycleCtx(mctx, lc)
 
-		fps, err := storage.NewWindowedPoStScheduler(api, fc, as, sealer, verif, sealer, j, maddr)
-		if err != nil {
-			return nil, err
-		}
-
 		sm, err := storage.NewMiner(api, maddr, ds, sealer, sc, verif, prover, gsd, fc, j, as)
 		if err != nil {
 			return nil, err
@@ -252,7 +244,6 @@ func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*st
 
 		lc.Append(fx.Hook{
 			OnStart: func(context.Context) error {
-				go fps.Run(ctx)
 				return sm.Run(ctx)
 			},
 			OnStop: sm.Stop,
@@ -262,6 +253,37 @@ func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*st
 	}
 }
 
+func WindowPostScheduler(fc config.MinerFeeConfig) func(params StorageMinerParams) (*storage.WindowPoStScheduler, error) {
+	return func(params StorageMinerParams) (*storage.WindowPoStScheduler, error) {
+		var (
+			mctx   = params.MetricsCtx
+			lc     = params.Lifecycle
+			api    = params.API
+			sealer = params.Sealer
+			verif  = params.Verifier
+			j      = params.Journal
+			as     = params.AddrSel
+			maddr  = address.Address(params.Maddr)
+		)
+
+		ctx := helpers.LifecycleCtx(mctx, lc)
+
+		fps, err := storage.NewWindowedPoStScheduler(api, fc, as, sealer, verif, sealer, j, maddr)
+		if err != nil {
+			return nil, err
+		}
+
+		lc.Append(fx.Hook{
+			OnStart: func(context.Context) error {
+				go fps.Run(ctx)
+				return nil
+			},
+		})
+
+		return fps, nil
+	}
+}
+
 func HandleRetrieval(host host.Host, lc fx.Lifecycle, m retrievalmarket.RetrievalProvider, j journal.Journal) {
 	m.OnReady(marketevents.ReadyLogger("retrieval provider"))
 	lc.Append(fx.Hook{
@@ -717,11 +739,11 @@ func LocalStorage(mctx helpers.MetricsCtx, lc fx.Lifecycle, ls stores.LocalStora
 	return stores.NewLocal(ctx, ls, si, urls)
 }
 
-func RemoteStorage(lstor *stores.Local, si stores.SectorIndex, sa sectorstorage.StorageAuth, sc sectorstorage.SealerConfig) *stores.Remote {
+func RemoteStorage(lstor *stores.Local, si stores.SectorIndex, sa sectorstorage.StorageAuth, sc sectorstorage.Config) *stores.Remote {
 	return stores.NewRemote(lstor, si, http.Header(sa), sc.ParallelFetchLimit, &stores.DefaultPartialFileHandler{})
 }
 
-func SectorStorage(mctx helpers.MetricsCtx, lc fx.Lifecycle, lstor *stores.Local, stor stores.Store, ls stores.LocalStorage, si stores.SectorIndex, sc sectorstorage.SealerConfig, ds dtypes.MetadataDS) (*sectorstorage.Manager, error) {
+func SectorStorage(mctx helpers.MetricsCtx, lc fx.Lifecycle, lstor *stores.Local, stor stores.Store, ls stores.LocalStorage, si stores.SectorIndex, sc sectorstorage.Config, ds dtypes.MetadataDS) (*sectorstorage.Manager, error) {
 	ctx := helpers.LifecycleCtx(mctx, lc)
 
 	wsts := statestore.New(namespace.Wrap(ds, WorkerCallsPrefix))

storage/wdpost_run.go

@@ -93,7 +93,7 @@ func (s *WindowPoStScheduler) runGeneratePoST(
 	ctx, span := trace.StartSpan(ctx, "WindowPoStScheduler.generatePoST")
 	defer span.End()
 
-	posts, err := s.runPoStCycle(ctx, *deadline, ts)
+	posts, err := s.runPoStCycle(ctx, false, *deadline, ts)
 	if err != nil {
 		log.Errorf("runPoStCycle failed: %+v", err)
 		return nil, err
@@ -449,19 +449,8 @@ func (s *WindowPoStScheduler) declareFaults(ctx context.Context, dlIdx uint64, p
 	return faults, sm, nil
 }
 
-// runPoStCycle runs a full cycle of the PoSt process:
-//
-//  1. performs recovery declarations for the next deadline.
-//  2. performs fault declarations for the next deadline.
-//  3. computes and submits proofs, batching partitions and making sure they
-//     don't exceed message capacity.
-func (s *WindowPoStScheduler) runPoStCycle(ctx context.Context, di dline.Info, ts *types.TipSet) ([]miner.SubmitWindowedPoStParams, error) {
-	ctx, span := trace.StartSpan(ctx, "storage.runPoStCycle")
-	defer span.End()
-
+func (s *WindowPoStScheduler) asyncFaultRecover(di dline.Info, ts *types.TipSet) {
 	go func() {
-		// TODO: extract from runPoStCycle, run on fault cutoff boundaries
-
 		// check faults / recoveries for the *next* deadline. It's already too
 		// late to declare them for this deadline
 		declDeadline := (di.Index + 2) % di.WPoStPeriodDeadlines
@@ -520,6 +509,24 @@ func (s *WindowPoStScheduler) runPoStCycle(ctx context.Context, di dline.Info, t
 			}
 		})
 	}()
+}
+
+// runPoStCycle runs a full cycle of the PoSt process:
+//
+//  1. performs recovery declarations for the next deadline.
+//  2. performs fault declarations for the next deadline.
+//  3. computes and submits proofs, batching partitions and making sure they
+//     don't exceed message capacity.
+//
+// When `manual` is set, no messages (fault/recover) will be automatically sent
+func (s *WindowPoStScheduler) runPoStCycle(ctx context.Context, manual bool, di dline.Info, ts *types.TipSet) ([]miner.SubmitWindowedPoStParams, error) {
+	ctx, span := trace.StartSpan(ctx, "storage.runPoStCycle")
+	defer span.End()
+
+	if !manual {
+		// TODO: extract from runPoStCycle, run on fault cutoff boundaries
+		s.asyncFaultRecover(di, ts)
+	}
 
 	buf := new(bytes.Buffer)
 	if err := s.actor.MarshalCBOR(buf); err != nil {
@@ -941,3 +948,24 @@ func (s *WindowPoStScheduler) prepareMessage(ctx context.Context, msg *types.Mes
 	}
 	return nil
 }
+
+func (s *WindowPoStScheduler) ComputePoSt(ctx context.Context, dlIdx uint64, ts *types.TipSet) ([]miner.SubmitWindowedPoStParams, error) {
+	dl, err := s.api.StateMinerProvingDeadline(ctx, s.actor, ts.Key())
+	if err != nil {
+		return nil, xerrors.Errorf("getting deadline: %w", err)
+	}
+	curIdx := dl.Index
+	dl.Index = dlIdx
+	dlDiff := dl.Index - curIdx
+	if dl.Index > curIdx {
+		dlDiff -= dl.WPoStPeriodDeadlines
+		dl.PeriodStart -= dl.WPoStProvingPeriod
+	}
+
+	epochDiff := (dl.WPoStProvingPeriod / abi.ChainEpoch(dl.WPoStPeriodDeadlines)) * abi.ChainEpoch(dlDiff)
+
+	// runPoStCycle only needs dl.Index and dl.Challenge
+	dl.Challenge += epochDiff
+
+	return s.runPoStCycle(ctx, true, *dl, ts)
+}