Parallelize the walk and commit per operand
All checks were successful
check / check (push) Successful in 5s
All checks were successful
check / check (push) Successful in 5s
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.
This commit is contained in:
160
scan_test.go
160
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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user