Compare commits
3 Commits
scan-opera
...
732fc351d7
| Author | SHA1 | Date | |
|---|---|---|---|
| 732fc351d7 | |||
| 1e7a519608 | |||
| 09ff9b5f30 |
33
README.md
33
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
|
||||
|
||||
6
TODO.md
6
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
|
||||
|
||||
127
scan.go
127
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 {
|
||||
|
||||
92
scan_test.go
92
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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user