diff --git a/src/Fable.Build/Test/Python.fs b/src/Fable.Build/Test/Python.fs index 87081b3596..caaaffb692 100644 --- a/src/Fable.Build/Test/Python.fs +++ b/src/Fable.Build/Test/Python.fs @@ -23,6 +23,10 @@ let handle (args: string list) = Directory.clean buildDir + // Native runtime tests use controlled timers and real threads alongside generated F# regressions. + for testFile in Directory.GetFiles(sourceDir, "test_*.py") do + File.Copy(testFile, Path.Combine(buildDir, Path.GetFileName(testFile)), overwrite = true) + Command.Run("uv", "sync") // Install local fable-library as editable package for testing diff --git a/src/Fable.Transforms/Python/Replacements.fs b/src/Fable.Transforms/Python/Replacements.fs index 2f18e27862..f67b8fb70a 100644 --- a/src/Fable.Transforms/Python/Replacements.fs +++ b/src/Fable.Transforms/Python/Replacements.fs @@ -2851,7 +2851,10 @@ let exceptions (com: ICompiler) (ctx: Context) r t (i: CallInfo) (thisArg: Expr Helper.ConstructorCall(e, t, args, ?loc = r) |> Some | "get_Message", Some e -> Helper.GlobalCall("str", t, [ thisArg.Value ], ?loc = r) |> Some | "get_StackTrace", Some e -> getFieldWith r t e "stack" |> Some - | "get_InnerException", Some e -> getFieldWith r t e "inner_exception" |> Some + | "get_InnerException", Some e -> + // Caught exceptions are typed as Python's built-in Exception, which lacks this attribute. + Helper.GlobalCall("getattr", t, [ e; makeStrConst "inner_exception" ], ?loc = r) + |> Some | _ -> None let unchecked (com: ICompiler) (ctx: Context) r t (i: CallInfo) (_: Expr option) (args: Expr list) = diff --git a/src/fable-library-py/fable_library/async_.py b/src/fable-library-py/fable_library/async_.py index 1b72efd6ee..b708149a0d 100644 --- a/src/fable-library-py/fable_library/async_.py +++ b/src/fable-library-py/fable_library/async_.py @@ -58,7 +58,7 @@ def cont(ctx: IAsyncContext[T]): def create_cancellation_token(arg: int | bool | None = None) -> CancellationToken: cancelled = arg if isinstance(arg, bool) else False token = CancellationToken(cancelled) - if isinstance(arg, int): + if isinstance(arg, int) and not isinstance(arg, bool): timer = Timer(arg / 1000.0, token.cancel) timer.start() @@ -112,6 +112,8 @@ def cancel() -> None: ctx.on_cancel(OperationCanceledError()) token_id[0] = ctx.cancel_token.add_listener(cancel) + if ctx.cancel_token.is_cancelled: + return event.AddHandler(handler) return protected_cont(cont) diff --git a/src/fable-library-py/fable_library/async_builder.py b/src/fable-library-py/fable_library/async_builder.py index eb098c1c95..30334b32dc 100644 --- a/src/fable-library-py/fable_library/async_builder.py +++ b/src/fable-library-py/fable_library/async_builder.py @@ -4,16 +4,15 @@ from abc import abstractmethod from collections.abc import Callable from dataclasses import dataclass, field -from threading import Lock, RLock +from threading import Condition, Lock, RLock, get_ident from typing import ( Any, Literal, - Protocol, overload, ) from .protocols import IEnumerable_1 -from .util import Disposable, IDisposable +from .util import IDisposable class OperationCanceledError(Exception): @@ -28,58 +27,105 @@ def __init__(self, msg: str | None = None) -> None: ] -class _Listener(Protocol): - def __call__(self, __state: Any | None = None) -> None: ... +_NO_STATE = object() + + +class CancellationCallbackError(ExceptionGroup[Exception]): + @property + def inner_exception(self) -> Exception: + # invariant: F# InnerException exposes the first failure without discarding the remaining errors + return self.exceptions[0] class CancellationToken: - __slots__ = "cancelled", "idx", "listeners", "lock" + __slots__ = "_condition", "_running", "cancelled", "idx", "listeners", "lock" - def __init__(self, cancelled: bool = False): + def __init__(self, cancelled: bool = False) -> None: self.cancelled = cancelled self.listeners: dict[int, Callable[[], None]] = {} self.idx = 0 self.lock = RLock() + self._condition = Condition(self.lock) + self._running: dict[int, int] = {} @property - def is_cancelled(self): - return self.cancelled + def is_cancelled(self) -> bool: + with self.lock: + return self.cancelled is_cancellation_requested = is_cancelled def cancel(self) -> None: - cancel = False with self.lock: - if not self.cancelled: - cancel = True - self.cancelled = True + if self.cancelled: + return + self.cancelled = True + # decision: snapshots IDs newest first to match .NET while allowing disposal of pending callbacks + pending = list(reversed(self.listeners)) + + errors: list[Exception] = [] + for listener_id in pending: + try: + self._invoke_listener(listener_id) + except Exception as error: + errors.append(error) + if errors: + raise CancellationCallbackError("Cancellation callbacks failed", errors) - if cancel: - for listener in self.listeners.values(): - listener() + def _invoke_listener(self, listener_id: int) -> None: + with self.lock: + listener = self.listeners.pop(listener_id, None) + if listener is None: + return + self._running[listener_id] = get_ident() + # invariant: user callbacks execute outside the registry lock + try: + listener() + finally: + with self._condition: + del self._running[listener_id] + self._condition.notify_all() def add_listener(self, f: Callable[[], None]) -> int: with self.lock: - id = self.idx - self.idx = self.idx + 1 - self.listeners[id] = f - - return id + listener_id = self.idx + self.idx += 1 + if not self.cancelled: + self.listeners[listener_id] = f + return listener_id + # .NET invokes late registrations synchronously and does not retain them. + f() + return listener_id def remove_listener(self, id: int) -> None: - with self.lock: - del self.listeners[id] + with self._condition: + self.listeners.pop(id, None) + # invariant: disposal waits for a running callback except when that callback disposes itself + while id in self._running and self._running[id] != get_ident(): + self._condition.wait() + + @overload + def register(self, f: Callable[[], None]) -> IDisposable: ... + + @overload + def register[S](self, f: Callable[[S], None], state: S) -> IDisposable: ... + + def register(self, f: Callable[..., None], state: object = _NO_STATE) -> IDisposable: + listener_id = self.add_listener(f if state is _NO_STATE else lambda: f(state)) + return _CancellationRegistration(self, listener_id) + + def Dispose(self) -> None: + # Source disposal is currently a no-op, as in the compiler replacement and JS runtime. + pass - def register(self, f: _Listener, state: Any | None = None) -> None: - if state: - id = self.add_listener(lambda: f(state)) - else: - id = self.add_listener(f) - def dispose(): - self.remove_listener(id) +class _CancellationRegistration: + def __init__(self, token: CancellationToken, listener_id: int) -> None: + self._token = token + self._listener_id = listener_id - Disposable.create(dispose) + def Dispose(self) -> None: + self._token.remove_listener(self._listener_id) class IAsyncContext[T]: @@ -217,6 +263,7 @@ def protected_cont[T](f: Async[T]) -> Async[T]: def _protected_cont(ctx: IAsyncContext[T]): if ctx.cancel_token and ctx.cancel_token.is_cancelled: ctx.on_cancel(OperationCanceledError()) + return def fn(): try: diff --git a/src/fable-library-ts/Async.ts b/src/fable-library-ts/Async.ts index eb0fb63a6b..ec9e58c4aa 100644 --- a/src/fable-library-ts/Async.ts +++ b/src/fable-library-ts/Async.ts @@ -104,6 +104,7 @@ export function awaitEvent(event: IEvent$2, can if (cancelAction != null) { cancelAction(); } ctx.onCancel(new OperationCanceledException()); }); + if (ctx.cancelToken.isCancelled) { return; } event.AddHandler(handler); }); } diff --git a/src/fable-library-ts/AsyncBuilder.ts b/src/fable-library-ts/AsyncBuilder.ts index 86d59e4ce4..56912579f3 100644 --- a/src/fable-library-ts/AsyncBuilder.ts +++ b/src/fable-library-ts/AsyncBuilder.ts @@ -12,6 +12,16 @@ export type Continuations = [ Continuation ]; +// AggregateError requires ES2021; the runtime also supports ES2020 hosts. +export class CancellationCallbackError extends Exception { + public readonly errors: unknown[]; + constructor(errors: unknown[]) { + // invariant: F# InnerException exposes the first failure without discarding the remaining errors + super("Cancellation callbacks failed", errors[0] as Exception); + this.errors = errors; + } +} + export class CancellationToken implements IDisposable { private _id: number; private _cancelled: boolean; @@ -27,14 +37,23 @@ export class CancellationToken implements IDisposable { public cancel() { if (!this._cancelled) { this._cancelled = true; - for (const [, listener] of this._listeners) { - listener(); + // decision: visit newest registrations first, matching .NET; disposal can skip pending callbacks + const pending = Array.from(this._listeners.keys()).reverse(); + const errors: unknown[] = []; + for (const id of pending) { + const listener = this._listeners.get(id); + this._listeners.delete(id); + if (listener) { + try { listener(); } catch (error) { errors.push(error); } + } } + if (errors.length > 0) { throw new CancellationCallbackError(errors); } } } public addListener(f: () => void) { const id = this._id; - this._listeners.set(this._id++, f); + this._id++; + if (this._cancelled) { f(); } else { this._listeners.set(id, f); } return id; } public removeListener(id: number) { diff --git a/tests/Js/Main/AsyncTests.fs b/tests/Js/Main/AsyncTests.fs index 761d32daa4..5127510cf8 100644 --- a/tests/Js/Main/AsyncTests.fs +++ b/tests/Js/Main/AsyncTests.fs @@ -6,6 +6,13 @@ open Util.Testing #if FABLE_COMPILER open Fable.Core + +type private Deferred = + [] + static member startDeferred(computation: Async, token: System.Threading.CancellationToken, onSuccess: int -> unit, onError: exn -> unit, onCancel: exn -> unit): unit -> unit = jsNative + + [] + static member handlerCount(event: Event): int = jsNative #endif type DisposableAction(f) = @@ -44,6 +51,30 @@ let tests = async { return () } |> Async.StartImmediate + testCase "Cancellation callback errors expose the first InnerException" <| fun () -> + for count in [1; 2] do + use cts = new System.Threading.CancellationTokenSource() + let errors = Array.init count (fun i -> exn (string i)) + let mutable calls = 0 + errors |> Array.iter (fun error -> + cts.Token.Register(fun () -> + calls <- calls + 1 + raise error) |> ignore) + let caught = + try + cts.Cancel() + None + with ex -> Some ex + match caught with + | Some ex -> + equal errors[count - 1].Message ex.InnerException.Message + equal true (obj.ReferenceEquals(errors[count - 1], ex.InnerException)) + | None -> failwith "Expected cancellation callback errors" + equal count calls + equal true cts.IsCancellationRequested + cts.Cancel() + equal count calls + testCase "Async while binding works correctly" <| fun () -> let mutable result = 0 async { @@ -684,6 +715,21 @@ let tests = cts.Cancel() async { equal true cancelCalled } +#if FABLE_COMPILER + testCase "Async.AwaitEvent cancelled before its body runs does not subscribe" <| fun () -> + let ev = Event() + let cts = new System.Threading.CancellationTokenSource() + let calls = ResizeArray() + let resume = + Deferred.startDeferred(Async.AwaitEvent ev.Publish, cts.Token, + (fun v -> calls.Add $"success {v}"), (fun _ -> calls.Add "error"), (fun _ -> calls.Add "cancel")) + cts.Cancel() + resume () + ev.Trigger(42) + equal [ "cancel" ] (List.ofSeq calls) + equal 0 (Deferred.handlerCount ev) +#endif + testCase "Async try .. with does not run 'with' branch when body succeeds" <| fun () -> let work = async { try diff --git a/tests/Js/Main/js/async-deferred.js b/tests/Js/Main/js/async-deferred.js new file mode 100644 index 0000000000..07cb2ad500 --- /dev/null +++ b/tests/Js/Main/js/async-deferred.js @@ -0,0 +1,15 @@ +export function startDeferred(computation, token, onSuccess, onError, onCancel) { + let body; + computation({ + onSuccess, + onError, + onCancel, + cancelToken: token, + trampoline: { completed: false, incrementAndCheck: () => true, hijack: f => { body = f; } }, + }); + return () => body(); +} + +export function handlerCount(event) { + return event.delegates.length; +} diff --git a/tests/Python/TestAsync.fs b/tests/Python/TestAsync.fs index 6db0fdf8c3..3ff0be83fb 100644 --- a/tests/Python/TestAsync.fs +++ b/tests/Python/TestAsync.fs @@ -32,6 +32,31 @@ let ``test Simple async translates without exception`` () = async { return () } |> Async.StartImmediate +[] +let ``test Cancellation callback errors expose the first InnerException`` () = + for count in [1; 2] do + use cts = new System.Threading.CancellationTokenSource() + let errors = Array.init count (fun i -> exn (string i)) + let mutable calls = 0 + errors |> Array.iter (fun error -> + cts.Token.Register(fun () -> + calls <- calls + 1 + raise error) |> ignore) + let caught = + try + cts.Cancel() + None + with ex -> Some ex + match caught with + | Some ex -> + equal errors[count - 1].Message ex.InnerException.Message + equal true (obj.ReferenceEquals(errors[count - 1], ex.InnerException)) + | None -> failwith "Expected cancellation callback errors" + equal count calls + equal true cts.IsCancellationRequested + cts.Cancel() + equal count calls + [] let ``test Async while binding works correctly`` () = diff --git a/tests/Python/test_async_runtime.py b/tests/Python/test_async_runtime.py new file mode 100644 index 0000000000..aba116e32e --- /dev/null +++ b/tests/Python/test_async_runtime.py @@ -0,0 +1,193 @@ +import asyncio +from threading import Barrier, Condition, Event, Thread + +import pytest +from fable_library.async_ import await_event, sleep, start_with_continuations +from fable_library.async_builder import CancellationCallbackError, CancellationToken, IAsyncContext, Trampoline +from fable_library.event import Event as FableEvent + + +def test_registration_handle_and_disposal(): + token = CancellationToken() + calls: list[str] = [] + first = token.register(lambda: calls.append("first")) + second = token.register(lambda: calls.append("second")) + assert first is not None + first.Dispose() + first.Dispose() + token.cancel() + second.Dispose() + assert calls == ["second"] + assert not token.listeners + + +def test_late_registration_and_falsy_state(): + token = CancellationToken(True) + calls: list[object] = [] + handle = token.register(lambda state: calls.append(state), 0) + assert calls == [0] + token.register(lambda state: calls.append(state), None).Dispose() + assert calls == [0, None] + handle.Dispose() + assert not token.listeners + + +def test_self_disposal_and_disposal_of_pending_callback(): + token = CancellationToken() + calls: list[str] = [] + first = token.register(lambda: calls.append("first")) + + def callback(): + second.Dispose() + first.Dispose() + calls.append("second") + + second = token.register(callback) + token.cancel() + token.cancel() + assert calls == ["second"] + assert not token.listeners + + +def test_callback_failures_do_not_skip_other_callbacks(): + token = CancellationToken() + calls: list[str] = [] + token.register(lambda: calls.append("other")) + + def failing(): + raise ValueError("callback") + + token.register(failing) + with pytest.raises(ExceptionGroup) as errors: + token.cancel() + assert len(errors.value.exceptions) == 1 + assert calls == ["other"] + assert not token.listeners + + +@pytest.mark.parametrize("count", [1, 2]) +def test_callback_errors_preserve_all_errors_and_first_inner_exception(count: int) -> None: + token = CancellationToken() + errors = [ValueError(str(i)) for i in range(count)] + + def failing(error: Exception) -> None: + raise error + + for error in errors: + token.register(failing, error) + + with pytest.raises(CancellationCallbackError) as caught: + token.cancel() + assert isinstance(caught.value, ExceptionGroup) + assert caught.value.exceptions == tuple(reversed(errors)) + assert caught.value.inner_exception is errors[-1] + assert not token.listeners + token.cancel() + + +def test_dispose_waits_for_running_callback_and_late_registration_runs_outside_lock(): + token = CancellationToken() + entered, release, disposed, late = Event(), Event(), Event(), Event() + + def callback(): + entered.set() + assert release.wait(5) + + waiting = Event() + + class ObservedCondition(Condition): + def wait(self, timeout=None): + waiting.set() + return super().wait(timeout) + + token._condition = ObservedCondition(token.lock) + handle = token.register(callback) + cancel_thread = Thread(target=token.cancel) + cancel_thread.start() + assert entered.wait(5) + + # Registration after cancellation must run immediately, even while an earlier callback is blocked. + register_thread = Thread(target=lambda: token.register(late.set).Dispose()) + register_thread.start() + assert late.wait(5) + register_thread.join(5) + + def dispose(): + handle.Dispose() + disposed.set() + + dispose_thread = Thread(target=dispose) + dispose_thread.start() + assert waiting.wait(5) + assert not disposed.is_set() + release.set() + cancel_thread.join(5) + dispose_thread.join(5) + assert disposed.is_set() + assert not token.listeners + + +def test_concurrent_registration_and_cancellation(): + token = CancellationToken() + gate = Barrier(17) + calls = [0] * 16 + + def register(index: int): + gate.wait(5) + + def callback(): + calls[index] += 1 + + token.register(callback).Dispose() + + threads = [Thread(target=register, args=(i,)) for i in range(16)] + for thread in threads: + thread.start() + gate.wait(5) + token.cancel() + for thread in threads: + thread.join(5) + assert not thread.is_alive() + assert all(count in (0, 1) for count in calls) + assert not token.listeners + assert not token._running + + +def test_sleep_on_cancelled_token_settles_once(): + calls: list[str] = [] + start_with_continuations( + sleep(10), + lambda _: calls.append("success"), + lambda _: calls.append("error"), + lambda _: calls.append("cancel"), + CancellationToken(True), + ) + assert calls == ["cancel"] + + +class _DeferringTrampoline(Trampoline): + def increment_and_check(self) -> bool: + return True + + +def test_await_event_cancelled_before_its_body_runs_does_not_subscribe(): + event: FableEvent[int] = FableEvent() + token = CancellationToken() + calls: list[str] = [] + + async def run() -> None: + ctx = IAsyncContext.create( + _DeferringTrampoline(), + token, + lambda value: calls.append(f"success {value}"), + lambda _: calls.append("error"), + lambda _: calls.append("cancel"), + ) + await_event(event)(ctx) + token.cancel() + await asyncio.sleep(0) + + asyncio.run(run()) + event.Trigger(42) + assert calls == ["cancel"] + assert not event.delegates