From dced5cf0d2bc74aac1e19e72f770bd1add95ab07 Mon Sep 17 00:00:00 2001 From: sneak Date: Fri, 24 Jul 2026 07:44:18 +0700 Subject: [PATCH] Parallelize the walk and commit per operand MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the single-goroutine WalkDir traversal with a per-directory worker pool: workers read directories concurrently and lstat entries while each directory is fresh in cache, recording size and mtime during the walk. This folds the separate stat pass away (halving metadata I/O per run) and overlaps metadata latency, which dominated on busy pools — a sequential walk of a ~22M-file tree was observed taking over 4 hours. Each PATH operand now loads its scope, walks, hashes, and commits in its own transaction, so an interrupted scan keeps every operand completed so far; a later overlapping operand sees the records committed by earlier ones and reuses them unchanged. --- TODO.md | 19 +-- scan.go | 436 +++++++++++++++++++++++++++++---------------------- scan_test.go | 160 ++++++++++--------- 3 files changed, 344 insertions(+), 271 deletions(-) diff --git a/TODO.md b/TODO.md index bf2c863..edd38fc 100644 --- a/TODO.md +++ b/TODO.md @@ -14,16 +14,19 @@ # Next Step -- parallel walk (branch `parallel-walk`): the walk pass is a single - goroutine and takes hours at ~20M files on a busy pool (observed: - 22M files in 4h on a ZFS server); make it a per-directory - worker-pool traversal that records size/mtime during the walk - (folding away the separate stat pass, halving metadata I/O), and - commit each `PATH` operand in its own transaction so an interrupted - scan keeps completed operands +- convert Makefile targets to scripts-to-rule-them-all `script/` + entrypoints like the other managed repos # Completed Steps +- parallel walk (2026-07-24, branch `parallel-walk`): the walk pass + was a single goroutine and took hours at ~20M files on a busy pool + (observed: 22M files in 4h on a ZFS server); it is now a + per-directory worker-pool traversal that records size/mtime during + the walk (folding away the separate stat pass, halving metadata + I/O), and each `PATH` operand commits in its own transaction so an + interrupted scan keeps completed operands + - persistent scan database (2026-07-24, branch `persistent-database`): `scan` now maintains a SQLite database (`modernc.org/sqlite`, pure Go, cgo stays disabled) keyed by absolute path that survives between @@ -49,8 +52,6 @@ # Future Steps -- convert Makefile targets to scripts-to-rule-them-all `script/` - entrypoints like the other managed repos - possible later features (explicitly out of scope per README): full-content verification of candidates, removal-script helpers diff --git a/scan.go b/scan.go index b70faf3..5ac6958 100644 --- a/scan.go +++ b/scan.go @@ -11,6 +11,7 @@ import ( "path/filepath" "slices" "strings" + "sync" "syscall" ) @@ -21,8 +22,8 @@ const chunk = 1024 // and hash worker pools. const workQueueDepth = 1024 -// errNotRegular reports a path that stopped being a regular file -// between the walk and stat passes. +// errNotRegular reports a path whose type changed between the +// directory read and its lstat. var errNotRegular = errors.New("no longer a regular file") // fileRec carries one file between the stat and hash passes. @@ -98,27 +99,42 @@ type scanStats struct { skipped int } -// syncScan synchronizes the database with the filesystem under roots: -// walk and stat everything, hash only new or changed files, and apply -// the resulting record insertions, updates, and deletions in a single -// transaction. Records outside the roots are never touched. +// syncScan synchronizes the database with the filesystem under roots. +// Operands are processed in order, each one walked, hashed, and +// committed independently, so an interrupted scan keeps every operand +// completed so far. Records outside the roots are never touched. func syncScan(db *sql.DB, roots []string, workers int, oneFS bool, ) (scanStats, error) { var st scanStats - existing, err := loadScopedRows(db, roots) - if err != nil { - return st, err + for _, root := range roots { + err := syncRoot(db, root, workers, oneFS, &st) + if err != nil { + return st, err + } } - paths, walkErrs := walkPass(roots, oneFS) - recs, statErrs := statPass(uniquePaths(paths), workers) - toHash, unchanged := partitionChanged(recs, existing) + return st, nil +} + +// syncRoot walks one PATH operand, hashes its new and changed files, +// and commits the operand's record insertions, updates, and deletions +// in a single transaction. +func syncRoot(db *sql.DB, root string, workers int, oneFS bool, + st *scanStats, +) error { + existing, err := loadScopedRows(db, root) + if err != nil { + return err + } + + walked, walkErrs := walkPass(root, oneFS, workers) + toHash, unchanged := partitionChanged(walked, existing) hashed, hashErrs := hashPass(toHash, workers) - st.unchanged = len(unchanged) - st.skipped = walkErrs + statErrs + hashErrs + st.unchanged += len(unchanged) + st.skipped += walkErrs + hashErrs for _, r := range hashed { if _, ok := existing[r.path]; ok { @@ -129,15 +145,15 @@ func syncScan(db *sql.DB, roots []string, workers int, } deletes := collectDeletes(existing, unchanged, hashed) - st.removed = len(deletes) + st.removed += len(deletes) - return st, applyPass(db, hashed, deletes) + return applyPass(db, hashed, deletes) } -// loadScopedRows loads the database records whose paths lie under any -// of the scan roots, keyed by path. Records outside the roots belong -// to other trees and are left untouched by this scan. -func loadScopedRows(db *sql.DB, roots []string) (map[string]scanRec, error) { +// loadScopedRows loads the database records whose paths lie under +// root, keyed by path. Records outside the scanned operands belong to +// other trees and are left untouched. +func loadScopedRows(db *sql.DB, root string) (map[string]scanRec, error) { all, err := loadFileRows(db) if err != nil { return nil, err @@ -146,7 +162,7 @@ func loadScopedRows(db *sql.DB, roots []string) (map[string]scanRec, error) { scoped := make(map[string]scanRec) for _, r := range all { - if underAnyRoot(r.path, roots) { + if underRoot(r.path, root) { scoped[r.path] = r } } @@ -154,18 +170,6 @@ func loadScopedRows(db *sql.DB, roots []string) (map[string]scanRec, error) { return scoped, nil } -// underAnyRoot reports whether path is any of the roots or lies under -// one of them. -func underAnyRoot(path string, roots []string) bool { - for _, root := range roots { - if underRoot(path, root) { - return true - } - } - - return false -} - // underRoot reports whether path is root itself or lies under it. Both // must be absolute and lexically clean. func underRoot(path, root string) bool { @@ -181,26 +185,6 @@ func underRoot(path, root string) bool { return strings.HasPrefix(path, prefix) } -// uniquePaths deduplicates the walked paths, preserving order. -// Overlapping operands can reach the same file more than once, but the -// database keys records by path, so each file is processed once. -func uniquePaths(paths []string) []string { - seen := make(map[string]bool, len(paths)) - out := make([]string, 0, len(paths)) - - for _, p := range paths { - if seen[p] { - continue - } - - seen[p] = true - - out = append(out, p) - } - - return out -} - // partitionChanged splits the stat results into files that must be // hashed (new, or changed) and files whose existing records are reused // without reading them: a file is unchanged when its statted size @@ -270,105 +254,247 @@ func applyPass(db *sql.DB, upserts []scanRec, deletes []string) error { return err } -// treeWalker carries the walk-pass state shared by all PATH operands. -type treeWalker struct { - prog *progress - oneFS bool - paths []string - errs int +// dirJob is one directory awaiting traversal by the walk workers. It +// carries its operand's filesystem device so -x can stop at +// filesystem boundaries. +type dirJob struct { + path string + rootDev uint64 + rootDevOK bool } -// walkPass enumerates every regular file under each root operand in -// order. 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 its root operand. -func walkPass(roots []string, oneFS bool) ([]string, int) { - w := &treeWalker{prog: newProgress("walk", -1), oneFS: oneFS} - - for _, root := range roots { - w.walkRoot(root) - } - - w.prog.finish() - - return w.paths, w.errs +// walkEvent is one walk result delivered to the main goroutine: a +// regular file's stat record, or a warning when fail is set. +type walkEvent struct { + rec fileRec + warn string + fail bool } -// walkRoot walks a single PATH operand, appending regular-file paths. -func (w *treeWalker) walkRoot(root string) { - rootDev, rootDevOK := deviceOf(root) +// 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 +// directory on a different filesystem than root. +func walkPass(root string, oneFS bool, workers int) ([]fileRec, int) { + prog := newProgress("walk", -1) - walkErr := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error { - if err != nil { - w.errs++ + recs, initial, errs := seedRoot(root, prog) + jobs, subdirs, events := startWalkWorkers(workers, oneFS) - w.prog.warnf("walk %s: %v", p, err) + dispatchDirs(initial, jobs, subdirs) - if d != nil && d.IsDir() { - return filepath.SkipDir - } + for ev := range events { + if ev.fail { + errs++ - return nil + prog.warnf("%s", ev.warn) + + continue } - if d.IsDir() { - return w.dirAction(p, d, rootDev, rootDevOK) + recs = append(recs, ev.rec) + + prog.increment() + } + + prog.finish() + + return recs, errs +} + +// seedRoot turns the PATH operand into the walk's starting state: a +// regular-file operand becomes a record 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) { + fi, err := os.Lstat(root) + if err != nil { + prog.warnf("walk %s: %v", root, err) + + return nil, nil, 1 + } + + switch { + case fi.IsDir(): + if filepath.Base(root) == ".zfs" { + return nil, nil, 0 + } + + dev, ok := deviceOfInfo(fi) + + 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 + default: + return nil, nil, 0 + } +} + +// startWalkWorkers starts the walk worker pool. Each worker processes +// one directory at a time, emitting an event per regular file and +// handing discovered subdirectories back to the dispatcher; events is +// closed once every worker has finished. +func startWalkWorkers(workers int, + oneFS bool, +) (chan dirJob, chan []dirJob, chan walkEvent) { + jobs := make(chan dirJob, workQueueDepth) + subdirs := make(chan []dirJob, workers) + events := make(chan walkEvent, workQueueDepth) + + var wg sync.WaitGroup + + for range workers { + wg.Go(func() { + for job := range jobs { + subdirs <- walkOneDir(job, oneFS, events) + } + }) + } + + go func() { + wg.Wait() + close(events) + }() + + return jobs, subdirs, events +} + +// dispatchDirs feeds directory jobs to the walk workers, queueing +// newly discovered subdirectories (newest first, which keeps the +// frontier small) until every directory has been processed, then +// closes jobs. +func dispatchDirs(initial []dirJob, jobs chan<- dirJob, + subdirs <-chan []dirJob, +) { + go func() { + queue := slices.Clone(initial) + pending := len(queue) + + for pending > 0 { + var ( + out chan<- dirJob + next dirJob + ) + + if len(queue) > 0 { + out = jobs + next = queue[len(queue)-1] + } + + select { + case out <- next: + queue = queue[:len(queue)-1] + case subs := <-subdirs: + pending += len(subs) - 1 + queue = append(queue, subs...) + } + } + + close(jobs) + }() +} + +// walkOneDir reads one directory, emitting an event per regular-file +// entry and a warning event per unreadable one, and returns the +// subdirectories to descend into. +func walkOneDir(job dirJob, oneFS bool, events chan<- walkEvent) []dirJob { + entries, err := os.ReadDir(job.path) + if err != nil { + events <- walkEvent{ + warn: fmt.Sprintf("walk %s: %v", job.path, err), + fail: true, + } + + return nil + } + + var subs []dirJob + + for _, e := range entries { + p := filepath.Join(job.path, e.Name()) + + if e.IsDir() { + if sub, ok := subdirJob(p, e, job, oneFS, events); ok { + subs = append(subs, sub) + } + + continue } // Regular files only: skip symlinks, sockets, FIFOs, and // device nodes. - if !d.Type().IsRegular() { - return nil + if !e.Type().IsRegular() { + continue } - w.paths = append(w.paths, p) + events <- fileEvent(p, e) + } - w.prog.increment() + return subs +} - return nil - }) - if walkErr != nil { - fatalf("walk %s: %v", root, walkErr) +// 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(), + }} } } -// dirAction decides whether the walk descends into directory p. -func (w *treeWalker) dirAction(p string, d fs.DirEntry, rootDev uint64, rootDevOK bool) error { - // ZFS snapshot pseudo-dirs would list every file once per - // snapshot; never descend. - if d.Name() == ".zfs" { - return filepath.SkipDir +// 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 +// filesystem than its operand. +func subdirJob(p string, e fs.DirEntry, parent dirJob, oneFS bool, + events chan<- walkEvent, +) (dirJob, bool) { + if e.Name() == ".zfs" { + return dirJob{}, false } - if !w.oneFS || !rootDevOK { - return nil + job := dirJob{path: p, rootDev: parent.rootDev, rootDevOK: parent.rootDevOK} + if !oneFS || !parent.rootDevOK { + return job, true } - info, err := d.Info() + info, err := e.Info() if err != nil { - w.errs++ + events <- walkEvent{ + warn: fmt.Sprintf("walk %s: %v", p, err), + fail: true, + } - w.prog.warnf("walk %s: %v", p, err) - - return filepath.SkipDir + return dirJob{}, false } - if dev, ok := deviceOfInfo(info); ok && dev != rootDev { - return filepath.SkipDir + if dev, ok := deviceOfInfo(info); ok && dev != parent.rootDev { + return dirJob{}, false } - return nil -} - -// deviceOf returns the filesystem device ID of path without following -// symlinks. -func deviceOf(path string) (uint64, bool) { - fi, err := os.Lstat(path) - if err != nil { - return 0, false - } - - return deviceOfInfo(fi) + return job, true } // deviceOfInfo extracts the filesystem device ID from a FileInfo, when @@ -382,72 +508,6 @@ 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 2a3130d..ca3b708 100644 --- a/scan_test.go +++ b/scan_test.go @@ -5,6 +5,7 @@ import ( "crypto/sha256" "database/sql" "encoding/hex" + "fmt" "os" "path/filepath" "slices" @@ -109,6 +110,33 @@ 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() @@ -130,37 +158,46 @@ func TestWalkPass(t *testing.T) { t.Fatal(err) } - paths, errs := walkPass([]string{dir}, false) + recs, errs := walkPass(dir, false, 4) if errs != 0 { t.Fatalf("errs = %d, want 0", errs) } - slices.Sort(paths) + if got := walkedPaths(recs); !slices.Equal(got, want) { + t.Fatalf("paths = %q, want %q", got, want) + } - if !slices.Equal(paths, want) { - t.Fatalf("paths = %q, want %q", paths, want) + // 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) } } -func TestWalkPassMultipleRoots(t *testing.T) { +func TestWalkPassDeepAndWide(t *testing.T) { t.Parallel() - rootA := t.TempDir() - rootB := t.TempDir() - want := []string{ - writeFile(t, rootA, "a1", []byte("1")), - writeFile(t, rootA, "sub/a2", []byte("2")), - writeFile(t, rootB, "b1", []byte("3")), + // Exercise the dispatcher with more directories than workers and + // with nesting deeper than the worker count. + dir := t.TempDir() + deep := "deep" + strings.Repeat("/d", 30) + + want := make([]string, 0, 41) + want = append(want, writeFile(t, dir, deep+"/f", []byte("x"))) + + for i := range 40 { + want = append(want, writeFile(t, dir, + fmt.Sprintf("wide/%02d/f", i), []byte("y"))) } - paths, errs := walkPass([]string{rootA, rootB}, false) + slices.Sort(want) + + recs, errs := walkPass(dir, false, 8) if errs != 0 { t.Fatalf("errs = %d, want 0", errs) } - // Operands are walked in the order given. - if !slices.Equal(paths, want) { - t.Fatalf("paths = %q, want %q", paths, want) + if got := walkedPaths(recs); !slices.Equal(got, want) { + t.Fatalf("walked %d paths, want %d", len(got), len(want)) } } @@ -178,15 +215,15 @@ func TestWalkPassFileAndSymlinkOperands(t *testing.T) { } // A regular-file operand is emitted as itself. - paths, errs := walkPass([]string{f}, false) - if errs != 0 || !slices.Equal(paths, []string{f}) { - t.Fatalf("file operand: paths = %q, errs = %d", paths, errs) + 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) } // A symlink operand is not followed and yields nothing. - paths, errs = walkPass([]string{link}, false) - if errs != 0 || len(paths) != 0 { - t.Fatalf("symlink operand: paths = %q, errs = %d", paths, errs) + recs, errs = walkPass(link, false, 2) + if errs != 0 || len(recs) != 0 { + t.Fatalf("symlink operand: recs = %+v, errs = %d", recs, errs) } } @@ -200,58 +237,37 @@ func TestWalkPassOneFilesystemSameFS(t *testing.T) { writeFile(t, dir, "sub/deep/b", []byte("b")), } - paths, errs := walkPass([]string{dir}, true) + recs, errs := walkPass(dir, true, 4) if errs != 0 { t.Fatalf("errs = %d, want 0", errs) } - slices.Sort(paths) - - if !slices.Equal(paths, want) { - t.Fatalf("paths = %q, want %q", paths, want) + if got := walkedPaths(recs); !slices.Equal(got, want) { + t.Fatalf("paths = %q, want %q", got, want) } } -func TestDeviceOf(t *testing.T) { +func TestDeviceOfInfo(t *testing.T) { t.Parallel() dir := t.TempDir() - dev1, ok1 := deviceOf(dir) + fi1, err := os.Lstat(dir) + if err != nil { + t.Fatal(err) + } - dev2, ok2 := deviceOf(dir) + fi2, err := os.Lstat(dir) + if err != nil { + t.Fatal(err) + } + + dev1, ok1 := deviceOfInfo(fi1) + + dev2, ok2 := deviceOfInfo(fi2) if !ok1 || !ok2 || dev1 != dev2 { - t.Fatalf("deviceOf unstable: %d/%v vs %d/%v", dev1, ok1, dev2, ok2) - } - - if _, ok := deviceOf(filepath.Join(dir, "missing")); ok { - t.Fatal("deviceOf reported ok for a missing path") - } -} - -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) + t.Fatalf("deviceOfInfo unstable: %d/%v vs %d/%v", + dev1, ok1, dev2, ok2) } } @@ -607,10 +623,17 @@ func TestSyncScanOverlappingRoots(t *testing.T) { writeFile(t, dir, "sub/f", pattern(1, 10)) - // A file reachable via two overlapping operands yields one record. + // A file reachable via two overlapping operands yields one + // record: the second operand sees the rows the first operand + // committed and reuses them unchanged, which also proves each + // operand commits before the next starts. st := syncTree(t, db, dir, filepath.Join(dir, "sub")) - if st.added != 1 { - t.Fatalf("stats = %+v, want 1 added", st) + if st != (scanStats{added: 1, unchanged: 1}) { + t.Fatalf("stats = %+v, want 1 added 1 unchanged", st) + } + + if got := recordPaths(dbRecords(t, db)); len(got) != 1 { + t.Fatalf("records = %q, want exactly one", got) } } @@ -659,14 +682,3 @@ func TestUnderRoot(t *testing.T) { } } } - -func TestUniquePaths(t *testing.T) { - t.Parallel() - - got := uniquePaths([]string{"/a", "/b", "/a", "/c", "/b"}) - - want := []string{"/a", "/b", "/c"} - if !slices.Equal(got, want) { - t.Fatalf("uniquePaths = %q, want %q", got, want) - } -}