add configurable cef message pump mode

commit: 95bfd157c53129cdce1ecde17c3f14be12e82b6c
author: brice · 7mo ago · 2026-03-20 13:27 UTC
committer: brice · 6mo ago · 2026-03-28 04:36 UTC
parent: 93202e9
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +115 -2

M internal/infrastructure/cef/engine.go +79 -1


@@ -29,6 +29,9 @@ type Engine struct {
29	29		factory *WebViewFactory
30	30		pool *WebViewPool
31	31
	32	+	externalMessagePump bool
	33	+	manualPumpInterval int64
	34	+
32	35		// nextWorkAtMs is the next deadline, in Unix milliseconds, when
33	36		// CefDoMessageLoopWork should run. It is written from any Chromium thread
34	37		// via OnScheduleMessagePumpWork and read on the GTK main thread.

@@ -148,7 +151,10 @@ func (e *Engine) ConfigureDownloads(_ context.Context, _ string, _ port.Download
148	151		// Starts the persistent pump timer on the GTK main thread.
149	152		func (e *Engine) OnToolkitReady(_ context.Context) error {
150	153		log := logging.FromContext(e.ctx)
151		-	log.Debug().Msg("cef: OnToolkitReady called, starting message pump")
	154	+	log.Debug().
	155	+	Bool("external_message_pump", e.externalMessagePump).
	156	+	Int64("manual_pump_interval_ms", e.manualPumpInterval).
	157	+	Msg("cef: OnToolkitReady called, starting message pump")
152	158		e.startMessagePump()
153	159		return nil
154	160		}

@@ -289,6 +295,10 @@ func (e *Engine) maybeLogBrowserCreateStall() {
289	295		// thread. It updates the earliest due deadline atomically and signals the GTK
290	296		// main thread when it needs to re-evaluate timer state.
291	297		func (e *Engine) scheduleMessagePumpWork(delayMs int64) {
	298	+	if !e.externalMessagePump {
	299	+	return
	300	+	}
	301	+
292	302		nowMs := time.Now().UnixMilli()
293	303		dueAtMs := nowMs
294	304		if delayMs > 0 {

@@ -334,6 +344,14 @@ func (e *Engine) scheduleMessagePumpWork(delayMs int64) {
334	344		}
335	345
336	346		func (e *Engine) startMessagePump() {
	347	+	if e.externalMessagePump {
	348	+	e.startExternalMessagePump()
	349	+	return
	350	+	}
	351	+	e.startManualMessagePump()
	352	+	}
	353	+
	354	+	func (e *Engine) startExternalMessagePump() {
337	355		if e.pumpEnabled.Load() {
338	356		return
339	357		}

@@ -379,6 +397,66 @@ func (e *Engine) startMessagePump() {
379	397		e.signalPump()
380	398		}
381	399
	400	+	func (e *Engine) startManualMessagePump() {
	401	+	if e.pumpEnabled.Load() {
	402	+	return
	403	+	}
	404	+
	405	+	intervalMs := e.manualPumpInterval
	406	+	if intervalMs <= 0 {
	407	+	intervalMs = 10
	408	+	}
	409	+
	410	+	cb := glib.SourceFunc(func(_ uintptr) bool {
	411	+	if e.pumpClosing.Load() \|\| !e.pumpEnabled.Load() {
	412	+	return glib.SOURCE_REMOVE
	413	+	}
	414	+
	415	+	nowMs := time.Now().UnixMilli()
	416	+	e.pumpWakeCount++
	417	+	e.pumpWorkCount++
	418	+	e.maybeLogBrowserCreateStall()
	419	+	if e.pumpWorkCount <= 20 \|\| e.pumpWorkCount%100 == 0 {
	420	+	logging.FromContext(e.ctx).Debug().
	421	+	Uint64("wake", e.pumpWakeCount).
	422	+	Uint64("work", e.pumpWorkCount).
	423	+	Int64("interval_ms", intervalMs).
	424	+	Int64("now_ms", nowMs).
	425	+	Msg("cef: manual pump doing work")
	426	+	}
	427	+
	428	+	workStartedAt := time.Now()
	429	+	purecef.DoMessageLoopWork()
	430	+	workDuration := time.Since(workStartedAt)
	431	+	e.pumpWorkTotalNs += uint64(workDuration)
	432	+	if workDuration.Nanoseconds() > e.pumpWorkMaxNs {
	433	+	e.pumpWorkMaxNs = workDuration.Nanoseconds()
	434	+	}
	435	+	if workDuration >= 2*time.Millisecond {
	436	+	e.pumpSlowCount++
	437	+	}
	438	+	e.maybeLogPumpDiagnostics()
	439	+
	440	+	return glib.SOURCE_CONTINUE
	441	+	})
	442	+
	443	+	sourceID := glib.TimeoutAdd(uint(intervalMs), &cb, 0)
	444	+	if sourceID == 0 {
	445	+	logging.FromContext(e.ctx).Error().
	446	+	Int64("interval_ms", intervalMs).
	447	+	Msg("cef: failed to start manual pump timer")
	448	+	return
	449	+	}
	450	+
	451	+	e.pumpClosing.Store(false)
	452	+	e.pumpEnabled.Store(true)
	453	+	e.pumpTimerSource.Store(uint64(sourceID))
	454	+	e.pumpTimerDueAtMs.Store(time.Now().UnixMilli() + intervalMs)
	455	+	logging.FromContext(e.ctx).Info().
	456	+	Int64("interval_ms", intervalMs).
	457	+	Msg("cef: started manual CEF pump")
	458	+	}
	459	+
382	460		func (e *Engine) stopMessagePump() {
383	461		e.pumpClosing.Store(true)
384	462		e.pumpEnabled.Store(false)


@@ -29,6 +29,9 @@ type Engine struct {
29		factory *WebViewFactory	29		factory *WebViewFactory
30		pool *WebViewPool	30		pool *WebViewPool
31			31
			32	+	externalMessagePump bool
			33	+	manualPumpInterval int64
			34	+
32		// nextWorkAtMs is the next deadline, in Unix milliseconds, when	35		// nextWorkAtMs is the next deadline, in Unix milliseconds, when
33		// CefDoMessageLoopWork should run. It is written from any Chromium thread	36		// CefDoMessageLoopWork should run. It is written from any Chromium thread
34		// via OnScheduleMessagePumpWork and read on the GTK main thread.	37		// via OnScheduleMessagePumpWork and read on the GTK main thread.

@@ -148,7 +151,10 @@ func (e *Engine) ConfigureDownloads(_ context.Context, _ string, _ port.Download
148		// Starts the persistent pump timer on the GTK main thread.	151		// Starts the persistent pump timer on the GTK main thread.
149		func (e *Engine) OnToolkitReady(_ context.Context) error {	152		func (e *Engine) OnToolkitReady(_ context.Context) error {
150		log := logging.FromContext(e.ctx)	153		log := logging.FromContext(e.ctx)
151	-	log.Debug().Msg("cef: OnToolkitReady called, starting message pump")	154	+	log.Debug().
			155	+	Bool("external_message_pump", e.externalMessagePump).
			156	+	Int64("manual_pump_interval_ms", e.manualPumpInterval).
			157	+	Msg("cef: OnToolkitReady called, starting message pump")
152		e.startMessagePump()	158		e.startMessagePump()
153		return nil	159		return nil
154		}	160		}

@@ -289,6 +295,10 @@ func (e *Engine) maybeLogBrowserCreateStall() {
289		// thread. It updates the earliest due deadline atomically and signals the GTK	295		// thread. It updates the earliest due deadline atomically and signals the GTK
290		// main thread when it needs to re-evaluate timer state.	296		// main thread when it needs to re-evaluate timer state.
291		func (e *Engine) scheduleMessagePumpWork(delayMs int64) {	297		func (e *Engine) scheduleMessagePumpWork(delayMs int64) {
			298	+	if !e.externalMessagePump {
			299	+	return
			300	+	}
			301	+
292		nowMs := time.Now().UnixMilli()	302		nowMs := time.Now().UnixMilli()
293		dueAtMs := nowMs	303		dueAtMs := nowMs
294		if delayMs > 0 {	304		if delayMs > 0 {

@@ -334,6 +344,14 @@ func (e *Engine) scheduleMessagePumpWork(delayMs int64) {
334		}	344		}
335			345
336		func (e *Engine) startMessagePump() {	346		func (e *Engine) startMessagePump() {
			347	+	if e.externalMessagePump {
			348	+	e.startExternalMessagePump()
			349	+	return
			350	+	}
			351	+	e.startManualMessagePump()
			352	+	}
			353	+
			354	+	func (e *Engine) startExternalMessagePump() {
337		if e.pumpEnabled.Load() {	355		if e.pumpEnabled.Load() {
338		return	356		return
339		}	357		}

@@ -379,6 +397,66 @@ func (e *Engine) startMessagePump() {
379		e.signalPump()	397		e.signalPump()
380		}	398		}
381			399
			400	+	func (e *Engine) startManualMessagePump() {
			401	+	if e.pumpEnabled.Load() {
			402	+	return
			403	+	}
			404	+
			405	+	intervalMs := e.manualPumpInterval
			406	+	if intervalMs <= 0 {
			407	+	intervalMs = 10
			408	+	}
			409	+
			410	+	cb := glib.SourceFunc(func(_ uintptr) bool {
			411	+	if e.pumpClosing.Load() \|\| !e.pumpEnabled.Load() {
			412	+	return glib.SOURCE_REMOVE
			413	+	}
			414	+
			415	+	nowMs := time.Now().UnixMilli()
			416	+	e.pumpWakeCount++
			417	+	e.pumpWorkCount++
			418	+	e.maybeLogBrowserCreateStall()
			419	+	if e.pumpWorkCount <= 20 \|\| e.pumpWorkCount%100 == 0 {
			420	+	logging.FromContext(e.ctx).Debug().
			421	+	Uint64("wake", e.pumpWakeCount).
			422	+	Uint64("work", e.pumpWorkCount).
			423	+	Int64("interval_ms", intervalMs).
			424	+	Int64("now_ms", nowMs).
			425	+	Msg("cef: manual pump doing work")
			426	+	}
			427	+
			428	+	workStartedAt := time.Now()
			429	+	purecef.DoMessageLoopWork()
			430	+	workDuration := time.Since(workStartedAt)
			431	+	e.pumpWorkTotalNs += uint64(workDuration)
			432	+	if workDuration.Nanoseconds() > e.pumpWorkMaxNs {
			433	+	e.pumpWorkMaxNs = workDuration.Nanoseconds()
			434	+	}
			435	+	if workDuration >= 2*time.Millisecond {
			436	+	e.pumpSlowCount++
			437	+	}
			438	+	e.maybeLogPumpDiagnostics()
			439	+
			440	+	return glib.SOURCE_CONTINUE
			441	+	})
			442	+
			443	+	sourceID := glib.TimeoutAdd(uint(intervalMs), &cb, 0)
			444	+	if sourceID == 0 {
			445	+	logging.FromContext(e.ctx).Error().
			446	+	Int64("interval_ms", intervalMs).
			447	+	Msg("cef: failed to start manual pump timer")
			448	+	return
			449	+	}
			450	+
			451	+	e.pumpClosing.Store(false)
			452	+	e.pumpEnabled.Store(true)
			453	+	e.pumpTimerSource.Store(uint64(sourceID))
			454	+	e.pumpTimerDueAtMs.Store(time.Now().UnixMilli() + intervalMs)
			455	+	logging.FromContext(e.ctx).Info().
			456	+	Int64("interval_ms", intervalMs).
			457	+	Msg("cef: started manual CEF pump")
			458	+	}
			459	+
382		func (e *Engine) stopMessagePump() {	460		func (e *Engine) stopMessagePump() {
383		e.pumpClosing.Store(true)	461		e.pumpClosing.Store(true)
384		e.pumpEnabled.Store(false)	462		e.pumpEnabled.Store(false)

M internal/infrastructure/cef/engine_init.go +36 -1


@@ -5,6 +5,7 @@ import (
5	5		"fmt"
6	6		"os"
7	7		"path/filepath"
	8	+	"strconv"
8	9
9	10		purecef "github.com/bnema/purego-cef/cef"
10	11

@@ -17,6 +18,7 @@ func NewEngine(ctx context.Context, cfg config.CEFEngineConfig) (*Engine, error)
17	18		logger := logging.FromContext(ctx)
18	19		// 1. Initialize CEF.
19	20		settings := purecef.DefaultSettings()
	21	+	settings.ExternalMessagePump = parseBoolEnv("DUMBER_CEF_EXTERNAL_PUMP", settings.ExternalMessagePump)
20	22		if cfg.CEFDir != "" {
21	23		settings.CEFDir = cfg.CEFDir
22	24		}

@@ -50,9 +52,16 @@ func NewEngine(ctx context.Context, cfg config.CEFEngineConfig) (*Engine, error)
50	52
51	53		// 2. Build the engine early so the App can reference it for the pump callback.
52	54		eng := &Engine{
53		-	ctx: ctx,
	55	+	ctx: ctx,
	56	+	externalMessagePump: settings.ExternalMessagePump,
	57	+	manualPumpInterval: parseInt64Env("DUMBER_CEF_MANUAL_PUMP_MS", 10),
54	58		}
55	59
	60	+	logger.Info().
	61	+	Bool("external_message_pump", settings.ExternalMessagePump).
	62	+	Int64("manual_pump_interval_ms", eng.manualPumpInterval).
	63	+	Msg("cef: configured message pump mode")
	64	+
56	65		// Create the CEF App with a BrowserProcessHandler that drives the
57	66		// adaptive message pump via OnScheduleMessagePumpWork.
58	67		app := newDumberApp(eng)

@@ -108,3 +117,29 @@ func findHelperBinary() string {
108	117		}
109	118		return ""
110	119		}
	120	+
	121	+	func parseBoolEnv(name string, fallback bool) bool {
	122	+	value, ok := os.LookupEnv(name)
	123	+	if !ok \|\| value == "" {
	124	+	return fallback
	125	+	}
	126	+
	127	+	parsed, err := strconv.ParseBool(value)
	128	+	if err != nil {
	129	+	return fallback
	130	+	}
	131	+	return parsed
	132	+	}
	133	+
	134	+	func parseInt64Env(name string, fallback int64) int64 {
	135	+	value, ok := os.LookupEnv(name)
	136	+	if !ok \|\| value == "" {
	137	+	return fallback
	138	+	}
	139	+
	140	+	parsed, err := strconv.ParseInt(value, 10, 64)
	141	+	if err != nil \|\| parsed <= 0 {
	142	+	return fallback
	143	+	}
	144	+	return parsed
	145	+	}


@@ -5,6 +5,7 @@ import (
5		"fmt"	5		"fmt"
6		"os"	6		"os"
7		"path/filepath"	7		"path/filepath"
			8	+	"strconv"
8			9
9		purecef "github.com/bnema/purego-cef/cef"	10		purecef "github.com/bnema/purego-cef/cef"
10			11

@@ -17,6 +18,7 @@ func NewEngine(ctx context.Context, cfg config.CEFEngineConfig) (*Engine, error)
17		logger := logging.FromContext(ctx)	18		logger := logging.FromContext(ctx)
18		// 1. Initialize CEF.	19		// 1. Initialize CEF.
19		settings := purecef.DefaultSettings()	20		settings := purecef.DefaultSettings()
			21	+	settings.ExternalMessagePump = parseBoolEnv("DUMBER_CEF_EXTERNAL_PUMP", settings.ExternalMessagePump)
20		if cfg.CEFDir != "" {	22		if cfg.CEFDir != "" {
21		settings.CEFDir = cfg.CEFDir	23		settings.CEFDir = cfg.CEFDir
22		}	24		}

@@ -50,9 +52,16 @@ func NewEngine(ctx context.Context, cfg config.CEFEngineConfig) (*Engine, error)
50			52
51		// 2. Build the engine early so the App can reference it for the pump callback.	53		// 2. Build the engine early so the App can reference it for the pump callback.
52		eng := &Engine{	54		eng := &Engine{
53	-	ctx: ctx,	55	+	ctx: ctx,
			56	+	externalMessagePump: settings.ExternalMessagePump,
			57	+	manualPumpInterval: parseInt64Env("DUMBER_CEF_MANUAL_PUMP_MS", 10),
54		}	58		}
55			59
			60	+	logger.Info().
			61	+	Bool("external_message_pump", settings.ExternalMessagePump).
			62	+	Int64("manual_pump_interval_ms", eng.manualPumpInterval).
			63	+	Msg("cef: configured message pump mode")
			64	+
56		// Create the CEF App with a BrowserProcessHandler that drives the	65		// Create the CEF App with a BrowserProcessHandler that drives the
57		// adaptive message pump via OnScheduleMessagePumpWork.	66		// adaptive message pump via OnScheduleMessagePumpWork.
58		app := newDumberApp(eng)	67		app := newDumberApp(eng)

@@ -108,3 +117,29 @@ func findHelperBinary() string {
108		}	117		}
109		return ""	118		return ""
110		}	119		}
			120	+
			121	+	func parseBoolEnv(name string, fallback bool) bool {
			122	+	value, ok := os.LookupEnv(name)
			123	+	if !ok \|\| value == "" {
			124	+	return fallback
			125	+	}
			126	+
			127	+	parsed, err := strconv.ParseBool(value)
			128	+	if err != nil {
			129	+	return fallback
			130	+	}
			131	+	return parsed
			132	+	}
			133	+
			134	+	func parseInt64Env(name string, fallback int64) int64 {
			135	+	value, ok := os.LookupEnv(name)
			136	+	if !ok \|\| value == "" {
			137	+	return fallback
			138	+	}
			139	+
			140	+	parsed, err := strconv.ParseInt(value, 10, 64)
			141	+	if err != nil \|\| parsed <= 0 {
			142	+	return fallback
			143	+	}
			144	+	return parsed
			145	+	}