From 8233ddcfba1f5451d3250df4829bd4f63319cb83 Mon Sep 17 00:00:00 2001 From: Andy Dixon Date: Fri, 2 Oct 2026 08:02:32 +0100 Subject: [PATCH] Add bounded concurrent scanner Single-writer controller with a select loop that offers jobs and accepts results simultaneously, so bounded channels cannot deadlock. statx/fstatat relative to the directory fd on Linux/macOS, portable fallback elsewhere. Hard links counted once, symlink loops detected, virtual filesystems skipped, errors classified and surfaced. --- internal/platform/mode_unix.go | 36 ++ internal/platform/platform.go | 151 +++++++++ internal/platform/readdir_bsd.go | 82 +++++ internal/platform/readdir_linux.go | 138 ++++++++ internal/platform/readdir_other.go | 31 ++ internal/platform/sys_nowindows.go | 7 + internal/platform/sys_windows.go | 12 + internal/scan/scan.go | 521 +++++++++++++++++++++++++++++ internal/scan/scan_test.go | 282 ++++++++++++++++ 9 files changed, 1260 insertions(+) create mode 100644 internal/platform/mode_unix.go create mode 100644 internal/platform/platform.go create mode 100644 internal/platform/readdir_bsd.go create mode 100644 internal/platform/readdir_linux.go create mode 100644 internal/platform/readdir_other.go create mode 100644 internal/platform/sys_nowindows.go create mode 100644 internal/platform/sys_windows.go create mode 100644 internal/scan/scan.go create mode 100644 internal/scan/scan_test.go diff --git a/internal/platform/mode_unix.go b/internal/platform/mode_unix.go new file mode 100644 index 0000000..54476c7 --- /dev/null +++ b/internal/platform/mode_unix.go @@ -0,0 +1,36 @@ +//go:build unix + +package platform + +import ( + "io/fs" + "syscall" +) + +func modeFromUnix(m uint32) fs.FileMode { + mode := fs.FileMode(m & 0o777) + switch m & syscall.S_IFMT { + case syscall.S_IFDIR: + mode |= fs.ModeDir + case syscall.S_IFLNK: + mode |= fs.ModeSymlink + case syscall.S_IFIFO: + mode |= fs.ModeNamedPipe + case syscall.S_IFSOCK: + mode |= fs.ModeSocket + case syscall.S_IFCHR: + mode |= fs.ModeDevice | fs.ModeCharDevice + case syscall.S_IFBLK: + mode |= fs.ModeDevice + } + if m&syscall.S_ISUID != 0 { + mode |= fs.ModeSetuid + } + if m&syscall.S_ISGID != 0 { + mode |= fs.ModeSetgid + } + if m&syscall.S_ISVTX != 0 { + mode |= fs.ModeSticky + } + return mode +} diff --git a/internal/platform/platform.go b/internal/platform/platform.go new file mode 100644 index 0000000..6e59396 --- /dev/null +++ b/internal/platform/platform.go @@ -0,0 +1,151 @@ +// Package platform hides OS-specific filesystem metadata behind a small +// common API. Files are split by build tag; nothing else in the program +// switches on GOOS. +package platform + +import ( + "errors" + "io" + "io/fs" + "os" + "os/user" + "strconv" + "sync" +) + +// Meta is the metadata of one directory entry. +type Meta struct { + Name string + Mode fs.FileMode + Size int64 + Alloc int64 + AllocKnown bool + MTime int64 // unix nanoseconds + BTime int64 // birth time, 0 if unknown + UID, GID uint32 + Nlink uint32 + Dev, Ino uint64 + HasIno bool // Dev/Ino are meaningful + Err error // stat failed; only Name is valid +} + +// ChunkFunc receives successive batches of entries. The slice is owned by +// the callee. Returning an error stops the read. +type ChunkFunc func([]Meta) error + +// Lstat returns metadata for path without following a final symlink. +func Lstat(path string) (Meta, error) { return statPath(path, false) } + +// Stat returns metadata for path, following symlinks. +func Stat(path string) (Meta, error) { return statPath(path, true) } + +var ( + nameMu sync.Mutex + userNames = map[uint32]string{} + groupNames = map[uint32]string{} +) + +// UserName resolves a uid to a name (cached; falls back to the number). +func UserName(uid uint32) string { + return lookupName(userNames, uid, func(s string) (string, error) { + u, err := user.LookupId(s) + if err != nil { + return "", err + } + return u.Username, nil + }) +} + +// GroupName resolves a gid to a name (cached; falls back to the number). +func GroupName(gid uint32) string { + return lookupName(groupNames, gid, func(s string) (string, error) { + g, err := user.LookupGroupId(s) + if err != nil { + return "", err + } + return g.Name, nil + }) +} + +func lookupName(cache map[uint32]string, id uint32, look func(string) (string, error)) string { + nameMu.Lock() + defer nameMu.Unlock() + if n, ok := cache[id]; ok { + return n + } + s := strconv.FormatUint(uint64(id), 10) + if n, err := look(s); err == nil { + s = n + } + cache[id] = s + return s +} + +// fromFileInfo fills the portable fields of Meta. +func fromFileInfo(fi fs.FileInfo) Meta { + return Meta{ + Name: fi.Name(), + Mode: fi.Mode(), + Size: fi.Size(), + Alloc: fi.Size(), + MTime: fi.ModTime().UnixNano(), + Nlink: 1, + } +} + +// readDirGeneric is the portable fallback: os.ReadDir in chunks plus Lstat +// per entry by path. +func readDirGeneric(path string, chunk int, fn ChunkFunc, stat func(string) (Meta, error)) error { + f, err := os.Open(path) + if err != nil { + return err + } + defer f.Close() + for { + des, err := f.ReadDir(chunk) + if len(des) > 0 { + out := make([]Meta, len(des)) + for i, de := range des { + m, serr := stat(joinPath(path, de.Name())) + if serr != nil { + m = Meta{Err: serr} + } + m.Name = de.Name() + out[i] = m + } + if ferr := fn(out); ferr != nil { + return ferr + } + } + if err != nil { + if errors.Is(err, io.EOF) { + return nil + } + return err + } + } +} + +func joinPath(dir, name string) string { + if len(dir) > 0 && os.IsPathSeparator(dir[len(dir)-1]) { + return dir + name + } + return dir + string(os.PathSeparator) + name +} + +// JoinPath joins a directory and an entry name without cleaning. +func JoinPath(dir, name string) string { return joinPath(dir, name) } + +func baseName(path string) string { + i := len(path) - 1 + for i > 0 && os.IsPathSeparator(path[i]) { + i-- + } + path = path[:i+1] + for j := len(path) - 1; j >= 0; j-- { + if os.IsPathSeparator(path[j]) { + return path[j+1:] + } + } + return path +} diff --git a/internal/platform/readdir_bsd.go b/internal/platform/readdir_bsd.go new file mode 100644 index 0000000..f221031 --- /dev/null +++ b/internal/platform/readdir_bsd.go @@ -0,0 +1,82 @@ +//go:build darwin || freebsd + +package platform + +import ( + "errors" + "io" + "os" + + "golang.org/x/sys/unix" +) + +// ReadDir streams the entries of a directory in chunks, stat-ing each entry +// relative to the directory fd. +func ReadDir(path string, chunk int, fn ChunkFunc) error { + fd, err := unix.Open(path, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC, 0) + if err != nil { + return &os.PathError{Op: "open", Path: path, Err: err} + } + f := os.NewFile(uintptr(fd), path) + defer f.Close() + for { + des, err := f.ReadDir(chunk) + if len(des) > 0 { + out := make([]Meta, len(des)) + for i, de := range des { + var st unix.Stat_t + if serr := unix.Fstatat(fd, de.Name(), &st, unix.AT_SYMLINK_NOFOLLOW); serr != nil { + out[i] = Meta{Name: de.Name(), Err: &os.PathError{Op: "stat", Path: de.Name(), Err: serr}} + continue + } + out[i] = fromStat(&st) + out[i].Name = de.Name() + } + if ferr := fn(out); ferr != nil { + return ferr + } + } + if err != nil { + if errors.Is(err, io.EOF) { + return nil + } + return err + } + } +} + +func fromStat(st *unix.Stat_t) Meta { + return Meta{ + Mode: modeFromUnix(uint32(st.Mode)), + Size: st.Size, + Alloc: st.Blocks * 512, + AllocKnown: true, + MTime: st.Mtim.Nano(), + BTime: st.Btim.Nano(), + UID: st.Uid, + GID: st.Gid, + Nlink: uint32(st.Nlink), + Dev: uint64(st.Dev), + Ino: uint64(st.Ino), + HasIno: true, + } +} + +func statPath(path string, follow bool) (Meta, error) { + var st unix.Stat_t + var err error + if follow { + err = unix.Stat(path, &st) + } else { + err = unix.Lstat(path, &st) + } + if err != nil { + return Meta{}, &os.PathError{Op: "stat", Path: path, Err: err} + } + m := fromStat(&st) + m.Name = baseName(path) + return m, nil +} + +// IsVirtualFS is Linux-specific; elsewhere nothing is skipped. +func IsVirtualFS(string) bool { return false } diff --git a/internal/platform/readdir_linux.go b/internal/platform/readdir_linux.go new file mode 100644 index 0000000..167f49e --- /dev/null +++ b/internal/platform/readdir_linux.go @@ -0,0 +1,138 @@ +package platform + +import ( + "errors" + "io" + "os" + "sync/atomic" + + "golang.org/x/sys/unix" +) + +// noStatx is set if the kernel lacks statx (pre-4.11); we fall back to fstatat. +var noStatx atomic.Bool + +const statxMask = unix.STATX_BASIC_STATS | unix.STATX_BTIME + +// ReadDir streams the entries of a directory in chunks, stat-ing each entry +// relative to the directory fd (no repeated path resolution). +func ReadDir(path string, chunk int, fn ChunkFunc) error { + fd, err := unix.Open(path, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC, 0) + if err != nil { + return &os.PathError{Op: "open", Path: path, Err: err} + } + f := os.NewFile(uintptr(fd), path) + defer f.Close() + for { + des, err := f.ReadDir(chunk) + if len(des) > 0 { + out := make([]Meta, len(des)) + for i, de := range des { + m, serr := statAt(fd, de.Name(), unix.AT_SYMLINK_NOFOLLOW) + if serr != nil { + m = Meta{Err: &os.PathError{Op: "stat", Path: de.Name(), Err: serr}} + } + m.Name = de.Name() + out[i] = m + } + if ferr := fn(out); ferr != nil { + return ferr + } + } + if err != nil { + if errors.Is(err, io.EOF) { + return nil + } + return err + } + } +} + +func statAt(dirfd int, name string, flags int) (Meta, error) { + if !noStatx.Load() { + var st unix.Statx_t + err := unix.Statx(dirfd, name, flags|unix.AT_STATX_DONT_SYNC, statxMask, &st) + if err == nil { + return fromStatx(&st), nil + } + if !errors.Is(err, unix.ENOSYS) { + return Meta{}, err + } + noStatx.Store(true) + } + var st unix.Stat_t + if err := unix.Fstatat(dirfd, name, &st, flags); err != nil { + return Meta{}, err + } + return Meta{ + Mode: modeFromUnix(st.Mode), Size: st.Size, Alloc: st.Blocks * 512, AllocKnown: true, + MTime: st.Mtim.Nano(), UID: st.Uid, GID: st.Gid, Nlink: uint32(st.Nlink), + Dev: st.Dev, Ino: st.Ino, HasIno: true, + }, nil +} + +func fromStatx(st *unix.Statx_t) Meta { + m := Meta{ + Mode: modeFromUnix(uint32(st.Mode)), + Size: int64(st.Size), + Alloc: int64(st.Blocks) * 512, + AllocKnown: true, + MTime: st.Mtime.Sec*1e9 + int64(st.Mtime.Nsec), + UID: st.Uid, + GID: st.Gid, + Nlink: st.Nlink, + Dev: unix.Mkdev(st.Dev_major, st.Dev_minor), + Ino: st.Ino, + HasIno: true, + } + if st.Mask&unix.STATX_BTIME != 0 { + m.BTime = st.Btime.Sec*1e9 + int64(st.Btime.Nsec) + } + return m +} + +func statPath(path string, follow bool) (Meta, error) { + flags := unix.AT_SYMLINK_NOFOLLOW + if follow { + flags = 0 + } + m, err := statAt(unix.AT_FDCWD, path, flags) + if err != nil { + return Meta{}, &os.PathError{Op: "stat", Path: path, Err: err} + } + m.Name = baseName(path) + return m, nil +} + +// Linux pseudo-filesystems whose contents are not disk usage. +var virtualFS = map[int64]bool{ + 0x9fa0: true, // proc + 0x62656572: true, // sysfs + 0x1cd1: true, // devpts + 0x27e0eb: true, // cgroup + 0x63677270: true, // cgroup2 + 0x64626720: true, // debugfs + 0x74726163: true, // tracefs + 0x73636673: true, // securityfs + 0x6165676c: true, // pstore + 0xcafe4a11: true, // bpf + 0x62656570: true, // configfs + 0x65735543: true, // fusectl + 0x19800202: true, // mqueue + 0xde5e81e4: true, // efivarfs + 0x42494e4d: true, // binfmt_misc + 0xf97cff8c: true, // selinuxfs + 0x958458f6: true, // hugetlbfs + 0x6e736673: true, // nsfs + 0x50495045: true, // pipefs +} + +// IsVirtualFS reports whether path is the root of a pseudo filesystem +// (proc, sysfs, cgroup, …) that should not be scanned as disk usage. +func IsVirtualFS(path string) bool { + var st unix.Statfs_t + if err := unix.Statfs(path, &st); err != nil { + return false + } + return virtualFS[int64(st.Type)] +} diff --git a/internal/platform/readdir_other.go b/internal/platform/readdir_other.go new file mode 100644 index 0000000..025bcd7 --- /dev/null +++ b/internal/platform/readdir_other.go @@ -0,0 +1,31 @@ +//go:build !linux && !darwin && !freebsd + +package platform + +import "os" + +// ReadDir streams the entries of a directory using the portable os API. +// Allocated size, inode identity and link counts are not available here. +func ReadDir(path string, chunk int, fn ChunkFunc) error { + return readDirGeneric(path, chunk, fn, func(p string) (Meta, error) { return statPath(p, false) }) +} + +func statPath(path string, follow bool) (Meta, error) { + var fi os.FileInfo + var err error + if follow { + fi, err = os.Stat(path) + } else { + fi, err = os.Lstat(path) + } + if err != nil { + return Meta{}, err + } + m := fromFileInfo(fi) + m.Name = baseName(path) + fillSys(&m, fi) + return m, nil +} + +// IsVirtualFS is Linux-specific; elsewhere nothing is skipped. +func IsVirtualFS(string) bool { return false } diff --git a/internal/platform/sys_nowindows.go b/internal/platform/sys_nowindows.go new file mode 100644 index 0000000..4973a88 --- /dev/null +++ b/internal/platform/sys_nowindows.go @@ -0,0 +1,7 @@ +//go:build !windows && !linux && !darwin && !freebsd + +package platform + +import "os" + +func fillSys(*Meta, os.FileInfo) {} diff --git a/internal/platform/sys_windows.go b/internal/platform/sys_windows.go new file mode 100644 index 0000000..c866bec --- /dev/null +++ b/internal/platform/sys_windows.go @@ -0,0 +1,12 @@ +package platform + +import ( + "os" + "syscall" +) + +func fillSys(m *Meta, fi os.FileInfo) { + if d, ok := fi.Sys().(*syscall.Win32FileAttributeData); ok { + m.BTime = d.CreationTime.Nanoseconds() + } +} diff --git a/internal/scan/scan.go b/internal/scan/scan.go new file mode 100644 index 0000000..2a2de94 --- /dev/null +++ b/internal/scan/scan.go @@ -0,0 +1,521 @@ +// Package scan implements the bounded concurrent filesystem scanner. +// +// Architecture (see ARCHITECTURE.md): a single controller goroutine owns the +// inventory.Tree and a FIFO of pending directories. A fixed pool of workers +// performs directory I/O and sends result chunks back. The controller's +// select loop offers jobs and accepts results at the same time, so bounded +// channels can never deadlock: workers never enqueue work themselves. +package scan + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "runtime" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/andydixon/mapsize/internal/inventory" + "github.com/andydixon/mapsize/internal/platform" +) + +// FollowMode controls symlink traversal. +type FollowMode int + +const ( + FollowNone FollowMode = iota + FollowSameFS + FollowAll +) + +func (f FollowMode) String() string { + return [...]string{"none", "same-filesystem", "all"}[f] +} + +// ParseFollow parses a --follow-symlinks value. +func ParseFollow(s string) (FollowMode, error) { + switch s { + case "none", "": + return FollowNone, nil + case "same-filesystem", "same-fs": + return FollowSameFS, nil + case "all": + return FollowAll, nil + } + return 0, fmt.Errorf("invalid --follow-symlinks %q (want none, same-filesystem or all)", s) +} + +// Workers resolves a concurrency mode name or number to a worker count. +func Workers(mode string) (int, error) { + cpu := runtime.NumCPU() + switch mode { + case "conservative": + return 2, nil + case "balanced", "": + return min(8, max(2, 2*cpu)), nil + case "aggressive": + return min(64, max(4, 8*cpu)), nil + } + var n int + if _, err := fmt.Sscan(mode, &n); err != nil || n < 1 || n > 1024 { + return 0, fmt.Errorf("invalid workers %q (conservative, balanced, aggressive or 1-1024)", mode) + } + return n, nil +} + +// Options configures a scan. +type Options struct { + Workers int + Follow FollowMode + OneFileSystem bool + Excludes []string + + // Tuning knobs, mostly for tests. Zero means default. + ChunkSize int // entries per result chunk + JobBuffer int // capacity of the job channel + ResultBuf int // capacity of the result channel + ApplyBatch int // max results applied per write-lock hold +} + +func (o *Options) defaults() { + if o.Workers <= 0 { + o.Workers, _ = Workers("balanced") + } + if o.ChunkSize <= 0 { + o.ChunkSize = 1024 + } + if o.JobBuffer <= 0 { + o.JobBuffer = o.Workers * 2 + } + if o.ResultBuf <= 0 { + o.ResultBuf = o.Workers * 4 + } + if o.ApplyBatch <= 0 { + o.ApplyBatch = 64 + } +} + +// Progress is a cheap, lock-free view of scan activity. +type Progress struct { + Entries int64 + Pending int64 // directories discovered but not yet dispatched + Inflight int64 // directories being read + Workers int +} + +// Scanner runs one scan. Tree is safe to read under Tree.RLock while the +// scan runs. +type Scanner struct { + Tree *inventory.Tree + opts Options + + entries, pending, inflight atomic.Int64 + done chan struct{} +} + +type job struct { + id inventory.NodeID + path string + dev uint64 + // hints derived once per directory + inCache, inLog bool +} + +type result struct { + job *job + entries []platform.Meta + final bool + err error +} + +type inodeKey struct{ dev, ino uint64 } + +// Start stats root, creates the tree and begins scanning in the background. +func Start(ctx context.Context, root string, opts Options) (*Scanner, error) { + opts.defaults() + abs, err := filepath.Abs(root) + if err != nil { + return nil, err + } + m, err := platform.Stat(abs) + if err != nil { + return nil, err + } + kind := kindOf(m) + t := inventory.New(abs, kind) + t.Stats.Workers = opts.Workers + t.Stats.Excludes = opts.Excludes + t.Stats.OneFileSystem = opts.OneFileSystem + t.Stats.Follow = opts.Follow.String() + r := t.Node(t.Root()) + setMeta(r, m) + r.TotSize, r.TotAlloc = m.Size, m.Alloc + s := &Scanner{Tree: t, opts: opts, done: make(chan struct{})} + if kind != inventory.KindDir { + t.Stats.Files = 1 + t.Stats.Complete = true + t.Stats.End = time.Now() + close(s.done) + return s, nil + } + t.Stats.Dirs = 1 + c := &controller{s: s, t: t, opts: opts, links: map[inodeKey]struct{}{}} + if opts.Follow != FollowNone { + c.visited = map[inodeKey]struct{}{{m.Dev, m.Ino}: {}} + } + c.rootDev = m.Dev + c.pending = []pendingDir{{id: t.Root(), dev: m.Dev}} + s.pending.Store(1) + go c.run(ctx) + return s, nil +} + +// Done is closed when the scan has finished or been cancelled and drained. +func (s *Scanner) Done() <-chan struct{} { return s.done } + +// Progress returns live counters without locking the tree. +func (s *Scanner) Progress() Progress { + return Progress{Entries: s.entries.Load(), Pending: s.pending.Load(), Inflight: s.inflight.Load(), Workers: s.opts.Workers} +} + +// maxEntriesPerLock bounds how long the controller holds the write lock, so +// the UI's read lock is never starved for more than a few milliseconds. +const maxEntriesPerLock = 8192 + +type pendingDir struct { + id inventory.NodeID + dev uint64 +} + +type controller struct { + s *Scanner + t *inventory.Tree + opts Options + rootDev uint64 + pending []pendingDir // FIFO; head index advances, compacted occasionally + head int + links map[inodeKey]struct{} + visited map[inodeKey]struct{} // only when following symlinks +} + +func (c *controller) run(ctx context.Context) { + defer close(c.s.done) + jobs := make(chan *job, c.opts.JobBuffer) + results := make(chan result, c.opts.ResultBuf) + var wg sync.WaitGroup + for range c.opts.Workers { + wg.Add(1) + go func() { + defer wg.Done() + worker(ctx, jobs, results, c.opts.ChunkSize) + }() + } + + inflight := 0 + ctxDone := ctx.Done() + cancelled := false + var next *job + batch := make([]result, 0, c.opts.ApplyBatch) + for { + if next == nil && !cancelled && c.head < len(c.pending) { + next = c.makeJob(c.pending[c.head]) + } + if inflight == 0 && next == nil { + break + } + var jobCh chan<- *job + if next != nil { + jobCh = jobs + } + select { + case jobCh <- next: + next = nil + c.head++ + if c.head > 4096 && c.head*2 > len(c.pending) { + c.pending = append(c.pending[:0], c.pending[c.head:]...) + c.head = 0 + } + inflight++ + c.s.inflight.Store(int64(inflight)) + c.s.pending.Store(int64(len(c.pending) - c.head)) + case r := <-results: + batch = append(batch[:0], r) + n := len(r.entries) + drain: + for len(batch) < cap(batch) && n < maxEntriesPerLock { + select { + case r := <-results: + batch = append(batch, r) + n += len(r.entries) + default: + break drain + } + } + c.t.Lock() + for i := range batch { + if c.apply(&batch[i], cancelled) { + inflight-- + } + batch[i] = result{} + } + c.t.Unlock() + c.s.inflight.Store(int64(inflight)) + c.s.pending.Store(int64(len(c.pending) - c.head)) + case <-ctxDone: + cancelled = true + ctxDone = nil + next = nil + } + } + close(jobs) + wg.Wait() + + c.t.Lock() + st := &c.t.Stats + if cancelled { + st.Cancelled = true + st.Unscanned = int64(len(c.pending) - c.head) + for _, p := range c.pending[c.head:] { + c.t.Node(p.id).Flags |= inventory.FlagIncomplete + } + } + st.Complete = true + st.End = time.Now() + c.t.Unlock() + c.s.pending.Store(0) + c.s.inflight.Store(0) +} + +func (c *controller) makeJob(p pendingDir) *job { + path := c.t.Path(p.id) + j := &job{id: p.id, path: path, dev: p.dev} + j.inCache, j.inLog = inventory.PathHints(path) + return j +} + +func worker(ctx context.Context, jobs <-chan *job, results chan<- result, chunk int) { + for j := range jobs { + var err error + if err = ctx.Err(); err == nil { + err = platform.ReadDir(j.path, chunk, func(ms []platform.Meta) error { + if err := ctx.Err(); err != nil { + return err + } + results <- result{job: j, entries: ms} + return nil + }) + } + results <- result{job: j, final: true, err: err} + } +} + +func kindOf(m platform.Meta) inventory.Kind { + switch { + case m.Mode.IsDir(): + return inventory.KindDir + case m.Mode.IsRegular(): + return inventory.KindFile + case m.Mode&os.ModeSymlink != 0: + return inventory.KindSymlink + } + return inventory.KindOther +} + +func setMeta(n *inventory.Node, m platform.Meta) { + n.Size, n.Alloc = m.Size, m.Alloc + n.MTime = m.MTime + n.Mode = uint32(m.Mode) + n.UID, n.GID = m.UID, m.GID + n.Nlink = m.Nlink + if !m.AllocKnown { + n.Flags |= inventory.FlagAllocUnknown + } +} + +// apply merges one result into the tree. It reports whether the result +// completed its directory. Called with the tree write-locked. +func (c *controller) apply(r *result, cancelled bool) bool { + t := c.t + j := r.job + dir := j.id + var d inventory.Delta + c.s.entries.Add(int64(len(r.entries))) + for _, m := range r.entries { + c.addEntry(j, m, &d) + } + if r.final { + dn := t.Node(dir) + switch { + case r.err == nil: + dn.Flags |= inventory.FlagScanned + case errors.Is(r.err, context.Canceled) || errors.Is(r.err, context.DeadlineExceeded) || cancelled: + dn.Flags |= inventory.FlagIncomplete + t.Stats.Unscanned++ + default: + dn.Flags |= inventory.FlagError | inventory.FlagIncomplete + t.Stats.AddError(inventory.ErrorRecord{Node: dir, Kind: inventory.Classify(r.err), Msg: errMsg(r.err)}) + d.Errors++ + } + } + if d != (inventory.Delta{}) { + t.Propagate(dir, d) + } + return r.final +} + +func errMsg(err error) string { + var pe *os.PathError + if errors.As(err, &pe) { + return pe.Op + ": " + pe.Err.Error() + } + return err.Error() +} + +func (c *controller) excluded(path, name string) bool { + for _, p := range c.opts.Excludes { + if strings.ContainsAny(p, `/\`) { + pp := filepath.Clean(p) + if path == pp || strings.HasPrefix(path, pp+string(os.PathSeparator)) { + return true + } + if ok, _ := filepath.Match(pp, path); ok { + return true + } + } else if ok, _ := filepath.Match(p, name); ok { + return true + } + } + return false +} + +func (c *controller) addEntry(j *job, m platform.Meta, d *inventory.Delta) { + t := c.t + st := &t.Stats + if m.Err != nil { + k := inventory.Classify(m.Err) + if k == inventory.ErrVanished { + st.AddError(inventory.ErrorRecord{Node: j.id, Name: m.Name, Kind: k, Msg: "vanished between listing and stat"}) + d.Errors++ + return + } + id := t.Add(j.id, m.Name, inventory.KindOther) + n := t.Node(id) + n.Flags |= inventory.FlagError + n.Errors = 1 + st.AddError(inventory.ErrorRecord{Node: id, Kind: k, Msg: errMsg(m.Err)}) + d.Errors++ + d.Files++ + st.Others++ + return + } + var path string + if len(c.opts.Excludes) > 0 { + path = platform.JoinPath(j.path, m.Name) + if c.excluded(path, m.Name) { + st.Excluded++ + return + } + } + kind := kindOf(m) + var target platform.Meta + followed, broken := false, false + if kind == inventory.KindSymlink { + if path == "" { + path = platform.JoinPath(j.path, m.Name) + } + tm, err := platform.Stat(path) + switch { + case err != nil: + st.BrokenLinks++ + broken = true + case tm.Mode.IsDir() && (c.opts.Follow == FollowAll || c.opts.Follow == FollowSameFS && tm.Dev == j.dev): + target, followed = tm, true + kind = inventory.KindDir + } + } + + id := t.Add(j.id, m.Name, kind) + n := t.Node(id) + setMeta(n, m) + if broken { + n.Flags |= inventory.FlagBrokenLink + } + + switch kind { + case inventory.KindDir: + st.Dirs++ + d.Dirs++ + d.Size += n.Size + d.Alloc += n.Alloc + n.TotSize, n.TotAlloc = n.Size, n.Alloc + dev := m.Dev + if followed { + n.Flags |= inventory.FlagFollowed + dev = target.Dev + } + if c.visited != nil { + key := inodeKey{m.Dev, m.Ino} + if followed { + key = inodeKey{target.Dev, target.Ino} + } + if _, seen := c.visited[key]; seen { + n.Flags |= inventory.FlagLoop + st.LoopsSkipped++ + return + } + c.visited[key] = struct{}{} + } + if m.HasIno && dev != j.dev { + n.Flags |= inventory.FlagMountPoint + if c.opts.OneFileSystem { + n.Flags |= inventory.FlagSkippedFS + st.SkippedMounts++ + return + } + if path == "" { + path = platform.JoinPath(j.path, m.Name) + } + if platform.IsVirtualFS(path) { + n.Flags |= inventory.FlagVirtualFS + st.VirtualSkipped++ + return + } + } + c.pending = append(c.pending, pendingDir{id: id, dev: dev}) + case inventory.KindFile: + st.Files++ + d.Files++ + if m.Nlink > 1 && m.HasIno { + n.Flags |= inventory.FlagHardlinked + key := inodeKey{m.Dev, m.Ino} + if _, seen := c.links[key]; seen { + n.Flags |= inventory.FlagHardlinkDup + st.HardlinkDups++ + return + } + c.links[key] = struct{}{} + } + if m.AllocKnown && m.Alloc < m.Size && m.Size-m.Alloc >= 4096 { + n.Flags |= inventory.FlagSparse + } + n.Cat = inventory.CategoryFor(t.ExtName(n), j.inCache, j.inLog) + n.TotSize, n.TotAlloc = n.Size, n.Alloc + d.Size += n.Size + d.Alloc += n.Alloc + d.Cat[n.Cat] += n.Alloc + default: + if kind == inventory.KindSymlink { + st.Symlinks++ + } else { + st.Others++ + } + d.Files++ + n.TotSize, n.TotAlloc = n.Size, n.Alloc + d.Size += n.Size + d.Alloc += n.Alloc + } +} diff --git a/internal/scan/scan_test.go b/internal/scan/scan_test.go new file mode 100644 index 0000000..0c9932a --- /dev/null +++ b/internal/scan/scan_test.go @@ -0,0 +1,282 @@ +package scan + +import ( + "context" + "fmt" + "os" + "path/filepath" + "runtime" + "testing" + "time" + + "github.com/andydixon/mapsize/internal/inventory" +) + +// mkTree creates dirs×files under root: root/d0/d1/... fan-out tree. +func mkTree(t testing.TB, root string, fanout, depth, files int) (nfiles, ndirs int64, bytes int64) { + var rec func(dir string, d int) + rec = func(dir string, d int) { + for i := range files { + sz := (i + 1) * 100 + if err := os.WriteFile(filepath.Join(dir, fmt.Sprintf("f%d.txt", i)), make([]byte, sz), 0o644); err != nil { + t.Fatal(err) + } + nfiles++ + bytes += int64(sz) + } + if d == depth { + return + } + for i := range fanout { + sub := filepath.Join(dir, fmt.Sprintf("d%d", i)) + if err := os.Mkdir(sub, 0o755); err != nil { + t.Fatal(err) + } + ndirs++ + rec(sub, d+1) + } + } + rec(root, 0) + return +} + +func runScan(t testing.TB, root string, o Options) *inventory.Tree { + s, err := Start(context.Background(), root, o) + if err != nil { + t.Fatal(err) + } + select { + case <-s.Done(): + case <-time.After(60 * time.Second): + t.Fatal("scan did not finish (deadlock?)") + } + return s.Tree +} + +func TestScanCountsAndAggregates(t *testing.T) { + root := t.TempDir() + nf, nd, bytes := mkTree(t, root, 3, 3, 4) + tr := runScan(t, root, Options{Workers: 4}) + r := tr.Node(tr.Root()) + if int64(r.Files) != nf || int64(r.Dirs) != nd { + t.Fatalf("files=%d dirs=%d, want %d %d", r.Files, r.Dirs, nf, nd) + } + // Logical total = file bytes + directory entries' own sizes. + var dirBytes int64 + for i := 0; i < tr.Len(); i++ { + if n := tr.Node(inventory.NodeID(i)); n.IsDir() { + dirBytes += n.Size + } + } + if r.TotSize != bytes+dirBytes { + t.Fatalf("TotSize=%d want %d", r.TotSize, bytes+dirBytes) + } + if !tr.Stats.Complete || tr.Stats.Incomplete() { + t.Fatalf("stats: %+v", tr.Stats) + } + // Every directory's aggregate equals the sum of its children. + checkAggregates(t, tr) + // Incremental aggregation must match a full recompute. + before := snapshotTotals(tr) + tr.Recompute() + if after := snapshotTotals(tr); fmt.Sprint(before) != fmt.Sprint(after) { + t.Fatal("incremental aggregates differ from recompute") + } +} + +func snapshotTotals(tr *inventory.Tree) [][4]int64 { + out := make([][4]int64, tr.Len()) + for i := range out { + n := tr.Node(inventory.NodeID(i)) + out[i] = [4]int64{n.TotSize, n.TotAlloc, int64(n.Files), int64(n.Dirs)} + } + return out +} + +func checkAggregates(t *testing.T, tr *inventory.Tree) { + t.Helper() + for i := 0; i < tr.Len(); i++ { + id := inventory.NodeID(i) + n := tr.Node(id) + if !n.IsDir() { + continue + } + sum := n.Size + tr.Children(id, func(_ inventory.NodeID, c *inventory.Node) { sum += c.TotSize }) + if sum != n.TotSize { + t.Fatalf("%s: TotSize %d != own+children %d", tr.Path(id), n.TotSize, sum) + } + } +} + +// TestBackpressureNoDeadlock forces maximum contention: one worker, channel +// capacities of one, tiny chunks, and thousands of directories. +func TestBackpressureNoDeadlock(t *testing.T) { + root := t.TempDir() + nf, nd, _ := mkTree(t, root, 6, 4, 3) // 1554 dirs + for _, w := range []int{1, 2, 16} { + tr := runScan(t, root, Options{Workers: w, JobBuffer: 1, ResultBuf: 1, ChunkSize: 1, ApplyBatch: 1}) + r := tr.Node(tr.Root()) + if int64(r.Files) != nf || int64(r.Dirs) != nd { + t.Fatalf("workers=%d: files=%d dirs=%d want %d %d", w, r.Files, r.Dirs, nf, nd) + } + } +} + +func TestCancel(t *testing.T) { + root := t.TempDir() + mkTree(t, root, 8, 3, 2) + ctx, cancel := context.WithCancel(context.Background()) + s, err := Start(ctx, root, Options{Workers: 2, JobBuffer: 1, ResultBuf: 1, ChunkSize: 1}) + if err != nil { + t.Fatal(err) + } + cancel() + select { + case <-s.Done(): + case <-time.After(10 * time.Second): + t.Fatal("cancel did not stop scan") + } + tr := s.Tree + tr.RLock() + defer tr.RUnlock() + if !tr.Stats.Cancelled || !tr.Stats.Incomplete() { + t.Fatalf("expected cancelled+incomplete: %+v", tr.Stats) + } + // Workers must have exited: give the runtime a moment and check goroutines. + time.Sleep(50 * time.Millisecond) + if n := runtime.NumGoroutine(); n > 10 { + t.Logf("goroutines after cancel: %d", n) + } +} + +func TestSymlinkLoops(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("symlinks need privileges on Windows") + } + root := t.TempDir() + a := filepath.Join(root, "a") + os.MkdirAll(filepath.Join(a, "b"), 0o755) + os.WriteFile(filepath.Join(a, "b", "x"), make([]byte, 5000), 0o644) + os.Symlink("..", filepath.Join(a, "b", "up")) // loop + os.Symlink("../a", filepath.Join(a, "self")) // loop + os.Symlink("nowhere", filepath.Join(root, "broken")) + for _, f := range []FollowMode{FollowNone, FollowSameFS, FollowAll} { + tr := runScan(t, root, Options{Workers: 4, Follow: f}) + r := tr.Node(tr.Root()) + if r.Files == 0 || r.TotSize <= 5000 { + t.Fatalf("follow=%v: bad totals %+v", f, r) + } + if tr.Stats.BrokenLinks != 1 { + t.Fatalf("follow=%v: broken links %d", f, tr.Stats.BrokenLinks) + } + if f != FollowNone && tr.Stats.LoopsSkipped == 0 { + t.Fatalf("follow=%v: expected loops to be detected", f) + } + // The 5000-byte file must be counted exactly once. + var count int + for i := 0; i < tr.Len(); i++ { + if n := tr.Node(inventory.NodeID(i)); n.Name == "x" { + count++ + } + } + if count != 1 { + t.Fatalf("follow=%v: file x seen %d times", f, count) + } + } +} + +func TestHardlinksCountedOnce(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("hard-link identity not available on Windows") + } + root := t.TempDir() + p := filepath.Join(root, "orig") + os.WriteFile(p, make([]byte, 100000), 0o644) + os.Mkdir(filepath.Join(root, "d"), 0o755) + if err := os.Link(p, filepath.Join(root, "d", "link")); err != nil { + t.Skip(err) + } + tr := runScan(t, root, Options{Workers: 2}) + r := tr.Node(tr.Root()) + if tr.Stats.HardlinkDups != 1 { + t.Fatalf("dups=%d", tr.Stats.HardlinkDups) + } + if r.TotSize >= 200000 { + t.Fatalf("hard link double counted: %d", r.TotSize) + } +} + +func TestPermissionErrorsSurface(t *testing.T) { + if runtime.GOOS == "windows" || os.Geteuid() == 0 { + t.Skip("needs non-root unix") + } + root := t.TempDir() + locked := filepath.Join(root, "locked") + os.Mkdir(locked, 0o755) + os.WriteFile(filepath.Join(locked, "f"), []byte("x"), 0o644) + os.Chmod(locked, 0) + defer os.Chmod(locked, 0o755) + tr := runScan(t, root, Options{Workers: 2}) + if tr.Stats.ErrCounts[inventory.ErrPermission] != 1 || !tr.Stats.Incomplete() { + t.Fatalf("expected one permission error: %+v", tr.Stats.ErrCounts) + } + if tr.Node(tr.Root()).Errors != 1 { + t.Fatal("error not aggregated to root") + } +} + +func TestExcludes(t *testing.T) { + root := t.TempDir() + os.MkdirAll(filepath.Join(root, "keep"), 0o755) + os.MkdirAll(filepath.Join(root, "skip"), 0o755) + os.WriteFile(filepath.Join(root, "keep", "a.tmp"), []byte("x"), 0o644) + os.WriteFile(filepath.Join(root, "keep", "b.txt"), []byte("x"), 0o644) + os.WriteFile(filepath.Join(root, "skip", "c.txt"), []byte("x"), 0o644) + tr := runScan(t, root, Options{Workers: 2, Excludes: []string{"*.tmp", filepath.Join(root, "skip")}}) + if tr.Stats.Excluded != 2 || tr.Node(0).Files != 1 { + t.Fatalf("excluded=%d files=%d", tr.Stats.Excluded, tr.Node(0).Files) + } +} + +func TestUnicodeAndHostileNames(t *testing.T) { + root := t.TempDir() + names := []string{"日本語.txt", "emoji-🎉.png", "esc\x1b[31mred", "nl\nname", "sp ace"} + if runtime.GOOS == "windows" { + names = names[:2] + } + for _, n := range names { + if err := os.WriteFile(filepath.Join(root, n), []byte("x"), 0o644); err != nil { + t.Fatal(err) + } + } + tr := runScan(t, root, Options{Workers: 2}) + got := map[string]bool{} + tr.Children(0, func(_ inventory.NodeID, n *inventory.Node) { got[n.Name] = true }) + for _, n := range names { + if !got[n] { + t.Errorf("missing %q (raw names must be preserved)", n) + } + } +} + +func TestSingleFileRoot(t *testing.T) { + p := filepath.Join(t.TempDir(), "file") + os.WriteFile(p, make([]byte, 1234), 0o644) + tr := runScan(t, p, Options{}) + if tr.Node(0).TotSize != 1234 || tr.Node(0).Kind != inventory.KindFile { + t.Fatalf("%+v", tr.Node(0)) + } +} + +func BenchmarkScan(b *testing.B) { + root := b.TempDir() + mkTree(b, root, 5, 4, 20) + for _, w := range []int{1, 4, 16} { + b.Run(fmt.Sprintf("workers=%d", w), func(b *testing.B) { + for b.Loop() { + runScan(b, root, Options{Workers: w}) + } + }) + } +}