diff --git a/daemon/process/inotify/events.go b/daemon/process/inotify/events.go index f40392bd..d221ed7a 100644 --- a/daemon/process/inotify/events.go +++ b/daemon/process/inotify/events.go @@ -4,34 +4,240 @@ import ( "context" "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 } func (m modEvent) Mode() string { return fmt.Sprintf("%o", m.FileMode) } -const syncEventScript = `/bin/chmod "$1" "$2" && : >> "$2"` +// syncEventsScript syncs the files passed as arguments after their mode. +// Files that do not exist in the VM are skipped. +// +// 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 + +// 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. +const maxSyncArgsSize = 64 * 1024 + +// 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() + if _, ok := byMode[mode]; !ok { + modes = append(modes, mode) + } + byMode[mode] = append(byMode[mode], ev.path) + } + 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 +} + +// 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 + handling := false + done := make(chan struct{}, 1) + + start := func() { + paths := pending + pending = nil + seen = map[string]struct{}{} + handling = true + go func() { + handle(paths) + done <- struct{}{} + }() + } + + for { + select { + case <-ctx.Done(): + return + + case path, ok := <-mod: + if !ok { + return + } + if _, ok := seen[path]; ok { + continue + } + seen[path] = struct{}{} + pending = append(pending, path) + if !handling && flush == nil { + flush = time.After(delay) + } + + case <-flush: + flush = nil + start() + + case <-done: + handling = false + if len(pending) > 0 { + start() + } + } + } +} + +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) + } + } -func syncEventCmd(ev modEvent) []string { - return []string{"sudo", "/bin/sh", "-c", syncEventScript, "sh", ev.Mode(), ev.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) { + 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) } - var last time.Time + go batchEvents(ctx, mod, syncBatchDelay, newSyncer(f.guest, log).sync) + var cancelWatch context.CancelFunc var currentVols []string @@ -47,14 +253,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 @@ -74,42 +277,11 @@ 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)) } }(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..84a0b2a0 100644 --- a/daemon/process/inotify/events_test.go +++ b/daemon/process/inotify/events_test.go @@ -1,32 +1,206 @@ package inotify import ( + "context" + "fmt" + "io" + "os" + "path/filepath" "reflect" + "sort" + "strings" + "sync" "testing" + "time" + + "github.com/sirupsen/logrus" ) -func Test_syncEventCmd(t *testing.T) { +func Test_syncEventsCmds(t *testing.T) { + cmd := func(mode string, paths ...string) []string { + return append([]string{"sudo", "/bin/sh", "-c", syncEventsScript, "sh", mode}, paths...) + } + 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: "one command per mode", + evs: []modEvent{ + {path: "/Users/someone/project/main.go", FileMode: 0o644}, + {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`), + }, }, } 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 { + size := 0 + for _, path := range cmd[6:] { + size += len(path) + 1 + } + if size > maxSyncArgsSize { + t.Errorf("command arguments size %d exceeds %d", size, maxSyncArgsSize) + } + paths = append(paths, cmd[6:]...) + } + 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 string) + var mu sync.Mutex + var batches [][]string + release := make(chan struct{}) + + go batchEvents(ctx, mod, 500*time.Millisecond, func(paths []string) { + mu.Lock() + batches = append(batches, paths) + first := len(batches) == 1 + mu.Unlock() + if first { + <-release + } + }) + + 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): + } + } + } + + for i := range 100 { + mod <- fmt.Sprintf("/a/%d", i) + } + mod <- "/a/0" + waitBatches(1) + + for i := range 50 { + mod <- fmt.Sprintf("/b/%d", i) + } + mod <- "/b/0" + close(release) + waitBatches(2) + + mu.Lock() + defer mu.Unlock() + if len(batches[0]) != 100 { + t.Errorf("first batch has %d paths, want 100", len(batches[0])) + } + if len(batches[1]) != 50 { + t.Errorf("paths received during a sync: got batch of %d, want 50", len(batches[1])) + } +} + +type fakeRunner struct { + mu sync.Mutex + cmds [][]string +} + +func (f *fakeRunner) RunQuiet(args ...string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.cmds = append(f.cmds, args) + return nil +} + +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) + } + + guest := &fakeRunner{} + logger := logrus.New() + logger.SetOutput(io.Discard) + s := newSyncer(guest, logrus.NewEntry(logger)) + synced := func(paths ...string) [][]string { + t.Helper() + 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 + } + + 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) + } + + if got := synced(file, script); len(got) != 0 { + t.Errorf("unchanged files were synced again: %q", got) + } + + time.Sleep(10 * time.Millisecond) + 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 46118f49..fabeca29 100644 --- a/daemon/process/inotify/watch.go +++ b/daemon/process/inotify/watch.go @@ -3,13 +3,14 @@ package inotify import ( "context" "fmt" - "os" "github.com/abiosoft/colima/util" "github.com/rjeczalik/notify" "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. @@ -18,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 { @@ -26,9 +27,11 @@ 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 - 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) @@ -41,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 { @@ -58,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()} + mod <- path } } }(ctx, c, mod)