Split the stat pass back out of the walk

Phases are strictly sequential again — walk, stat, hash, update per
operand — with parallelism only inside each phase. The walk
enumerates paths with per-directory workers (no lstat of file
entries); the stat pass lstats every collected path with per-file
workers, restoring its exact-total/ETA progress bar and per-file
parallelism inside wide flat directories.

scan.go

@@ -135,12 +135,13 @@ func syncRoot(db *sql.DB, root string, workers int, oneFS bool,
 		return err
 	}
 
-	walked, walkErrs := walkPass(root, oneFS, workers)
-	toHash, unchanged := partitionChanged(walked, existing)
+	paths, walkErrs := walkPass(root, oneFS, workers)
+	recs, statErrs := statPass(paths, workers)
+	toHash, unchanged := partitionChanged(recs, existing)
 	hashed, hashErrs := hashPass(toHash, workers)
 
 	st.unchanged += len(unchanged)
-	st.skipped += walkErrs + hashErrs
+	st.skipped += walkErrs + statErrs + hashErrs
 
 	for _, r := range hashed {
 		if _, ok := existing[r.path]; ok {
@@ -270,23 +271,22 @@ type dirJob struct {
 }
 
 // walkEvent is one walk result delivered to the main goroutine: a
-// regular file's stat record, or a warning when fail is set.
+// regular-file path, or a warning when fail is set.
 type walkEvent struct {
-	rec  fileRec
+	path string
 	warn string
 	fail bool
 }
 
 // walkPass enumerates every regular file under root with a
-// per-directory worker pool, recording size and mtime from lstat
-// while each directory is fresh in cache. It never follows symlinks,
-// never descends into directories named .zfs, and warns and continues
-// on any per-path error. With oneFS set it never descends into a
+// per-directory worker pool. It never follows symlinks, never
+// descends into directories named .zfs, and warns and continues on
+// any per-path error. With oneFS set it never descends into a
 // directory on a different filesystem than root.
-func walkPass(root string, oneFS bool, workers int) ([]fileRec, int) {
+func walkPass(root string, oneFS bool, workers int) ([]string, int) {
 	prog := newProgress("walk", -1)
 
-	recs, initial, errs := seedRoot(root, prog)
+	paths, initial, errs := seedRoot(root, prog)
 	jobs, subdirs, events := startWalkWorkers(workers, oneFS)
 
 	dispatchDirs(initial, jobs, subdirs)
@@ -300,22 +300,22 @@ func walkPass(root string, oneFS bool, workers int) ([]fileRec, int) {
 			continue
 		}
 
-		recs = append(recs, ev.rec)
+		paths = append(paths, ev.path)
 
 		prog.increment()
 	}
 
 	prog.finish()
 
-	return recs, errs
+	return paths, errs
 }
 
 // seedRoot turns the PATH operand into the walk's starting state: a
-// regular-file operand becomes a record directly, a directory operand
+// regular-file operand becomes a path directly, a directory operand
 // becomes the initial job, and a symlink or other non-regular operand
 // yields nothing (symlinks are never followed, including as
 // operands).
-func seedRoot(root string, prog *progress) ([]fileRec, []dirJob, int) {
+func seedRoot(root string, prog *progress) ([]string, []dirJob, int) {
 	fi, err := os.Lstat(root)
 	if err != nil {
 		prog.warnf("walk %s: %v", root, err)
@@ -333,11 +333,9 @@ func seedRoot(root string, prog *progress) ([]fileRec, []dirJob, int) {
 
 		return nil, []dirJob{{path: root, rootDev: dev, rootDevOK: ok}}, 0
 	case fi.Mode().IsRegular():
-		rec := fileRec{path: root, size: fi.Size(), mtime: fi.ModTime().Unix()}
-
 		prog.increment()
 
-		return []fileRec{rec}, nil, 0
+		return []string{root}, nil, 0
 	default:
 		return nil, nil, 0
 	}
@@ -439,37 +437,12 @@ func walkOneDir(job dirJob, oneFS bool, events chan<- walkEvent) []dirJob {
 			continue
 		}
 
-		events <- fileEvent(p, e)
+		events <- walkEvent{path: p}
 	}
 
 	return subs
 }
 
-// fileEvent lstats one regular-file directory entry into its walk
-// event.
-func fileEvent(p string, e fs.DirEntry) walkEvent {
-	fi, err := e.Info()
-
-	switch {
-	case err != nil:
-		return walkEvent{
-			warn: fmt.Sprintf("walk %s: %v", p, err),
-			fail: true,
-		}
-	case !fi.Mode().IsRegular():
-		return walkEvent{
-			warn: fmt.Sprintf("walk %s: %v", p, errNotRegular),
-			fail: true,
-		}
-	default:
-		return walkEvent{rec: fileRec{
-			path:  p,
-			size:  fi.Size(),
-			mtime: fi.ModTime().Unix(),
-		}}
-	}
-}
-
 // subdirJob applies the descent rules to directory p: never enter
 // .zfs (ZFS snapshot pseudo-dirs would list every file once per
 // snapshot), and with -x never enter a directory on a different
@@ -514,6 +487,72 @@ func deviceOfInfo(fi fs.FileInfo) (uint64, bool) {
 	return statDev(st), true
 }
 
+// statPass lstats every collected path in a worker pool, recording size
+// and mtime. Paths that fail to stat (or are no longer regular files)
+// are warned about and dropped.
+func statPass(paths []string, workers int) ([]fileRec, int) {
+	type result struct {
+		rec fileRec
+		err error
+	}
+
+	jobs := make(chan string, workQueueDepth)
+	results := make(chan result, workQueueDepth)
+
+	for range workers {
+		go func() {
+			for p := range jobs {
+				fi, err := os.Lstat(p)
+				switch {
+				case err != nil:
+					results <- result{rec: fileRec{path: p}, err: err}
+				case !fi.Mode().IsRegular():
+					results <- result{
+						rec: fileRec{path: p},
+						err: errNotRegular,
+					}
+				default:
+					results <- result{rec: fileRec{
+						path:  p,
+						size:  fi.Size(),
+						mtime: fi.ModTime().Unix(),
+					}}
+				}
+			}
+		}()
+	}
+
+	go func() {
+		for _, p := range paths {
+			jobs <- p
+		}
+
+		close(jobs)
+	}()
+
+	prog := newProgress("stat", int64(len(paths)))
+
+	var errs int
+
+	recs := make([]fileRec, 0, len(paths))
+	for range paths {
+		r := <-results
+		if r.err != nil {
+			errs++
+
+			prog.warnf("stat %s: %v", r.rec.path, r.err)
+		} else {
+			recs = append(recs, r.rec)
+		}
+
+		prog.increment()
+	}
+
+	prog.finish()
+
+	return recs, errs
+}
+
 // hashResult carries one file's head/tail hashes (or the error that
 // prevented hashing it) from the hash workers to the main goroutine.
 type hashResult struct {