From df50ab82c48c910845e0bfd87aef836ba01e48d6 Mon Sep 17 00:00:00 2001 From: Frederic Morin Date: Thu, 8 Oct 2026 08:50:22 -0400 Subject: [PATCH 1/4] vm: inotify: sync file events in batches Every file event was synced with two `limactl shell` commands run in sequence (a `stat`, then the sync), about 100ms each. Events were handled at roughly 4 per second while the watcher waited, so changing a few hundred files (switching branches, formatting a project) took close to a minute to reach the containers. Events above 50 unique files per 500ms were dropped. - Collect events for 100ms, deduplicated by path, and sync them with a single command. The command is split when its arguments would exceed 64KiB. - Check that the file exists in the VM within the sync command instead of running a separate `stat`. - Run the sync in the background so events keep being received while it runs. Events received meanwhile are synced right after. - Remove the rate limit, which dropped events: batching bounds the number of commands run in the VM. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Frederic Morin --- daemon/process/inotify/events.go | 148 +++++++++++++++----- daemon/process/inotify/events_linux_test.go | 102 +++++++++----- daemon/process/inotify/events_test.go | 145 +++++++++++++++++-- 3 files changed, 308 insertions(+), 87 deletions(-) diff --git a/daemon/process/inotify/events.go b/daemon/process/inotify/events.go index f40392bd..076c46ae 100644 --- a/daemon/process/inotify/events.go +++ b/daemon/process/inotify/events.go @@ -14,10 +14,115 @@ type modEvent struct { func (m modEvent) Mode() string { return fmt.Sprintf("%o", m.FileMode) } -const syncEventScript = `/bin/chmod "$1" "$2" && : >> "$2"` +// syncEventsScript syncs each (mode, path) pair passed as arguments. +// Files that do not exist in the VM are skipped. +const syncEventsScript = `while [ $# -gt 1 ]; do [ -e "$2" ] && /bin/chmod "$1" "$2" && : >> "$2"; shift 2; done; true` + +// syncBatchDelay is how long events are collected before they are synced. +const syncBatchDelay = 100 * time.Millisecond + +// maxSyncArgsSize caps the size of the arguments of a single sync command, +// well below the 128KiB limit of a single argument on Linux, as the command +// is passed to the VM's shell as one string. +const maxSyncArgsSize = 64 * 1024 + +// syncEventsCmds returns the commands that sync evs, splitting them so that +// each command stays within maxSyncArgsSize. +func syncEventsCmds(evs []modEvent) [][]string { + var cmds [][]string + var args []string + size := 0 + + flush := func() { + if len(args) == 0 { + return + } + cmd := append([]string{"sudo", "/bin/sh", "-c", syncEventsScript, "sh"}, args...) + cmds = append(cmds, cmd) + args = nil + size = 0 + } + + for _, ev := range evs { + mode := ev.Mode() + n := len(mode) + len(ev.path) + 2 + if size+n > maxSyncArgsSize { + flush() + } + args = append(args, mode, ev.path) + size += n + } + flush() + + return cmds +} + +// batchEvents collects events from mod, deduplicated by path, and passes +// them to sync in batches. A batch is synced delay after its first event, +// or as soon as the previous sync returns. sync runs in the +// background so that events keep being received while it runs. +func batchEvents(ctx context.Context, mod <-chan modEvent, delay time.Duration, sync func([]modEvent)) { + var pending []modEvent + index := map[string]int{} + var flush <-chan time.Time + syncing := false + done := make(chan struct{}, 1) + + start := func() { + evs := pending + pending = nil + index = map[string]int{} + syncing = true + go func() { + sync(evs) + done <- struct{}{} + }() + } + + for { + select { + case <-ctx.Done(): + return + + case ev, ok := <-mod: + if !ok { + return + } + if i, ok := index[ev.path]; ok { + pending[i] = ev + continue + } + index[ev.path] = len(pending) + pending = append(pending, ev) + if !syncing && flush == nil { + flush = time.After(delay) + } + + case <-flush: + flush = nil + start() + + case <-done: + syncing = false + if len(pending) > 0 { + start() + } + } + } +} + +func (f *inotifyProcess) syncEvents(evs []modEvent) { + log := f.log + for _, ev := range evs { + log.Tracef("syncing inotify event for %s", ev.path) + } + log.Infof("syncing inotify events for %d file(s)", len(evs)) -func syncEventCmd(ev modEvent) []string { - return []string{"sudo", "/bin/sh", "-c", syncEventScript, "sh", ev.Mode(), ev.path} + for _, cmd := range syncEventsCmds(evs) { + if err := f.guest.RunQuiet(cmd...); err != nil { + log.Trace(fmt.Errorf("error syncing inotify events: %w", err)) + } + } } func (f *inotifyProcess) handleEvents(ctx context.Context, watcher dirWatcher) error { @@ -31,7 +136,8 @@ func (f *inotifyProcess) handleEvents(ctx context.Context, watcher dirWatcher) e return fmt.Errorf("error watching container volumes: %w", err) } - var last time.Time + go batchEvents(ctx, mod, syncBatchDelay, f.syncEvents) + var cancelWatch context.CancelFunc var currentVols []string @@ -47,14 +153,11 @@ func (f *inotifyProcess) handleEvents(ctx context.Context, watcher dirWatcher) e return false } - cache := map[string]struct{}{} - for { select { // exit signal case <-ctx.Done(): - close(mod) return ctx.Err() // watch only container volumes @@ -79,37 +182,6 @@ func (f *inotifyProcess) handleEvents(ctx context.Context, watcher dirWatcher) e log.Error(fmt.Errorf("error running watcher: %w", err)) } }(ctx, vols, mod) - - // handle modification events - case ev := <-mod: - now := time.Now() - - // rate limit, handle at most 50 unique items every 500 ms - if now.Sub(last) < time.Millisecond*500 { - if _, ok := cache[ev.path]; ok { - continue // handled, ignore - } - if len(cache) > 50 { - continue - } - } else { - last = now - cache = map[string]struct{}{} // >500ms, reset unique cache - } - - // cache current event - cache[ev.path] = struct{}{} - - // validate that file exists - if err := f.guest.RunQuiet("stat", ev.path); err != nil { - log.Trace(fmt.Errorf("cannot stat '%s': %w", ev.path, err)) - continue - } - - log.Infof("syncing inotify event for %s ", ev.path) - if err := f.guest.RunQuiet(syncEventCmd(ev)...); err != nil { - log.Trace(fmt.Errorf("error syncing inotify event: %w", err)) - } } } } diff --git a/daemon/process/inotify/events_linux_test.go b/daemon/process/inotify/events_linux_test.go index 79500e5d..884457c5 100644 --- a/daemon/process/inotify/events_linux_test.go +++ b/daemon/process/inotify/events_linux_test.go @@ -12,9 +12,9 @@ import ( "unsafe" ) -func readInotifyMasks(t *testing.T, fd int) uint32 { +func readInotifyMasks(t *testing.T, fd int) map[int32]uint32 { t.Helper() - var mask uint32 + masks := map[int32]uint32{} buf := make([]byte, 64*(syscall.SizeofInotifyEvent+syscall.NAME_MAX+1)) deadline := time.Now().Add(2 * time.Second) for time.Now().Before(deadline) { @@ -28,62 +28,88 @@ func readInotifyMasks(t *testing.T, fd int) uint32 { } for offset := 0; offset < n; { ev := (*syscall.InotifyEvent)(unsafe.Pointer(&buf[offset])) - mask |= ev.Mask + masks[ev.Wd] |= ev.Mask offset += syscall.SizeofInotifyEvent + int(ev.Len) } } - return mask + return masks } -func Test_syncEventCmd_emitsAttribAndCloseWriteWithoutModifyingFile(t *testing.T) { +func Test_syncEventsCmds_emitsAttribAndCloseWriteWithoutModifyingFiles(t *testing.T) { dir := t.TempDir() - path := filepath.Join(dir, "file with spaces.go") content := []byte("package main\n") - if err := os.WriteFile(path, content, 0o644); err != nil { - t.Fatal(err) - } - before, err := os.Stat(path) - if err != nil { - t.Fatal(err) - } + paths := []string{filepath.Join(dir, "file with spaces.go"), filepath.Join(dir, `quote"d.go`)} + missing := filepath.Join(dir, "missing.go") fd, err := syscall.InotifyInit1(syscall.IN_NONBLOCK | syscall.IN_CLOEXEC) if err != nil { t.Fatal(err) } defer func() { _ = syscall.Close(fd) }() - if _, err := syscall.InotifyAddWatch(fd, path, syscall.IN_ATTRIB|syscall.IN_MODIFY|syscall.IN_CLOSE_WRITE); err != nil { - t.Fatal(err) - } - args := syncEventCmd(modEvent{path: path, FileMode: before.Mode()})[1:] - if out, err := exec.Command(args[0], args[1:]...).CombinedOutput(); err != nil { - t.Fatalf("sync command failed: %v: %s", err, out) + var evs []modEvent + before := map[string]os.FileInfo{} + for _, path := range paths { + if err := os.WriteFile(path, content, 0o644); err != nil { + t.Fatal(err) + } + stat, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + before[path] = stat + evs = append(evs, modEvent{path: path, FileMode: stat.Mode()}) } + evs = append([]modEvent{{path: missing, FileMode: 0o644}}, evs...) - mask := readInotifyMasks(t, fd) - if mask&syscall.IN_ATTRIB == 0 { - t.Errorf("expected IN_ATTRIB, got mask %#x", mask) - } - if mask&syscall.IN_CLOSE_WRITE == 0 { - t.Errorf("expected IN_CLOSE_WRITE, got mask %#x", mask) - } - if mask&syscall.IN_MODIFY != 0 { - t.Errorf("expected no IN_MODIFY, got mask %#x", mask) + watches := map[int32]string{} + for _, path := range paths { + wd, err := syscall.InotifyAddWatch(fd, path, syscall.IN_ATTRIB|syscall.IN_MODIFY|syscall.IN_CLOSE_WRITE) + if err != nil { + t.Fatal(err) + } + watches[int32(wd)] = path } - after, err := os.Stat(path) - if err != nil { - t.Fatal(err) + cmds := syncEventsCmds(evs) + if len(cmds) != 1 { + t.Fatalf("expected a single command, got %d", len(cmds)) } - if !after.ModTime().Equal(before.ModTime()) { - t.Errorf("modification time changed: %v -> %v", before.ModTime(), after.ModTime()) + args := cmds[0][1:] + if out, err := exec.Command(args[0], args[1:]...).CombinedOutput(); err != nil { + t.Fatalf("sync command failed: %v: %s", err, out) } - got, err := os.ReadFile(path) - if err != nil { - t.Fatal(err) + + masks := readInotifyMasks(t, fd) + for wd, path := range watches { + mask := masks[wd] + if mask&syscall.IN_ATTRIB == 0 { + t.Errorf("%s: expected IN_ATTRIB, got mask %#x", path, mask) + } + if mask&syscall.IN_CLOSE_WRITE == 0 { + t.Errorf("%s: expected IN_CLOSE_WRITE, got mask %#x", path, mask) + } + if mask&syscall.IN_MODIFY != 0 { + t.Errorf("%s: expected no IN_MODIFY, got mask %#x", path, mask) + } + + after, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if !after.ModTime().Equal(before[path].ModTime()) { + t.Errorf("%s: modification time changed: %v -> %v", path, before[path].ModTime(), after.ModTime()) + } + got, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if string(got) != string(content) { + t.Errorf("%s: content changed: %q -> %q", path, content, got) + } } - if string(got) != string(content) { - t.Errorf("content changed: %q -> %q", content, got) + + if _, err := os.Stat(missing); !os.IsNotExist(err) { + t.Errorf("missing file was created: %v", err) } } diff --git a/daemon/process/inotify/events_test.go b/daemon/process/inotify/events_test.go index 66de694a..6d54e784 100644 --- a/daemon/process/inotify/events_test.go +++ b/daemon/process/inotify/events_test.go @@ -1,32 +1,155 @@ package inotify import ( + "context" + "fmt" "reflect" + "strings" + "sync" "testing" + "time" ) -func Test_syncEventCmd(t *testing.T) { +func Test_syncEventsCmds(t *testing.T) { + prefix := []string{"sudo", "/bin/sh", "-c", syncEventsScript, "sh"} + tests := []struct { name string - ev modEvent - want []string + evs []modEvent + want [][]string }{ { - name: "regular file", - ev: modEvent{path: "/Users/someone/project/main.go", FileMode: 0o644}, - want: []string{"sudo", "/bin/sh", "-c", syncEventScript, "sh", "644", "/Users/someone/project/main.go"}, + name: "no events", + evs: nil, + want: nil, }, { - name: "path with spaces and quotes stays a single argument", - ev: modEvent{path: `/Users/someone/my "project"/a b.go`, FileMode: 0o755}, - want: []string{"sudo", "/bin/sh", "-c", syncEventScript, "sh", "755", `/Users/someone/my "project"/a b.go`}, + name: "events are passed as mode and path pairs", + evs: []modEvent{ + {path: "/Users/someone/project/main.go", FileMode: 0o644}, + {path: `/Users/someone/my "project"/a b.go`, FileMode: 0o755}, + }, + want: [][]string{append(append([]string{}, prefix...), + "644", "/Users/someone/project/main.go", + "755", `/Users/someone/my "project"/a b.go`, + )}, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if got := syncEventCmd(tt.ev); !reflect.DeepEqual(got, tt.want) { - t.Errorf("syncEventCmd() = %q, want %q", got, tt.want) + if got := syncEventsCmds(tt.evs); !reflect.DeepEqual(got, tt.want) { + t.Errorf("syncEventsCmds() = %q, want %q", got, tt.want) } }) } } + +func Test_syncEventsCmds_splitsLargeBatches(t *testing.T) { + var evs []modEvent + for i := range 2000 { + evs = append(evs, modEvent{path: fmt.Sprintf("/Users/someone/project/%s/file%d.go", strings.Repeat("d", 60), i), FileMode: 0o644}) + } + + cmds := syncEventsCmds(evs) + if len(cmds) < 2 { + t.Fatalf("expected the batch to be split, got %d command(s)", len(cmds)) + } + + var paths []string + for _, cmd := range cmds { + args := cmd[5:] + size := 0 + for _, arg := range args { + size += len(arg) + 1 + } + if size > maxSyncArgsSize { + t.Errorf("command arguments size %d exceeds %d", size, maxSyncArgsSize) + } + for i := 1; i < len(args); i += 2 { + paths = append(paths, args[i]) + } + } + if len(paths) != len(evs) { + t.Fatalf("expected %d paths, got %d", len(evs), len(paths)) + } + for i, ev := range evs { + if paths[i] != ev.path { + t.Fatalf("path %d = %s, want %s", i, paths[i], ev.path) + } + } +} + +func Test_batchEvents(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + mod := make(chan modEvent) + var mu sync.Mutex + var batches [][]modEvent + release := make(chan struct{}) + synced := make(chan struct{}, 10) + + go batchEvents(ctx, mod, 500*time.Millisecond, func(evs []modEvent) { + mu.Lock() + batches = append(batches, evs) + first := len(batches) == 1 + mu.Unlock() + if first { + <-release + } + synced <- struct{}{} + }) + + for i := range 100 { + mod <- modEvent{path: fmt.Sprintf("/a/%d", i), FileMode: 0o644} + } + mod <- modEvent{path: "/a/0", FileMode: 0o600} + + waitBatches := func(n int) { + t.Helper() + deadline := time.After(2 * time.Second) + for { + mu.Lock() + got := len(batches) + mu.Unlock() + if got >= n { + return + } + select { + case <-deadline: + t.Fatalf("expected %d batch(es), got %d", n, got) + case <-time.After(10 * time.Millisecond): + } + } + } + + waitBatches(1) + + for i := range 50 { + mod <- modEvent{path: fmt.Sprintf("/b/%d", i), FileMode: 0o644} + } + mod <- modEvent{path: "/b/0", FileMode: 0o600} + close(release) + + waitBatches(2) + <-synced + <-synced + + mu.Lock() + defer mu.Unlock() + if len(batches) != 2 { + t.Fatalf("expected 2 batches, got %d", len(batches)) + } + if len(batches[0]) != 100 { + t.Errorf("first batch has %d events, want 100", len(batches[0])) + } + if got := batches[0][0]; got.path != "/a/0" || got.Mode() != "600" { + t.Errorf("deduplicated event = (%s, %s), want latest (/a/0, 600)", got.path, got.Mode()) + } + if len(batches[1]) != 50 { + t.Errorf("events received during a sync: got batch of %d, want 50", len(batches[1])) + } + if got := batches[1][0]; got.path != "/b/0" || got.Mode() != "600" { + t.Errorf("deduplicated event = (%s, %s), want latest (/b/0, 600)", got.path, got.Mode()) + } +} From 1ee5da7e4f7ed5616410d8a7d8350f7713fb8331 Mon Sep 17 00:00:00 2001 From: Frederic Morin Date: Thu, 8 Oct 2026 09:06:31 -0400 Subject: [PATCH 2/4] vm: inotify: skip events for files unchanged since their last sync On macOS, a chmod run in the VM on a file that was previously opened for writing from the VM emits a write event on the host. Since #1643 every sync does both, so syncing a file emits an event for that same file, which is synced again, endlessly. The rate limit only slowed this loop down; batching removes it. Remember the state of the files on the host (inode, size, mode and modification time) when they are synced, and skip events for files whose state has not changed since. These events are either caused by the sync itself, or carry nothing that the VM has not already been notified of. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Frederic Morin --- daemon/process/inotify/events.go | 40 ++++++++++++++++++++ daemon/process/inotify/events_test.go | 53 +++++++++++++++++++++++++++ daemon/process/inotify/watch.go | 2 +- 3 files changed, 94 insertions(+), 1 deletion(-) diff --git a/daemon/process/inotify/events.go b/daemon/process/inotify/events.go index 076c46ae..51c438bf 100644 --- a/daemon/process/inotify/events.go +++ b/daemon/process/inotify/events.go @@ -4,12 +4,14 @@ import ( "context" "fmt" "io/fs" + "os" "time" ) type modEvent struct { path string // filename fs.FileMode + info fs.FileInfo // state of the file on the host, if known } func (m modEvent) Mode() string { return fmt.Sprintf("%o", m.FileMode) } @@ -21,6 +23,9 @@ const syncEventsScript = `while [ $# -gt 1 ]; do [ -e "$2" ] && /bin/chmod "$1" // syncBatchDelay is how long events are collected before they are synced. const syncBatchDelay = 100 * time.Millisecond +// syncedTTL is how long the state of a synced file is remembered. +const syncedTTL = time.Minute + // maxSyncArgsSize caps the size of the arguments of a single sync command, // well below the 128KiB limit of a single argument on Linux, as the command // is passed to the VM's shell as one string. @@ -57,13 +62,32 @@ func syncEventsCmds(evs []modEvent) [][]string { return cmds } +// unchanged reports whether a and b describe the same, unmodified file. +func unchanged(a, b fs.FileInfo) bool { + return a != nil && b != nil && + os.SameFile(a, b) && + a.Size() == b.Size() && + a.Mode() == b.Mode() && + a.ModTime().Equal(b.ModTime()) +} + +type syncedFile struct { + info fs.FileInfo + at time.Time +} + // batchEvents collects events from mod, deduplicated by path, and passes // them to sync in batches. A batch is synced delay after its first event, // or as soon as the previous sync returns. sync runs in the // background so that events keep being received while it runs. +// +// Events for files left unchanged since they were last synced are skipped. +// Syncing a file from the VM can itself emit an event on the host, which +// would otherwise sync the file again, endlessly. func batchEvents(ctx context.Context, mod <-chan modEvent, delay time.Duration, sync func([]modEvent)) { var pending []modEvent index := map[string]int{} + synced := map[string]syncedFile{} var flush <-chan time.Time syncing := false done := make(chan struct{}, 1) @@ -73,6 +97,19 @@ func batchEvents(ctx context.Context, mod <-chan modEvent, delay time.Duration, pending = nil index = map[string]int{} syncing = true + + now := time.Now() + for path, s := range synced { + if now.Sub(s.at) > syncedTTL { + delete(synced, path) + } + } + for _, ev := range evs { + if ev.info != nil { + synced[ev.path] = syncedFile{info: ev.info, at: now} + } + } + go func() { sync(evs) done <- struct{}{} @@ -88,6 +125,9 @@ func batchEvents(ctx context.Context, mod <-chan modEvent, delay time.Duration, if !ok { return } + if s, ok := synced[ev.path]; ok && unchanged(s.info, ev.info) { + continue + } if i, ok := index[ev.path]; ok { pending[i] = ev continue diff --git a/daemon/process/inotify/events_test.go b/daemon/process/inotify/events_test.go index 6d54e784..b1eaa4f2 100644 --- a/daemon/process/inotify/events_test.go +++ b/daemon/process/inotify/events_test.go @@ -3,6 +3,8 @@ package inotify import ( "context" "fmt" + "os" + "path/filepath" "reflect" "strings" "sync" @@ -153,3 +155,54 @@ func Test_batchEvents(t *testing.T) { t.Errorf("deduplicated event = (%s, %s), want latest (/b/0, 600)", got.path, got.Mode()) } } + +func Test_batchEvents_skipsUnchangedFiles(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + path := filepath.Join(t.TempDir(), "main.go") + event := func(content string, rewrite bool) modEvent { + t.Helper() + if rewrite { + if err := os.WriteFile(path, []byte(content), 0o644); err != nil { + t.Fatal(err) + } + } + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + return modEvent{path: path, FileMode: info.Mode(), info: info} + } + + mod := make(chan modEvent) + batches := make(chan []modEvent, 10) + go batchEvents(ctx, mod, 10*time.Millisecond, func(evs []modEvent) { batches <- evs }) + + expectBatch := func(want bool) { + t.Helper() + select { + case evs := <-batches: + if !want { + t.Fatalf("unexpected sync of %d event(s)", len(evs)) + } + case <-time.After(200 * time.Millisecond): + if want { + t.Fatal("expected a sync") + } + } + } + + mod <- event("package main\n", true) + expectBatch(true) + + mod <- event("", false) + expectBatch(false) + + time.Sleep(10 * time.Millisecond) + mod <- event("package main\n\nfunc main() {}\n", true) + expectBatch(true) + + mod <- modEvent{path: path, FileMode: 0o644} + expectBatch(true) +} diff --git a/daemon/process/inotify/watch.go b/daemon/process/inotify/watch.go index 46118f49..62b3bcbc 100644 --- a/daemon/process/inotify/watch.go +++ b/daemon/process/inotify/watch.go @@ -70,7 +70,7 @@ func (d *defaultWatcher) Watch(ctx context.Context, dirs []string, mod chan<- mo } // send modification event - mod <- modEvent{path: path, FileMode: stat.Mode()} + mod <- modEvent{path: path, FileMode: stat.Mode(), info: stat} } } }(ctx, c, mod) From 348d6d744f83d53386692ee599122f0a9592893f Mon Sep 17 00:00:00 2001 From: Frederic Morin Date: Thu, 8 Oct 2026 09:26:32 -0400 Subject: [PATCH 3/4] vm: inotify: buffer the file events channel notify drops events when the channel it sends to is full. The channel had a buffer of one, so most events of a burst of file changes were lost: rewriting 2000 files from the host delivered fewer than half of them to the containers. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Frederic Morin --- daemon/process/inotify/watch.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/daemon/process/inotify/watch.go b/daemon/process/inotify/watch.go index 62b3bcbc..1c16c801 100644 --- a/daemon/process/inotify/watch.go +++ b/daemon/process/inotify/watch.go @@ -10,6 +10,8 @@ import ( "github.com/sirupsen/logrus" ) +const eventBufferSize = 1024 + type dirWatcher interface { // Watch watches directories recursively for changes and sends message via c on // modifications to files within the watched directories. @@ -28,7 +30,9 @@ type defaultWatcher struct { // Watch implements dirWatcher func (d *defaultWatcher) Watch(ctx context.Context, dirs []string, mod chan<- modEvent) error { log := d.log - c := make(chan notify.EventInfo, 1) + // notify drops events when the channel is full, which happens during + // bursts of file changes. + c := make(chan notify.EventInfo, eventBufferSize) for _, dir := range dirs { dir, err := util.CleanPath(dir) From f316f605cc0a909e81566dd8af5cf6c87a39ab0d Mon Sep 17 00:00:00 2001 From: Frederic Morin Date: Thu, 8 Oct 2026 09:55:59 -0400 Subject: [PATCH 4/4] vm: inotify: keep up with large bursts of file events Syncing the 27k files of a project checkout took 36s after the last change, and some events were lost on the host. - The watcher only forwards the paths of the events. Checking the files on the host (stat, directory, unchanged since the last sync) moves to the sync, once per path and batch. Events are received as fast as notify sends them, and it drops those that are not received in time. - chmod runs once per batch and file mode instead of once per file: starting a process for each file is what took most of the time in the VM. - Up to 4 sync commands run at the same time. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Frederic Morin --- daemon/process/inotify/events.go | 234 ++++++++++++++++---------- daemon/process/inotify/events_test.go | 158 +++++++++-------- daemon/process/inotify/watch.go | 20 +-- 3 files changed, 229 insertions(+), 183 deletions(-) diff --git a/daemon/process/inotify/events.go b/daemon/process/inotify/events.go index 51c438bf..d221ed7a 100644 --- a/daemon/process/inotify/events.go +++ b/daemon/process/inotify/events.go @@ -5,20 +5,28 @@ import ( "fmt" "io/fs" "os" + "sort" + "sync" "time" + + "github.com/sirupsen/logrus" ) type modEvent struct { path string // filename fs.FileMode - info fs.FileInfo // state of the file on the host, if known + info fs.FileInfo // state of the file on the host } func (m modEvent) Mode() string { return fmt.Sprintf("%o", m.FileMode) } -// syncEventsScript syncs each (mode, path) pair passed as arguments. +// syncEventsScript syncs the files passed as arguments after their mode. // Files that do not exist in the VM are skipped. -const syncEventsScript = `while [ $# -gt 1 ]; do [ -e "$2" ] && /bin/chmod "$1" "$2" && : >> "$2"; shift 2; done; true` +// +// chmod emits IN_ATTRIB, and opening and closing a file for writing emits +// IN_CLOSE_WRITE. chmod runs once for all files, as starting a process for +// each file is what takes most of the time. +const syncEventsScript = `m=$1; shift; /bin/chmod "$m" "$@" 2>/dev/null; for f; do [ -e "$f" ] && : >> "$f"; done; true` // syncBatchDelay is how long events are collected before they are synced. const syncBatchDelay = 100 * time.Millisecond @@ -31,87 +39,67 @@ const syncedTTL = time.Minute // is passed to the VM's shell as one string. const maxSyncArgsSize = 64 * 1024 -// syncEventsCmds returns the commands that sync evs, splitting them so that -// each command stays within maxSyncArgsSize. -func syncEventsCmds(evs []modEvent) [][]string { - var cmds [][]string - var args []string - size := 0 - - flush := func() { - if len(args) == 0 { - return - } - cmd := append([]string{"sudo", "/bin/sh", "-c", syncEventsScript, "sh"}, args...) - cmds = append(cmds, cmd) - args = nil - size = 0 - } +// syncParallelism is the number of sync commands run at the same time. +const syncParallelism = 4 +// syncEventsCmds returns the commands that sync evs, one per file mode, +// split so that each command stays within maxSyncArgsSize. +func syncEventsCmds(evs []modEvent) [][]string { + byMode := map[string][]string{} + var modes []string for _, ev := range evs { mode := ev.Mode() - n := len(mode) + len(ev.path) + 2 - if size+n > maxSyncArgsSize { - flush() + if _, ok := byMode[mode]; !ok { + modes = append(modes, mode) } - args = append(args, mode, ev.path) - size += n + byMode[mode] = append(byMode[mode], ev.path) } - flush() + sort.Strings(modes) + var cmds [][]string + for _, mode := range modes { + var paths []string + size := 0 + flush := func() { + if len(paths) == 0 { + return + } + cmd := append([]string{"sudo", "/bin/sh", "-c", syncEventsScript, "sh", mode}, paths...) + cmds = append(cmds, cmd) + paths = nil + size = 0 + } + for _, path := range byMode[mode] { + if size+len(path)+1 > maxSyncArgsSize { + flush() + } + paths = append(paths, path) + size += len(path) + 1 + } + flush() + } return cmds } -// unchanged reports whether a and b describe the same, unmodified file. -func unchanged(a, b fs.FileInfo) bool { - return a != nil && b != nil && - os.SameFile(a, b) && - a.Size() == b.Size() && - a.Mode() == b.Mode() && - a.ModTime().Equal(b.ModTime()) -} - -type syncedFile struct { - info fs.FileInfo - at time.Time -} - -// batchEvents collects events from mod, deduplicated by path, and passes -// them to sync in batches. A batch is synced delay after its first event, -// or as soon as the previous sync returns. sync runs in the -// background so that events keep being received while it runs. -// -// Events for files left unchanged since they were last synced are skipped. -// Syncing a file from the VM can itself emit an event on the host, which -// would otherwise sync the file again, endlessly. -func batchEvents(ctx context.Context, mod <-chan modEvent, delay time.Duration, sync func([]modEvent)) { - var pending []modEvent - index := map[string]int{} - synced := map[string]syncedFile{} +// batchEvents collects paths from mod, deduplicated, and passes them to +// handle in batches. A batch is handled delay after its first path, or as +// soon as the previous batch has been handled. handle runs in the +// background so that paths keep being received while it runs: notify drops +// events that are not received in time. +func batchEvents(ctx context.Context, mod <-chan string, delay time.Duration, handle func([]string)) { + var pending []string + seen := map[string]struct{}{} var flush <-chan time.Time - syncing := false + handling := false done := make(chan struct{}, 1) start := func() { - evs := pending + paths := pending pending = nil - index = map[string]int{} - syncing = true - - now := time.Now() - for path, s := range synced { - if now.Sub(s.at) > syncedTTL { - delete(synced, path) - } - } - for _, ev := range evs { - if ev.info != nil { - synced[ev.path] = syncedFile{info: ev.info, at: now} - } - } - + seen = map[string]struct{}{} + handling = true go func() { - sync(evs) + handle(paths) done <- struct{}{} }() } @@ -121,20 +109,16 @@ func batchEvents(ctx context.Context, mod <-chan modEvent, delay time.Duration, case <-ctx.Done(): return - case ev, ok := <-mod: + case path, ok := <-mod: if !ok { return } - if s, ok := synced[ev.path]; ok && unchanged(s.info, ev.info) { + if _, ok := seen[path]; ok { continue } - if i, ok := index[ev.path]; ok { - pending[i] = ev - continue - } - index[ev.path] = len(pending) - pending = append(pending, ev) - if !syncing && flush == nil { + seen[path] = struct{}{} + pending = append(pending, path) + if !handling && flush == nil { flush = time.After(delay) } @@ -143,7 +127,7 @@ func batchEvents(ctx context.Context, mod <-chan modEvent, delay time.Duration, start() case <-done: - syncing = false + handling = false if len(pending) > 0 { start() } @@ -151,32 +135,108 @@ func batchEvents(ctx context.Context, mod <-chan modEvent, delay time.Duration, } } -func (f *inotifyProcess) syncEvents(evs []modEvent) { - log := f.log +type runner interface { + RunQuiet(args ...string) error +} + +type syncedFile struct { + info fs.FileInfo + at time.Time +} + +// syncer syncs batches of changed paths to the VM. It handles one batch at +// a time. +type syncer struct { + guest runner + log *logrus.Entry + synced map[string]syncedFile +} + +func newSyncer(guest runner, log *logrus.Entry) *syncer { + return &syncer{guest: guest, log: log, synced: map[string]syncedFile{}} +} + +// unchanged reports whether a and b describe the same, unmodified file. +func unchanged(a, b fs.FileInfo) bool { + return os.SameFile(a, b) && + a.Size() == b.Size() && + a.Mode() == b.Mode() && + a.ModTime().Equal(b.ModTime()) +} + +// events returns the events to sync for paths. +// +// Files left unchanged since they were last synced are skipped. Syncing a +// file from the VM can itself emit an event on the host, which would +// otherwise sync the file again, endlessly. +func (s *syncer) events(paths []string) []modEvent { + now := time.Now() + for path, f := range s.synced { + if now.Sub(f.at) > syncedTTL { + delete(s.synced, path) + } + } + + var evs []modEvent + for _, path := range paths { + info, err := os.Stat(path) + if err != nil { + s.log.Trace(fmt.Errorf("unable to stat inotify file '%s': %w", path, err)) + continue + } + if info.IsDir() { + s.log.Tracef("'%s' is directory, ignoring.", path) + continue + } + if f, ok := s.synced[path]; ok && unchanged(f.info, info) { + s.log.Tracef("'%s' is unchanged since its last sync, ignoring.", path) + continue + } + s.synced[path] = syncedFile{info: info, at: now} + evs = append(evs, modEvent{path: path, FileMode: info.Mode(), info: info}) + } + return evs +} + +func (s *syncer) sync(paths []string) { + log := s.log + evs := s.events(paths) + if len(evs) == 0 { + return + } for _, ev := range evs { log.Tracef("syncing inotify event for %s", ev.path) } log.Infof("syncing inotify events for %d file(s)", len(evs)) + var wg sync.WaitGroup + sem := make(chan struct{}, syncParallelism) for _, cmd := range syncEventsCmds(evs) { - if err := f.guest.RunQuiet(cmd...); err != nil { - log.Trace(fmt.Errorf("error syncing inotify events: %w", err)) - } + wg.Add(1) + sem <- struct{}{} + go func(cmd []string) { + defer wg.Done() + defer func() { <-sem }() + if err := s.guest.RunQuiet(cmd...); err != nil { + log.Trace(fmt.Errorf("error syncing inotify events: %w", err)) + } + }(cmd) } + wg.Wait() } func (f *inotifyProcess) handleEvents(ctx context.Context, watcher dirWatcher) error { log := f.log log.Trace("begin inotify event handler") - mod := make(chan modEvent) + mod := make(chan string) vols := make(chan []string) if err := f.monitorContainerVolumes(ctx, vols); err != nil { return fmt.Errorf("error watching container volumes: %w", err) } - go batchEvents(ctx, mod, syncBatchDelay, f.syncEvents) + go batchEvents(ctx, mod, syncBatchDelay, newSyncer(f.guest, log).sync) var cancelWatch context.CancelFunc var currentVols []string @@ -217,7 +277,7 @@ func (f *inotifyProcess) handleEvents(ctx context.Context, watcher dirWatcher) e ctx, cancel := context.WithCancel(ctx) cancelWatch = cancel - go func(ctx context.Context, vols []string, mod chan<- modEvent) { + go func(ctx context.Context, vols []string, mod chan<- string) { if err := watcher.Watch(ctx, vols, mod); err != nil { log.Error(fmt.Errorf("error running watcher: %w", err)) } diff --git a/daemon/process/inotify/events_test.go b/daemon/process/inotify/events_test.go index b1eaa4f2..84a0b2a0 100644 --- a/daemon/process/inotify/events_test.go +++ b/daemon/process/inotify/events_test.go @@ -3,17 +3,23 @@ package inotify import ( "context" "fmt" + "io" "os" "path/filepath" "reflect" + "sort" "strings" "sync" "testing" "time" + + "github.com/sirupsen/logrus" ) func Test_syncEventsCmds(t *testing.T) { - prefix := []string{"sudo", "/bin/sh", "-c", syncEventsScript, "sh"} + cmd := func(mode string, paths ...string) []string { + return append([]string{"sudo", "/bin/sh", "-c", syncEventsScript, "sh", mode}, paths...) + } tests := []struct { name string @@ -26,15 +32,16 @@ func Test_syncEventsCmds(t *testing.T) { want: nil, }, { - name: "events are passed as mode and path pairs", + name: "one command per mode", evs: []modEvent{ {path: "/Users/someone/project/main.go", FileMode: 0o644}, - {path: `/Users/someone/my "project"/a b.go`, FileMode: 0o755}, + {path: `/Users/someone/my "project"/run.sh`, FileMode: 0o755}, + {path: `/Users/someone/my "project"/a b.go`, FileMode: 0o644}, + }, + want: [][]string{ + cmd("644", "/Users/someone/project/main.go", `/Users/someone/my "project"/a b.go`), + cmd("755", `/Users/someone/my "project"/run.sh`), }, - want: [][]string{append(append([]string{}, prefix...), - "644", "/Users/someone/project/main.go", - "755", `/Users/someone/my "project"/a b.go`, - )}, }, } for _, tt := range tests { @@ -59,17 +66,14 @@ func Test_syncEventsCmds_splitsLargeBatches(t *testing.T) { var paths []string for _, cmd := range cmds { - args := cmd[5:] size := 0 - for _, arg := range args { - size += len(arg) + 1 + for _, path := range cmd[6:] { + size += len(path) + 1 } if size > maxSyncArgsSize { t.Errorf("command arguments size %d exceeds %d", size, maxSyncArgsSize) } - for i := 1; i < len(args); i += 2 { - paths = append(paths, args[i]) - } + paths = append(paths, cmd[6:]...) } if len(paths) != len(evs) { t.Fatalf("expected %d paths, got %d", len(evs), len(paths)) @@ -85,28 +89,21 @@ func Test_batchEvents(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - mod := make(chan modEvent) + mod := make(chan string) var mu sync.Mutex - var batches [][]modEvent + var batches [][]string release := make(chan struct{}) - synced := make(chan struct{}, 10) - go batchEvents(ctx, mod, 500*time.Millisecond, func(evs []modEvent) { + go batchEvents(ctx, mod, 500*time.Millisecond, func(paths []string) { mu.Lock() - batches = append(batches, evs) + batches = append(batches, paths) first := len(batches) == 1 mu.Unlock() if first { <-release } - synced <- struct{}{} }) - for i := range 100 { - mod <- modEvent{path: fmt.Sprintf("/a/%d", i), FileMode: 0o644} - } - mod <- modEvent{path: "/a/0", FileMode: 0o600} - waitBatches := func(n int) { t.Helper() deadline := time.After(2 * time.Second) @@ -125,84 +122,85 @@ func Test_batchEvents(t *testing.T) { } } + for i := range 100 { + mod <- fmt.Sprintf("/a/%d", i) + } + mod <- "/a/0" waitBatches(1) for i := range 50 { - mod <- modEvent{path: fmt.Sprintf("/b/%d", i), FileMode: 0o644} + mod <- fmt.Sprintf("/b/%d", i) } - mod <- modEvent{path: "/b/0", FileMode: 0o600} + mod <- "/b/0" close(release) - waitBatches(2) - <-synced - <-synced mu.Lock() defer mu.Unlock() - if len(batches) != 2 { - t.Fatalf("expected 2 batches, got %d", len(batches)) - } if len(batches[0]) != 100 { - t.Errorf("first batch has %d events, want 100", len(batches[0])) - } - if got := batches[0][0]; got.path != "/a/0" || got.Mode() != "600" { - t.Errorf("deduplicated event = (%s, %s), want latest (/a/0, 600)", got.path, got.Mode()) + t.Errorf("first batch has %d paths, want 100", len(batches[0])) } if len(batches[1]) != 50 { - t.Errorf("events received during a sync: got batch of %d, want 50", len(batches[1])) - } - if got := batches[1][0]; got.path != "/b/0" || got.Mode() != "600" { - t.Errorf("deduplicated event = (%s, %s), want latest (/b/0, 600)", got.path, got.Mode()) + t.Errorf("paths received during a sync: got batch of %d, want 50", len(batches[1])) } } -func Test_batchEvents_skipsUnchangedFiles(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() +type fakeRunner struct { + mu sync.Mutex + cmds [][]string +} - path := filepath.Join(t.TempDir(), "main.go") - event := func(content string, rewrite bool) modEvent { - t.Helper() - if rewrite { - if err := os.WriteFile(path, []byte(content), 0o644); err != nil { - t.Fatal(err) - } - } - info, err := os.Stat(path) - if err != nil { - t.Fatal(err) - } - return modEvent{path: path, FileMode: info.Mode(), info: info} - } +func (f *fakeRunner) RunQuiet(args ...string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.cmds = append(f.cmds, args) + return nil +} - mod := make(chan modEvent) - batches := make(chan []modEvent, 10) - go batchEvents(ctx, mod, 10*time.Millisecond, func(evs []modEvent) { batches <- evs }) +func Test_syncer(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "main.go") + script := filepath.Join(dir, "run.sh") + sub := filepath.Join(dir, "pkg") + if err := os.WriteFile(file, []byte("package main\n"), 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(script, []byte("#!/bin/sh\n"), 0o755); err != nil { + t.Fatal(err) + } + if err := os.Mkdir(sub, 0o755); err != nil { + t.Fatal(err) + } - expectBatch := func(want bool) { + guest := &fakeRunner{} + logger := logrus.New() + logger.SetOutput(io.Discard) + s := newSyncer(guest, logrus.NewEntry(logger)) + synced := func(paths ...string) [][]string { t.Helper() - select { - case evs := <-batches: - if !want { - t.Fatalf("unexpected sync of %d event(s)", len(evs)) - } - case <-time.After(200 * time.Millisecond): - if want { - t.Fatal("expected a sync") - } - } + guest.cmds = nil + s.sync(paths) + sort.Slice(guest.cmds, func(i, j int) bool { + return strings.Join(guest.cmds[i], " ") < strings.Join(guest.cmds[j], " ") + }) + return guest.cmds } - mod <- event("package main\n", true) - expectBatch(true) + want := syncEventsCmds([]modEvent{{path: file, FileMode: 0o644}, {path: script, FileMode: 0o755}}) + if got := synced(file, script, sub, filepath.Join(dir, "missing.go")); !reflect.DeepEqual(got, want) { + t.Errorf("first sync = %q, want %q", got, want) + } - mod <- event("", false) - expectBatch(false) + if got := synced(file, script); len(got) != 0 { + t.Errorf("unchanged files were synced again: %q", got) + } time.Sleep(10 * time.Millisecond) - mod <- event("package main\n\nfunc main() {}\n", true) - expectBatch(true) - - mod <- modEvent{path: path, FileMode: 0o644} - expectBatch(true) + if err := os.WriteFile(file, []byte("package main\n\nfunc main() {}\n"), 0o644); err != nil { + t.Fatal(err) + } + want = syncEventsCmds([]modEvent{{path: file, FileMode: 0o644}}) + if got := synced(file, script); !reflect.DeepEqual(got, want) { + t.Errorf("sync after a change = %q, want %q", got, want) + } } diff --git a/daemon/process/inotify/watch.go b/daemon/process/inotify/watch.go index 1c16c801..fabeca29 100644 --- a/daemon/process/inotify/watch.go +++ b/daemon/process/inotify/watch.go @@ -3,7 +3,6 @@ package inotify import ( "context" "fmt" - "os" "github.com/abiosoft/colima/util" "github.com/rjeczalik/notify" @@ -20,7 +19,7 @@ type dirWatcher interface { // An error is returned when the watcher can not be started in background. // // The watcher terminates on fatal error or when ctx is done. - Watch(ctx context.Context, dirs []string, c chan<- modEvent) error + Watch(ctx context.Context, dirs []string, c chan<- string) error } type defaultWatcher struct { @@ -28,7 +27,7 @@ type defaultWatcher struct { } // Watch implements dirWatcher -func (d *defaultWatcher) Watch(ctx context.Context, dirs []string, mod chan<- modEvent) error { +func (d *defaultWatcher) Watch(ctx context.Context, dirs []string, mod chan<- string) error { log := d.log // notify drops events when the channel is full, which happens during // bursts of file changes. @@ -45,7 +44,7 @@ func (d *defaultWatcher) Watch(ctx context.Context, dirs []string, mod chan<- mo } } - go func(ctx context.Context, c chan notify.EventInfo, mod chan<- modEvent) { + go func(ctx context.Context, c chan notify.EventInfo, mod chan<- string) { for { select { @@ -62,19 +61,8 @@ func (d *defaultWatcher) Watch(ctx context.Context, dirs []string, mod chan<- mo log.Tracef("received event %s for %s", e.Event().String(), path) - stat, err := os.Stat(path) - if err != nil { - log.Trace(fmt.Errorf("unable to stat inotify file '%s': %w", path, err)) - continue - } - - if stat.IsDir() { - log.Tracef("'%s' is directory, ignoring.", path) - continue - } - // send modification event - mod <- modEvent{path: path, FileMode: stat.Mode(), info: stat} + mod <- path } } }(ctx, c, mod)