Stat in the walk, hash only shared sizes, flush batches mid-scan
Restructure scan into three phases: walk+stat, hash, update. The stat pass is folded into the walk workers: each regular file is lstatted as its directory is read, while the metadata is hot. The walk builds a scan-wide size census (walked files plus records outside the scan roots), and unchanged already-hashed files resolve during the walk without further work. Only files whose size at least one other file shares are ever read: a size-unique file cannot be a duplicate, so it is recorded without hashes (head and tail empty). When a later scan makes its size shared, the file is hashed then, even if otherwise unchanged. report excludes unhashed records; trees gives them a never-matching signature so a tree containing one never compares equal to another. Hashed records are committed in batched transactions while the hash phase runs, so an interrupted scan keeps everything hashed so far and the next run resumes cheaply. The hash phase total is exact, giving a meaningful ETA. Memory drops accordingly: the existing-record index holds only path, size, mtime, and a hashed flag (no hash values); the walk carries one small record per candidate file; overlapping operands are pruned up front instead of deduplicating every walked path in a scan-wide set. Files no bigger than one chunk are hashed with a single read.
This commit is contained in:
714
scan.go
714
scan.go
@@ -4,7 +4,6 @@ import (
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
@@ -18,25 +17,34 @@ import (
|
||||
// chunk is the number of bytes hashed from each end of a file.
|
||||
const chunk = 1024
|
||||
|
||||
// workQueueDepth bounds the job and result channels feeding the stat
|
||||
// workQueueDepth bounds the job and result channels feeding the walk
|
||||
// and hash worker pools.
|
||||
const workQueueDepth = 1024
|
||||
|
||||
// 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.
|
||||
// fileRec carries one statted file between the scan phases.
|
||||
type fileRec struct {
|
||||
path string
|
||||
size int64
|
||||
mtime int64
|
||||
}
|
||||
|
||||
// runScan implements the scan subcommand: four sequential passes
|
||||
// (walk, stat, hash, update) that synchronize the persistent database
|
||||
// with the filesystem state under the PATH operands. Flag parsing and
|
||||
// the at-least-one-operand check are done by cobra.
|
||||
// fileMeta is the in-memory index entry for one existing database
|
||||
// record: just enough for change detection, plus whether the record
|
||||
// carries hashes. Hashes stay on disk; at tens of millions of records
|
||||
// they would dominate the scan's memory.
|
||||
type fileMeta struct {
|
||||
size int64
|
||||
mtime int64
|
||||
hashed bool
|
||||
}
|
||||
|
||||
// runScan implements the scan subcommand: three sequential phases —
|
||||
// walk (which stats each file as it is discovered), hash, update —
|
||||
// that synchronize the persistent database with the filesystem state
|
||||
// under the PATH operands. Only files whose size at least one other
|
||||
// file shares are ever hashed: a size-unique file cannot be a
|
||||
// duplicate. Flag parsing and the at-least-one-operand check are done
|
||||
// by cobra.
|
||||
func runScan(roots []string, workers int, oneFS bool) {
|
||||
if workers < 1 {
|
||||
workers = 1
|
||||
@@ -89,6 +97,33 @@ func resolveRoots(roots []string) []string {
|
||||
return abs
|
||||
}
|
||||
|
||||
// pruneRoots drops operands already covered by another operand:
|
||||
// duplicates and any operand lying under another one. Every remaining
|
||||
// file is then reachable through exactly one root, so walked paths are
|
||||
// unique without keeping a scan-wide set of every path seen.
|
||||
func pruneRoots(roots []string) []string {
|
||||
// Shorter-first ordering guarantees an ancestor is kept before any
|
||||
// operand under it is considered.
|
||||
sorted := slices.Clone(roots)
|
||||
slices.SortFunc(sorted, func(a, b string) int {
|
||||
if len(a) != len(b) {
|
||||
return len(a) - len(b)
|
||||
}
|
||||
|
||||
return strings.Compare(a, b)
|
||||
})
|
||||
|
||||
kept := make([]string, 0, len(sorted))
|
||||
|
||||
for _, r := range sorted {
|
||||
if !underAnyRoot(r, kept) {
|
||||
kept = append(kept, r)
|
||||
}
|
||||
}
|
||||
|
||||
return kept
|
||||
}
|
||||
|
||||
// scanStats summarizes one scan's database synchronization for the
|
||||
// final stderr summary.
|
||||
type scanStats struct {
|
||||
@@ -99,60 +134,289 @@ type scanStats struct {
|
||||
skipped int
|
||||
}
|
||||
|
||||
// syncScan synchronizes the database with the filesystem under roots:
|
||||
// four scan-wide passes — walk, stat, hash, update — each parallel
|
||||
// internally, run strictly in sequence over all operands together.
|
||||
// Records outside the roots are never touched.
|
||||
// scanState carries one scan's evolving state across its phases.
|
||||
// Entries consumed from existing mark files verified this run;
|
||||
// whatever remains after the walk and hash phases is deleted by the
|
||||
// update phase.
|
||||
type scanState struct {
|
||||
db *sql.DB
|
||||
existing map[string]fileMeta
|
||||
sizes []int64 // size census: every walked file, plus records outside the roots
|
||||
toHash []fileRec // files whose size is shared: must be read
|
||||
sentinels []fileRec // new/changed size-unique files: recorded without hashes
|
||||
batch []scanRec
|
||||
st scanStats
|
||||
}
|
||||
|
||||
// syncScan synchronizes the database with the filesystem under roots
|
||||
// in three sequential phases: walk (enumerate and stat every file,
|
||||
// building a complete size census), hash (read only the new or
|
||||
// changed — or previously unhashed — files whose size at least one
|
||||
// other file shares, committing results in batches as they arrive),
|
||||
// and update (record the size-unique files without reading them, and
|
||||
// delete the records the scan no longer verifies). Records outside
|
||||
// the roots are never touched.
|
||||
func syncScan(db *sql.DB, roots []string, workers int,
|
||||
oneFS bool,
|
||||
) (scanStats, error) {
|
||||
var st scanStats
|
||||
roots = pruneRoots(roots)
|
||||
|
||||
existing, err := loadScopedRows(db, roots)
|
||||
s := &scanState{db: db}
|
||||
|
||||
err := s.loadIndex(roots)
|
||||
if err != nil {
|
||||
return st, err
|
||||
return s.st, err
|
||||
}
|
||||
|
||||
paths, walkErrs := walkPass(roots, oneFS, workers)
|
||||
recs, statErrs := statPass(uniquePaths(paths), workers)
|
||||
toHash, unchanged := partitionChanged(recs, existing)
|
||||
hashed, hashErrs := hashPass(toHash, workers)
|
||||
changed, unhashed := s.walkPhase(startWalk(roots, oneFS, workers))
|
||||
|
||||
st.unchanged = len(unchanged)
|
||||
st.skipped = walkErrs + statErrs + hashErrs
|
||||
s.partition(changed, unhashed)
|
||||
|
||||
for _, r := range hashed {
|
||||
if _, ok := existing[r.path]; ok {
|
||||
st.updated++
|
||||
} else {
|
||||
st.added++
|
||||
}
|
||||
err = s.hashPhase(workers)
|
||||
if err != nil {
|
||||
return s.st, err
|
||||
}
|
||||
|
||||
deletes := collectDeletes(existing, unchanged, hashed)
|
||||
st.removed = len(deletes)
|
||||
|
||||
return st, applyPass(db, hashed, deletes)
|
||||
return s.st, s.updatePhase()
|
||||
}
|
||||
|
||||
// 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.
|
||||
func loadScopedRows(db *sql.DB, roots []string) (map[string]scanRec, error) {
|
||||
all, err := loadFileRows(db)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// loadIndex indexes the database records under the scan roots for
|
||||
// change detection and collects the sizes of every record outside
|
||||
// them: out-of-scope records join the size census so a scanned file
|
||||
// can be recognized as a possible duplicate of a tree scanned
|
||||
// separately into the same database.
|
||||
func (s *scanState) loadIndex(roots []string) error {
|
||||
s.existing = make(map[string]fileMeta)
|
||||
|
||||
scoped := make(map[string]scanRec)
|
||||
return loadFileMeta(s.db,
|
||||
func(path string, size, mtime int64, hashed bool) {
|
||||
if underAnyRoot(path, roots) {
|
||||
s.existing[path] = fileMeta{
|
||||
size: size, mtime: mtime, hashed: hashed,
|
||||
}
|
||||
|
||||
for _, r := range all {
|
||||
if underAnyRoot(r.path, roots) {
|
||||
scoped[r.path] = r
|
||||
return
|
||||
}
|
||||
|
||||
s.sizes = append(s.sizes, size)
|
||||
})
|
||||
}
|
||||
|
||||
// walkPhase drains the walk, appending every walked file's size to
|
||||
// the census and resolving what it can immediately: an unchanged file
|
||||
// whose record already has hashes needs nothing further. It returns
|
||||
// the new-or-changed files and the unchanged files whose records lack
|
||||
// hashes; both remain candidates until the census decides whether
|
||||
// their sizes are shared.
|
||||
func (s *scanState) walkPhase(
|
||||
events <-chan walkEvent,
|
||||
) ([]fileRec, []fileRec) {
|
||||
prog := newProgress("walk", -1)
|
||||
|
||||
var changed, unhashed []fileRec
|
||||
|
||||
for ev := range events {
|
||||
if ev.fail {
|
||||
s.st.skipped++
|
||||
|
||||
prog.warnf("%s", ev.warn)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
s.sizes = append(s.sizes, ev.rec.size)
|
||||
|
||||
prog.increment()
|
||||
|
||||
old, ok := s.existing[ev.rec.path]
|
||||
|
||||
switch {
|
||||
case !ok || old.size != ev.rec.size || old.mtime < ev.rec.mtime:
|
||||
changed = append(changed, ev.rec)
|
||||
case old.hashed:
|
||||
delete(s.existing, ev.rec.path)
|
||||
|
||||
s.st.unchanged++
|
||||
default:
|
||||
unhashed = append(unhashed, ev.rec)
|
||||
}
|
||||
}
|
||||
|
||||
return scoped, nil
|
||||
prog.finish()
|
||||
|
||||
return changed, unhashed
|
||||
}
|
||||
|
||||
// partition decides each candidate's disposition now that the size
|
||||
// census is complete. A file whose size no other file shares cannot
|
||||
// be a duplicate and is never read: a new or changed one is recorded
|
||||
// without hashes, an unchanged unhashed one keeps its record. Every
|
||||
// file with a shared size queues for the hash phase.
|
||||
func (s *scanState) partition(changed, unhashed []fileRec) {
|
||||
slices.Sort(s.sizes)
|
||||
|
||||
for _, rec := range changed {
|
||||
if s.sizeShared(rec.size) {
|
||||
s.toHash = append(s.toHash, rec)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
s.sentinels = append(s.sentinels, rec)
|
||||
s.resolve(rec.path)
|
||||
}
|
||||
|
||||
for _, rec := range unhashed {
|
||||
if s.sizeShared(rec.size) {
|
||||
s.toHash = append(s.toHash, rec)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
delete(s.existing, rec.path)
|
||||
|
||||
s.st.unchanged++
|
||||
}
|
||||
|
||||
s.sizes = nil
|
||||
}
|
||||
|
||||
// sizeShared reports whether at least two census entries have this
|
||||
// size. Every candidate's own size is in the census exactly once, so
|
||||
// a second entry means another file (or a record outside the scan
|
||||
// roots) could share its content.
|
||||
func (s *scanState) sizeShared(size int64) bool {
|
||||
i, found := slices.BinarySearch(s.sizes, size)
|
||||
|
||||
return found && i+1 < len(s.sizes) && s.sizes[i+1] == size
|
||||
}
|
||||
|
||||
// resolve counts one written record as added or updated and marks its
|
||||
// path verified.
|
||||
func (s *scanState) resolve(path string) {
|
||||
if _, ok := s.existing[path]; ok {
|
||||
s.st.updated++
|
||||
|
||||
delete(s.existing, path)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
s.st.added++
|
||||
}
|
||||
|
||||
// hashPhase hashes every queued file with the worker pool, committing
|
||||
// completed records to the database in batches as results arrive, so
|
||||
// a long scan persists its progress as it goes (an interrupted scan
|
||||
// resumes cheaply: the next run skips everything already recorded).
|
||||
// The total is exact, so the bar shows a real ETA. Files that fail to
|
||||
// hash are warned about and skipped; their stale records, if any, are
|
||||
// deleted by the update phase.
|
||||
func (s *scanState) hashPhase(workers int) error {
|
||||
jobs := make(chan fileRec, workQueueDepth)
|
||||
results := make(chan hashResult, workQueueDepth)
|
||||
|
||||
startHashWorkers(jobs, results, workers)
|
||||
|
||||
// The feeder ranges over its own reference: s.toHash is released
|
||||
// below while the feeder may still be running.
|
||||
toHash := s.toHash
|
||||
s.toHash = nil
|
||||
|
||||
go func() {
|
||||
for _, rec := range toHash {
|
||||
jobs <- rec
|
||||
}
|
||||
|
||||
close(jobs)
|
||||
}()
|
||||
|
||||
prog := newProgress("hash", int64(len(toHash)))
|
||||
defer prog.finish()
|
||||
|
||||
for range toHash {
|
||||
r := <-results
|
||||
|
||||
prog.increment()
|
||||
|
||||
if r.err != nil {
|
||||
s.st.skipped++
|
||||
|
||||
prog.warnf("hash %s: %v", r.rec.path, r.err)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
s.resolve(r.rec.path)
|
||||
|
||||
s.batch = append(s.batch, scanRec{
|
||||
size: r.rec.size,
|
||||
mtime: r.rec.mtime,
|
||||
head: r.head,
|
||||
tail: r.tail,
|
||||
path: r.rec.path,
|
||||
})
|
||||
|
||||
if len(s.batch) < updateBatchSize {
|
||||
continue
|
||||
}
|
||||
|
||||
err := applyBatch(s.db, s.batch, nil, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.batch = s.batch[:0]
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// updatePhase writes the scan's tail under one progress display: the
|
||||
// final partial batch of hashed records, a hash-less record for every
|
||||
// size-unique new or changed file, and deletions for every record the
|
||||
// scan did not verify (vanished files, plus paths that failed to stat
|
||||
// or hash).
|
||||
func (s *scanState) updatePhase() error {
|
||||
deletes := make([]string, 0, len(s.existing))
|
||||
for path := range s.existing {
|
||||
deletes = append(deletes, path)
|
||||
}
|
||||
|
||||
// Sorted deletes keep the update phase deterministic.
|
||||
slices.Sort(deletes)
|
||||
|
||||
s.st.removed = len(deletes)
|
||||
|
||||
total := len(s.batch) + len(s.sentinels) + len(deletes)
|
||||
prog := newProgress("update", int64(total))
|
||||
|
||||
defer prog.finish()
|
||||
|
||||
err := applyChanges(s.db, s.batch, nil, prog)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.batch = nil
|
||||
|
||||
// Sentinel records are converted in batch-sized chunks rather than
|
||||
// materialized all at once; a first scan can have millions.
|
||||
for chunk := range slices.Chunk(s.sentinels, updateBatchSize) {
|
||||
recs := make([]scanRec, 0, len(chunk))
|
||||
for _, rec := range chunk {
|
||||
recs = append(recs, scanRec{
|
||||
size: rec.size, mtime: rec.mtime, path: rec.path,
|
||||
})
|
||||
}
|
||||
|
||||
err = applyBatch(s.db, recs, nil, prog)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return applyChanges(s.db, nil, deletes, prog)
|
||||
}
|
||||
|
||||
// underAnyRoot reports whether path is any of the roots or lies under
|
||||
@@ -167,26 +431,6 @@ func underAnyRoot(path string, roots []string) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
// underRoot reports whether path is root itself or lies under it. Both
|
||||
// must be absolute and lexically clean.
|
||||
func underRoot(path, root string) bool {
|
||||
@@ -202,75 +446,6 @@ func underRoot(path, root string) bool {
|
||||
return strings.HasPrefix(path, prefix)
|
||||
}
|
||||
|
||||
// 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
|
||||
// equals the recorded size and its statted mtime is not newer than the
|
||||
// recorded mtime.
|
||||
func partitionChanged(recs []fileRec,
|
||||
existing map[string]scanRec,
|
||||
) ([]fileRec, []scanRec) {
|
||||
var (
|
||||
toHash []fileRec
|
||||
unchanged []scanRec
|
||||
)
|
||||
|
||||
for _, rec := range recs {
|
||||
old, ok := existing[rec.path]
|
||||
if ok && old.size == rec.size && old.mtime >= rec.mtime {
|
||||
unchanged = append(unchanged, old)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
toHash = append(toHash, rec)
|
||||
}
|
||||
|
||||
return toHash, unchanged
|
||||
}
|
||||
|
||||
// collectDeletes returns the existing record paths that were not
|
||||
// successfully processed this run: vanished files, plus paths that
|
||||
// failed to stat or hash. The database keeps only records verified by
|
||||
// the latest scan covering them. The result is sorted so the update
|
||||
// pass is deterministic.
|
||||
func collectDeletes(existing map[string]scanRec,
|
||||
unchanged, hashed []scanRec,
|
||||
) []string {
|
||||
kept := make(map[string]bool, len(unchanged)+len(hashed))
|
||||
|
||||
for _, r := range unchanged {
|
||||
kept[r.path] = true
|
||||
}
|
||||
|
||||
for _, r := range hashed {
|
||||
kept[r.path] = true
|
||||
}
|
||||
|
||||
var deletes []string
|
||||
|
||||
for p := range existing {
|
||||
if !kept[p] {
|
||||
deletes = append(deletes, p)
|
||||
}
|
||||
}
|
||||
|
||||
slices.Sort(deletes)
|
||||
|
||||
return deletes
|
||||
}
|
||||
|
||||
// applyPass writes the scan's changes to the database under an update
|
||||
// progress display (one item per insertion, update, or deletion).
|
||||
func applyPass(db *sql.DB, upserts []scanRec, deletes []string) error {
|
||||
prog := newProgress("update", int64(len(upserts)+len(deletes)))
|
||||
err := applyChanges(db, upserts, deletes, prog)
|
||||
|
||||
prog.finish()
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// dirJob is one directory awaiting traversal by the walk workers. It
|
||||
// carries its operand's filesystem device so -x can stop at
|
||||
// filesystem boundaries.
|
||||
@@ -280,87 +455,68 @@ type dirJob struct {
|
||||
rootDevOK bool
|
||||
}
|
||||
|
||||
// walkEvent is one walk result delivered to the main goroutine: a
|
||||
// regular-file path, or a warning when fail is set.
|
||||
// walkEvent is one walk result delivered to the walk phase: a regular
|
||||
// file's statted record, or a warning when fail is set.
|
||||
type walkEvent struct {
|
||||
path string
|
||||
rec fileRec
|
||||
warn string
|
||||
fail bool
|
||||
}
|
||||
|
||||
// walkPass enumerates every regular file under all roots concurrently
|
||||
// with a per-directory worker pool; every operand seeds the shared
|
||||
// queue. 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, workers int) ([]string, int) {
|
||||
prog := newProgress("walk", -1)
|
||||
|
||||
var (
|
||||
paths []string
|
||||
initial []dirJob
|
||||
errs int
|
||||
)
|
||||
|
||||
for _, root := range roots {
|
||||
p, jobs, e := seedRoot(root, prog)
|
||||
paths = append(paths, p...)
|
||||
initial = append(initial, jobs...)
|
||||
errs += e
|
||||
}
|
||||
|
||||
// startWalk seeds every root into the shared walk worker pool and
|
||||
// returns the event stream: one record per regular file, one warning
|
||||
// event per per-path error. The channel is closed when the walk
|
||||
// completes.
|
||||
func startWalk(roots []string, oneFS bool, workers int) <-chan walkEvent {
|
||||
jobs, subdirs, events := startWalkWorkers(workers, oneFS)
|
||||
|
||||
dispatchDirs(initial, jobs, subdirs)
|
||||
|
||||
for ev := range events {
|
||||
if ev.fail {
|
||||
errs++
|
||||
|
||||
prog.warnf("%s", ev.warn)
|
||||
|
||||
continue
|
||||
go func() {
|
||||
initial := make([]dirJob, 0, len(roots))
|
||||
for _, root := range roots {
|
||||
initial = append(initial, seedRoot(root, events)...)
|
||||
}
|
||||
|
||||
paths = append(paths, ev.path)
|
||||
dispatchDirs(initial, jobs, subdirs)
|
||||
}()
|
||||
|
||||
prog.increment()
|
||||
}
|
||||
|
||||
prog.finish()
|
||||
|
||||
return paths, errs
|
||||
return events
|
||||
}
|
||||
|
||||
// seedRoot turns the PATH operand into the walk's starting state: a
|
||||
// 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
|
||||
// seedRoot turns one PATH operand into the walk's starting state: a
|
||||
// regular-file operand is statted and emitted directly, a directory
|
||||
// operand becomes an 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) ([]string, []dirJob, int) {
|
||||
func seedRoot(root string, events chan<- walkEvent) []dirJob {
|
||||
fi, err := os.Lstat(root)
|
||||
if err != nil {
|
||||
prog.warnf("walk %s: %v", root, err)
|
||||
events <- walkEvent{
|
||||
warn: fmt.Sprintf("walk %s: %v", root, err),
|
||||
fail: true,
|
||||
}
|
||||
|
||||
return nil, nil, 1
|
||||
return nil
|
||||
}
|
||||
|
||||
switch {
|
||||
case fi.IsDir():
|
||||
if filepath.Base(root) == ".zfs" {
|
||||
return nil, nil, 0
|
||||
return nil
|
||||
}
|
||||
|
||||
dev, ok := deviceOfInfo(fi)
|
||||
|
||||
return nil, []dirJob{{path: root, rootDev: dev, rootDevOK: ok}}, 0
|
||||
return []dirJob{{path: root, rootDev: dev, rootDevOK: ok}}
|
||||
case fi.Mode().IsRegular():
|
||||
prog.increment()
|
||||
events <- walkEvent{rec: fileRec{
|
||||
path: root,
|
||||
size: fi.Size(),
|
||||
mtime: fi.ModTime().Unix(),
|
||||
}}
|
||||
|
||||
return []string{root}, nil, 0
|
||||
return nil
|
||||
default:
|
||||
return nil, nil, 0
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -460,12 +616,39 @@ func walkOneDir(job dirJob, oneFS bool, events chan<- walkEvent) []dirJob {
|
||||
continue
|
||||
}
|
||||
|
||||
events <- walkEvent{path: p}
|
||||
emitFile(p, e, events)
|
||||
}
|
||||
|
||||
return subs
|
||||
}
|
||||
|
||||
// emitFile stats one regular directory entry and emits its record.
|
||||
// The lstat happens here in the walk worker, while the directory's
|
||||
// metadata is still hot; a path that fails to stat (or stops being a
|
||||
// regular file) between the directory read and the lstat is warned
|
||||
// about and skipped.
|
||||
func emitFile(p string, e fs.DirEntry, events chan<- walkEvent) {
|
||||
info, err := e.Info()
|
||||
if err != nil {
|
||||
events <- walkEvent{
|
||||
warn: fmt.Sprintf("stat %s: %v", p, err),
|
||||
fail: true,
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if !info.Mode().IsRegular() {
|
||||
return
|
||||
}
|
||||
|
||||
events <- walkEvent{rec: fileRec{
|
||||
path: p,
|
||||
size: info.Size(),
|
||||
mtime: info.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
|
||||
@@ -510,74 +693,8 @@ 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.
|
||||
// prevented hashing it) from the hash workers to the hash phase.
|
||||
type hashResult struct {
|
||||
rec fileRec
|
||||
head string
|
||||
@@ -585,76 +702,28 @@ type hashResult struct {
|
||||
err error
|
||||
}
|
||||
|
||||
// startHashWorkers starts the hash worker pool over recs and returns
|
||||
// the channel its results arrive on (one per record, in completion
|
||||
// order).
|
||||
func startHashWorkers(recs []fileRec, workers int) <-chan hashResult {
|
||||
jobs := make(chan fileRec, workQueueDepth)
|
||||
results := make(chan hashResult, workQueueDepth)
|
||||
|
||||
// startHashWorkers starts the hash worker pool: workers read jobs,
|
||||
// write one result per record, and exit when jobs is closed.
|
||||
func startHashWorkers(jobs <-chan fileRec, results chan<- hashResult,
|
||||
workers int,
|
||||
) {
|
||||
for range workers {
|
||||
go func() {
|
||||
for rec := range jobs {
|
||||
head, tail, err := hashHeadTail(rec.path, rec.size)
|
||||
results <- hashResult{rec: rec, head: head, tail: tail, err: err}
|
||||
results <- hashResult{
|
||||
rec: rec, head: head, tail: tail, err: err,
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
go func() {
|
||||
for _, rec := range recs {
|
||||
jobs <- rec
|
||||
}
|
||||
|
||||
close(jobs)
|
||||
}()
|
||||
|
||||
return results
|
||||
}
|
||||
|
||||
// hashPass hashes the first and last chunk bytes of every file in a
|
||||
// worker pool and collects the resulting records on the main
|
||||
// goroutine. Files that fail to open or read are warned about and
|
||||
// dropped. Only new or changed files reach this pass.
|
||||
func hashPass(recs []fileRec, workers int) ([]scanRec, int) {
|
||||
results := startHashWorkers(recs, workers)
|
||||
prog := newProgress("hash", int64(len(recs)))
|
||||
out := make([]scanRec, 0, len(recs))
|
||||
|
||||
var errs int
|
||||
|
||||
for range recs {
|
||||
r := <-results
|
||||
if r.err != nil {
|
||||
errs++
|
||||
|
||||
prog.warnf("hash %s: %v", r.rec.path, r.err)
|
||||
prog.increment()
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
out = append(out, scanRec{
|
||||
size: r.rec.size,
|
||||
mtime: r.rec.mtime,
|
||||
head: r.head,
|
||||
tail: r.tail,
|
||||
path: r.rec.path,
|
||||
})
|
||||
|
||||
prog.increment()
|
||||
}
|
||||
|
||||
prog.finish()
|
||||
|
||||
return out, errs
|
||||
}
|
||||
|
||||
// hashHeadTail returns the lowercase-hex SHA-256 of the first
|
||||
// min(chunk, size) bytes and of the last min(chunk, size) bytes of the
|
||||
// file at path. The two reads overlap when size < 2*chunk; for
|
||||
// size == 0 both hashes are of the empty input. size is the value
|
||||
// recorded by the stat pass.
|
||||
// recorded when the file was statted.
|
||||
func hashHeadTail(path string, size int64) (string, string, error) {
|
||||
//nolint:gosec // hashing operator-supplied paths is the tool's purpose
|
||||
f, err := os.Open(path)
|
||||
@@ -676,11 +745,18 @@ func hashHeadTail(path string, size int64) (string, string, error) {
|
||||
|
||||
h := sha256.Sum256(buf)
|
||||
|
||||
if n > 0 {
|
||||
_, err = f.ReadAt(buf, size-n)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
// When the whole file fits in one chunk the tail window is exactly
|
||||
// the bytes just read: reuse the head hash instead of issuing a
|
||||
// second read for every small file.
|
||||
if size <= int64(chunk) {
|
||||
hh := hex.EncodeToString(h[:])
|
||||
|
||||
return hh, hh, nil
|
||||
}
|
||||
|
||||
_, err = f.ReadAt(buf, size-n)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
|
||||
t := sha256.Sum256(buf)
|
||||
|
||||
Reference in New Issue
Block a user