Add mechanism to wake native loop

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;
 		}
 

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:Null<Void->Void>):Void {}
 }

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<Float> = 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
 }