allow sealing bench to run parallel seals

cmd/lotus-bench/main.go

@@ -139,6 +139,10 @@ var sealBenchCmd = &cli.Command{
 			Name:  "num-sectors",
 			Value: 1,
 		},
+		&cli.IntFlag{
+			Name:  "parallel",
+			Value: 1,
+		},
 	},
 	Action: func(c *cli.Context) error {
 		if c.Bool("no-gpu") {
@@ -235,7 +239,7 @@ var sealBenchCmd = &cli.Command{
 
 		if robench == "" {
 			var err error
-			sealTimings, sealedSectors, err = runSeals(sb, sbfs, c.Int("num-sectors"), mid, sectorSize, []byte(c.String("ticket-preimage")), c.String("save-commit2-input"), c.Bool("skip-commit2"), c.Bool("skip-unseal"))
+			sealTimings, sealedSectors, err = runSeals(sb, sbfs, c.Int("num-sectors"), c.Int("parallel"), mid, sectorSize, []byte(c.String("ticket-preimage")), c.String("save-commit2-input"), c.Bool("skip-commit2"), c.Bool("skip-unseal"))
 			if err != nil {
 				return xerrors.Errorf("failed to run seals: %w", err)
 			}
@@ -447,9 +451,14 @@ var sealBenchCmd = &cli.Command{
 	},
 }
 
-func runSeals(sb *ffiwrapper.Sealer, sbfs *basicfs.Provider, numSectors int, mid abi.ActorID, sectorSize abi.SectorSize, ticketPreimage []byte, saveC2inp string, skipc2, skipunseal bool) ([]SealingResult, []abi.SectorInfo, error) {
+func runSeals(sb *ffiwrapper.Sealer, sbfs *basicfs.Provider, numSectors, parallelism int, mid abi.ActorID, sectorSize abi.SectorSize, ticketPreimage []byte, saveC2inp string, skipc2, skipunseal bool) ([]SealingResult, []abi.SectorInfo, error) {
-	var sealTimings []SealingResult
+	var pieces []abi.PieceInfo
-	var sealedSectors []abi.SectorInfo
+	sealTimings := make([]SealingResult, numSectors)
+	sealedSectors := make([]abi.SectorInfo, numSectors)
+
+	if numSectors%parallelism != 0 {
+		return nil, nil, fmt.Errorf("parallelism factor must cleanly divide numSectors")
+	}
 
 	for i := abi.SectorNumber(1); i <= abi.SectorNumber(numSectors); i++ {
 		sid := abi.SectorID{
@@ -458,7 +467,7 @@ func runSeals(sb *ffiwrapper.Sealer, sbfs *basicfs.Provider, numSectors int, mid
 		}
 
 		start := time.Now()
-		log.Info("Writing piece into sector...")
+		log.Infof("[%d] Writing piece into sector...", i)
 
 		r := rand.New(rand.NewSource(100 + int64(i)))
 
@@ -467,48 +476,68 @@ func runSeals(sb *ffiwrapper.Sealer, sbfs *basicfs.Provider, numSectors int, mid
 			return nil, nil, err
 		}
 
-		addpiece := time.Now()
+		pieces = append(pieces, pi)
+
+		sealTimings[i-1].AddPiece = time.Since(start)
+	}
+
+	sectorsPerWorker := numSectors / parallelism
+
+	errs := make(chan error, parallelism)
+	for wid := 0; wid < parallelism; wid++ {
+		go func(worker int) {
+			sealerr := func() error {
+				start := 1 + (worker * sectorsPerWorker)
+				end := start + sectorsPerWorker
+				for i := abi.SectorNumber(start); i < abi.SectorNumber(end); i++ {
+					ix := int(i - 1)
+					sid := abi.SectorID{
+						Miner:  mid,
+						Number: i,
+					}
+
+					start := time.Now()
 
 					trand := blake2b.Sum256(ticketPreimage)
 					ticket := abi.SealRandomness(trand[:])
 
-		log.Info("Running replication(1)...")
+					log.Infof("[%d] Running replication(1)...", i)
-		pieces := []abi.PieceInfo{pi}
+					pieces := []abi.PieceInfo{pieces[ix]}
 					pc1o, err := sb.SealPreCommit1(context.TODO(), sid, ticket, pieces)
 					if err != nil {
-			return nil, nil, xerrors.Errorf("commit: %w", err)
+						return xerrors.Errorf("commit: %w", err)
 					}
 
 					precommit1 := time.Now()
 
-		log.Info("Running replication(2)...")
+					log.Infof("[%d] Running replication(2)...", i)
 					cids, err := sb.SealPreCommit2(context.TODO(), sid, pc1o)
 					if err != nil {
-			return nil, nil, xerrors.Errorf("commit: %w", err)
+						return xerrors.Errorf("commit: %w", err)
 					}
 
 					precommit2 := time.Now()
 
-		sealedSectors = append(sealedSectors, abi.SectorInfo{
+					sealedSectors[ix] = abi.SectorInfo{
 						SealProof:    sb.SealProofType(),
 						SectorNumber: i,
 						SealedCID:    cids.Sealed,
-		})
+					}
 
 					seed := lapi.SealSeed{
 						Epoch: 101,
 						Value: []byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 255},
 					}
 
-		log.Info("Generating PoRep for sector (1)")
+					log.Infof("[%d] Generating PoRep for sector (1)", i)
 					c1o, err := sb.SealCommit1(context.TODO(), sid, ticket, seed.Value, pieces, cids)
 					if err != nil {
-			return nil, nil, err
+						return err
 					}
 
 					sealcommit1 := time.Now()
 
-		log.Info("Generating PoRep for sector (2)")
+					log.Infof("[%d] Generating PoRep for sector (2)", i)
 
 					if saveC2inp != "" {
 						c2in := Commit2In{
@@ -519,7 +548,7 @@ func runSeals(sb *ffiwrapper.Sealer, sbfs *basicfs.Provider, numSectors int, mid
 
 						b, err := json.Marshal(&c2in)
 						if err != nil {
-				return nil, nil, err
+							return err
 						}
 
 						if err := ioutil.WriteFile(saveC2inp, b, 0664); err != nil {
@@ -531,7 +560,7 @@ func runSeals(sb *ffiwrapper.Sealer, sbfs *basicfs.Provider, numSectors int, mid
 					if !skipc2 {
 						proof, err = sb.SealCommit2(context.TODO(), sid, c1o)
 						if err != nil {
-				return nil, nil, err
+							return err
 						}
 					}
 
@@ -551,45 +580,58 @@ func runSeals(sb *ffiwrapper.Sealer, sbfs *basicfs.Provider, numSectors int, mid
 
 						ok, err := ffiwrapper.ProofVerifier.VerifySeal(svi)
 						if err != nil {
-				return nil, nil, err
+							return err
 						}
 						if !ok {
-				return nil, nil, xerrors.Errorf("porep proof for sector %d was invalid", i)
+							return xerrors.Errorf("porep proof for sector %d was invalid", i)
 						}
 					}
 
 					verifySeal := time.Now()
 
 					if !skipunseal {
-			log.Info("Unsealing sector")
+						log.Infof("[%d] Unsealing sector", i)
 						{
 							p, done, err := sbfs.AcquireSector(context.TODO(), abi.SectorID{Miner: mid, Number: 1}, stores.FTUnsealed, stores.FTNone, true)
 							if err != nil {
-					return nil, nil, xerrors.Errorf("acquire unsealed sector for removing: %w", err)
+								return xerrors.Errorf("acquire unsealed sector for removing: %w", err)
 							}
 							done()
 
 							if err := os.Remove(p.Unsealed); err != nil {
-					return nil, nil, xerrors.Errorf("removing unsealed sector: %w", err)
+								return xerrors.Errorf("removing unsealed sector: %w", err)
 							}
 						}
 
 						err := sb.UnsealPiece(context.TODO(), abi.SectorID{Miner: mid, Number: 1}, 0, abi.PaddedPieceSize(sectorSize).Unpadded(), ticket, cids.Unsealed)
 						if err != nil {
-				return nil, nil, err
+							return err
 						}
 					}
 					unseal := time.Now()
 
-		sealTimings = append(sealTimings, SealingResult{
+					sealTimings[ix].PreCommit1 = precommit1.Sub(start)
-			AddPiece:   addpiece.Sub(start),
+					sealTimings[ix].PreCommit2 = precommit2.Sub(precommit1)
-			PreCommit1: precommit1.Sub(addpiece),
+					sealTimings[ix].Commit1 = sealcommit1.Sub(precommit2)
-			PreCommit2: precommit2.Sub(precommit1),
+					sealTimings[ix].Commit2 = sealcommit2.Sub(sealcommit1)
-			Commit1:    sealcommit1.Sub(precommit2),
+					sealTimings[ix].Verify = verifySeal.Sub(sealcommit2)
-			Commit2:    sealcommit2.Sub(sealcommit1),
+					sealTimings[ix].Unseal = unseal.Sub(verifySeal)
-			Verify:     verifySeal.Sub(sealcommit2),
+				}
-			Unseal:     unseal.Sub(verifySeal),
+				return nil
-		})
+			}()
+			if sealerr != nil {
+				errs <- sealerr
+				return
+			}
+			errs <- nil
+		}(wid)
+	}
+
+	for i := 0; i < parallelism; i++ {
+		err := <-errs
+		if err != nil {
+			return nil, nil, err
+		}
 	}
 
 	return sealTimings, sealedSectors, nil

go.mod

@@ -126,4 +126,6 @@ require (
 
 replace github.com/golangci/golangci-lint => github.com/golangci/golangci-lint v1.18.0
 
+replace github.com/filecoin-project/specs-actors => ../specs-actors
+
 replace github.com/filecoin-project/filecoin-ffi => ./extern/filecoin-ffi

go.sum

@@ -215,14 +215,11 @@ github.com/fatih/color v1.8.0/go.mod h1:3l45GVGkyrnYNl9HoIjnp2NnNWvh6hLAqD8yTfGj
 github.com/fd/go-nat v1.0.0/go.mod h1:BTBu/CKvMmOMUPkKVef1pngt2WFH/lg7E6yQnulfp6E=
 github.com/filecoin-project/chain-validation v0.0.6-0.20200615191232-6be1a8c6ed09 h1:GuiNSEZ9nc05LUpKhABw/SO6t9wqCfsJX1D0ByWQjkc=
 github.com/filecoin-project/chain-validation v0.0.6-0.20200615191232-6be1a8c6ed09/go.mod h1:HEJn6kOXMNhCNBYNTO/lrEI7wSgqCOR6hN5ecfYUnC8=
-github.com/filecoin-project/go-address v0.0.0-20200107215422-da8eea2842b5/go.mod h1:SAOwJoakQ8EPjwNIsiakIQKsoKdkcbx8U3IapgCg9R0=
 github.com/filecoin-project/go-address v0.0.2-0.20200218010043-eb9bb40ed5be/go.mod h1:SAOwJoakQ8EPjwNIsiakIQKsoKdkcbx8U3IapgCg9R0=
 github.com/filecoin-project/go-address v0.0.2-0.20200504173055-8b6f2fb2b3ef h1:Wi5E+P1QfHP8IF27eUiTx5vYfqQZwfPxzq3oFEq8w8U=
 github.com/filecoin-project/go-address v0.0.2-0.20200504173055-8b6f2fb2b3ef/go.mod h1:SrA+pWVoUivqKOfC+ckVYbx41hWz++HxJcrlmHNnebU=
-github.com/filecoin-project/go-amt-ipld/v2 v2.0.1-0.20200131012142-05d80eeccc5e/go.mod h1:boRtQhzmxNocrMxOXo1NYn4oUc1NGvR8tEa79wApNXg=
 github.com/filecoin-project/go-amt-ipld/v2 v2.0.1-0.20200424220931-6263827e49f2 h1:jamfsxfK0Q9yCMHt8MPWx7Aa/O9k2Lve8eSc6FILYGQ=
 github.com/filecoin-project/go-amt-ipld/v2 v2.0.1-0.20200424220931-6263827e49f2/go.mod h1:boRtQhzmxNocrMxOXo1NYn4oUc1NGvR8tEa79wApNXg=
-github.com/filecoin-project/go-bitfield v0.0.0-20200416002808-b3ee67ec9060/go.mod h1:iodsLxOFZnqKtjj2zkgqzoGNrv6vUqj69AT/J8DKXEw=
 github.com/filecoin-project/go-bitfield v0.0.1/go.mod h1:Ry9/iUlWSyjPUzlAvdnfy4Gtvrq4kWmWDztCU1yEgJY=
 github.com/filecoin-project/go-bitfield v0.0.2-0.20200518150651-562fdb554b6e h1:gkG/7G+iKy4He+IiQNeQn+nndFznb/vCoOR8iRQsm60=
 github.com/filecoin-project/go-bitfield v0.0.2-0.20200518150651-562fdb554b6e/go.mod h1:Ry9/iUlWSyjPUzlAvdnfy4Gtvrq4kWmWDztCU1yEgJY=
@@ -255,12 +252,6 @@ github.com/filecoin-project/go-storedcounter v0.0.0-20200421200003-1c99c62e8a5b/
 github.com/filecoin-project/sector-storage v0.0.0-20200615154852-728a47ab99d6/go.mod h1:M59QnAeA/oV+Z8oHFLoNpGMv0LZ8Rll+vHVXX7GirPM=
 github.com/filecoin-project/sector-storage v0.0.0-20200615192001-42c9e08595b7 h1:cjsOpQKvZosPx9/qqq2bucHVdRyXzvBR1f37atiR3/0=
 github.com/filecoin-project/sector-storage v0.0.0-20200615192001-42c9e08595b7/go.mod h1:M59QnAeA/oV+Z8oHFLoNpGMv0LZ8Rll+vHVXX7GirPM=
-github.com/filecoin-project/specs-actors v0.0.0-20200210130641-2d1fbd8672cf/go.mod h1:xtDZUB6pe4Pksa/bAJbJ693OilaC5Wbot9jMhLm3cZA=
-github.com/filecoin-project/specs-actors v0.3.0/go.mod h1:nQYnFbQ7Y0bHZyq6HDEuVlCPR+U3z5Q3wMOQ+2aiV+Y=
-github.com/filecoin-project/specs-actors v0.6.0 h1:IepUsmDGY60QliENVTkBTAkwqGWw9kNbbHOcU/9oiC0=
-github.com/filecoin-project/specs-actors v0.6.0/go.mod h1:dRdy3cURykh2R8O/DKqy8olScl70rmIS7GrB4hB1IDY=
-github.com/filecoin-project/specs-actors v0.6.1 h1:rhHlEzqcuuQU6oKc4csuq+/kQBDZ4EXtSomoN2XApCA=
-github.com/filecoin-project/specs-actors v0.6.1/go.mod h1:dRdy3cURykh2R8O/DKqy8olScl70rmIS7GrB4hB1IDY=
 github.com/filecoin-project/specs-storage v0.1.0 h1:PkDgTOT5W5Ao7752onjDl4QSv+sgOVdJbvFjOnD5w94=
 github.com/filecoin-project/specs-storage v0.1.0/go.mod h1:Pr5ntAaxsh+sLG/LYiL4tKzvA83Vk5vLODYhfNwOg7k=
 github.com/filecoin-project/storage-fsm v0.0.0-20200615162749-494c3bc48743 h1:a8f1p6UdeD+ZINBKJN4FhEos8uaKeASOAabq5RCpQdg=
@@ -1330,7 +1321,6 @@ github.com/whyrusleeping/bencher v0.0.0-20190829221104-bb6607aa8bba/go.mod h1:CH
 github.com/whyrusleeping/cbor-gen v0.0.0-20191212224538-d370462a7e8a/go.mod h1:xdlJQaiqipF0HW+Mzpg7XRM3fWbGvfgFlcppuvlkIvY=
 github.com/whyrusleeping/cbor-gen v0.0.0-20191216205031-b047b6acb3c0/go.mod h1:xdlJQaiqipF0HW+Mzpg7XRM3fWbGvfgFlcppuvlkIvY=
 github.com/whyrusleeping/cbor-gen v0.0.0-20200123233031-1cdf64d27158/go.mod h1:Xj/M2wWU+QdTdRbu/L/1dIZY8/Wb2K9pAhtroQuxJJI=
-github.com/whyrusleeping/cbor-gen v0.0.0-20200206220010-03c9665e2a66/go.mod h1:Xj/M2wWU+QdTdRbu/L/1dIZY8/Wb2K9pAhtroQuxJJI=
 github.com/whyrusleeping/cbor-gen v0.0.0-20200402171437-3d27c146c105/go.mod h1:Xj/M2wWU+QdTdRbu/L/1dIZY8/Wb2K9pAhtroQuxJJI=
 github.com/whyrusleeping/cbor-gen v0.0.0-20200414195334-429a0b5e922e/go.mod h1:Xj/M2wWU+QdTdRbu/L/1dIZY8/Wb2K9pAhtroQuxJJI=
 github.com/whyrusleeping/cbor-gen v0.0.0-20200501014322-5f9941ef88e0/go.mod h1:Xj/M2wWU+QdTdRbu/L/1dIZY8/Wb2K9pAhtroQuxJJI=