diff --git a/v3/pkg/application/event_ordering_test.go b/v3/pkg/application/event_ordering_test.go new file mode 100644 index 00000000000..0a7d7a5dd49 --- /dev/null +++ b/v3/pkg/application/event_ordering_test.go @@ -0,0 +1,351 @@ +package application + +import ( + "fmt" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// threadProbeApp stands in for the platform layer so a test can control which +// thread it appears to be on and when queued work actually runs. +// +// Embedding platformApp supplies nil-bodied placeholders for everything the +// dispatch path never touches. +type threadProbeApp struct { + platformApp + + onMain atomic.Bool + mu sync.Mutex + pending []uint +} + +func (f *threadProbeApp) isOnMainThread() bool { return f.onMain.Load() } + +// dispatchOnMainThread records the work instead of running it, so a test can +// hold the UI thread still and then release it. +func (f *threadProbeApp) dispatchOnMainThread(id uint) { + f.mu.Lock() + f.pending = append(f.pending, id) + f.mu.Unlock() +} + +// runPending executes the deferred work in the order it was dispatched, which +// is what the real main-thread queue does. +func (f *threadProbeApp) runPending() { + for { + f.mu.Lock() + if len(f.pending) == 0 { + f.mu.Unlock() + return + } + id := f.pending[0] + f.pending = f.pending[1:] + f.mu.Unlock() + + mainThreadFunctionStoreLock.Lock() + fn := mainThreadFunctionStore[id] + delete(mainThreadFunctionStore, id) + mainThreadFunctionStoreLock.Unlock() + + if fn == nil { + continue + } + // Anything dispatched this way is, by definition, running on the UI + // thread — so report that while it runs. Without this, an InvokeSync + // inside the callback would try to hop to a thread it is already on + // and wait forever. + prev := f.onMain.Load() + f.onMain.Store(true) + fn() + f.onMain.Store(prev) + } +} + +// stubWindowImpl records what actually reached the webview, in the order it +// got there. That order is the thing under test. +type stubWindowImpl struct { + webviewWindowImpl + mu sync.Mutex + seen []string +} + +func (s *stubWindowImpl) execJS(js string) { + s.mu.Lock() + s.seen = append(s.seen, js) + s.mu.Unlock() +} + +func (s *stubWindowImpl) delivered() []string { + s.mu.Lock() + defer s.mu.Unlock() + return append([]string(nil), s.seen...) +} + +func (f *threadProbeApp) pendingCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.pending) +} + +// newOrderingProbe wires a window to a controllable platform layer. Leaving +// runtimeLoaded false makes ExecJS record into pendingJS instead of hopping to +// the UI thread, so pendingJS is the delivery order. +func newOrderingProbe(t *testing.T) (*threadProbeApp, *WebviewWindow, func()) { + t.Helper() + + probe := &threadProbeApp{} + prev := globalApplication + globalApplication = &App{impl: probe} + + // runtimeLoaded must be true: with it false ExecJS short-circuits into + // pendingJS and never touches the main-thread dispatch that this is all + // about, which would make the test pass whether or not the bug is present. + win := &WebviewWindow{ + id: 1, + impl: &stubWindowImpl{}, + runtimeLoaded: true, + } + + return probe, win, func() { globalApplication = prev } +} + +// A UI-thread emit must not overtake an event a goroutine queued earlier. +// +// This is the regression test for the inline fast path in +// App.dispatchOnMainThread: when the caller is already on the UI thread the +// work runs immediately, so before the queue existed a main-thread emit +// executed its eval while an earlier goroutine emit was still waiting to be +// dispatched. Measured at ~4.4% of events inverted under two concurrent +// emitters on macOS, Linux and Windows. +func TestEventFromMainThreadDoesNotOvertakeQueuedEvent(t *testing.T) { + probe, win, restore := newOrderingProbe(t) + defer restore() + impl := win.impl.(*stubWindowImpl) + + // Emitted from a goroutine. Without the queue this parks in InvokeSync + // waiting for the UI thread, so it runs in its own goroutine either way. + probe.onMain.Store(false) + firstReturned := make(chan struct{}) + go func() { + defer close(firstReturned) + win.DispatchWailsEvent(&CustomEvent{Name: "first"}) + }() + + // Wait until it has reached the UI-thread dispatch queue. + deadline := time.Now().Add(5 * time.Second) + for probe.pendingCount() == 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if probe.pendingCount() == 0 { + t.Fatal("the goroutine emit never reached the main-thread dispatch queue") + } + + // Now emit from the UI thread, while that one is still outstanding. + // dispatchOnMainThread runs inline here, which is exactly how the later + // event used to overtake the earlier one. + probe.onMain.Store(true) + win.DispatchWailsEvent(&CustomEvent{Name: "second"}) + + // Release the UI thread. + probe.runPending() + + select { + case <-firstReturned: + case <-time.After(5 * time.Second): + t.Fatal("the goroutine emit never completed") + } + + got := impl.delivered() + if len(got) != 2 { + t.Fatalf("delivered %d events, want 2: %v", len(got), got) + } + if !strings.Contains(got[0], `"first"`) || !strings.Contains(got[1], `"second"`) { + t.Fatalf("the later main-thread event overtook the earlier queued one\n 1st delivered: %s\n 2nd delivered: %s", got[0], got[1]) + } +} + +// Events emitted in sequence from one goroutine must arrive in that sequence. +func TestEventQueuePreservesSingleProducerOrder(t *testing.T) { + probe, win, restore := newOrderingProbe(t) + defer restore() + + probe.onMain.Store(false) + + // More events than the queue holds, so the drain has to run concurrently + // with the producer; emitting them all first would (correctly) block on the + // bound with nobody there to relieve it. + const n = eventQueueCapacity * 4 + + stop := make(chan struct{}) + drained := make(chan struct{}) + go func() { + defer close(drained) + for { + probe.runPending() + select { + case <-stop: + probe.runPending() // final sweep + return + default: + } + } + }() + + for i := 0; i < n; i++ { + win.enqueueEventJS(fmt.Sprintf("e%d", i)) + } + close(stop) + <-drained + + deadline := time.Now().Add(10 * time.Second) + for { + delivered := len(win.impl.(*stubWindowImpl).delivered()) + if delivered == n || time.Now().After(deadline) { + break + } + probe.runPending() + } + + got := win.impl.(*stubWindowImpl).delivered() + + if len(got) != n { + t.Fatalf("delivered %d events, want %d", len(got), n) + } + for i := 0; i < n; i++ { + if want := fmt.Sprintf("e%d", i); got[i] != want { + t.Fatalf("event %d = %q, want %q", i, got[i], want) + } + } +} + +// A UI-thread emitter must never block, however many events it emits: it is +// the drainer, so blocking it could not be relieved by anyone. Emitting far +// more than the queue holds must still complete, in order. +// +// In practice it never even approaches the bound, because the drain scheduled +// by InvokeAsync runs inline when already on the UI thread — so a main-thread +// emitter enqueues and immediately drains. The bound exists for goroutine +// emitters racing a busy UI thread. +func TestEventQueueDoesNotBlockMainThreadWhenFull(t *testing.T) { + probe, win, restore := newOrderingProbe(t) + defer restore() + + probe.onMain.Store(true) + + const n = eventQueueCapacity * 3 + + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < n; i++ { + win.enqueueEventJS(fmt.Sprintf("e%d", i)) + } + }() + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("a main-thread emitter blocked on a full queue; this deadlocks in a real app") + } + + got := win.impl.(*stubWindowImpl).delivered() + + if len(got) != n { + t.Fatalf("delivered %d events, want %d", len(got), n) + } + for i := 0; i < n; i++ { + if want := fmt.Sprintf("e%d", i); got[i] != want { + t.Fatalf("event %d = %q, want %q", i, got[i], want) + } + } + + // Draining inline is what keeps the UI thread from ever waiting. + if high := win.eventQueueHighWater(); high > eventQueueCapacity { + t.Errorf("high water = %d; a main-thread emitter should drain inline, not accumulate", high) + } +} + +// Destroying a window must release an emitter waiting for queue space rather +// than leaving the goroutine parked forever. +func TestEventQueueCloseReleasesBlockedProducer(t *testing.T) { + probe, win, restore := newOrderingProbe(t) + defer restore() + + probe.onMain.Store(false) + + // Fill to capacity so the next append has to wait. + for i := 0; i < eventQueueCapacity; i++ { + win.enqueueEventJS(fmt.Sprintf("e%d", i)) + } + + // Wait until the producer is genuinely parked before closing, otherwise + // close can win the race and the test passes without ever exercising the + // Broadcast that is the thing under test. + blocked := make(chan struct{}) + go func() { + defer close(blocked) + win.enqueueEventJS("waits for space") + }() + + deadline := time.Now().Add(5 * time.Second) + for { + win.eventQueueMu.Lock() + waiting := len(win.eventQueue) >= eventQueueCapacity + win.eventQueueMu.Unlock() + + select { + case <-blocked: + t.Fatal("the producer returned without waiting; the queue was not full") + default: + } + if waiting && time.Now().After(deadline.Add(-4900*time.Millisecond)) { + break // queue is full and the producer has had a chance to park + } + if time.Now().After(deadline) { + t.Fatal("producer never reached the full queue") + } + time.Sleep(2 * time.Millisecond) + } + + win.closeEventQueue() + + select { + case <-blocked: + case <-time.After(10 * time.Second): + t.Fatal("closing the queue did not wake the blocked emitter") + } +} + +// A payload parked just as the window goes away must not be stranded. The +// dispatcher can pass the destroyed check, markAsDestroyed can then close the +// queue and run dropWindow, and only then does put succeed — so dropWindow +// cannot see it and the queue refuses the event that would have fetched it. +func TestOrphanedPayloadIsReclaimedWhenQueueRefuses(t *testing.T) { + probe, win, restore := newOrderingProbe(t) + defer restore() + probe.onMain.Store(false) + + store := newEventPayloadStore() + globalApplication.eventPayloads = store + + // Simulate the interleaving: the queue is already closed by the time the + // dispatcher reaches it. + win.closeEventQueue() + + big := make([]byte, maxInlineEventPayload+1) + for i := range big { + big[i] = 'x' + } + win.DispatchWailsEvent(&CustomEvent{Name: "big", Data: string(big)}) + + store.mu.Lock() + parked, bytes := len(store.items), store.bytes + store.mu.Unlock() + + if parked != 0 || bytes != 0 { + t.Errorf("payload left stranded in the store: %d entries, %d bytes", parked, bytes) + } +} diff --git a/v3/pkg/application/webview_window.go b/v3/pkg/application/webview_window.go index 25dbf5e6335..689d035124f 100644 --- a/v3/pkg/application/webview_window.go +++ b/v3/pkg/application/webview_window.go @@ -189,6 +189,15 @@ type WebviewWindow struct { menuBindings map[string]*MenuItem menuBindingsLock sync.RWMutex + // Events are queued and drained by a single consumer on the UI thread so + // that delivery order matches emit order. See enqueueEventJS. + eventQueueMu sync.Mutex + eventQueueCond *sync.Cond + eventQueue []string + eventDraining bool + eventQueueClosed bool + eventQueueHigh int + // Indicates that the window is destroyed destroyed bool destroyedLock sync.RWMutex @@ -268,8 +277,12 @@ func (w *WebviewWindow) onApplicationEvent( func (w *WebviewWindow) markAsDestroyed() { w.destroyedLock.Lock() - defer w.destroyedLock.Unlock() w.destroyed = true + w.destroyedLock.Unlock() + + // Release anyone blocked on a full queue. Done outside destroyedLock so the + // lock order between it and eventQueueMu is always one-way. + w.closeEventQueue() // Anything parked for this window will never be fetched now. TTL would // eventually reclaim it, but the window is gone, so release immediately. @@ -1425,7 +1438,14 @@ func (w *WebviewWindow) DispatchWailsEvent(event *CustomEvent) { if len(payload) > maxInlineEventPayload { if store := globalApplication.eventPayloads; store != nil { if id, ok := store.put(w.id, []byte(payload)); ok { - w.ExecJS(fmt.Sprintf(refEventJS, eventPayloadPath+id)) + if w.enqueueEventJS(fmt.Sprintf(refEventJS, eventPayloadPath+id)) { + return + } + // The window closed between put and enqueue, so nothing will + // ever fetch this. dropWindow has already run by then and + // cannot see it, so reclaim it here rather than leaving it for + // the TTL sweep. + store.take(id, w.id) return } } @@ -1433,7 +1453,135 @@ func (w *WebviewWindow) DispatchWailsEvent(event *CustomEvent) { // event pays the out-of-line retention cost, which beats dropping it. } - w.ExecJS(fmt.Sprintf(inlineEventJS, payload)) + w.enqueueEventJS(fmt.Sprintf(inlineEventJS, payload)) +} + +// eventQueueCapacity bounds how many events may be waiting for the UI thread +// before an emitting goroutine is made to wait. Typical apps emit far below +// this, so the queue stays empty and the bound never engages; it exists to stop +// a runaway producer from growing the queue without limit. +// +// A main-thread emitter is never blocked by it — see enqueueEventJS. +// +// Chosen by measurement (v3/tests/event-performance, macOS, 25s runs). Ordering +// held at every size, and 5000 ev/s was sustained at every size, so the only +// thing that moved was tail latency, which grows with depth because an event +// waits behind whatever is already queued: +// +// capacity 5000 ev/s mixed-source p99 +// 16 5000/s, 0 inv 179us +// 64 4991/s, 0 inv 257us +// 256 4995/s, 0 inv 329us +// 1024 5000/s, 0 inv 427us +// +// Shallow is better on both counts, so this is deliberately small: enough to +// absorb a burst without making a producer wait, not so much that an event can +// sit behind a long backlog. Raise it only with a measurement that shows a +// producer being throttled. +const eventQueueCapacity = 64 + +// enqueueEventJS appends an event's JavaScript to the window's queue and makes +// sure a drain is scheduled. +// +// This exists because dispatchOnMainThread runs inline when the caller is +// already on the UI thread. Without a queue, an event emitted from the UI +// thread executes its eval immediately while an event emitted moments earlier +// from a goroutine is still sitting in the dispatch queue, so the later event +// overtakes the earlier one. TestEventFromMainThreadDoesNotOvertakeQueuedEvent +// reproduces it deterministically. +// +// Appending under a mutex makes the queue the single ordering authority: the +// order events are added is the order one drainer emits them, whichever thread +// each came from. +// +// Returns false when the queue has been closed because the window is going +// away, so a caller that has already parked a payload can reclaim it. +func (w *WebviewWindow) enqueueEventJS(js string) bool { + // Whether we are on the UI thread decides if we may block below, so it must + // be read before taking the queue lock. + onMainThread := globalApplication != nil && + globalApplication.impl != nil && + globalApplication.impl.isOnMainThread() + + w.eventQueueMu.Lock() + if w.eventQueueCond == nil { + w.eventQueueCond = sync.NewCond(&w.eventQueueMu) + } + + // Backpressure for goroutine emitters only. The UI thread is the drainer, + // so blocking it on a full queue could never be relieved — it would be a + // guaranteed deadlock rather than a slow path. It is allowed past the bound + // instead, and the drain it schedules runs inline immediately afterwards. + for !onMainThread && !w.eventQueueClosed && len(w.eventQueue) >= eventQueueCapacity { + w.eventQueueCond.Wait() + } + if w.eventQueueClosed { + w.eventQueueMu.Unlock() + return false + } + + w.eventQueue = append(w.eventQueue, js) + if n := len(w.eventQueue); n > w.eventQueueHigh { + w.eventQueueHigh = n + } + + scheduleDrain := !w.eventDraining + if scheduleDrain { + w.eventDraining = true + } + w.eventQueueMu.Unlock() + + if scheduleDrain { + // Runs inline when already on the UI thread, which is what lets a + // main-thread emitter make progress past the bound. + InvokeAsync(w.drainEventQueue) + } + return true +} + +// drainEventQueue empties the queue in order. It always runs on the UI thread, +// and only one drain is ever in flight per window. +func (w *WebviewWindow) drainEventQueue() { + for { + w.eventQueueMu.Lock() + if len(w.eventQueue) == 0 || w.eventQueueClosed { + w.eventDraining = false + w.eventQueueMu.Unlock() + return + } + // Take the whole batch rather than popping one at a time: repeatedly + // reslicing the front would keep the original backing array alive. + batch := w.eventQueue + w.eventQueue = nil + w.eventQueueCond.Broadcast() // space is available again + w.eventQueueMu.Unlock() + + for _, js := range batch { + if w.isDestroyed() { + return + } + w.ExecJS(js) + } + } +} + +// closeEventQueue discards anything still queued and wakes blocked emitters. +func (w *WebviewWindow) closeEventQueue() { + w.eventQueueMu.Lock() + w.eventQueueClosed = true + w.eventQueue = nil + if w.eventQueueCond != nil { + w.eventQueueCond.Broadcast() + } + w.eventQueueMu.Unlock() +} + +// eventQueueHighWater reports the deepest the window's event queue has been. +// Diagnostics for tests; deliberately unexported so it is not public API. +func (w *WebviewWindow) eventQueueHighWater() int { + w.eventQueueMu.Lock() + defer w.eventQueueMu.Unlock() + return w.eventQueueHigh } func (w *WebviewWindow) dispatchWindowEvent(id uint) { diff --git a/v3/tests/event-performance/main.go b/v3/tests/event-performance/main.go index f4b9ac46f84..fa80bd0ca47 100644 --- a/v3/tests/event-performance/main.go +++ b/v3/tests/event-performance/main.go @@ -386,12 +386,31 @@ func (h *harness) startEmitter(win *application.WebviewWindow, sc Scenario) func case <-done: return case <-ticker.C: - seq := sc.nextSeq() - ev := &application.CustomEvent{ - Name: "perf", - Data: perfPayload{Seq: seq, TMS: h.nowMS(), Pad: pad}, - } - application.InvokeAsync(func() { win.DispatchWailsEvent(ev) }) + // The sequence number must be taken inside the callback, + // at the moment DispatchWailsEvent is actually called. + // Taking it out here and deferring the dispatch would make + // seq order differ from call order by construction, and + // every reorder the JS side counted would be this loop's + // own race rather than anything the framework did. + // Acquired here, on the ticker goroutine, and released by + // the UI callback below. Locking inside the callback + // instead would put the UI thread behind a lock that a + // goroutine can be holding while it waits for queue + // capacity — and since the UI thread is the drainer, that + // wait could never be relieved. Handing the lock over means + // only this goroutine ever waits for it. + // + // sync.Mutex permits unlocking from a different goroutine. + sc.emitMu.Lock() + application.InvokeAsync(func() { + defer sc.emitMu.Unlock() + seq := sc.nextSeq() + ev := &application.CustomEvent{ + Name: "perf", + Data: perfPayload{Seq: seq, TMS: h.nowMS(), Pad: pad}, + } + win.DispatchWailsEvent(ev) + }) } } }() @@ -421,6 +440,9 @@ func (h *harness) startEmitter(win *application.WebviewWindow, sc Scenario) func n := int(credit) credit -= float64(n) for i := 0; i < n; i++ { + if sc.MixedSource { + sc.emitMu.Lock() + } seq := sc.nextSeq() body := pad if seq%2 == 1 { @@ -431,7 +453,14 @@ func (h *harness) startEmitter(win *application.WebviewWindow, sc Scenario) func Data: perfPayload{Seq: seq, TMS: h.nowMS(), Pad: body}, } t := time.Now() - win.DispatchWailsEvent(ev) + func() { + // A panic in dispatch must not leave emitMu held, or + // every later emit in this scenario would wedge. + if sc.MixedSource { + defer sc.emitMu.Unlock() + } + win.DispatchWailsEvent(ev) + }() us := float64(time.Since(t).Nanoseconds()) / 1e3 h.mu.Lock() diff --git a/v3/tests/event-performance/scenarios.go b/v3/tests/event-performance/scenarios.go index 03770073f91..1e8b85d3e43 100644 --- a/v3/tests/event-performance/scenarios.go +++ b/v3/tests/event-performance/scenarios.go @@ -2,6 +2,7 @@ package main import ( "strings" + "sync" "sync/atomic" "time" ) @@ -31,6 +32,13 @@ type Scenario struct { AltPayloadBytes int counter *atomic.Int64 // shared across copies of the struct + + // emitMu serialises taking a sequence number with the dispatch call that + // carries it. Without it, two emitters can take seq N and N+1 and then + // reach the framework in the other order, and the reorders the JS side + // counts are this harness racing itself rather than anything the framework + // did. Only used by MixedSource scenarios. + emitMu *sync.Mutex } func (s Scenario) nextSeq() int64 { return s.counter.Add(1) - 1 } @@ -39,6 +47,7 @@ func (s Scenario) resetCounter() { s.counter.Store(0) } func newScenario(s Scenario) Scenario { s.counter = &atomic.Int64{} + s.emitMu = &sync.Mutex{} return s }