Merge remote-tracking branch 'origin/master' into conformance-tests

extern/sector-storage/ffiwrapper/partialfile.go

@@ -118,7 +118,7 @@ func openPartialFile(maxPieceSize abi.PaddedPieceSize, path string) (*partialFil
 		trailerLen := binary.LittleEndian.Uint32(tlen[:])
 		expectLen := int64(trailerLen) + int64(len(tlen)) + int64(maxPieceSize)
 		if expectLen != st.Size() {
-			return xerrors.Errorf("file '%d' has inconsistent length; has %d bytes; expected %d (%d trailer, %d sector data)", path, st.Size(), expectLen, int64(trailerLen)+int64(len(tlen)), maxPieceSize)
+			return xerrors.Errorf("file '%s' has inconsistent length; has %d bytes; expected %d (%d trailer, %d sector data)", path, st.Size(), expectLen, int64(trailerLen)+int64(len(tlen)), maxPieceSize)
 		}
 		if trailerLen > veryLargeRle {
 			log.Warnf("Partial file '%s' has a VERY large trailer with %d bytes", path, trailerLen)

extern/sector-storage/fsutil/statfs_unix.go

@@ -12,6 +12,8 @@ func Statfs(path string) (FsStat, error) {
 		return FsStat{}, xerrors.Errorf("statfs: %w", err)
 	}
 
+	// force int64 to handle platform specific differences
+	//nolint:unconvert
 	return FsStat{
 		Capacity:  int64(stat.Blocks) * int64(stat.Bsize),
 		Available: int64(stat.Bavail) * int64(stat.Bsize),

extern/sector-storage/mock/mock.go

@@ -438,7 +438,7 @@ func (m mockVerif) VerifyWindowPoSt(ctx context.Context, info abi.WindowPoStVeri
 		}
 
 		if !bytes.Equal(b1, b2) {
-			return false, xerrors.Errorf("proven and challenged sector sets didn't match: %s != !s", string(b1), string(b2))
+			return false, xerrors.Errorf("proven and challenged sector sets didn't match: %s != %s", string(b1), string(b2))
 		}
 	}
 

extern/sector-storage/sched.go

@@ -287,67 +287,93 @@ func (sh *scheduler) trySched() {
 
 	sh.workersLk.RLock()
 	defer sh.workersLk.RUnlock()
+	if len(sh.openWindows) == 0 {
+		// nothing to schedule on
+		return
+	}
 
 	// Step 1
-	for sqi := 0; sqi < sh.schedQueue.Len(); sqi++ {
-		task := (*sh.schedQueue)[sqi]
-		needRes := ResourceTable[task.taskType][sh.spt]
+	concurrency := len(sh.openWindows)
+	throttle := make(chan struct{}, concurrency)
 
-		task.indexHeap = sqi
-		for wnd, windowRequest := range sh.openWindows {
-			worker := sh.workers[windowRequest.worker]
+	var wg sync.WaitGroup
+	wg.Add(sh.schedQueue.Len())
 
-			// TODO: allow bigger windows
-			if !windows[wnd].allocated.canHandleRequest(needRes, windowRequest.worker, worker.info.Resources) {
-				continue
+	for i := 0; i < sh.schedQueue.Len(); i++ {
+		throttle <- struct{}{}
+
+		go func(sqi int) {
+			defer wg.Done()
+			defer func() {
+				<-throttle
+			}()
+
+			task := (*sh.schedQueue)[sqi]
+			needRes := ResourceTable[task.taskType][sh.spt]
+
+			task.indexHeap = sqi
+			for wnd, windowRequest := range sh.openWindows {
+				worker, ok := sh.workers[windowRequest.worker]
+				if !ok {
+					log.Errorf("worker referenced by windowRequest not found (worker: %d)", windowRequest.worker)
+					// TODO: How to move forward here?
+					continue
+				}
+
+				// TODO: allow bigger windows
+				if !windows[wnd].allocated.canHandleRequest(needRes, windowRequest.worker, worker.info.Resources) {
+					continue
+				}
+
+				rpcCtx, cancel := context.WithTimeout(task.ctx, SelectorTimeout)
+				ok, err := task.sel.Ok(rpcCtx, task.taskType, sh.spt, worker)
+				cancel()
+				if err != nil {
+					log.Errorf("trySched(1) req.sel.Ok error: %+v", err)
+					continue
+				}
+
+				if !ok {
+					continue
+				}
+
+				acceptableWindows[sqi] = append(acceptableWindows[sqi], wnd)
 			}
 
-			rpcCtx, cancel := context.WithTimeout(task.ctx, SelectorTimeout)
-			ok, err := task.sel.Ok(rpcCtx, task.taskType, sh.spt, worker)
-			cancel()
-			if err != nil {
-				log.Errorf("trySched(1) req.sel.Ok error: %+v", err)
-				continue
+			if len(acceptableWindows[sqi]) == 0 {
+				return
 			}
 
-			if !ok {
-				continue
-			}
+			// Pick best worker (shuffle in case some workers are equally as good)
+			rand.Shuffle(len(acceptableWindows[sqi]), func(i, j int) {
+				acceptableWindows[sqi][i], acceptableWindows[sqi][j] = acceptableWindows[sqi][j], acceptableWindows[sqi][i] // nolint:scopelint
+			})
+			sort.SliceStable(acceptableWindows[sqi], func(i, j int) bool {
+				wii := sh.openWindows[acceptableWindows[sqi][i]].worker // nolint:scopelint
+				wji := sh.openWindows[acceptableWindows[sqi][j]].worker // nolint:scopelint
 
-			acceptableWindows[sqi] = append(acceptableWindows[sqi], wnd)
-		}
+				if wii == wji {
+					// for the same worker prefer older windows
+					return acceptableWindows[sqi][i] < acceptableWindows[sqi][j] // nolint:scopelint
+				}
 
-		if len(acceptableWindows[sqi]) == 0 {
-			continue
-		}
+				wi := sh.workers[wii]
+				wj := sh.workers[wji]
 
-		// Pick best worker (shuffle in case some workers are equally as good)
-		rand.Shuffle(len(acceptableWindows[sqi]), func(i, j int) {
-			acceptableWindows[sqi][i], acceptableWindows[sqi][j] = acceptableWindows[sqi][j], acceptableWindows[sqi][i] // nolint:scopelint
-		})
-		sort.SliceStable(acceptableWindows[sqi], func(i, j int) bool {
-			wii := sh.openWindows[acceptableWindows[sqi][i]].worker // nolint:scopelint
-			wji := sh.openWindows[acceptableWindows[sqi][j]].worker // nolint:scopelint
+				rpcCtx, cancel := context.WithTimeout(task.ctx, SelectorTimeout)
+				defer cancel()
 
-			if wii == wji {
-				// for the same worker prefer older windows
-				return acceptableWindows[sqi][i] < acceptableWindows[sqi][j] // nolint:scopelint
-			}
-
-			wi := sh.workers[wii]
-			wj := sh.workers[wji]
-
-			rpcCtx, cancel := context.WithTimeout(task.ctx, SelectorTimeout)
-			defer cancel()
-
-			r, err := task.sel.Cmp(rpcCtx, task.taskType, wi, wj)
-			if err != nil {
-				log.Error("selecting best worker: %s", err)
-			}
-			return r
-		})
+				r, err := task.sel.Cmp(rpcCtx, task.taskType, wi, wj)
+				if err != nil {
+					log.Error("selecting best worker: %s", err)
+				}
+				return r
+			})
+		}(i)
 	}
 
+	wg.Wait()
+
 	log.Debugf("SCHED windows: %+v", windows)
 	log.Debugf("SCHED Acceptable win: %+v", acceptableWindows)
 

extern/sector-storage/sched_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/ipfs/go-cid"
+	logging "github.com/ipfs/go-log/v2"
 	"github.com/stretchr/testify/require"
 
 	"github.com/filecoin-project/specs-actors/actors/abi"
@@ -100,16 +101,18 @@ func (s *schedTestWorker) Paths(ctx context.Context) ([]stores.StoragePath, erro
 	return s.paths, nil
 }
 
+var decentWorkerResources = storiface.WorkerResources{
+	MemPhysical: 128 << 30,
+	MemSwap:     200 << 30,
+	MemReserved: 2 << 30,
+	CPUs:        32,
+	GPUs:        []string{"a GPU"},
+}
+
 func (s *schedTestWorker) Info(ctx context.Context) (storiface.WorkerInfo, error) {
 	return storiface.WorkerInfo{
-		Hostname: s.name,
-		Resources: storiface.WorkerResources{
-			MemPhysical: 128 << 30,
-			MemSwap:     200 << 30,
-			MemReserved: 2 << 30,
-			CPUs:        32,
-			GPUs:        []string{"a GPU"},
-		},
+		Hostname:  s.name,
+		Resources: decentWorkerResources,
 	}, nil
 }
 
@@ -451,3 +454,67 @@ func TestSched(t *testing.T) {
 		}))
 	}
 }
+
+type slowishSelector bool
+
+func (s slowishSelector) Ok(ctx context.Context, task sealtasks.TaskType, spt abi.RegisteredSealProof, a *workerHandle) (bool, error) {
+	time.Sleep(200 * time.Microsecond)
+	return bool(s), nil
+}
+
+func (s slowishSelector) Cmp(ctx context.Context, task sealtasks.TaskType, a, b *workerHandle) (bool, error) {
+	time.Sleep(100 * time.Microsecond)
+	return true, nil
+}
+
+var _ WorkerSelector = slowishSelector(true)
+
+func BenchmarkTrySched(b *testing.B) {
+	spt := abi.RegisteredSealProof_StackedDrg32GiBV1
+	logging.SetAllLoggers(logging.LevelInfo)
+	defer logging.SetAllLoggers(logging.LevelDebug)
+	ctx := context.Background()
+
+	test := func(windows, queue int) func(b *testing.B) {
+		return func(b *testing.B) {
+			for i := 0; i < b.N; i++ {
+				b.StopTimer()
+
+				sched := newScheduler(spt)
+				sched.workers[0] = &workerHandle{
+					w: nil,
+					info: storiface.WorkerInfo{
+						Hostname:  "t",
+						Resources: decentWorkerResources,
+					},
+					preparing: &activeResources{},
+					active:    &activeResources{},
+				}
+
+				for i := 0; i < windows; i++ {
+					sched.openWindows = append(sched.openWindows, &schedWindowRequest{
+						worker: 0,
+						done:   make(chan *schedWindow, 1000),
+					})
+				}
+
+				for i := 0; i < queue; i++ {
+					sched.schedQueue.Push(&workerRequest{
+						taskType: sealtasks.TTCommit2,
+						sel:      slowishSelector(true),
+						ctx:      ctx,
+					})
+				}
+
+				b.StartTimer()
+
+				sched.trySched()
+			}
+		}
+	}
+
+	b.Run("1w-1q", test(1, 1))
+	b.Run("500w-1q", test(500, 1))
+	b.Run("1w-500q", test(1, 500))
+	b.Run("200w-400q", test(200, 400))
+}

extern/sector-storage/tarutil/systar.go

@@ -36,6 +36,8 @@ func ExtractTar(body io.Reader, dir string) error {
 			return xerrors.Errorf("creating file %s: %w", filepath.Join(dir, header.Name), err)
 		}
 
+		// This data is coming from a trusted source, no need to check the size.
+		//nolint:gosec
 		if _, err := io.Copy(f, tr); err != nil {
 			return err
 		}

extern/storage-sealing/checks.go

@@ -79,7 +79,7 @@ func checkPrecommit(ctx context.Context, maddr address.Address, si SectorInfo, t
 		return &ErrApi{xerrors.Errorf("calling StateComputeDataCommitment: %w", err)}
 	}
 
-	if !commD.Equals(*si.CommD) {
+	if si.CommD == nil || !commD.Equals(*si.CommD) {
 		return &ErrBadCommD{xerrors.Errorf("on chain CommD differs from sector: %s != %s", commD, si.CommD)}
 	}
 

extern/storage-sealing/fsm.go

@@ -348,7 +348,9 @@ func (m *Sealing) restartSectors(ctx context.Context) error {
 				timer := time.NewTimer(cfg.WaitDealsDelay)
 				go func() {
 					<-timer.C
-					m.StartPacking(sector.SectorNumber)
+					if err := m.StartPacking(sector.SectorNumber); err != nil {
+						log.Errorf("starting sector %d: %+v", sector.SectorNumber, err)
+					}
 				}()
 			}
 		}

extern/storage-sealing/nullreader.go

@@ -3,8 +3,8 @@ package sealing
 import (
 	"io"
 
+	nr "github.com/filecoin-project/lotus/extern/storage-sealing/lib/nullreader"
 	"github.com/filecoin-project/specs-actors/actors/abi"
-	nr "github.com/filecoin-project/storage-fsm/lib/nullreader"
 )
 
 type NullReader struct {

extern/storage-sealing/sealing.go

@@ -355,7 +355,9 @@ func (m *Sealing) newDealSector() (abi.SectorNumber, error) {
 		timer := time.NewTimer(cf.WaitDealsDelay)
 		go func() {
 			<-timer.C
-			m.StartPacking(sid)
+			if err := m.StartPacking(sid); err != nil {
+				log.Errorf("starting sector %d: %+v", sid, err)
+			}
 		}()
 	}
 
@@ -396,7 +398,6 @@ func (m *Sealing) Address() address.Address {
 func getDealPerSectorLimit(size abi.SectorSize) uint64 {
 	if size < 64<<30 {
 		return 256
-	} else {
-		return 512
 	}
+	return 512
 }

extern/storage-sealing/states_sealing.go

@@ -308,6 +308,10 @@ func (m *Sealing) handleCommitting(ctx statemachine.Context, sector SectorInfo)
 
 	log.Infof("KOMIT %d %x(%d); %x(%d); %v; r:%x; d:%x", sector.SectorNumber, sector.TicketValue, sector.TicketEpoch, sector.SeedValue, sector.SeedEpoch, sector.pieceInfos(), sector.CommR, sector.CommD)
 
+	if sector.CommD == nil || sector.CommR == nil {
+		return ctx.Send(SectorCommitFailed{xerrors.Errorf("sector had nil commR or commD")})
+	}
+
 	cids := storage.SectorCids{
 		Unsealed: *sector.CommD,
 		Sealed:   *sector.CommR,

extern/storage-sealing/types.go

@@ -3,6 +3,7 @@ package sealing
 import (
 	"bytes"
 	"context"
+
 	"github.com/ipfs/go-cid"
 
 	"github.com/filecoin-project/specs-actors/actors/abi"

extern/storage-sealing/upgrade_queue.go

@@ -10,6 +10,13 @@ import (
 	"github.com/filecoin-project/specs-actors/actors/builtin/miner"
 )
 
+func (m *Sealing) IsMarkedForUpgrade(id abi.SectorNumber) bool {
+	m.upgradeLk.Lock()
+	_, found := m.toUpgrade[id]
+	m.upgradeLk.Unlock()
+	return found
+}
+
 func (m *Sealing) MarkForUpgrade(id abi.SectorNumber) error {
 	m.upgradeLk.Lock()
 	defer m.upgradeLk.Unlock()
@@ -44,6 +51,9 @@ func (m *Sealing) MarkForUpgrade(id abi.SectorNumber) error {
 }
 
 func (m *Sealing) tryUpgradeSector(ctx context.Context, params *miner.SectorPreCommitInfo) big.Int {
+	if len(params.DealIDs) == 0 {
+		return big.Zero()
+	}
 	replace := m.maybeUpgradableSector()
 	if replace != nil {
 		loc, err := m.api.StateSectorPartition(ctx, m.maddr, *replace, nil)