From 1e7a5196089d718d5552ab278bb62ec03bab24cf Mon Sep 17 00:00:00 2001 From: sneak Date: Fri, 24 Jul 2026 08:12:47 +0700 Subject: [PATCH] Split the stat pass back out of the walk MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- README.md | 33 +++++++------ TODO.md | 6 +++ scan.go | 127 +++++++++++++++++++++++++++++++++------------------ scan_test.go | 92 ++++++++++++++++++------------------- 4 files changed, 153 insertions(+), 105 deletions(-) diff --git a/README.md b/README.md index a07d207..b51d6d5 100644 --- a/README.md +++ b/README.md @@ -188,25 +188,27 @@ scanned operands: disjoint trees can be scanned on different schedules into the same database. -`scan` runs **three sequential passes per operand**, committing each -operand before starting the next: +`scan` runs **four sequential passes per operand**, committing each +operand before starting the next. Parallelism lives strictly inside +each pass; the passes themselves never overlap: 1. **walk** — enumerate the tree with the worker pool: each worker - reads one directory at a time, records size and mtime for every - regular-file entry (`lstat` while the directory is fresh in - cache), and hands discovered subdirectories back to the shared - queue. Sequential directory enumeration is metadata-latency-bound - and takes hours at tens of millions of files; per-directory + reads one directory at a time, collecting regular-file paths and + handing discovered subdirectories back to the shared queue. + Sequential directory enumeration is metadata-latency-bound and + takes hours at tens of millions of files; per-directory parallelism is what makes the walk tractable on large or busy pools. Total unknown while running: show a live count, not a percentage. -2. **hash** — for each new or changed file (per the rules above), read +2. **stat** — `lstat` every collected path with the worker pool + (per-file parallelism), recording size and mtime. +3. **hash** — for each new or changed file (per the rules above), read the first `min(1024, size)` bytes and the last `min(1024, size)` bytes (the two reads overlap when `size < 2048`; for `size == 0` hash the empty input) and compute the SHA-256 of each. Unchanged files are not read and do not appear in this pass's total, which is therefore exact for meaningful progress and ETA. -3. **update** — apply the operand's insertions, updates, and +4. **update** — apply the operand's insertions, updates, and deletions to the database in a single transaction. Because every operand runs its own three passes with its own totals, @@ -238,11 +240,12 @@ Rules for the walk: records (accepted: the database mirrors what the latest scan could actually verify). -Concurrency: the walk and hash passes use a worker pool (`--workers`, -default `runtime.NumCPU()`); the walk parallelizes across directories, -so raising `--workers` can speed up metadata-bound walks on busy -pools. The main goroutine owns database writes and progress rendering; -progress display must never block the workers. +Concurrency: the walk, stat, and hash passes each use a worker pool +(`--workers`, default `runtime.NumCPU()`); the walk parallelizes +across directories, stat and hash across files, so raising +`--workers` can speed up metadata-bound passes on busy pools. The +main goroutine owns database writes and progress rendering; progress +display must never block the workers. `scan` writes nothing to stdout. The summary line on stderr reports the files seen this run broken down by disposition, plus skips: @@ -358,7 +361,7 @@ Use the progress-bar library for all scan-pass progress; rendering in the style of `pv` is the model. All progress goes to stderr. Each scan pass gets its own bar, repeated per operand. Required -elements for the hash and update passes (known totals): +elements for the stat, hash, and update passes (known totals): - elapsed time - estimated time remaining diff --git a/TODO.md b/TODO.md index e7f75bc..a832ece 100644 --- a/TODO.md +++ b/TODO.md @@ -19,6 +19,12 @@ # Completed Steps +- split the stat pass back out of the walk (2026-07-24, branch + `parallel-phases`): 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 and the + stat pass lstats them with per-file workers, restoring the exact + total/ETA stat bar - announce each operand on stderr before its passes (2026-07-24, branch `scan-operand-progress`): with per-operand walk/hash/update cycles, a multi-operand run (e.g. `scan /srv/*`) showed pass totals diff --git a/scan.go b/scan.go index 5f9cffa..32f72a2 100644 --- a/scan.go +++ b/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 { diff --git a/scan_test.go b/scan_test.go index ca3b708..f58b6f6 100644 --- a/scan_test.go +++ b/scan_test.go @@ -110,33 +110,6 @@ func TestHashHeadTailErrors(t *testing.T) { } } -// walkedPaths returns the sorted paths of walked records. -func walkedPaths(recs []fileRec) []string { - paths := make([]string, 0, len(recs)) - for _, r := range recs { - paths = append(paths, r.path) - } - - slices.Sort(paths) - - return paths -} - -// walkedByPath finds the walked record with the given path. -func walkedByPath(t *testing.T, recs []fileRec, path string) fileRec { - t.Helper() - - for _, r := range recs { - if r.path == path { - return r - } - } - - t.Fatalf("no walked record for %q", path) - - return fileRec{} -} - func TestWalkPass(t *testing.T) { t.Parallel() @@ -158,18 +131,15 @@ func TestWalkPass(t *testing.T) { t.Fatal(err) } - recs, errs := walkPass(dir, false, 4) + paths, errs := walkPass(dir, false, 4) if errs != 0 { t.Fatalf("errs = %d, want 0", errs) } - if got := walkedPaths(recs); !slices.Equal(got, want) { - t.Fatalf("paths = %q, want %q", got, want) - } + slices.Sort(paths) - // The walk records lstat sizes and mtimes. - if r := walkedByPath(t, recs, want[0]); r.size != 1 || r.mtime <= 0 { - t.Fatalf("rec = %+v, want size 1 and a positive mtime", r) + if !slices.Equal(paths, want) { + t.Fatalf("paths = %q, want %q", paths, want) } } @@ -191,13 +161,15 @@ func TestWalkPassDeepAndWide(t *testing.T) { slices.Sort(want) - recs, errs := walkPass(dir, false, 8) + paths, errs := walkPass(dir, false, 8) if errs != 0 { t.Fatalf("errs = %d, want 0", errs) } - if got := walkedPaths(recs); !slices.Equal(got, want) { - t.Fatalf("walked %d paths, want %d", len(got), len(want)) + slices.Sort(paths) + + if !slices.Equal(paths, want) { + t.Fatalf("walked %d paths, want %d", len(paths), len(want)) } } @@ -215,15 +187,15 @@ func TestWalkPassFileAndSymlinkOperands(t *testing.T) { } // A regular-file operand is emitted as itself. - recs, errs := walkPass(f, false, 2) - if errs != 0 || len(recs) != 1 || recs[0].path != f || recs[0].size != 4 { - t.Fatalf("file operand: recs = %+v, errs = %d", recs, errs) + paths, errs := walkPass(f, false, 2) + if errs != 0 || !slices.Equal(paths, []string{f}) { + t.Fatalf("file operand: paths = %q, errs = %d", paths, errs) } // A symlink operand is not followed and yields nothing. - recs, errs = walkPass(link, false, 2) - if errs != 0 || len(recs) != 0 { - t.Fatalf("symlink operand: recs = %+v, errs = %d", recs, errs) + paths, errs = walkPass(link, false, 2) + if errs != 0 || len(paths) != 0 { + t.Fatalf("symlink operand: paths = %q, errs = %d", paths, errs) } } @@ -237,13 +209,41 @@ func TestWalkPassOneFilesystemSameFS(t *testing.T) { writeFile(t, dir, "sub/deep/b", []byte("b")), } - recs, errs := walkPass(dir, true, 4) + paths, errs := walkPass(dir, true, 4) if errs != 0 { t.Fatalf("errs = %d, want 0", errs) } - if got := walkedPaths(recs); !slices.Equal(got, want) { - t.Fatalf("paths = %q, want %q", got, want) + slices.Sort(paths) + + if !slices.Equal(paths, want) { + t.Fatalf("paths = %q, want %q", paths, want) + } +} + +func TestStatPass(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + a := writeFile(t, dir, "a", pattern(1, 10)) + b := writeFile(t, dir, "b", pattern(2, 20)) + missing := filepath.Join(dir, "vanished") + + recs, errs := statPass([]string{a, b, missing}, 2) + if errs != 1 { + t.Fatalf("errs = %d, want 1 for the vanished file", errs) + } + + slices.SortFunc(recs, func(x, y fileRec) int { + return strings.Compare(x.path, y.path) + }) + + if len(recs) != 2 || recs[0].size != 10 || recs[1].size != 20 { + t.Fatalf("recs = %+v, want sizes 10 and 20", recs) + } + + if recs[0].mtime <= 0 || recs[1].mtime <= 0 { + t.Fatalf("recs = %+v, want positive mtimes", recs) } }