align cef linux pump with wakeup pipe scheduling

commit: 05695999b1addbefa007658cbc748fa1a0c39825
author: brice · 7mo ago · 2026-03-20 10:58 UTC
committer: brice · 6mo ago · 2026-03-28 04:36 UTC
parent: 52e0095
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

1 file · +92 -36

M internal/infrastructure/cef/engine.go +92 -36


@@ -2,6 +2,7 @@ package cef
2	2
3	3		import (
4	4		"context"
	5	+	"os"
5	6		"sync/atomic"
6	7		"time"
7	8

@@ -39,8 +40,10 @@ type Engine struct {
39	40		pumpEnabled atomic.Bool
40	41		pumpClosing atomic.Bool
41	42
42		-	pumpWakeQueued atomic.Bool
43		-	pumpIdleSourceID atomic.Uint64
	43	+	pumpSignalPending atomic.Bool
	44	+	pumpSignalSource atomic.Uint64
	45	+	pumpSignalRead *os.File
	46	+	pumpSignalWrite *os.File
44	47
45	48		pumpTimerDueAtMs atomic.Int64
46	49		pumpTimerSource atomic.Uint64

@@ -266,14 +269,15 @@ func (e *Engine) maybeLogBrowserCreateStall() {
266	269		// CEF message pump — thread-safe via atomic deadlines
267	270		//
268	271		// OnScheduleMessagePumpWork can be called from ANY Chromium thread (UI, IO,
269		-	// GPU compositor, etc.). It updates the next due time atomically and coalesces
270		-	// a wake-up on the GTK main loop. When work is not yet due, a one-shot GLib
271		-	// timeout is armed for the exact deadline instead of polling every few ms.
	272	+	// GPU compositor, etc.). It updates the next due time atomically and signals a
	273	+	// persistent GLib fd watch via a wakeup pipe. This closely follows the Linux
	274	+	// external-message-pump strategy used by upstream cefclient and avoids
	275	+	// creating/destroying GLib sources from Chromium threads.
272	276		// ---------------------------------------------------------------------------
273	277
274	278		// scheduleMessagePumpWork is called by OnScheduleMessagePumpWork from any
275		-	// thread. It updates the earliest due deadline atomically and queues a wake-up
276		-	// if the main loop needs to re-evaluate timer state.
	279	+	// thread. It updates the earliest due deadline atomically and signals the GTK
	280	+	// main thread when it needs to re-evaluate timer state.
277	281		func (e *Engine) scheduleMessagePumpWork(delayMs int64) {
278	282		nowMs := time.Now().UnixMilli()
279	283		dueAtMs := nowMs

@@ -315,7 +319,7 @@ func (e *Engine) scheduleMessagePumpWork(delayMs int64) {
315	319
316	320		armedDueAtMs := e.pumpTimerDueAtMs.Load()
317	321		if dueAtMs <= nowMs \|\| armedDueAtMs == 0 \|\| dueAtMs < armedDueAtMs {
318		-	e.queuePumpWake()
	322	+	e.signalPump()
319	323		}
320	324		}
321	325

@@ -324,10 +328,45 @@ func (e *Engine) startMessagePump() {
324	328		return
325	329		}
326	330
	331	+	readFile, writeFile, err := os.Pipe()
	332	+	if err != nil {
	333	+	logging.FromContext(e.ctx).Error().Err(err).Msg("cef: failed to create wakeup pipe")
	334	+	return
	335	+	}
	336	+
	337	+	cb := glib.UnixFDSourceFunc(func(_ int, _ glib.IOCondition, _ uintptr) bool {
	338	+	e.pumpSignalPending.Store(false)
	339	+	e.drainPumpSignal()
	340	+	if e.pumpClosing.Load() \|\| !e.pumpEnabled.Load() {
	341	+	return glib.SOURCE_REMOVE
	342	+	}
	343	+	e.onPumpWake()
	344	+	return glib.SOURCE_CONTINUE
	345	+	})
	346	+
	347	+	sourceID := glib.UnixFdAddFull(
	348	+	glib.PRIORITY_DEFAULT_IDLE,
	349	+	int(readFile.Fd()),
	350	+	glib.GIoInValue\|glib.GIoHupValue\|glib.GIoErrValue,
	351	+	&cb,
	352	+	0,
	353	+	nil,
	354	+	)
	355	+
	356	+	if sourceID == 0 {
	357	+	_ = readFile.Close()
	358	+	_ = writeFile.Close()
	359	+	logging.FromContext(e.ctx).Error().Msg("cef: failed to attach wakeup pipe source")
	360	+	return
	361	+	}
	362	+
	363	+	e.pumpSignalRead = readFile
	364	+	e.pumpSignalWrite = writeFile
	365	+	e.pumpSignalSource.Store(uint64(sourceID))
327	366		e.pumpClosing.Store(false)
328	367		e.pumpEnabled.Store(true)
329	368		e.nextWorkAtMs.Store(time.Now().UnixMilli())
330		-	e.queuePumpWake()
	369	+	e.signalPump()
331	370		}
332	371
333	372		func (e *Engine) stopMessagePump() {

@@ -335,12 +374,20 @@ func (e *Engine) stopMessagePump() {
335	374		e.pumpEnabled.Store(false)
336	375		e.nextWorkAtMs.Store(0)
337	376		e.pumpTimerDueAtMs.Store(0)
338		-	if sourceID := uint(e.pumpIdleSourceID.Swap(0)); sourceID != 0 {
	377	+	if sourceID := uint(e.pumpTimerSource.Swap(0)); sourceID != 0 {
339	378		glib.SourceRemove(sourceID)
340	379		}
341		-	if sourceID := uint(e.pumpTimerSource.Swap(0)); sourceID != 0 {
	380	+	if sourceID := uint(e.pumpSignalSource.Swap(0)); sourceID != 0 {
342	381		glib.SourceRemove(sourceID)
343	382		}
	383	+	if e.pumpSignalRead != nil {
	384	+	_ = e.pumpSignalRead.Close()
	385	+	e.pumpSignalRead = nil
	386	+	}
	387	+	if e.pumpSignalWrite != nil {
	388	+	_ = e.pumpSignalWrite.Close()
	389	+	e.pumpSignalWrite = nil
	390	+	}
344	391		}
345	392
346	393		func (e *Engine) scheduleFallbackPumpWork(nowMs int64) {

@@ -350,25 +397,33 @@ func (e *Engine) scheduleFallbackPumpWork(nowMs int64) {
350	397		e.nextWorkAtMs.CompareAndSwap(0, nowMs+cefFallbackPumpDelayMS)
351	398		}
352	399
353		-	func (e *Engine) queuePumpWake() {
	400	+	func (e *Engine) signalPump() {
354	401		if e.pumpClosing.Load() \|\| !e.pumpEnabled.Load() {
355	402		return
356	403		}
357		-	if !e.pumpWakeQueued.CompareAndSwap(false, true) {
	404	+	if !e.pumpSignalPending.CompareAndSwap(false, true) {
358	405		return
359	406		}
360	407
361		-	cb := glib.SourceOnceFunc(func(_ uintptr) {
362		-	e.pumpIdleSourceID.Store(0)
363		-	e.pumpWakeQueued.Store(false)
364		-	if e.pumpClosing.Load() \|\| !e.pumpEnabled.Load() {
365		-	return
366		-	}
367		-	e.onPumpWake()
368		-	})
	408	+	if e.pumpSignalWrite == nil {
	409	+	e.pumpSignalPending.Store(false)
	410	+	return
	411	+	}
369	412
370		-	sourceID := glib.IdleAddOnce(&cb, 0)
371		-	e.pumpIdleSourceID.Store(uint64(sourceID))
	413	+	var wakeByte [1]byte
	414	+	wakeByte[0] = 1
	415	+	if _, err := e.pumpSignalWrite.Write(wakeByte[:]); err != nil {
	416	+	e.pumpSignalPending.Store(false)
	417	+	}
	418	+	}
	419	+
	420	+	func (e *Engine) drainPumpSignal() {
	421	+	if e.pumpSignalRead == nil {
	422	+	return
	423	+	}
	424	+
	425	+	var buf [64]byte
	426	+	_, _ = e.pumpSignalRead.Read(buf[:])
372	427		}
373	428
374	429		func (e *Engine) armPumpTimer(dueAtMs, nowMs int64) {

@@ -389,6 +444,9 @@ func (e *Engine) armPumpTimer(dueAtMs, nowMs int64) {
389	444		if delayMs <= 0 {
390	445		delayMs = 1
391	446		}
	447	+	if delayMs > cefFallbackPumpDelayMS {
	448	+	delayMs = cefFallbackPumpDelayMS
	449	+	}
392	450
393	451		cb := glib.SourceOnceFunc(func(_ uintptr) {
394	452		e.pumpTimerSource.Store(0)

@@ -401,7 +459,7 @@ func (e *Engine) armPumpTimer(dueAtMs, nowMs int64) {
401	459
402	460		sourceID := glib.TimeoutAddOnce(uint(delayMs), &cb, 0)
403	461		e.pumpTimerSource.Store(uint64(sourceID))
404		-	e.pumpTimerDueAtMs.Store(dueAtMs)
	462	+	e.pumpTimerDueAtMs.Store(nowMs + delayMs)
405	463		}
406	464
407	465		func (e *Engine) disarmPumpTimer() {

@@ -442,12 +500,16 @@ func (e *Engine) onPumpWake() {
442	500		Int64("now_ms", nowMs).
443	501		Msg("cef: pump doing work")
444	502		}
445		-	e.performMessageLoopWork()
	503	+	wasReentrant := e.performMessageLoopWork()
446	504
447	505		// Match the upstream external-pump reference behavior: if CEF did not
448	506		// request another deadline, keep a placeholder wakeup so async work such as
449	507		// browser creation can continue progressing.
450	508		nowMs = time.Now().UnixMilli()
	509	+	if wasReentrant {
	510	+	e.signalPump()
	511	+	return
	512	+	}
451	513		e.scheduleFallbackPumpWork(nowMs)
452	514
453	515		nextDueAtMs := e.nextWorkAtMs.Load()

@@ -455,17 +517,18 @@ func (e *Engine) onPumpWake() {
455	517		return
456	518		}
457	519		if nextDueAtMs <= nowMs {
458		-	e.queuePumpWake()
	520	+	e.signalPump()
459	521		return
460	522		}
461	523		e.armPumpTimer(nextDueAtMs, nowMs)
462	524		}
463	525
464	526		// performMessageLoopWork calls DoMessageLoopWork with reentrancy protection.
465		-	func (e *Engine) performMessageLoopWork() {
	527	+	// It returns true when re-entrant work was detected and should be reposted.
	528	+	func (e *Engine) performMessageLoopWork() bool {
466	529		if e.pumpActive {
467	530		e.pumpReentry = true
468		-	return
	531	+	return false
469	532		}
470	533
471	534		e.pumpActive = true

@@ -473,12 +536,5 @@ func (e *Engine) performMessageLoopWork() {
473	536		purecef.DoMessageLoopWork()
474	537		e.pumpActive = false
475	538
476		-	// If CEF re-entered during the call, run one more pass immediately. Any
477		-	// follow-up callback will have already published a fresh deadline.
478		-	if e.pumpReentry {
479		-	e.pumpActive = true
480		-	e.pumpReentry = false
481		-	purecef.DoMessageLoopWork()
482		-	e.pumpActive = false
483		-	}
	539	+	return e.pumpReentry
484	540		}

 import (
 	"context"
+	"os"
 	"sync/atomic"
 	"time"
 	pumpEnabled atomic.Bool
 	pumpClosing atomic.Bool
-	pumpWakeQueued   atomic.Bool
+	pumpSignalPending atomic.Bool
-	pumpIdleSourceID atomic.Uint64
+	pumpSignalSource  atomic.Uint64
+	pumpSignalRead    *os.File
+	pumpSignalWrite   *os.File
 	pumpTimerDueAtMs atomic.Int64
 	pumpTimerSource  atomic.Uint64
 // CEF message pump — thread-safe via atomic deadlines
 //
 // OnScheduleMessagePumpWork can be called from ANY Chromium thread (UI, IO,
-// GPU compositor, etc.). It updates the next due time atomically and coalesces
+// GPU compositor, etc.). It updates the next due time atomically and signals a
-// a wake-up on the GTK main loop. When work is not yet due, a one-shot GLib
+// persistent GLib fd watch via a wakeup pipe. This closely follows the Linux
-// timeout is armed for the exact deadline instead of polling every few ms.
+// external-message-pump strategy used by upstream cefclient and avoids
+// creating/destroying GLib sources from Chromium threads.
 // ---------------------------------------------------------------------------
 // scheduleMessagePumpWork is called by OnScheduleMessagePumpWork from any
-// thread. It updates the earliest due deadline atomically and queues a wake-up
+// thread. It updates the earliest due deadline atomically and signals the GTK
-// if the main loop needs to re-evaluate timer state.
+// main thread when it needs to re-evaluate timer state.
 func (e *Engine) scheduleMessagePumpWork(delayMs int64) {
 	nowMs := time.Now().UnixMilli()
 	dueAtMs := nowMs
 	armedDueAtMs := e.pumpTimerDueAtMs.Load()
 	if dueAtMs <= nowMs || armedDueAtMs == 0 || dueAtMs < armedDueAtMs {
-		e.queuePumpWake()
+		e.signalPump()
 	}
 }
 		return
 	}
+	readFile, writeFile, err := os.Pipe()
+	if err != nil {
+		logging.FromContext(e.ctx).Error().Err(err).Msg("cef: failed to create wakeup pipe")
+		return
+	}
+	cb := glib.UnixFDSourceFunc(func(_ int, _ glib.IOCondition, _ uintptr) bool {
+		e.pumpSignalPending.Store(false)
+		e.drainPumpSignal()
+		if e.pumpClosing.Load() || !e.pumpEnabled.Load() {
+			return glib.SOURCE_REMOVE
+		}
+		e.onPumpWake()
+		return glib.SOURCE_CONTINUE
+	})
+	sourceID := glib.UnixFdAddFull(
+		glib.PRIORITY_DEFAULT_IDLE,
+		int(readFile.Fd()),
+		glib.GIoInValue|glib.GIoHupValue|glib.GIoErrValue,
+		&cb,
+		0,
+		nil,
+	)
+	if sourceID == 0 {
+		_ = readFile.Close()
+		_ = writeFile.Close()
+		logging.FromContext(e.ctx).Error().Msg("cef: failed to attach wakeup pipe source")
+		return
+	}
+	e.pumpSignalRead = readFile
+	e.pumpSignalWrite = writeFile
+	e.pumpSignalSource.Store(uint64(sourceID))
 	e.pumpClosing.Store(false)
 	e.pumpEnabled.Store(true)
 	e.nextWorkAtMs.Store(time.Now().UnixMilli())
-	e.queuePumpWake()
+	e.signalPump()
 }
 func (e *Engine) stopMessagePump() {
 	e.pumpEnabled.Store(false)
 	e.nextWorkAtMs.Store(0)
 	e.pumpTimerDueAtMs.Store(0)
-	if sourceID := uint(e.pumpIdleSourceID.Swap(0)); sourceID != 0 {
+	if sourceID := uint(e.pumpTimerSource.Swap(0)); sourceID != 0 {
 		glib.SourceRemove(sourceID)
 	}
-	if sourceID := uint(e.pumpTimerSource.Swap(0)); sourceID != 0 {
+	if sourceID := uint(e.pumpSignalSource.Swap(0)); sourceID != 0 {
 		glib.SourceRemove(sourceID)
 	}
+	if e.pumpSignalRead != nil {
+		_ = e.pumpSignalRead.Close()
+		e.pumpSignalRead = nil
+	}
+	if e.pumpSignalWrite != nil {
+		_ = e.pumpSignalWrite.Close()
+		e.pumpSignalWrite = nil
+	}
 }
 func (e *Engine) scheduleFallbackPumpWork(nowMs int64) {
 	e.nextWorkAtMs.CompareAndSwap(0, nowMs+cefFallbackPumpDelayMS)
 }
-func (e *Engine) queuePumpWake() {
+func (e *Engine) signalPump() {
 	if e.pumpClosing.Load() || !e.pumpEnabled.Load() {
 		return
 	}
-	if !e.pumpWakeQueued.CompareAndSwap(false, true) {
+	if !e.pumpSignalPending.CompareAndSwap(false, true) {
 		return
 	}
-	cb := glib.SourceOnceFunc(func(_ uintptr) {
+	if e.pumpSignalWrite == nil {
-		e.pumpIdleSourceID.Store(0)
+		e.pumpSignalPending.Store(false)
-		e.pumpWakeQueued.Store(false)
+		return
-		if e.pumpClosing.Load() || !e.pumpEnabled.Load() {
+	}
-			return
-		}
-		e.onPumpWake()
-	})
-	sourceID := glib.IdleAddOnce(&cb, 0)
+	var wakeByte [1]byte
-	e.pumpIdleSourceID.Store(uint64(sourceID))
+	wakeByte[0] = 1
+	if _, err := e.pumpSignalWrite.Write(wakeByte[:]); err != nil {
+		e.pumpSignalPending.Store(false)
+	}
+}
+func (e *Engine) drainPumpSignal() {
+	if e.pumpSignalRead == nil {
+		return
+	}
+	var buf [64]byte
+	_, _ = e.pumpSignalRead.Read(buf[:])
 }
 func (e *Engine) armPumpTimer(dueAtMs, nowMs int64) {
 	if delayMs <= 0 {
 		delayMs = 1
 	}
+	if delayMs > cefFallbackPumpDelayMS {
+		delayMs = cefFallbackPumpDelayMS
+	}
 	cb := glib.SourceOnceFunc(func(_ uintptr) {
 		e.pumpTimerSource.Store(0)
 	sourceID := glib.TimeoutAddOnce(uint(delayMs), &cb, 0)
 	e.pumpTimerSource.Store(uint64(sourceID))
-	e.pumpTimerDueAtMs.Store(dueAtMs)
+	e.pumpTimerDueAtMs.Store(nowMs + delayMs)
 }
 func (e *Engine) disarmPumpTimer() {
 			Int64("now_ms", nowMs).
 			Msg("cef: pump doing work")
 	}
-	e.performMessageLoopWork()
+	wasReentrant := e.performMessageLoopWork()
 	// Match the upstream external-pump reference behavior: if CEF did not
 	// request another deadline, keep a placeholder wakeup so async work such as
 	// browser creation can continue progressing.
 	nowMs = time.Now().UnixMilli()
+	if wasReentrant {
+		e.signalPump()
+		return
+	}
 	e.scheduleFallbackPumpWork(nowMs)
 	nextDueAtMs := e.nextWorkAtMs.Load()
 		return
 	}
 	if nextDueAtMs <= nowMs {
-		e.queuePumpWake()
+		e.signalPump()
 		return
 	}
 	e.armPumpTimer(nextDueAtMs, nowMs)
 }
 // performMessageLoopWork calls DoMessageLoopWork with reentrancy protection.
-func (e *Engine) performMessageLoopWork() {
+// It returns true when re-entrant work was detected and should be reposted.
+func (e *Engine) performMessageLoopWork() bool {
 	if e.pumpActive {
 		e.pumpReentry = true
-		return
+		return false
 	}
 	e.pumpActive = true
 	purecef.DoMessageLoopWork()
 	e.pumpActive = false
-	// If CEF re-entered during the call, run one more pass immediately. Any
+	return e.pumpReentry
-	// follow-up callback will have already published a fresh deadline.
-	if e.pumpReentry {
-		e.pumpActive = true
-		e.pumpReentry = false
-		purecef.DoMessageLoopWork()
-		e.pumpActive = false
-	}
 }