From 0afe3c672da76a21b6102079bec65ce98cacdf2d Mon Sep 17 00:00:00 2001 From: Kevin Leung Date: Tue, 21 Jul 2026 13:15:53 +1000 Subject: [PATCH] Add EventLoopDriver --- std/haxe/EventLoop.hx | 215 +++++++++++++---- std/haxe/EventLoopDriver.hx | 73 ++++++ std/haxe/HaxeEventLoopDriver.hx | 83 +++++++ std/haxe/Timer.hx | 7 +- std/hl/uv/Loop.hx | 67 +++--- std/hl/uv/UvEventLoopDriver.hx | 172 ++++++++++++++ tests/threads/src/cases/TestEvents.hx | 322 +++++++++++++++++++++++++- 7 files changed, 852 insertions(+), 87 deletions(-) create mode 100644 std/haxe/EventLoopDriver.hx create mode 100644 std/haxe/HaxeEventLoopDriver.hx create mode 100644 std/hl/uv/UvEventLoopDriver.hx diff --git a/std/haxe/EventLoop.hx b/std/haxe/EventLoop.hx index f7d9ea660a9..28bd6630948 100644 --- a/std/haxe/EventLoop.hx +++ b/std/haxe/EventLoop.hx @@ -65,14 +65,6 @@ class Event { } -private typedef NativeEventLoop = { - final allowsReentrancy:Bool; - function run():Void; - function close():Void; - function isAlive():Bool; -}; - - /** Handles async events for all threads **/ @@ -90,6 +82,16 @@ class EventLoop { **/ public static var current(get,never) : EventLoop; + /** + Factory used when `new EventLoop()` is called without an explicit driver. + Must be non-null before any `EventLoop` is constructed (including `main`). + **/ + public static var DEFAULT_DRIVER_FACTORY:(loop:EventLoop)->EventLoopDriver + #if !target.threaded + = function(_) return new HaxeEventLoopDriver() + #end + ; + #if target.threaded static var eventsTls:sys.thread.Tls; static var threadsToEventLoops:IntMap; @@ -99,24 +101,27 @@ class EventLoop { var events : Event; var queue : Event; var inLoop : Bool; + var inWait : Bool; var hasPendingRemove : Bool; var promiseCount : Int = 0; + var driver : EventLoopDriver; + var pendingDriver : Null; + var pendingOnSwapped : Null<()->Void>; #if target.threaded var mutex : sys.thread.Mutex; - var lockTime : sys.thread.Lock; final numPendingThreadTasks : AtomicInt; /** The reference thread for this loop. If set, the loop can only be run within this thread. **/ public var thread : sys.thread.Thread; #end - var nativeLoop : Null; + /** True while a non-reentrant driver is inside `wait` (e.g. UV callbacks). **/ var inNative : Bool = false; - public function new() { + public function new(?driver:EventLoopDriver) { + this.driver = driver ?? DEFAULT_DRIVER_FACTORY(this); #if target.threaded mutex = new sys.thread.Mutex(); - lockTime = new sys.thread.Lock(); numPendingThreadTasks = new AtomicInt(0); #end } @@ -126,7 +131,101 @@ class EventLoop { It is already automatically called for threads loops. **/ public function dispose() { - if( nativeLoop != null ) nativeLoop.close(); + lock(); + if( pendingDriver != null ) { + pendingDriver.close(); + pendingDriver = null; + pendingOnSwapped = null; + } + final d = driver; + unlock(); + d.close(); + // Leave a closed non-null driver so later wakeup() cannot NPE. + } + + /** + Install a new waiting driver. + + Applied synchronously when called on the loop thread (or when `thread` is + unset) and the loop is not inside `wait` / `loopOnce`. Otherwise the + driver is pending until `applyPendingSwap` (start of each `loop()` + iteration and end of `loopOnce`), and the current driver is woken so a + blocked `wait` can progress. + + Last-wins: a new pending swap immediately `close()`s any superseded + pending driver and drops its callback. + + `onSwapped` always runs on the loop thread after the new driver is + published (never on a foreign caller thread for deferred swaps). + **/ + public function swapDriver(driver:EventLoopDriver, ?onSwapped:()->Void):Void { + lock(); + if( pendingDriver != null ) { + pendingDriver.close(); + pendingDriver = null; + pendingOnSwapped = null; + } + #if target.threaded + final onLoopThread = thread == null || sys.thread.Thread.current() == thread; + #else + final onLoopThread = true; + #end + if( onLoopThread && !inWait && !inLoop ) { + final old = this.driver; + this.driver = driver; + old.close(); + unlock(); + if( onSwapped != null ) + onSwapped(); + return; + } + pendingDriver = driver; + pendingOnSwapped = onSwapped; + final current = this.driver; + unlock(); + current.wake(); + } + + /** + Currently installed driver (never null). + **/ + public function getDriver():EventLoopDriver { + lock(); + final d = driver; + unlock(); + return d; + } + + /** + Driver waiting to be applied by `applyPendingSwap`, or `null`. + **/ + public function getPendingDriver():Null { + lock(); + final d = pendingDriver; + unlock(); + return d; + } + + /** + Publish a pending `swapDriver` under the mutex: assign the new `driver`, + then `close()` the old one. Invokes `onSwapped` on the loop thread. + **/ + function applyPendingSwap() { + lock(); + if( pendingDriver == null ) { + unlock(); + return; + } + final next = pendingDriver; + final cb = pendingOnSwapped; + pendingDriver = null; + pendingOnSwapped = null; + final old = driver; + driver = next; + old.close(); + unlock(); + if( cb != null ) + cb(); } /** @@ -136,29 +235,33 @@ class EventLoop { public function loop() { checkThread(); while (true) { + applyPendingSwap(); 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 ) { - wait(time); - continue; - } + // Derive maxBlock; never pass getNextTick()'s 1e6 sentinel to the driver. + wait(deriveMaxBlock(getNextTick())); loopOnce(false); } else if (promiseCount > 0 || hasRunningThreadTasks()) { - #if target.threaded - // wait till we get notified - lockTime.wait(); - #else - // ? - #end + // wait till we get notified (maxBlock 0 = block until wake) + wait(0); } else { break; } } } + /** + Map `getNextTick()` to a driver `maxBlock` value. + `getNextTick()` returns `-1` (work due), a positive delta, or the `1e6` + idle sentinel when there are no Haxe events — never exact `0`. + **/ + inline function deriveMaxBlock(nextTick:Float):Float { + if( nextTick < 0 ) + return -1; + if( nextTick == 1e6 ) + return 0; // keep-alive only (external work / pending swap) + return nextTick; + } + /** Promise a possible future event addition. This will prevent the `loop()` from terminating until the matching number of `deliver()` calls have been made. **/ @@ -183,18 +286,28 @@ class EventLoop { wakeup(); } - inline function wakeup() { - #if target.threaded - lockTime.release(); - #end + function wakeup() { + lock(); + final d = driver; + unlock(); + d.wake(); } - inline function wait( time : Float ) { - #if target.threaded - lockTime.wait(time); - #elseif sys - Sys.sleep(time); - #end + function wait( time : Float ) { + lock(); + inWait = true; + final d = driver; + unlock(); + // Guard nested loopOnce while a non-reentrant driver runs callbacks in wait. + final guard = !d.allowsReentrancy; + if( guard ) + inNative = true; + d.wait(time); + if( guard ) + inNative = false; + lock(); + inWait = false; + unlock(); } inline function lock() { @@ -223,7 +336,9 @@ class EventLoop { public function loopOnce( threadCheck = true ) { if( threadCheck ) checkThread(); - if( inNative && !nativeLoop.allowsReentrancy ) throw "You cannot call EventLoop.loop() while in an event callback with a non-reentrant native loop"; + // Waiting (including UV run) lives only in loop() → wait(); loopOnce never blocks. + if( inNative && !driver.allowsReentrancy ) + throw "You cannot call EventLoop.loop() while in an event callback with a non-reentrant driver"; lock(); sortEvents(); @@ -231,12 +346,6 @@ class EventLoop { inLoop = true; unlock(); - if( nativeLoop != null ) { - inNative = true; - nativeLoop.run(); - inNative = false; - } - // if inLoop turns false, stop because we had reentrency var time = haxe.Timer.stamp(); while( inLoop && current != null ) { @@ -258,6 +367,7 @@ class EventLoop { } } unlock(); + applyPendingSwap(); } /** @@ -318,7 +428,7 @@ class EventLoop { events = e; e.prev = queue; queue = e; - wakeup(); + driver.wake(); // already under mutex unlock(); } @@ -347,7 +457,7 @@ class EventLoop { if( queue == e ) queue = e.prev; e.prev = null; - wakeup(); + driver.wake(); // already under mutex unlock(); } @@ -458,9 +568,16 @@ class EventLoop { /** Tells if the event loop has remaining events. If blocking is set to true, only check if it has remaining blocking events. + Also true when a pending `swapDriver` is waiting to apply or the driver + reports external work (e.g. libuv handles), so `loop()` cannot idle-exit + between UV attach and deferred apply. **/ public function hasEvents( blocking : Bool = true ) { - if( nativeLoop != null && nativeLoop.isAlive() ) + lock(); + final hasPending = pendingDriver != null; + final d = driver; + unlock(); + if( hasPending || d.hasExternalWork() ) return true; if( !blocking ) return events != null; @@ -537,6 +654,10 @@ class EventLoop { #if target.threaded static function __init__() { + // Assign before touching `main`: on JVM/Python static field initializers + // may run after `__init__`, so the field initializer alone is not enough. + DEFAULT_DRIVER_FACTORY = function(_) return new HaxeEventLoopDriver(); + eventsTls = new sys.thread.Tls(); threadsToEventLoops = new IntMap(); threadsToEventLoopsMutex = new sys.thread.Mutex(); diff --git a/std/haxe/EventLoopDriver.hx b/std/haxe/EventLoopDriver.hx new file mode 100644 index 00000000000..afc85904c3b --- /dev/null +++ b/std/haxe/EventLoopDriver.hx @@ -0,0 +1,73 @@ +/* + * Copyright (C)2005-2026 Haxe Foundation + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + */ + +package haxe; + +/** + Owns waiting for an `EventLoop`. The loop keeps bookkeeping (queues, timers, + promises); the driver performs `wait` / `wake` / `close`. + + `allowsReentrancy` is read-only-by-contract: implementations expose a fixed + value and callers must not attempt to mutate it. + + `wake` and `close` must be safe after `close` and from other threads + (no-op if already closed). +**/ +interface EventLoopDriver { + /** + When `false`, nested `loopOnce` during native/driver callbacks is forbidden. + Read-only by contract after construction. + **/ + final allowsReentrancy:Bool; + + /** + Block according to `maxBlock` (seconds): + + - `< 0`: do not block; poll only (Haxe work is already due) + - `0`: block until an event or `wake()` + - `> 0`: block at most this many seconds + + Callers must never pass the `EventLoop.getNextTick()` idle sentinel `1e6` + as a real deadline. `EventLoop.loop()` synthesizes `maxBlock` before + calling `wait` (work due → `-1`; next timer → positive delta; keep-alive + only → `0`). + **/ + function wait(maxBlock:Float):Void; + + /** + Wake a thread blocked in `wait`. No-op if the driver is closed. + Safe to call from other threads. + **/ + function wake():Void; + + /** + Release driver resources. Idempotent; subsequent `wake` / `close` / + `wait` are no-ops (or `wait` returns immediately). Safe from other threads. + **/ + function close():Void; + + /** + Whether the driver has external work that should keep the event loop alive + (for example native UV handles). The default Haxe driver always returns `false`. + **/ + function hasExternalWork():Bool; +} diff --git a/std/haxe/HaxeEventLoopDriver.hx b/std/haxe/HaxeEventLoopDriver.hx new file mode 100644 index 00000000000..2fd5ad4c38c --- /dev/null +++ b/std/haxe/HaxeEventLoopDriver.hx @@ -0,0 +1,83 @@ +/* + * Copyright (C)2005-2026 Haxe Foundation + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + */ + +package haxe; + +/** + Default `EventLoop` driver: threaded targets own a `sys.thread.Lock`; + non-threaded sys uses `Sys.sleep` for positive timeouts; JS/Flash-shaped + targets no-op `wait` / `wake`. +**/ +class HaxeEventLoopDriver implements EventLoopDriver { + public final allowsReentrancy:Bool = true; + + #if target.threaded + final lock:sys.thread.Lock; + #end + + var closed:Bool = false; + + public function new() { + #if target.threaded + lock = new sys.thread.Lock(); + #end + } + + public function wait(maxBlock:Float):Void { + if (closed || maxBlock < 0) + return; + #if target.threaded + if (maxBlock == 0) + lock.wait(); + else + lock.wait(maxBlock); + #elseif sys + if (maxBlock > 0) + Sys.sleep(maxBlock); + // maxBlock == 0: best-effort no-op without threads + #else + // JS / Flash: guaranteed no-ops + #end + } + + public function wake():Void { + if (closed) + return; + #if target.threaded + lock.release(); + #end + } + + public function close():Void { + if (closed) + return; + closed = true; + #if target.threaded + // Unblock any waiter that may still be in wait(0)/wait(timeout) + lock.release(); + #end + } + + public function hasExternalWork():Bool { + return false; + } +} diff --git a/std/haxe/Timer.hx b/std/haxe/Timer.hx index ccf04173aa7..da7d0c0ada0 100644 --- a/std/haxe/Timer.hx +++ b/std/haxe/Timer.hx @@ -39,9 +39,10 @@ import haxe.Int64; the child class. Notice for threaded targets: - `Timer` instances require threads they were created in to run with Haxe's event loops. - Main thread of a Haxe program always contains an event loop. For other cases use - `sys.thread.Thread.createWithEventLoop` and `sys.thread.Thread.runWithEventLoop` methods. + `Timer` instances require the thread they were created on to have a Haxe event loop. + The main thread always has one. Threads created with `sys.thread.Thread.create` also + get an event loop that runs after the job returns. For a custom loop, construct + `haxe.EventLoop` and call `loop()` on that thread. **/ class Timer { #if (flash || js) diff --git a/std/hl/uv/Loop.hx b/std/hl/uv/Loop.hx index 8a83830f590..57515d1984a 100644 --- a/std/hl/uv/Loop.hx +++ b/std/hl/uv/Loop.hx @@ -47,20 +47,38 @@ abstract Loop(hl.Abstract<"uv_loop">) { @:hlNative("uv", #if (hl_ver >= version("1.16.0")) "stop_wrap" #else "stop" #end) public function stop():Void {} + /** + Attach a libuv loop to `loop` as its waiting driver and return the UV + `Loop` **synchronously**. + + Install uses `swapDriver`: when called from an event callback (`inLoop`) + or another thread, apply is deferred until the next `applyPendingSwap`. + A pending UV swap counts as external work so `loop()` cannot idle-exit + before the driver is published. Idempotent: if a UV driver is already + current or pending for this EventLoop, its loop is reused. + **/ public static function getFromEventLoop(loop:haxe.EventLoop):Loop { - if (@:privateAccess loop.nativeLoop == null) { - if (loop == haxe.EventLoop.main) - @:privateAccess loop.nativeLoop = new LoopWrapper(default_loop()); - else { - #if (hl_ver < version("1.16.0")) - throw "Using libUV multithread requires -D hl-ver=1.16.0"; - #else - @:privateAccess loop.nativeLoop = new LoopWrapper(create()); - #end - } - } - final wrapped:LoopWrapper = cast @:privateAccess loop.nativeLoop; - return @:privateAccess wrapped.uvLoop; + final current = loop.getDriver(); + if (Std.isOfType(current, UvEventLoopDriver)) + return (cast current : UvEventLoopDriver).uvLoop; + final pending = loop.getPendingDriver(); + if (pending != null && Std.isOfType(pending, UvEventLoopDriver)) + return (cast pending : UvEventLoopDriver).uvLoop; + + final isDefault = loop == haxe.EventLoop.main; + final uvLoop = if (isDefault) { + default_loop(); + } else { + #if (hl_ver < version("1.16.0")) + throw "Using libUV multithread requires -D hl-ver=1.16.0"; + #else + create(); + #end + }; + final driver = new UvEventLoopDriver(uvLoop, isDefault); + // Often deferred when called inside a callback; sync only when rule #5 allows. + loop.swapDriver(driver); + return uvLoop; } public static function getCurrent():Loop { @@ -83,26 +101,3 @@ abstract Loop(hl.Abstract<"uv_loop">) { #end } - -private class LoopWrapper { - public final allowsReentrancy = false; - final uvLoop:Loop; - - public function new(loop:Loop) { - this.uvLoop = loop; - } - - public function run() { - uvLoop.run(NoWait); - } - - public function close() { - final result = uvLoop.close(); - if (result != 0) - Sys.println("Some async handlers have not been closed"); - } - - public function isAlive() { - return uvLoop.alive() > 0; - } -} diff --git a/std/hl/uv/UvEventLoopDriver.hx b/std/hl/uv/UvEventLoopDriver.hx new file mode 100644 index 00000000000..ca3dfc0349c --- /dev/null +++ b/std/hl/uv/UvEventLoopDriver.hx @@ -0,0 +1,172 @@ +/* + * Copyright (C)2005-2026 Haxe Foundation + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + */ + +package hl.uv; + +/** + LibUV-backed `haxe.EventLoopDriver`. + + Owns an async doorbell and a one-shot deadline timer on `uvLoop`. While + `wait(maxBlock)` blocks (`maxBlock >= 0`), the async handle is referenced so + `UV_RUN_ONCE` does not busy-spin when only driver handles exist. Outside + waits both handles stay unreferenced so they alone do not keep the loop + alive (`hasExternalWork`). + + `isDefault` must be `true` for the process-global `default_loop`: `close` + then only closes driver-owned handles and never calls `uv_loop_close`. +**/ +class UvEventLoopDriver implements haxe.EventLoopDriver { + public final allowsReentrancy = false; + + /** Underlying libuv loop (for `getFromEventLoop` idempotent reuse). **/ + public final uvLoop:Loop; + + final isDefault:Bool; + /** Closures stored in uv handle data are in hl_gc_alloc_raw; keep a Haxe ref. **/ + final keepAliveCb:Void->Void; + var asyncHandle:HandleData; + var timerHandle:HandleData; + var closed = false; + + /** + @param uvLoop libuv loop to drive + @param isDefault `true` when `uvLoop` is the process default loop + **/ + public function new(uvLoop:Loop, isDefault:Bool) { + this.uvLoop = uvLoop; + this.isDefault = isDefault; + keepAliveCb = function() {}; + asyncHandle = async_init(uvLoop, keepAliveCb); + if (asyncHandle == null) + throw "Failed to create uv_async_t wake handle"; + handle_unref(asyncHandle); + timerHandle = timer_init(uvLoop); + if (timerHandle == null) + throw "Failed to create uv_timer_t deadline handle"; + handle_unref(timerHandle); + } + + public function wait(maxBlock:Float):Void { + if (closed) + return; + if (maxBlock < 0) { + stopDeadlineTimer(); + uvLoop.run(NoWait); + return; + } + if (maxBlock > 0) + armDeadlineTimer(maxBlock); + else + stopDeadlineTimer(); + // Ref async for the blocking poll so wait(0)/wait(t) cannot busy-spin + // when only unref'd driver handles exist. + final async = asyncHandle; + if (async != null) + handle_ref(async); + uvLoop.run(Once); + if (async != null) + handle_unref(async); + stopDeadlineTimer(); + } + + public function wake():Void { + if (closed) + return; + final async = asyncHandle; + if (async != null) + async_send(async); + } + + public function close():Void { + 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 + uvLoop.run(NoWait); + if (!isDefault) { + final result = uvLoop.close(); + if (result != 0) + Sys.println("Some async handlers have not been closed"); + } + } + + public function hasExternalWork():Bool { + 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_ref_wrap") + static function handle_ref(h:HandleData):Void {} + + @: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..76fc740ee93 100644 --- a/tests/threads/src/cases/TestEvents.hx +++ b/tests/threads/src/cases/TestEvents.hx @@ -2,6 +2,8 @@ package cases; import utest.Assert; import haxe.EventLoop; +import haxe.EventLoopDriver; +import haxe.HaxeEventLoopDriver; @:timeout(2000) class TestEvents extends ThreadTestBase { @@ -103,4 +105,322 @@ class TestEvents extends ThreadTestBase { Assert.equals("ok", threadValue); } -} \ No newline at end of file + + /** + Sync `swapDriver` on the loop thread before `loop()` applies immediately + and runs `onSwapped` on that thread. + **/ + function testSwapDriverSync() { + final loop = new EventLoop(); + final newDriver = new HaxeEventLoopDriver(); + var swappedOn = null; + loop.swapDriver(newDriver, () -> swappedOn = Thread.current()); + isTrue(loop.getDriver() == newDriver); + isNull(loop.getPendingDriver()); + equals(Thread.current(), swappedOn); + } + + /** + Cross-thread `swapDriver` while the loop thread is blocked in `wait(0)` + must defer, wake, apply on the loop thread, and run `onSwapped` there. + **/ + function testSwapDriverWhileBlocked() { + final ready = new Lock(); + final swapped = new Lock(); + var loopThread:Thread = null; + var onSwappedThread:Thread = null; + final newDriver = new HaxeEventLoopDriver(); + + final child = Thread.create(() -> { + loopThread = Thread.current(); + EventLoop.current.promise(); + ready.release(); + }); + + isTrue(ready.wait(1.0)); + Sys.sleep(0.05); // let onJobDone enter loop() → wait(0) + + final loop = EventLoop.getThreadLoop(child); + isTrue(loop != null); + isTrue(loop.getDriver() != newDriver); + + loop.swapDriver(newDriver, () -> { + onSwappedThread = Thread.current(); + isTrue(loop.getDriver() == newDriver); + loop.deliver(); + swapped.release(); + }); + + isTrue(swapped.wait(1.0)); + equals(loopThread, onSwappedThread); + isTrue(loop.getDriver() == newDriver); + isNull(loop.getPendingDriver()); + } + + /** + Last-wins: a second pending `swapDriver` closes the superseded pending + driver and drops its `onSwapped` callback. + **/ + function testSwapDriverLastWins() { + final loop = new EventLoop(); + final first = new TrackingDriver(); + final second = new TrackingDriver(); + var firstCb = false; + var secondOn:Thread = null; + + loop.run(() -> { + loop.swapDriver(first, () -> firstCb = true); + isTrue(loop.getPendingDriver() == first); + loop.swapDriver(second, () -> secondOn = Thread.current()); + isTrue(first.closed); + isFalse(second.closed); + isTrue(loop.getPendingDriver() == second); + }); + loop.loopOnce(); + + isFalse(firstCb); + equals(Thread.current(), secondOn); + isTrue(loop.getDriver() == second); + isNull(loop.getPendingDriver()); + loop.dispose(); + } + + /** + After `dispose`, cross-thread `wakeup` / `run` must not NPE on the + closed non-null driver sentinel. + **/ + function testDisposeThenCrossThreadWakeup() { + final loop = new EventLoop(); + loop.dispose(); + final done = new Lock(); + Thread.create(() -> { + loop.run(() -> {}); + @:privateAccess loop.wakeup(); + loop.getDriver().wake(); + done.release(); + }); + isTrue(done.wait(1.0)); + loop.getDriver().close(); // idempotent + } + + #if (hl && hl_ver >= version("1.16.0")) + /** + `getFromEventLoop` returns a Loop synchronously, installs via `swapDriver`, + and is idempotent for current/pending UV drivers. + **/ + function testGetFromEventLoopAttach() { + final loop = new EventLoop(); + final uv1 = hl.uv.Loop.getFromEventLoop(loop); + isTrue(Std.isOfType(loop.getDriver(), hl.uv.UvEventLoopDriver)); + isTrue((cast loop.getDriver() : hl.uv.UvEventLoopDriver).uvLoop == uv1); + final uv2 = hl.uv.Loop.getFromEventLoop(loop); + isTrue(uv1 == uv2); + isTrue(Std.isOfType(loop.getDriver(), hl.uv.UvEventLoopDriver)); + isNull(loop.getPendingDriver()); + loop.dispose(); + } + + /** + Deferred attach from an event callback: pending UV swap keeps the loop + alive until apply; second getFromEventLoop reuses the pending driver. + **/ + function testGetFromEventLoopDeferredIdempotent() { + final loop = new EventLoop(); + var uvFromCallback:hl.uv.Loop = null; + var uvSecond:hl.uv.Loop = null; + var driverAfterFirst:haxe.EventLoopDriver = null; + loop.run(() -> { + uvFromCallback = hl.uv.Loop.getFromEventLoop(loop); + driverAfterFirst = loop.getPendingDriver(); + isTrue(Std.isOfType(driverAfterFirst, hl.uv.UvEventLoopDriver)); + // Still Haxe driver until applyPendingSwap at end of loopOnce. + isTrue(Std.isOfType(loop.getDriver(), HaxeEventLoopDriver)); + uvSecond = hl.uv.Loop.getFromEventLoop(loop); + isTrue(uvFromCallback == uvSecond); + isTrue(loop.getPendingDriver() == driverAfterFirst); + }); + loop.loopOnce(); + isTrue(Std.isOfType(loop.getDriver(), hl.uv.UvEventLoopDriver)); + isTrue((cast loop.getDriver() : hl.uv.UvEventLoopDriver).uvLoop == uvFromCallback); + isNull(loop.getPendingDriver()); + loop.dispose(); + } + + /** + Cross-thread `getFromEventLoop` while a child is in `wait(0)` must not + idle-exit before the UV driver applies (pending swap is external work). + **/ + function testGetFromEventLoopCrossThreadAttach() { + final ready = new Lock(); + final done = new Lock(); + var loopThread:Thread = null; + + final child = Thread.create(() -> { + loopThread = Thread.current(); + EventLoop.current.promise(); + ready.release(); + }); + + isTrue(ready.wait(1.0)); + Sys.sleep(0.05); + + final loop = EventLoop.getThreadLoop(child); + final uv = hl.uv.Loop.getFromEventLoop(loop); + isTrue(uv != null); + + loop.run(() -> { + isTrue(Std.isOfType(EventLoop.current.getDriver(), hl.uv.UvEventLoopDriver)); + equals(loopThread, Thread.current()); + EventLoop.current.deliver(); + done.release(); + }); + + isTrue(done.wait(2.0)); + isTrue(Std.isOfType(loop.getDriver(), hl.uv.UvEventLoopDriver)); + isNull(loop.getPendingDriver()); + } + + /** + Rewrite of the old `testNativeWake` pattern: express wake via child + `EventLoop.loop()` (after job) and cross-thread `run`, with a ref'd TCP + listener as user work. Does not use blocking `loopOnce(maxBlock)`. + **/ + @:timeout(3000) + function testUvWakeWithSocket() { + final ready = new Lock(); + final woke = new Lock(); + var tcp:hl.uv.Tcp = null; + var wokeAt:Null = null; + final t0 = haxe.Timer.stamp(); + + final child = Thread.create(() -> { + final loop = EventLoop.current; + final uv = hl.uv.Loop.getFromEventLoop(loop); + tcp = new hl.uv.Tcp(uv); + tcp.bind(new sys.net.Host("127.0.0.1"), 0); + tcp.listen(1, () -> {}); + ready.release(); + // job ends → onJobDone → loop(); TCP keeps hasExternalWork alive + }); + + isTrue(ready.wait(1.0)); + Sys.sleep(0.1); + + EventLoop.getThreadLoop(child).run(() -> { + wokeAt = haxe.Timer.stamp(); + tcp.close(); + woke.release(); + }); + + isTrue(woke.wait(2.0)); + final latency = wokeAt - t0; + isTrue(latency >= 0.05 && latency < 0.5, 'unexpected wake latency: ${latency}s'); + } + + /** + Busy-spin regression: UV-backed `wait(0)` with **no** ref'd user handle + must block until the async doorbell wakes it (not return immediately). + **/ + @:timeout(3000) + function testUvIdleNoBusySpinWake() { + final ready = new Lock(); + final finished = new Lock(); + var driver:EventLoopDriver = null; + var elapsed = 0.0; + + Thread.create(() -> { + final loop = EventLoop.current; + hl.uv.Loop.getFromEventLoop(loop); + driver = loop.getDriver(); + isTrue(Std.isOfType(driver, hl.uv.UvEventLoopDriver)); + isFalse(driver.hasExternalWork()); + ready.release(); + final t0 = haxe.Timer.stamp(); + // Direct driver.wait(0) on the owning thread — no TCP / user handle. + driver.wait(0); + elapsed = haxe.Timer.stamp() - t0; + finished.release(); + }); + + isTrue(ready.wait(1.0)); + Sys.sleep(0.1); + driver.wake(); + isTrue(finished.wait(2.0)); + isTrue(elapsed >= 0.08, 'wait(0) returned too fast (busy-spin?): ${elapsed}s'); + isTrue(elapsed < 0.5, 'wait(0) took too long: ${elapsed}s'); + } + + /** + Promise-idle path: UV driver with no user handle stays in `loop()` via + `promise`, blocks, and wakes via cross-thread `run` + doorbell. + **/ + @:timeout(3000) + function testUvPromiseIdleWake() { + final ready = new Lock(); + final done = new Lock(); + var wokeAt:Null = null; + final t0 = haxe.Timer.stamp(); + + final child = Thread.create(() -> { + final loop = EventLoop.current; + hl.uv.Loop.getFromEventLoop(loop); + isFalse(loop.getDriver().hasExternalWork()); + loop.promise(); + ready.release(); + // job ends → loop() → wait(0) on UV with only unref'd doorbell/timer + }); + + isTrue(ready.wait(1.0)); + Sys.sleep(0.1); + + EventLoop.getThreadLoop(child).run(() -> { + wokeAt = haxe.Timer.stamp(); + EventLoop.current.deliver(); + done.release(); + }); + + isTrue(done.wait(2.0)); + final latency = wokeAt - t0; + isTrue(latency >= 0.05 && latency < 0.5, 'unexpected promise-idle wake latency: ${latency}s'); + } + + /** + Positive `maxBlock` with no user handle must honor the deadline timer + (not spin-return from `UV_RUN_ONCE`). + **/ + function testUvWaitDeadlineWithoutUserHandle() { + final loop = new EventLoop(); + hl.uv.Loop.getFromEventLoop(loop); + final driver = loop.getDriver(); + isFalse(driver.hasExternalWork()); + final t0 = haxe.Timer.stamp(); + driver.wait(0.1); + final elapsed = haxe.Timer.stamp() - t0; + isTrue(elapsed >= 0.08, 'wait(0.1) returned too fast: ${elapsed}s'); + isTrue(elapsed < 0.5, 'wait(0.1) took too long: ${elapsed}s'); + loop.dispose(); + } + #end +} + +/** + Test helper: tracks `close` for last-wins `swapDriver` assertions. +**/ +private class TrackingDriver implements EventLoopDriver { + public final allowsReentrancy = true; + public var closed = false; + + public function new() {} + + public function wait(maxBlock:Float):Void {} + + public function wake():Void {} + + public function close():Void { + closed = true; + } + + public function hasExternalWork():Bool { + return false; + } +}