diff --git a/watch/burst.go b/watch/burst.go new file mode 100644 index 0000000..50716b3 --- /dev/null +++ b/watch/burst.go @@ -0,0 +1,67 @@ +package watch + +import ( + "time" + + "github.com/fsnotify/fsnotify" +) + +// burstDetector decides when the event loop is inside a checkout storm and +// holds the storm's events until it settles. It never reads the wall clock: +// every method takes `now`, so the loop passes time.Now() and tests pass a +// synthetic clock and get the same decisions every run (#201). +type burstDetector struct { + arrivals []time.Time // source-event times inside burstDetectWindow + queue []fsnotify.Event // the storm's events, first-seen order, last op per path + queued map[string]int // path -> index into queue + start time.Time // when the current burst opened +} + +// observe records one source-event arrival and reports whether the loop is +// in a burst: either one is already open, or more than burstThreshold +// arrivals fall inside burstDetectWindow ending now. +func (b *burstDetector) observe(now time.Time) bool { + if b.queued != nil { + return true + } + cut := 0 + for cut < len(b.arrivals) && now.Sub(b.arrivals[cut]) > burstDetectWindow { + cut++ + } + b.arrivals = append(b.arrivals[cut:], now) + return len(b.arrivals) > burstThreshold +} + +// enqueue adds an event to the open burst, keeping the last op per path, and +// returns how long the loop should wait before flushing: burstSettle of +// quiet, clipped so the burst never stays open past burstCeiling. +func (b *burstDetector) enqueue(event fsnotify.Event, now time.Time) time.Duration { + if b.queued == nil { + b.queued = make(map[string]int) + b.start = now + } + if i, ok := b.queued[event.Name]; ok { + b.queue[i] = event + } else { + b.queued[event.Name] = len(b.queue) + b.queue = append(b.queue, event) + } + if remaining := burstCeiling - now.Sub(b.start); remaining < burstSettle { + return max(remaining, 0) + } + return burstSettle +} + +// take closes the burst and returns its events; the next observe starts a +// fresh detection window. +func (b *burstDetector) take() []fsnotify.Event { + events := b.queue + b.queue, b.queued = nil, nil + b.arrivals = b.arrivals[:0] + return events +} + +// pending reports whether a burst is open with events waiting. +func (b *burstDetector) pending() bool { + return len(b.queue) > 0 +} diff --git a/watch/burst_test.go b/watch/burst_test.go index 1625e32..0fcb490 100644 --- a/watch/burst_test.go +++ b/watch/burst_test.go @@ -11,24 +11,93 @@ import ( "github.com/fsnotify/fsnotify" ) -// A rebase or branch switch writes every touched file through a temp file and -// a rename, so a 300-file checkout used to cost 300 git subprocesses, 300 -// state writes and 300 topology recomputes on the event loop (#201). The -// storm must coalesce into one batch. -func TestCheckoutStormCoalescesIntoOneBatch(t *testing.T) { - root := t.TempDir() - if err := os.Mkdir(filepath.Join(root, ".git"), 0o755); err != nil { - t.Fatal(err) - } +// stormEvents is a 300-file checkout the way git writes one: every file +// CREATEd, the first 200 also WRITTEN, 500 events in all. +func stormEvents(t *testing.T, root string) []fsnotify.Event { + t.Helper() const files = 300 - paths := make([]string, files) - for i := range paths { - paths[i] = filepath.Join(root, fmt.Sprintf("f%03d.go", i)) - if err := os.WriteFile(paths[i], []byte("package x\n"), 0o644); err != nil { + events := make([]fsnotify.Event, 0, 500) + for i := 0; i < files; i++ { + path := filepath.Join(root, fmt.Sprintf("f%03d.go", i)) + if err := os.WriteFile(path, []byte("package x\n"), 0o644); err != nil { t.Fatal(err) } + events = append(events, fsnotify.Event{Name: path, Op: fsnotify.Create}) + if i < 200 { + events = append(events, fsnotify.Event{Name: path, Op: fsnotify.Write}) + } } + return events +} + +// The detector takes its clock from the caller, so the same 500 arrivals +// 0.4 ms apart make the same decisions every run: the first burstThreshold +// are handled one by one, everything after is queued, and one settle +// closes the burst with one event per path. +func TestBurstDetectorOpensOnceForAStorm(t *testing.T) { + events := stormEvents(t, t.TempDir()) + var b burstDetector + clock := time.Unix(1_700_000_000, 0) + const gap = 400 * time.Microsecond // 500 events in 200 ms + perFile := 0 + for _, event := range events { + if !b.observe(clock) { + perFile++ + } else { + if delay := b.enqueue(event, clock); delay != burstSettle { + t.Fatalf("enqueue delay = %v, want burstSettle %v", delay, burstSettle) + } + } + clock = clock.Add(gap) + } + if perFile != burstThreshold { + t.Fatalf("events handled before the burst opened = %d, want burstThreshold %d", perFile, burstThreshold) + } + batch := b.take() + // The first 16 events are f000..f007 (CREATE+WRITE each); the batch holds + // the other 292 files once each, last op wins. + if len(batch) != 300-burstThreshold/2 { + t.Fatalf("batch = %d events, want %d (one per remaining file)", len(batch), 300-burstThreshold/2) + } + if got := batch[0]; filepath.Base(got.Name) != "f008.go" || got.Op != fsnotify.Write { + t.Fatalf("batch[0] = %s %v, want f008.go WRITE (last op for the path)", filepath.Base(got.Name), got.Op) + } + if b.pending() { + t.Fatal("take must close the burst") + } + // After the burst, a lone event starts a fresh window and is not batched. + if b.observe(clock.Add(time.Second)) { + t.Fatal("a single event after a flush must not open a burst") + } +} + +// The ceiling clips the settle delay so a storm that never goes quiet still +// flushes after burstCeiling. +func TestBurstDetectorCeilingClipsSettle(t *testing.T) { + var b burstDetector + clock := time.Unix(1_700_000_000, 0) + for i := 0; i <= burstThreshold; i++ { + b.observe(clock) + } + event := fsnotify.Event{Name: "a.go", Op: fsnotify.Create} + if delay := b.enqueue(event, clock); delay != burstSettle { + t.Fatalf("first enqueue delay = %v, want %v", delay, burstSettle) + } + late := clock.Add(burstCeiling - 100*time.Millisecond) + if delay := b.enqueue(event, late); delay != 100*time.Millisecond { + t.Fatalf("delay near the ceiling = %v, want 100ms", delay) + } + if delay := b.enqueue(event, clock.Add(burstCeiling+time.Second)); delay != 0 { + t.Fatalf("delay past the ceiling = %v, want 0", delay) + } +} + +func newBurstTestDaemon(t *testing.T, root string) *Daemon { + t.Helper() + if err := os.Mkdir(filepath.Join(root, ".git"), 0o755); err != nil { + t.Fatal(err) + } d, err := NewDaemon(root, false) if err != nil { t.Fatal(err) @@ -41,69 +110,70 @@ func TestCheckoutStormCoalescesIntoOneBatch(t *testing.T) { if err := os.MkdirAll(d.runtimeDir, 0o755); err != nil { t.Fatal(err) } - var gitCalls, topology, refreshes atomic.Int32 + return d +} + +type burstCounters struct { + git, topology, refreshes atomic.Int32 +} + +// stubBurstSeams counts the shared costs. The daemon must be stopped before +// the seams are restored, so callers register Stop after this. +func stubBurstSeams(t *testing.T) *burstCounters { + t.Helper() + c := &burstCounters{} origDirty, origDirtyPaths := isFileDirtyFn, dirtyPathsFn origTopology, origRefresh := daemonComputeTopology, daemonRefreshDependencies t.Cleanup(func() { isFileDirtyFn, dirtyPathsFn = origDirty, origDirtyPaths daemonComputeTopology, daemonRefreshDependencies = origTopology, origRefresh }) - // Registered after the seam restore so it runs first: the loop must be - // stopped before the seams change under it. - t.Cleanup(func() { d.Stop() }) - isFileDirtyFn = func(string, string) bool { gitCalls.Add(1); return false } + isFileDirtyFn = func(string, string) bool { c.git.Add(1); return false } dirtyPathsFn = func(_ string, rel []string) map[string]bool { - gitCalls.Add(1) + c.git.Add(1) + if len(rel) == 0 { + return nil + } return map[string]bool{rel[0]: true} } - daemonComputeTopology = func(*Daemon) { topology.Add(1) } - daemonRefreshDependencies = func(*Daemon) { refreshes.Add(1) } + daemonComputeTopology = func(*Daemon) { c.topology.Add(1) } + daemonRefreshDependencies = func(*Daemon) { c.refreshes.Add(1) } + return c +} - d.eventLoopWG.Add(1) - go func() { - defer d.eventLoopWG.Done() - d.eventLoop() - }() +// A flushed batch pays each shared cost exactly once, however many files it +// holds, and still records one event per file with the batched dirty flag. +func TestHandleBurstPaysEachSharedCostOnce(t *testing.T) { + root := t.TempDir() + events := stormEvents(t, root) + d := newBurstTestDaemon(t, root) + c := stubBurstSeams(t) + t.Cleanup(func() { d.Stop() }) - // 500 events for 300 files inside 200 ms: every file CREATEd, the first - // 200 also RENAMEd into place, the way git writes a checkout. - start := time.Now() - for i := range paths { - d.watcher.Events <- fsnotify.Event{Name: paths[i], Op: fsnotify.Create} - if i < 200 { - d.watcher.Events <- fsnotify.Event{Name: paths[i], Op: fsnotify.Write} - } - } - if elapsed := time.Since(start); elapsed > 200*time.Millisecond { - t.Fatalf("feeding the storm took %v, want under 200ms for the test to mean anything", elapsed) + var b burstDetector + clock := time.Unix(1_700_000_000, 0) + for _, event := range events { + b.observe(clock) + b.enqueue(event, clock) + clock = clock.Add(400 * time.Microsecond) } + d.handleBurst(b.take()) - waitForWatchCondition(t, 5*time.Second, func() bool { - return len(d.GetEvents(files+1)) >= files - }) - // Let any trailing per-file work surface before counting. - time.Sleep(burstSettle + 50*time.Millisecond) - - // The first burstThreshold events are handled one by one before the storm - // is recognised; everything after that is one batch. - const leading = burstThreshold - if got := gitCalls.Load(); got > leading+1 { - t.Fatalf("git subprocesses = %d, want at most %d (per-file before detection + one batch)", got, leading+1) + if got := c.git.Load(); got != 1 { + t.Fatalf("git subprocesses = %d, want exactly 1", got) } - if got := topology.Load(); got > leading+1 { - t.Fatalf("topology recomputes = %d, want at most %d", got, leading+1) + if got := c.topology.Load(); got != 1 { + t.Fatalf("topology recomputes = %d, want exactly 1", got) } - if got := refreshes.Load(); got > leading+1 { - t.Fatalf("dependency refreshes = %d, want at most %d", got, leading+1) + if got := c.refreshes.Load(); got != 1 { + t.Fatalf("dependency refreshes = %d, want exactly 1", got) } - // Files handled before detection may log both their CREATE and WRITE; - // every file in the batch logs exactly once, last op wins. - events := d.GetEvents(files + leading + 1) - if len(events) < files || len(events) > files+leading { - t.Fatalf("events recorded = %d, want between %d and %d", len(events), files, files+leading) + recorded := d.GetEvents(1000) + if len(recorded) != 300 { + t.Fatalf("events recorded = %d, want 300 (one per file)", len(recorded)) } dirty := 0 - for _, event := range events { + for _, event := range recorded { if event.Dirty { dirty++ } @@ -113,29 +183,54 @@ func TestCheckoutStormCoalescesIntoOneBatch(t *testing.T) { } } -// Below the threshold nothing changes: a handful of events are still handled -// one at a time with their own git call each. -func TestFewEventsAreNotBatched(t *testing.T) { +// End to end through the real event loop on the wall clock. CI load can +// split one storm into several bursts, so this asserts only that batching +// happened: per-file costs stay far below one per file. Exact counts live in +// the synthetic-clock tests above. +func TestCheckoutStormIsBatchedThroughTheEventLoop(t *testing.T) { root := t.TempDir() - if err := os.Mkdir(filepath.Join(root, ".git"), 0o755); err != nil { - t.Fatal(err) + events := stormEvents(t, root) + d := newBurstTestDaemon(t, root) + c := stubBurstSeams(t) + t.Cleanup(func() { d.Stop() }) + + d.eventLoopWG.Add(1) + go func() { + defer d.eventLoopWG.Done() + d.eventLoop() + }() + for _, event := range events { + d.watcher.Events <- event } - path := filepath.Join(root, "a.go") - if err := os.WriteFile(path, []byte("package x\n"), 0o644); err != nil { - t.Fatal(err) + waitForWatchCondition(t, 10*time.Second, func() bool { + return len(d.GetEvents(1000)) >= 300 + }) + time.Sleep(burstSettle + 50*time.Millisecond) + + // Each burst costs at most burstThreshold per-file calls before it is + // recognised plus one batched call; allow a slow runner to split the + // storm four ways. Unbatched, the count would be 500. + const bound = 4 * (burstThreshold + 1) + if got := c.git.Load(); got > bound { + t.Fatalf("git subprocesses = %d, want at most %d", got, bound) } - d, err := NewDaemon(root, false) - if err != nil { - t.Fatal(err) + if got := c.topology.Load(); got > bound { + t.Fatalf("topology recomputes = %d, want at most %d", got, bound) } - if err := d.watcher.Close(); err != nil { - t.Fatal(err) + if got := c.refreshes.Load(); got > bound { + t.Fatalf("dependency refreshes = %d, want at most %d", got, bound) } - d.watcher.Events = make(chan fsnotify.Event, 8) - d.watcher.Errors = make(chan error) - if err := os.MkdirAll(d.runtimeDir, 0o755); err != nil { +} + +// Below the threshold nothing changes: a lone event is handled at once with +// its own git call. +func TestFewEventsAreNotBatched(t *testing.T) { + root := t.TempDir() + path := filepath.Join(root, "a.go") + if err := os.WriteFile(path, []byte("package x\n"), 0o644); err != nil { t.Fatal(err) } + d := newBurstTestDaemon(t, root) var perFile, batched atomic.Int32 origDirty, origDirtyPaths := isFileDirtyFn, dirtyPathsFn origTopology, origRefresh := daemonComputeTopology, daemonRefreshDependencies @@ -143,11 +238,11 @@ func TestFewEventsAreNotBatched(t *testing.T) { isFileDirtyFn, dirtyPathsFn = origDirty, origDirtyPaths daemonComputeTopology, daemonRefreshDependencies = origTopology, origRefresh }) - t.Cleanup(func() { d.Stop() }) isFileDirtyFn = func(string, string) bool { perFile.Add(1); return false } dirtyPathsFn = func(string, []string) map[string]bool { batched.Add(1); return nil } daemonComputeTopology = func(*Daemon) {} daemonRefreshDependencies = func(*Daemon) {} + t.Cleanup(func() { d.Stop() }) d.eventLoopWG.Add(1) go func() { diff --git a/watch/events.go b/watch/events.go index 5470a9b..4a9821f 100644 --- a/watch/events.go +++ b/watch/events.go @@ -292,19 +292,14 @@ func (d *Daemon) eventLoop() { } }() - // Burst state: arrivals is the recent source-event timestamps used to - // detect a storm; queue holds the storm's events, last op per path, in - // first-seen order, until the burst settles or hits its ceiling. - var ( - arrivals []time.Time - queue []fsnotify.Event - queued map[string]int - burstStart time.Time - burstTimer = time.NewTimer(time.Hour) - burstC <-chan time.Time - ) + // A checkout storm is detected and queued by burstDetector, which takes + // the clock from this loop so tests can drive it deterministically. The + // loop owns only the timer that fires the flush. + var burst burstDetector + burstTimer := time.NewTimer(time.Hour) burstTimer.Stop() defer burstTimer.Stop() + var burstC <-chan time.Time armBurst := func(delay time.Duration) { if !burstTimer.Stop() { select { @@ -315,44 +310,13 @@ func (d *Daemon) eventLoop() { burstTimer.Reset(delay) burstC = burstTimer.C } - enqueue := func(event fsnotify.Event, now time.Time) { - if queued == nil { - queued = make(map[string]int) - burstStart = now - } - if i, ok := queued[event.Name]; ok { - queue[i] = event - } else { - queued[event.Name] = len(queue) - queue = append(queue, event) - } - if remaining := burstCeiling - now.Sub(burstStart); remaining < burstSettle { - armBurst(max(remaining, 0)) - } else { - armBurst(burstSettle) - } - } flushBurst := func() { burstC = nil - events := queue - queue, queued = nil, nil - arrivals = arrivals[:0] - d.handleBurst(events) - } - inBurst := func(now time.Time) bool { - if queued != nil { - return true - } - cut := 0 - for cut < len(arrivals) && now.Sub(arrivals[cut]) > burstDetectWindow { - cut++ - } - arrivals = append(arrivals[cut:], now) - return len(arrivals) > burstThreshold + d.handleBurst(burst.take()) } defer func() { - if len(queue) > 0 { - d.handleBurst(queue) + if burst.pending() { + d.handleBurst(burst.take()) } }() @@ -432,9 +396,9 @@ func (d *Daemon) eventLoop() { continue } - if inBurst(now) { + if burst.observe(now) { debouncer.cancelPending(event.Name) - enqueue(event, now) + armBurst(burst.enqueue(event, now)) continue }