From 2f535040c9062949fd5ed4574cf31aee2cbcee7c Mon Sep 17 00:00:00 2001 From: Kevin Leung Date: Fri, 17 Jul 2026 11:31:13 +1000 Subject: [PATCH] Add mechanism to wake native loop --- std/haxe/EventLoop.hx | 33 ++++++--- std/hl/uv/Loop.hx | 101 +++++++++++++++++++++++++- tests/threads/src/cases/TestEvents.hx | 41 +++++++++++ 3 files changed, 163 insertions(+), 12 deletions(-) diff --git a/std/haxe/EventLoop.hx b/std/haxe/EventLoop.hx index f7d9ea660a9..e6cb8b8a2b1 100644 --- a/std/haxe/EventLoop.hx +++ b/std/haxe/EventLoop.hx @@ -67,7 +67,15 @@ class Event { private typedef NativeEventLoop = { final allowsReentrancy:Bool; - function run():Void; + /** + Run one iteration of the native loop. + `maxBlock` is seconds until the next Haxe event is due: + - `< 0`: do not block (Haxe work is already due) + - `0`: block until a native event or wake (no Haxe deadline) + - `> 0`: block at most this long for the next Haxe deadline + **/ + function run(maxBlock:Float):Void; + function wake():Void; function close():Void; function isAlive():Bool; }; @@ -126,7 +134,10 @@ class EventLoop { It is already automatically called for threads loops. **/ public function dispose() { - if( nativeLoop != null ) nativeLoop.close(); + if( nativeLoop != null ) { + nativeLoop.close(); + nativeLoop = null; + } } /** @@ -138,14 +149,15 @@ class EventLoop { while (true) { if (hasEvents(true)) { var time = getNextTick(); - // disable wait if we have our native loop alive - if( nativeLoop != null && time > 0 && nativeLoop.isAlive() ) - time = -1; - if( time > 0 ) { + if( nativeLoop != null && nativeLoop.isAlive() ) { + // Sleep inside the native poller (with Haxe deadline / wake), not on lockTime + loopOnce(false, time); + } else if( time > 0 ) { wait(time); continue; + } else { + loopOnce(false); } - loopOnce(false); } else if (promiseCount > 0 || hasRunningThreadTasks()) { #if target.threaded // wait till we get notified @@ -186,6 +198,8 @@ class EventLoop { inline function wakeup() { #if target.threaded lockTime.release(); + if( nativeLoop != null && thread != null && thread != sys.thread.Thread.current() ) + nativeLoop.wake(); #end } @@ -219,8 +233,9 @@ class EventLoop { Perform an update of pending events. By default, an event loop from a thread can only be triggered from this thread. You can set `threadCheck` to false in the rare cases you might want otherwise. + `maxBlock` is forwarded to the native loop when present (see `NativeEventLoop.run`). **/ - public function loopOnce( threadCheck = true ) { + public function loopOnce( threadCheck = true, maxBlock = 0. ) { if( threadCheck ) checkThread(); if( inNative && !nativeLoop.allowsReentrancy ) throw "You cannot call EventLoop.loop() while in an event callback with a non-reentrant native loop"; @@ -233,7 +248,7 @@ class EventLoop { if( nativeLoop != null ) { inNative = true; - nativeLoop.run(); + nativeLoop.run(maxBlock); inNative = false; } diff --git a/std/hl/uv/Loop.hx b/std/hl/uv/Loop.hx index 8a83830f590..4f2239473e4 100644 --- a/std/hl/uv/Loop.hx +++ b/std/hl/uv/Loop.hx @@ -84,25 +84,120 @@ abstract Loop(hl.Abstract<"uv_loop">) { } +/** + NativeEventLoop adapter: blocking `UV_RUN_ONCE` with an async wake doorbell + and a one-shot UV timer for the next Haxe EventLoop deadline. +**/ private class LoopWrapper { public final allowsReentrancy = false; final uvLoop:Loop; + final keepAliveCb:Void->Void; + var asyncHandle:HandleData; + var timerHandle:HandleData; + var closed = false; public function new(loop:Loop) { this.uvLoop = loop; + // Closures stored in uv handle data are in hl_gc_alloc_raw; keep a Haxe reference. + keepAliveCb = function() {}; + asyncHandle = async_init(loop, keepAliveCb); + if (asyncHandle == null) + throw "Failed to create uv_async_t wake handle"; + handle_unref(asyncHandle); + timerHandle = timer_init(loop); + if (timerHandle == null) + throw "Failed to create uv_timer_t deadline handle"; + handle_unref(timerHandle); } - public function run() { - uvLoop.run(NoWait); + public function run(maxBlock:Float) { + if (closed) + return; + if (maxBlock < 0) { + // Haxe events already due: do not sleep in the poller + stopDeadlineTimer(); + uvLoop.run(NoWait); + return; + } + if (maxBlock > 0) + armDeadlineTimer(maxBlock); + else + stopDeadlineTimer(); + uvLoop.run(Once); + stopDeadlineTimer(); + } + + public function wake() { + if (asyncHandle != null) + async_send(asyncHandle); } public function close() { + if (closed) + return; + closed = true; + stopDeadlineTimer(); + if (asyncHandle != null) { + close_handle(asyncHandle, null); + asyncHandle = null; + } + if (timerHandle != null) { + close_handle(timerHandle, null); + timerHandle = null; + } + // Drain close callbacks so loop_close can succeed + uvLoop.run(NoWait); final result = uvLoop.close(); if (result != 0) Sys.println("Some async handlers have not been closed"); } public function isAlive() { - return uvLoop.alive() > 0; + return !closed && uvLoop.alive() > 0; + } + + function armDeadlineTimer(maxBlock:Float) { + if (timerHandle == null) + return; + var ms = Math.ceil(maxBlock * 1000); + if (ms < 1) + ms = 1; + if (ms > 2147483647) + ms = 2147483647; + timer_start(timerHandle, keepAliveCb, ms, 0); + } + + function stopDeadlineTimer() { + if (timerHandle != null) + timer_stop(timerHandle); + } + + @:hlNative("uv", "async_init_wrap") + static function async_init(loop:Loop, callb:Void->Void):HandleData { + return null; + } + + @:hlNative("uv", "async_send_wrap") + static function async_send(h:HandleData):Void {} + + @:hlNative("uv", "timer_init_wrap") + static function timer_init(loop:Loop):HandleData { + return null; } + + @:hlNative("uv", "timer_start_wrap") + static function timer_start(h:HandleData, callb:Void->Void, timeout:Int, repeat:Int):Bool { + return false; + } + + @:hlNative("uv", "timer_stop_wrap") + static function timer_stop(h:HandleData):Bool { + return false; + } + + @:hlNative("uv", "handle_unref_wrap") + static function handle_unref(h:HandleData):Void {} + + @:hlNative("uv", "close_handle") + static function close_handle(h:HandleData, callb:NullVoid>):Void {} } diff --git a/tests/threads/src/cases/TestEvents.hx b/tests/threads/src/cases/TestEvents.hx index babd89b8026..8e2e12b605d 100644 --- a/tests/threads/src/cases/TestEvents.hx +++ b/tests/threads/src/cases/TestEvents.hx @@ -103,4 +103,45 @@ class TestEvents extends ThreadTestBase { Assert.equals("ok", threadValue); } + + #if hl + /** + Cross-thread EventLoop.run must wake a thread blocked in UV_RUN_ONCE + via nativeLoop.wake() (uv_async_send), not by spinning on NoWait. + **/ + @:timeout(3000) + function testNativeWake(async:Async) { + final mainLoop = EventLoop.main; + final tcp = new hl.uv.Tcp(hl.uv.Loop.getFromEventLoop(mainLoop)); + tcp.bind(new sys.net.Host("127.0.0.1"), 0); + tcp.listen(1, () -> {}); + + final mainThread = Thread.current(); + var wokeAt:Null = null; + final t0 = haxe.Timer.stamp(); + + Thread.create(() -> { + Sys.sleep(0.1); + EventLoop.getThreadLoop(mainThread).run(() -> { + wokeAt = haxe.Timer.stamp(); + tcp.close(); + async.done(); + }); + }); + + while (wokeAt == null) { + final wait = @:privateAccess mainLoop.getNextTick(); + mainLoop.loopOnce(true, wait); + if (haxe.Timer.stamp() - t0 > 2.0) { + tcp.close(); + Assert.fail("native wake did not deliver cross-thread event within 2s"); + async.done(); + return; + } + } + + final latency = wokeAt - t0; + Assert.isTrue(latency >= 0.05 && latency < 0.5, 'unexpected wake latency: ${latency}s'); + } + #end } \ No newline at end of file