From 7bdbee7c6c6e6901b86b53d8d2fbd1c074e2f9e1 Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Sat, 3 Oct 2026 21:39:49 +0200 Subject: [PATCH 1/4] fix(js/ts/python): settle cancelled delays and finalizers once --- src/Fable.Build/Test/JavaScript.fs | 2 + src/fable-library-py/fable_library/async_.py | 71 +++- .../fable_library/async_builder.py | 22 +- .../fable_library/mailbox_processor.py | 4 +- src/fable-library-ts/Async.ts | 19 +- src/fable-library-ts/AsyncBuilder.ts | 13 +- src/fable-library-ts/MailboxProcessor.ts | 1 + tests/Js/Runtime/async-cancellation.mjs | 68 ++++ tests/Python/TestAsync.fs | 18 ++ tests/Python/test_async_runtime.py | 303 +++++++++++++++++- 10 files changed, 490 insertions(+), 31 deletions(-) create mode 100644 tests/Js/Runtime/async-cancellation.mjs diff --git a/src/Fable.Build/Test/JavaScript.fs b/src/Fable.Build/Test/JavaScript.fs index 4b978a26ad..e1440fc548 100644 --- a/src/Fable.Build/Test/JavaScript.fs +++ b/src/Fable.Build/Test/JavaScript.fs @@ -160,6 +160,8 @@ let private handleMainTests (isWatch: bool) (noDotnet: bool) = // Test the Main tests against JavaScript runMainTests JsDate false noDotnet + Command.Run("node", "tests/Js/Runtime/async-cancellation.mjs") + // Re-run them with the Temporal date/time representation enabled runMainTests Temporal false noDotnet diff --git a/src/fable-library-py/fable_library/async_.py b/src/fable-library-py/fable_library/async_.py index faed9113f8..3f515fc847 100644 --- a/src/fable-library-py/fable_library/async_.py +++ b/src/fable-library-py/fable_library/async_.py @@ -4,7 +4,7 @@ from asyncio import Future, ensure_future from collections.abc import Awaitable, Callable, Iterable from concurrent.futures import ThreadPoolExecutor -from threading import Timer +from threading import RLock, Timer from typing import ( Any, ) @@ -81,18 +81,65 @@ def is_cancellation_requested(token: CancellationToken) -> bool: def sleep(milliseconds_duetime: int | TimeSpan) -> Async[None]: - def cont(ctx: IAsyncContext[None]): - def cancel(): - ctx.on_cancel(OperationCanceledError()) - - token_id = ctx.cancel_token.add_listener(cancel) - - def timeout(): - ctx.cancel_token.remove_listener(token_id) - ctx.on_success(None) + def cont(ctx: IAsyncContext[None]) -> None: + loop = asyncio.get_running_loop() + gate = RLock() + completed = False + timer: asyncio.TimerHandle | None = None + token_id: int | None = None + + def settle(cancelled: bool) -> None: + nonlocal completed + with gate: + if completed: + return + cancelled = cancelled or ctx.cancel_token.is_cancelled + completed = True + if timer is not None: + timer.cancel() + listener_id = token_id + # Dispose outside the gate: another thread may already be running cancel(). + if listener_id is not None: + ctx.cancel_token.remove_listener(listener_id) + if cancelled: + ctx.on_cancel(OperationCanceledError()) + else: + ctx.on_success(None) - due_time = to_milliseconds(milliseconds_duetime) / 1000.0 - ctx.trampoline.run_later(timeout, due_time) + def cancel() -> None: + # decision: foreign-thread cancellation settles on the owning loop while it is alive + if loop.is_closed(): + settle(True) + return + try: + current_loop = asyncio.get_running_loop() + except RuntimeError: + current_loop = None + if current_loop is loop: + settle(True) + else: + try: + loop.call_soon_threadsafe(settle, True) + except RuntimeError: + if not loop.is_closed(): + raise + settle(True) + + # invariant: installation and settlement cannot interleave before both resources are owned + try: + with gate: + token_id = ctx.cancel_token.add_listener(cancel) + if not completed: + due_time = to_milliseconds(milliseconds_duetime) / 1000.0 + timer = ctx.trampoline.run_later(lambda: settle(False), due_time) + if completed: + timer.cancel() + except Exception: + with gate: + completed = True + if token_id is not None: + ctx.cancel_token.remove_listener(token_id) + raise 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 30334b32dc..377aa71481 100644 --- a/src/fable-library-py/fable_library/async_builder.py +++ b/src/fable-library-py/fable_library/async_builder.py @@ -245,9 +245,9 @@ def run_later( self, action: Callable[[], None], due_time: float = 0.0, - ): + ) -> asyncio.TimerHandle: loop = asyncio.get_running_loop() - loop.call_later(due_time, action) + return loop.call_later(due_time, action) def run(self, action: Callable[[], None]): loop = asyncio.get_running_loop() @@ -345,15 +345,27 @@ def ReturnFrom[T](self, computation: Async[T]) -> Async[T]: def TryFinally[T](self, computation: Async[T], compensation: Callable[[], None]) -> Async[T]: def cont(ctx: IAsyncContext[T]) -> None: def on_success(x: T) -> None: - compensation() + try: + compensation() + except Exception as error: + ctx.on_error(error) + return ctx.on_success(x) def on_error(x: Exception) -> None: - compensation() + try: + compensation() + except Exception as error: + ctx.on_error(error) + return ctx.on_error(x) def on_cancel(x: OperationCanceledError) -> None: - compensation() + # decision: ignores compensation failures during cancellation, matching .NET Async.TryFinally + try: + compensation() + except Exception: + pass ctx.on_cancel(x) ctx_ = IAsyncContext.create(ctx.trampoline, ctx.cancel_token, on_success, on_error, on_cancel) diff --git a/src/fable-library-py/fable_library/mailbox_processor.py b/src/fable-library-py/fable_library/mailbox_processor.py index 663635f589..15ef175e32 100644 --- a/src/fable-library-py/fable_library/mailbox_processor.py +++ b/src/fable-library-py/fable_library/mailbox_processor.py @@ -101,8 +101,8 @@ def receive(self) -> Async[Msg]: Returns: An asynchronous computation which will consume the first message in arrival order. No thread is blocked while - waiting for further messages. Raises a TimeoutException if - the timeout is exceeded. + waiting for further messages. Cancellation of the token alone + does not wake an idle receive; a post triggers the cancellation check. """ def callback(conts: Continuations[Any]): diff --git a/src/fable-library-ts/Async.ts b/src/fable-library-ts/Async.ts index ec9e58c4aa..23ab998b7e 100644 --- a/src/fable-library-ts/Async.ts +++ b/src/fable-library-ts/Async.ts @@ -159,15 +159,18 @@ export function sequential(computations: Iterable>) { export function sleep(millisecondsDueTime: number) { return protectedCont((ctx: IAsyncContext) => { - let tokenId: number; - const timeoutId = setTimeout(() => { - ctx.cancelToken.removeListener(tokenId); - ctx.onSuccess(void 0); - }, millisecondsDueTime); - tokenId = ctx.cancelToken.addListener(() => { + let completed = false; + let tokenId: number | undefined; + const settle = (cancelled: boolean) => { + if (completed) { return; } + completed = true; clearTimeout(timeoutId); - ctx.onCancel(new OperationCanceledException()); - }); + if (tokenId !== undefined) { ctx.cancelToken.removeListener(tokenId); } + if (cancelled) { ctx.onCancel(new OperationCanceledException()); } + else { ctx.onSuccess(void 0); } + }; + const timeoutId = setTimeout(() => settle(false), millisecondsDueTime); + tokenId = ctx.cancelToken.addListener(() => settle(true)); }); } diff --git a/src/fable-library-ts/AsyncBuilder.ts b/src/fable-library-ts/AsyncBuilder.ts index 56912579f3..2a89be285e 100644 --- a/src/fable-library-ts/AsyncBuilder.ts +++ b/src/fable-library-ts/AsyncBuilder.ts @@ -203,15 +203,22 @@ export class AsyncBuilder { return protectedCont((ctx: IAsyncContext) => { computation({ onSuccess: (x: T) => { - compensation(); + try { compensation(); } catch (error) { + ctx.onError(ensureErrorOrException(error)); + return; + } ctx.onSuccess(x); }, onError: (x: any) => { - compensation(); + try { compensation(); } catch (error) { + ctx.onError(ensureErrorOrException(error)); + return; + } ctx.onError(x); }, onCancel: (x: any) => { - compensation(); + // decision: cancellation remains terminal when compensation throws, matching .NET Async.TryFinally + try { compensation(); } catch { /* Cancellation takes precedence. */ } ctx.onCancel(x); }, cancelToken: ctx.cancelToken, diff --git a/src/fable-library-ts/MailboxProcessor.ts b/src/fable-library-ts/MailboxProcessor.ts index b78689d7c7..5ff3874435 100644 --- a/src/fable-library-ts/MailboxProcessor.ts +++ b/src/fable-library-ts/MailboxProcessor.ts @@ -79,6 +79,7 @@ export function startInstance($this: MailboxProcessor) { startImmediate($this.body($this), $this.cancellationToken); } +// Token cancellation does not wake an idle Receive; delivery is driven by posts. export function receive($this: MailboxProcessor) { return fromContinuations((conts: Continuations) => { if ($this.continuation) { diff --git a/tests/Js/Runtime/async-cancellation.mjs b/tests/Js/Runtime/async-cancellation.mjs new file mode 100644 index 0000000000..169be26ac3 --- /dev/null +++ b/tests/Js/Runtime/async-cancellation.mjs @@ -0,0 +1,68 @@ +import assert from "node:assert/strict"; +import { CancellationToken, CancellationCallbackError } from "../../../temp/fable-library-js/AsyncBuilder.js"; +import { sleep, startWithContinuations } from "../../../temp/fable-library-js/Async.js"; +import { singleton } from "../../../temp/fable-library-js/AsyncBuilder.js"; +import { MailboxProcessor, receive, post } from "../../../temp/fable-library-js/MailboxProcessor.js"; + +const token = new CancellationToken(); +const calls = []; +const first = token.register(() => calls.push("first")); +const second = token.register(() => { second.Dispose(); first.Dispose(); calls.push("second"); }); +token.cancel(); +token.register(state => calls.push(state), 0).Dispose(); +second.Dispose(); +assert.deepEqual(calls, ["second", 0]); +assert.equal(token._listeners.size, 0); + +const failing = new CancellationToken(); +let remaining = 0; +failing.register(() => remaining++); +failing.register(() => { throw new Error("callback"); }); +assert.throws(() => failing.cancel(), CancellationCallbackError); +assert.equal(remaining, 1); +assert.equal(failing._listeners.size, 0); + +const originalSetTimeout = globalThis.setTimeout; +const originalClearTimeout = globalThis.clearTimeout; +try { + for (const cancelFirst of [true, false]) { + for (const throws of [true, false]) { + let timeout, retired = 0, finalized = 0; + globalThis.setTimeout = callback => { timeout = callback; return 123; }; + globalThis.clearTimeout = id => { assert.equal(id, 123); retired++; }; + const source = new CancellationToken(); + const terminal = []; + const work = singleton.TryFinally(sleep(100), () => { + finalized++; + if (throws) { throw new Error("finalizer"); } + }); + const finish = tag => () => terminal.push(tag); + startWithContinuations(work, finish("success"), finish("error"), finish("cancel"), source); + if (cancelFirst) { source.cancel(); timeout(); } + else { + timeout(); + source.cancel(); + } + timeout(); + assert.equal(finalized, 1); + assert.deepEqual(terminal, cancelFirst ? ["cancel"] : throws ? ["error"] : ["success"]); + assert.equal(source._listeners.size, 0); + assert.equal(retired, 1); + } + } +} finally { + globalThis.setTimeout = originalSetTimeout; + globalThis.clearTimeout = originalClearTimeout; +} + +// JS Receive retains its success continuation while idle; token cancellation does not wake it. +const source = new CancellationToken(); +const mailbox = new MailboxProcessor(() => singleton.Zero(), source); +const terminal = []; +startWithContinuations(receive(mailbox), x => terminal.push(x), e => { throw e; }, () => terminal.push("cancel"), source); +source.cancel(); +assert.deepEqual(terminal, []); +post(mailbox, 42); +assert.deepEqual(terminal, [42]); +assert.equal(source._listeners.size, 0); +console.log("JS cancellation, controlled timers, and idle Receive regressions passed"); diff --git a/tests/Python/TestAsync.fs b/tests/Python/TestAsync.fs index cb701222d1..425a8c1b72 100644 --- a/tests/Python/TestAsync.fs +++ b/tests/Python/TestAsync.fs @@ -706,3 +706,21 @@ let ``test unit continuations preserve captured callbacks`` () = namedUnitCapture 3 2 equal 5 calls +[] +let ``test cooperative cancellation stops at the next async boundary`` () = + use source = new System.Threading.CancellationTokenSource() + let mutable finalizers = 0 + let mutable terminal = "" + let mutable reached = false + let work = async { + try + source.Cancel() + do! async { return () } + reached <- true + finally + finalizers <- finalizers + 1 + } + Async.StartWithContinuations(work, (fun () -> terminal <- "success"), (fun _ -> terminal <- "error"), (fun _ -> terminal <- "cancel"), source.Token) + equal false reached + equal 1 finalizers + equal "cancel" terminal diff --git a/tests/Python/test_async_runtime.py b/tests/Python/test_async_runtime.py index aba116e32e..905047a8a9 100644 --- a/tests/Python/test_async_runtime.py +++ b/tests/Python/test_async_runtime.py @@ -1,10 +1,14 @@ import asyncio +from collections.abc import Callable from threading import Barrier, Condition, Event, Thread import pytest +from fable_library import async_ as async_runtime from fable_library.async_ import await_event, sleep, start_with_continuations -from fable_library.async_builder import CancellationCallbackError, CancellationToken, IAsyncContext, Trampoline +from fable_library.async_builder import CancellationCallbackError, CancellationToken, IAsyncContext, Trampoline, singleton from fable_library.event import Event as FableEvent +from fable_library.mailbox_processor import MailboxProcessor +from fable_library.util import lock def test_registration_handle_and_disposal(): @@ -127,6 +131,111 @@ def dispose(): assert not token.listeners +class Timer(asyncio.TimerHandle): + def __init__(self, callback: Callable[[], None]): + super().__init__(0, callback, (), asyncio.get_running_loop()) + self.callback = callback + self.retired = False + self.cancel_count = 0 + + def cancel(self): + self.retired = True + self.cancel_count += 1 + + def fire(self): + # Deliberately fire even a retired callback to model an already queued timeout. + self.callback() + + +class Scheduler(Trampoline): + def __init__(self): + super().__init__() + self.timers: list[Timer] = [] + + def run_later(self, action: Callable[[], None], due_time: float = 0.0): + timer = Timer(action) + self.timers.append(timer) + return timer + + +@pytest.mark.parametrize("cancel_first", [True, False]) +def test_sleep_settles_once_and_cleans_up(cancel_first): + async def run(): + token, scheduler = CancellationToken(), Scheduler() + terminal: list[str] = [] + finalizers: list[str] = [] + ctx = IAsyncContext.create( + scheduler, + token, + lambda _: terminal.append("success"), + lambda _: terminal.append("error"), + lambda _: terminal.append("cancel"), + ) + work = singleton.TryFinally(async_runtime.sleep(100), lambda: finalizers.append("finally")) + work(ctx) + timer = scheduler.timers[0] + if cancel_first: + token.cancel() + timer.fire() + else: + timer.fire() + token.cancel() + timer.fire() + assert terminal == ["cancel" if cancel_first else "success"] + assert finalizers == ["finally"] + assert not token.listeners + assert timer.retired + assert timer.cancel_count == 1 + + asyncio.run(run()) + + +def test_precancelled_work_does_not_execute_or_schedule(): + async def run(): + token, scheduler = CancellationToken(True), Scheduler() + terminal: list[str] = [] + ctx = IAsyncContext.create( + scheduler, + token, + lambda _: terminal.append("success"), + lambda _: terminal.append("error"), + lambda _: terminal.append("cancel"), + ) + async_runtime.sleep(100)(ctx) + assert terminal == ["cancel"] + assert not scheduler.timers + assert not token.listeners + + asyncio.run(run()) + + +def test_cancellation_finalizer_failure_preserves_cancellation(): + async def run(): + token, scheduler = CancellationToken(), Scheduler() + terminal: list[str] = [] + finalizers: list[str] = [] + ctx = IAsyncContext.create( + scheduler, + token, + lambda _: terminal.append("success"), + lambda _: terminal.append("error"), + lambda _: terminal.append("cancel"), + ) + + def finalizer(): + finalizers.append("finally") + raise ValueError("finalizer") + + singleton.TryFinally(async_runtime.sleep(100), finalizer)(ctx) + token.cancel() + scheduler.timers[0].fire() + assert terminal == ["cancel"] + assert finalizers == ["finally"] + assert not token.listeners + + asyncio.run(run()) + + def test_concurrent_registration_and_cancellation(): token = CancellationToken() gate = Barrier(17) @@ -191,3 +300,195 @@ async def run() -> None: event.Trigger(42) assert calls == ["cancel"] assert not event.delegates + + +def test_idle_mailbox_cancellation_requires_an_event(): + # Passing a token alone does not wake Python Receive. A post triggers its cancellation check. + async def run(): + token = CancellationToken() + terminal: list[str] = [] + mailbox = MailboxProcessor(lambda _: singleton.Zero(), token) + ctx = IAsyncContext.create( + Trampoline(), + token, + lambda _: terminal.append("success"), + lambda _: terminal.append("error"), + lambda _: terminal.append("cancel"), + ) + mailbox.receive()(ctx) + token.cancel() + assert terminal == [] + assert mailbox.continuation is not None + mailbox.post(42) + assert terminal == ["cancel"] + assert mailbox.continuation is None + assert not token.listeners + + asyncio.run(run()) + + +def test_foreign_thread_sleep_cancellation_runs_on_owner_loop(): + async def run(): + token = CancellationToken() + loop = asyncio.get_running_loop() + finished = loop.create_future() + terminal: list[str] = [] + + def cancelled(_): + assert asyncio.get_running_loop() is loop + terminal.append("cancel") + finished.set_result(None) + + ctx = IAsyncContext.create( + Trampoline(), token, lambda _: terminal.append("success"), lambda _: terminal.append("error"), cancelled + ) + async_runtime.sleep(100000)(ctx) + thread = Thread(target=token.cancel) + thread.start() + await asyncio.wait_for(finished, 5) + thread.join(5) + assert terminal == ["cancel"] + assert not token.listeners + + asyncio.run(run()) + + +@pytest.mark.parametrize("winner", ["reply", "cancel", "timeout"]) +def test_concurrent_reply_cancel_timeout_settlement(winner): + async def run(): + token, scheduler = CancellationToken(), Scheduler() + key = object() + gate = Barrier(4) + winner_done = Event() + outcome: list[str] = [] + terminal: list[str] = [] + finalizers: list[str] = [] + loop = asyncio.get_running_loop() + finished = loop.create_future() + ctx = IAsyncContext.create( + scheduler, + token, + lambda _: terminal.append("success"), + lambda _: terminal.append("error"), + lambda _: terminal.append("cancel"), + ) + + def deadline(): + def callback(): + finalizers.append("finally") + choose("timeout") + + return singleton.TryFinally(async_runtime.sleep(100), callback) + + def choose(tag): + def settle(): + if outcome: + return + outcome.append(tag) + if tag == "reply": + # The winning reply retires its deadline before admitting competing signals. + token.cancel() + winner_done.set() + loop.call_soon_threadsafe(finished.set_result, None) + + lock(key, settle) + + registration = token.register(lambda: choose("cancel")) + deadline()(ctx) + timer = scheduler.timers[0] + + def compete(tag): + gate.wait(5) + if tag != winner: + assert winner_done.wait(5) + if tag == "reply": + choose("reply") + elif tag == "cancel": + token.cancel() + else: + loop.call_soon_threadsafe(timer.fire) + + threads = [Thread(target=compete, args=(tag,)) for tag in ("reply", "cancel", "timeout")] + for thread in threads: + thread.start() + gate.wait(5) + await asyncio.wait_for(finished, 5) + for thread in threads: + thread.join(5) + assert not thread.is_alive() + registration.Dispose() + token.cancel() + # Retire the deadline after any winner. Cancellation's finalizer must not settle again. + timer.fire() + await asyncio.sleep(0) + assert outcome == [winner] + assert terminal == (["success"] if winner == "timeout" else ["cancel"]) + assert finalizers == ["finally"] + assert not token.listeners + assert timer.cancel_count == 1 + + asyncio.run(run()) + + +@pytest.mark.parametrize("cancel_during_install", [True, False]) +def test_sleep_installation_releases_resources(cancel_during_install): + async def run(): + token = CancellationToken() + terminal: list[str] = [] + + class InstallingScheduler(Scheduler): + def run_later(self, action: Callable[[], None], due_time: float = 0.0): + if not cancel_during_install: + raise ValueError("scheduler") + timer = super().run_later(action, due_time) + token.cancel() + return timer + + scheduler = InstallingScheduler() + ctx = IAsyncContext.create( + scheduler, + token, + lambda _: terminal.append("success"), + lambda _: terminal.append("error"), + lambda _: terminal.append("cancel"), + ) + async_runtime.sleep(100)(ctx) + assert terminal == ["cancel" if cancel_during_install else "error"] + assert not token.listeners + if cancel_during_install: + timer = scheduler.timers[0] + timer.fire() + assert timer.cancel_count == 1 + assert terminal == ["cancel"] + + asyncio.run(run()) + + +@pytest.mark.parametrize("body_fails", [True, False]) +def test_finalizer_failure_overrides_success_or_error_once(body_fails): + async def run(): + ctx = IAsyncContext.create( + Trampoline(), + CancellationToken(), + lambda _: terminal.append("success"), + lambda error: terminal.append(str(error)), + lambda _: terminal.append("cancel"), + ) + terminal: list[str] = [] + finalizers: list[str] = [] + + def body(context): + if body_fails: + context.on_error(ValueError("body")) + else: + context.on_success(42) + + def finalizer(): + finalizers.append("finally") + raise ValueError("finalizer") + + singleton.TryFinally(body, finalizer)(ctx) + assert terminal == ["finalizer"] + assert finalizers == ["finally"] + + asyncio.run(run()) From 2683c3522d9024ac4db73f67ef6e8fb96e5b2d2c Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Sat, 3 Oct 2026 22:41:47 +0200 Subject: [PATCH 2/4] test(js/python): run async lifecycle regressions from F# suites --- src/Fable.Build/Test/JavaScript.fs | 2 - tests/Js/Main/AsyncTests.fs | 120 ++++++++++ tests/Js/Main/js/async-runtime.js | 36 +++ tests/Js/Runtime/async-cancellation.mjs | 68 ------ tests/Python/TestAsync.fs | 189 +++++++++++++++ tests/Python/py/async_runtime.py | 240 +++++++++++++++++++ tests/Python/test_async_runtime.py | 303 +----------------------- 7 files changed, 586 insertions(+), 372 deletions(-) create mode 100644 tests/Js/Main/js/async-runtime.js delete mode 100644 tests/Js/Runtime/async-cancellation.mjs create mode 100644 tests/Python/py/async_runtime.py diff --git a/src/Fable.Build/Test/JavaScript.fs b/src/Fable.Build/Test/JavaScript.fs index e1440fc548..4b978a26ad 100644 --- a/src/Fable.Build/Test/JavaScript.fs +++ b/src/Fable.Build/Test/JavaScript.fs @@ -160,8 +160,6 @@ let private handleMainTests (isWatch: bool) (noDotnet: bool) = // Test the Main tests against JavaScript runMainTests JsDate false noDotnet - Command.Run("node", "tests/Js/Runtime/async-cancellation.mjs") - // Re-run them with the Temporal date/time representation enabled runMainTests Temporal false noDotnet diff --git a/tests/Js/Main/AsyncTests.fs b/tests/Js/Main/AsyncTests.fs index 448326ad15..1348334f7c 100644 --- a/tests/Js/Main/AsyncTests.fs +++ b/tests/Js/Main/AsyncTests.fs @@ -13,6 +13,18 @@ type private Deferred = [] static member handlerCount(event: Event): int = jsNative + +module private Runtime = + open Fable.Core.JsInterop + + type Timer = + abstract retired: int + abstract invalidHandles: int + abstract fire: unit -> unit + + let withTimers (action: Timer -> 'T): 'T = importMember "./js/async-runtime.js" + let listenerCount (token: System.Threading.CancellationToken): int = importMember "./js/async-runtime.js" + let cancellationErrorCount (error: exn): int = importMember "./js/async-runtime.js" #endif type DisposableAction(f) = @@ -108,6 +120,114 @@ let tests = unitDeadline (fun () -> calls <- calls + 1) 2 equal 2 calls + testCase "Finalizer failure overrides success once" <| fun () -> + let mutable finalizers = 0 + let terminal = ResizeArray() + let work = async { + try return! successWork + finally + finalizers <- finalizers + 1 + failwith "finalizer" + } + Async.StartWithContinuations(work, (fun _ -> terminal.Add "success"), (fun error -> terminal.Add error.Message), (fun _ -> terminal.Add "cancel")) + equal [| "finalizer" |] (terminal.ToArray()) + equal 1 finalizers + + testCase "Finalizer failure overrides body failure once" <| fun () -> + let mutable finalizers = 0 + let terminal = ResizeArray() + let work = async { + try return! errorWork + finally + finalizers <- finalizers + 1 + failwith "finalizer" + } + Async.StartWithContinuations(work, (fun _ -> terminal.Add "success"), (fun error -> terminal.Add error.Message), (fun _ -> terminal.Add "cancel")) + equal [| "finalizer" |] (terminal.ToArray()) + equal 1 finalizers + + testCase "Finalizer failure preserves cancellation once" <| fun () -> + let mutable finalizers = 0 + let terminal = ResizeArray() + let work = async { + try return! cancelWork + finally + finalizers <- finalizers + 1 + failwith "finalizer" + } + Async.StartWithContinuations(work, (fun _ -> terminal.Add "success"), (fun _ -> terminal.Add "error"), (fun _ -> terminal.Add "cancel")) + equal [| "cancel" |] (terminal.ToArray()) + equal 1 finalizers + +#if FABLE_COMPILER + testCase "Cancellation registrations dispose pending callbacks and invoke late callbacks" <| fun () -> + use source = new System.Threading.CancellationTokenSource() + let calls = ResizeArray() + let first: IDisposable = source.Token.Register(fun () -> calls.Add 1) + let mutable second: IDisposable = null + second <- source.Token.Register(fun () -> + second.Dispose() + first.Dispose() + calls.Add 2) + source.Cancel() + let late: IDisposable = source.Token.Register((fun state -> calls.Add (unbox state)), box 0) + late.Dispose() + second.Dispose() + equal [| 2; 0 |] (calls.ToArray()) + equal 0 (Runtime.listenerCount source.Token) + + testCase "Cancellation callback failure preserves remaining listeners" <| fun () -> + use source = new System.Threading.CancellationTokenSource() + let mutable remaining = 0 + let mutable errors = -1 + source.Token.Register(fun () -> remaining <- remaining + 1) |> ignore + source.Token.Register(fun () -> failwith "callback") |> ignore + try source.Cancel() + with error -> errors <- Runtime.cancellationErrorCount error + equal 1 errors + equal 1 remaining + equal 0 (Runtime.listenerCount source.Token) + + testCase "Sleep settles once and retires its timer and listener" <| fun () -> + for cancelFirst in [ true; false ] do + for throws in [ true; false ] do + Runtime.withTimers (fun timer -> + use source = new System.Threading.CancellationTokenSource() + let terminal = ResizeArray() + let mutable finalizers = 0 + let work = async { + try do! Async.Sleep 100 + finally + finalizers <- finalizers + 1 + if throws then failwith "finalizer" + } + Async.StartWithContinuations(work, (fun () -> terminal.Add "success"), (fun _ -> terminal.Add "error"), (fun _ -> terminal.Add "cancel"), source.Token) + if cancelFirst then + source.Cancel() + timer.fire() + else + timer.fire() + source.Cancel() + timer.fire() + let expected = if cancelFirst then "cancel" elif throws then "error" else "success" + equal [| expected |] (terminal.ToArray()) + equal 1 finalizers + equal 0 (Runtime.listenerCount source.Token) + equal 1 timer.retired + equal 0 timer.invalidHandles) + + testCase "Idle Receive is not woken by cancellation and still delivers a post" <| fun () -> + use source = new System.Threading.CancellationTokenSource() + let mailbox = new MailboxProcessor((fun _ -> async { return () }), source.Token) + let terminal = ResizeArray() + Async.StartWithContinuations(mailbox.Receive(), (fun value -> terminal.Add (string value)), raise, (fun _ -> terminal.Add "cancel"), source.Token) + source.Cancel() + equal [||] (terminal.ToArray()) + mailbox.Post 42 + equal [| "42" |] (terminal.ToArray()) + equal 0 (Runtime.listenerCount source.Token) +#endif + testCase "Simple async translates without exception" <| fun () -> async { return () } |> Async.StartImmediate diff --git a/tests/Js/Main/js/async-runtime.js b/tests/Js/Main/js/async-runtime.js new file mode 100644 index 0000000000..34b8099935 --- /dev/null +++ b/tests/Js/Main/js/async-runtime.js @@ -0,0 +1,36 @@ +// Controlled native fixtures; test cases and assertions live in AsyncTests.fs. +export function listenerCount(token) { + return token._listeners.size; +} + +export function cancellationErrorCount(error) { + if (error.constructor.name !== "CancellationCallbackError" || error.name !== "AggregateException") { + return -1; + } + return error.errors.length; +} + +export function withTimers(action) { + const originalSetTimeout = globalThis.setTimeout; + const originalClearTimeout = globalThis.clearTimeout; + let callback; + const timer = { + retired: 0, + invalidHandles: 0, + fire() { + // decision: fires retired callbacks to model a timeout already queued at cancellation + callback(); + }, + }; + globalThis.setTimeout = f => { callback = f; return 123; }; + globalThis.clearTimeout = id => { + if (id !== 123) { timer.invalidHandles++; } + timer.retired++; + }; + try { + return action(timer); + } finally { + globalThis.setTimeout = originalSetTimeout; + globalThis.clearTimeout = originalClearTimeout; + } +} diff --git a/tests/Js/Runtime/async-cancellation.mjs b/tests/Js/Runtime/async-cancellation.mjs deleted file mode 100644 index 169be26ac3..0000000000 --- a/tests/Js/Runtime/async-cancellation.mjs +++ /dev/null @@ -1,68 +0,0 @@ -import assert from "node:assert/strict"; -import { CancellationToken, CancellationCallbackError } from "../../../temp/fable-library-js/AsyncBuilder.js"; -import { sleep, startWithContinuations } from "../../../temp/fable-library-js/Async.js"; -import { singleton } from "../../../temp/fable-library-js/AsyncBuilder.js"; -import { MailboxProcessor, receive, post } from "../../../temp/fable-library-js/MailboxProcessor.js"; - -const token = new CancellationToken(); -const calls = []; -const first = token.register(() => calls.push("first")); -const second = token.register(() => { second.Dispose(); first.Dispose(); calls.push("second"); }); -token.cancel(); -token.register(state => calls.push(state), 0).Dispose(); -second.Dispose(); -assert.deepEqual(calls, ["second", 0]); -assert.equal(token._listeners.size, 0); - -const failing = new CancellationToken(); -let remaining = 0; -failing.register(() => remaining++); -failing.register(() => { throw new Error("callback"); }); -assert.throws(() => failing.cancel(), CancellationCallbackError); -assert.equal(remaining, 1); -assert.equal(failing._listeners.size, 0); - -const originalSetTimeout = globalThis.setTimeout; -const originalClearTimeout = globalThis.clearTimeout; -try { - for (const cancelFirst of [true, false]) { - for (const throws of [true, false]) { - let timeout, retired = 0, finalized = 0; - globalThis.setTimeout = callback => { timeout = callback; return 123; }; - globalThis.clearTimeout = id => { assert.equal(id, 123); retired++; }; - const source = new CancellationToken(); - const terminal = []; - const work = singleton.TryFinally(sleep(100), () => { - finalized++; - if (throws) { throw new Error("finalizer"); } - }); - const finish = tag => () => terminal.push(tag); - startWithContinuations(work, finish("success"), finish("error"), finish("cancel"), source); - if (cancelFirst) { source.cancel(); timeout(); } - else { - timeout(); - source.cancel(); - } - timeout(); - assert.equal(finalized, 1); - assert.deepEqual(terminal, cancelFirst ? ["cancel"] : throws ? ["error"] : ["success"]); - assert.equal(source._listeners.size, 0); - assert.equal(retired, 1); - } - } -} finally { - globalThis.setTimeout = originalSetTimeout; - globalThis.clearTimeout = originalClearTimeout; -} - -// JS Receive retains its success continuation while idle; token cancellation does not wake it. -const source = new CancellationToken(); -const mailbox = new MailboxProcessor(() => singleton.Zero(), source); -const terminal = []; -startWithContinuations(receive(mailbox), x => terminal.push(x), e => { throw e; }, () => terminal.push("cancel"), source); -source.cancel(); -assert.deepEqual(terminal, []); -post(mailbox, 42); -assert.deepEqual(terminal, [42]); -assert.equal(source._listeners.size, 0); -console.log("JS cancellation, controlled timers, and idle Receive regressions passed"); diff --git a/tests/Python/TestAsync.fs b/tests/Python/TestAsync.fs index 425a8c1b72..3e7e53f70e 100644 --- a/tests/Python/TestAsync.fs +++ b/tests/Python/TestAsync.fs @@ -3,6 +3,35 @@ module Fable.Tests.Async open System open Util.Testing +#if FABLE_COMPILER +module private Runtime = + open Fable.Core.PyInterop + + type Observation = + abstract terminal: string array + abstract errors: string array + abstract before: string array + abstract outcome: string array + abstract listeners: int + abstract scheduled: int + abstract cancel_count: int + abstract retired: bool + abstract owner_loop: bool + abstract threads_finished: bool + abstract pending_before: bool + abstract pending_after: bool + + type ObserveSleep = delegate of Async * System.Threading.CancellationToken * string -> Observation + type ObserveMailbox = delegate of Async * MailboxProcessor * System.Threading.CancellationToken -> Observation + type ObserveOwnerLoop = delegate of Async * System.Threading.CancellationToken -> Observation + type ObserveSettlement = delegate of ((unit -> unit) -> Async) * System.Threading.CancellationToken * string -> Observation + + let observeSleep: ObserveSleep = import "observe_sleep" "./py/async_runtime.py" + let observeMailbox: ObserveMailbox = import "observe_mailbox" "./py/async_runtime.py" + let observeOwnerLoop: ObserveOwnerLoop = import "observe_owner_loop" "./py/async_runtime.py" + let observeSettlement: ObserveSettlement = import "observe_settlement" "./py/async_runtime.py" +#endif + type DisposableAction(f) = interface IDisposable with member _.Dispose() = f() @@ -724,3 +753,163 @@ let ``test cooperative cancellation stops at the next async boundary`` () = equal false reached equal 1 finalizers equal "cancel" terminal + + +[] +let ``test finalizer failure overrides success once`` () = + let mutable finalizers = 0 + let terminal = ResizeArray() + let work = async { + try return! successWork + finally + finalizers <- finalizers + 1 + failwith "finalizer" + } + Async.StartWithContinuations(work, (fun _ -> terminal.Add "success"), (fun error -> terminal.Add error.Message), (fun _ -> terminal.Add "cancel")) + equal [| "finalizer" |] (terminal.ToArray()) + equal 1 finalizers + +[] +let ``test finalizer failure overrides body failure once`` () = + let mutable finalizers = 0 + let terminal = ResizeArray() + let work = async { + try return! errorWork + finally + finalizers <- finalizers + 1 + failwith "finalizer" + } + Async.StartWithContinuations(work, (fun _ -> terminal.Add "success"), (fun error -> terminal.Add error.Message), (fun _ -> terminal.Add "cancel")) + equal [| "finalizer" |] (terminal.ToArray()) + equal 1 finalizers + +[] +let ``test finalizer failure preserves cancellation once`` () = + let mutable finalizers = 0 + let terminal = ResizeArray() + let work = async { + try return! cancelWork + finally + finalizers <- finalizers + 1 + failwith "finalizer" + } + Async.StartWithContinuations(work, (fun _ -> terminal.Add "success"), (fun _ -> terminal.Add "error"), (fun _ -> terminal.Add "cancel")) + equal [| "cancel" |] (terminal.ToArray()) + equal 1 finalizers + +#if FABLE_COMPILER +let private checkSleep cancelFirst throws = + use source = new System.Threading.CancellationTokenSource() + let mutable finalizers = 0 + let work = async { + try do! Async.Sleep 100 + finally + finalizers <- finalizers + 1 + if throws then failwith "finalizer" + } + let result = Runtime.observeSleep.Invoke(work, source.Token, if cancelFirst then "cancel" else "timeout") + let expected = if cancelFirst then "cancel" elif throws then "error" else "success" + equal [| expected |] result.terminal + equal 1 finalizers + equal 0 result.listeners + equal 1 result.scheduled + equal true result.retired + equal 1 result.cancel_count + if throws && not cancelFirst then equal [| "finalizer" |] result.errors + else equal [||] result.errors + +[] +let ``test cancelled Sleep ignores stale timers and cleans up`` () = + checkSleep true false + +[] +let ``test completed Sleep ignores cancellation and stale timers`` () = + checkSleep false false + +[] +let ``test cancelled Sleep ignores a throwing finalizer and stale timers`` () = + checkSleep true true + +[] +let ``test precancelled Sleep does not execute or schedule`` () = + use source = new System.Threading.CancellationTokenSource() + let mutable reached = false + let work = async { + reached <- true + do! Async.Sleep 100 + } + let result = Runtime.observeSleep.Invoke(work, source.Token, "precancel") + equal [| "cancel" |] result.terminal + equal false reached + equal 0 result.scheduled + equal 0 result.listeners + +[] +let ``test idle Receive observes cancellation only after a post`` () = + use source = new System.Threading.CancellationTokenSource() + let mailbox = new MailboxProcessor((fun _ -> async { return () }), source.Token) + let result = Runtime.observeMailbox.Invoke(mailbox.Receive(), mailbox, source.Token) + equal [||] result.before + equal true result.pending_before + equal [| "cancel" |] result.terminal + equal false result.pending_after + equal 0 result.listeners + +[] +let ``test Sleep cancellation from another thread runs on the owner loop`` () = + use source = new System.Threading.CancellationTokenSource() + let result = Runtime.observeOwnerLoop.Invoke(Async.Sleep 100000, source.Token) + equal [| "cancel" |] result.terminal + equal true result.owner_loop + equal true result.threads_finished + equal 0 result.listeners + +let private checkSettlement winner = + use source = new System.Threading.CancellationTokenSource() + let mutable finalizers = 0 + let makeWork timeout = async { + try do! Async.Sleep 100 + finally + finalizers <- finalizers + 1 + timeout () + } + let result = Runtime.observeSettlement.Invoke(makeWork, source.Token, winner) + equal [| winner |] result.outcome + equal [| if winner = "timeout" then "success" else "cancel" |] result.terminal + equal 1 finalizers + equal true result.threads_finished + equal 0 result.listeners + equal true result.retired + equal 1 result.cancel_count + +[] +let ``test reply wins concurrent reply cancellation and timeout settlement`` () = + checkSettlement "reply" + +[] +let ``test cancellation wins concurrent reply cancellation and timeout settlement`` () = + checkSettlement "cancel" + +[] +let ``test timeout wins concurrent reply cancellation and timeout settlement`` () = + checkSettlement "timeout" + +[] +let ``test cancellation during Sleep installation releases resources`` () = + use source = new System.Threading.CancellationTokenSource() + let result = Runtime.observeSleep.Invoke(Async.Sleep 100, source.Token, "install_cancel") + equal [| "cancel" |] result.terminal + equal 0 result.listeners + equal 1 result.scheduled + equal true result.retired + equal 1 result.cancel_count + +[] +let ``test failed Sleep installation releases its registration`` () = + use source = new System.Threading.CancellationTokenSource() + let result = Runtime.observeSleep.Invoke(Async.Sleep 100, source.Token, "install_error") + equal [| "error" |] result.terminal + equal [| "scheduler" |] result.errors + equal 0 result.listeners + equal 0 result.scheduled +#endif diff --git a/tests/Python/py/async_runtime.py b/tests/Python/py/async_runtime.py new file mode 100644 index 0000000000..370dd50b23 --- /dev/null +++ b/tests/Python/py/async_runtime.py @@ -0,0 +1,240 @@ +"""Controlled scheduler and thread fixtures for the generated F# async tests.""" + +import asyncio +from collections.abc import Callable +from dataclasses import dataclass, field +from threading import Barrier, Event, Thread + +from fable_library.array_ import Array +from fable_library.async_builder import Async, CancellationToken, IAsyncContext, Trampoline +from fable_library.mailbox_processor import MailboxProcessor +from fable_library.util import lock + + +@dataclass +class Observation: + _terminal: list[str] = field(default_factory=list) + _errors: list[str] = field(default_factory=list) + _before: list[str] = field(default_factory=list) + _outcome: list[str] = field(default_factory=list) + listeners: int = 0 + scheduled: int = 0 + cancel_count: int = 0 + retired: bool = False + owner_loop: bool = False + threads_finished: bool = False + pending_before: bool = False + pending_after: bool = False + + @property + def terminal(self) -> Array[str]: + return Array(self._terminal) + + @property + def errors(self) -> Array[str]: + return Array(self._errors) + + @property + def before(self) -> Array[str]: + return Array(self._before) + + @property + def outcome(self) -> Array[str]: + return Array(self._outcome) + + +class Timer(asyncio.TimerHandle): + def __init__(self, callback: Callable[[], None]): + super().__init__(0, callback, (), asyncio.get_running_loop()) + self.callback = callback + self.retired = False + self.cancel_count = 0 + + def cancel(self): + self.retired = True + self.cancel_count += 1 + + def fire(self): + # decision: fires retired callbacks to model a timeout already queued when cancellation wins + self.callback() + + +class Scheduler(Trampoline): + def __init__(self, token: CancellationToken, mode: str = ""): + super().__init__() + self.token = token + self.mode = mode + self.timers: list[Timer] = [] + + def run_later(self, action: Callable[[], None], due_time: float = 0.0): + if self.mode == "install_error": + raise ValueError("scheduler") + timer = Timer(action) + self.timers.append(timer) + if self.mode == "install_cancel": + self.token.cancel() + return timer + + +def context(scheduler: Trampoline, token: CancellationToken, result: Observation) -> IAsyncContext[None]: + def failed(error: Exception): + result._terminal.append("error") + result._errors.append(str(error)) + + return IAsyncContext.create( + scheduler, + token, + lambda _: result._terminal.append("success"), + failed, + lambda _: result._terminal.append("cancel"), + ) + + +def observe_sleep(work: Async[None], token: CancellationToken, mode: str) -> Observation: + async def run(): + result = Observation() + scheduler = Scheduler(token, mode) + if mode == "precancel": + token.cancel() + work(context(scheduler, token, result)) + if scheduler.timers: + timer = scheduler.timers[0] + if mode == "cancel": + token.cancel() + timer.fire() + elif mode == "timeout": + timer.fire() + token.cancel() + timer.fire() + result.retired = timer.retired + result.cancel_count = timer.cancel_count + result.scheduled = len(scheduler.timers) + result.listeners = len(token.listeners) + return result + + return asyncio.run(run()) + + +def observe_mailbox(work: Async[int], mailbox: MailboxProcessor[int], token: CancellationToken) -> Observation: + async def run(): + result = Observation() + ctx = IAsyncContext.create( + Trampoline(), + token, + lambda value: result._terminal.append(str(value)), + lambda _: result._terminal.append("error"), + lambda _: result._terminal.append("cancel"), + ) + work(ctx) + token.cancel() + result._before = result._terminal.copy() + result.pending_before = mailbox.continuation is not None + mailbox.post(42) + result.pending_after = mailbox.continuation is not None + result.listeners = len(token.listeners) + return result + + return asyncio.run(run()) + + +def observe_owner_loop(work: Async[None], token: CancellationToken) -> Observation: + async def run(): + result = Observation() + loop = asyncio.get_running_loop() + finished = loop.create_future() + + def cancelled(_): + result.owner_loop = asyncio.get_running_loop() is loop + result._terminal.append("cancel") + finished.set_result(None) + + ctx = IAsyncContext.create( + Trampoline(), + token, + lambda _: result._terminal.append("success"), + lambda _: result._terminal.append("error"), + cancelled, + ) + work(ctx) + thread = Thread(target=token.cancel) + thread.start() + try: + await asyncio.wait_for(finished, 5) + finally: + thread.join(5) + result.threads_finished = not thread.is_alive() + result.listeners = len(token.listeners) + return result + + return asyncio.run(run()) + + +def observe_settlement( + make_work: Callable[[Callable[[], None]], Async[None]], + token: CancellationToken, + winner: str, +) -> Observation: + async def run(): + result = Observation() + scheduler = Scheduler(token) + key = object() + gate = Barrier(4) + winner_done = Event() + loop = asyncio.get_running_loop() + finished = loop.create_future() + thread_errors: list[Exception] = [] + + def choose(tag: str): + def settle(): + if result._outcome: + return + result._outcome.append(tag) + if tag == "reply": + # invariant: the winning reply retires its deadline before competitors proceed + token.cancel() + winner_done.set() + loop.call_soon_threadsafe(finished.set_result, None) + + lock(key, settle) + + registration = token.register(lambda: choose("cancel")) + make_work(lambda: choose("timeout"))(context(scheduler, token, result)) + timer = scheduler.timers[0] + + def compete(tag: str): + try: + gate.wait(5) + if tag != winner and not winner_done.wait(5): + raise TimeoutError("settlement winner did not signal") + if tag == "reply": + choose("reply") + elif tag == "cancel": + token.cancel() + else: + loop.call_soon_threadsafe(timer.fire) + except Exception as error: + thread_errors.append(error) + + threads = [Thread(target=compete, args=(tag,)) for tag in ("reply", "cancel", "timeout")] + for thread in threads: + thread.start() + gate.wait(5) + try: + await asyncio.wait_for(finished, 5) + finally: + for thread in threads: + thread.join(5) + registration.Dispose() + token.cancel() + if thread_errors: + raise ExceptionGroup("fixture worker failed", thread_errors) + # Exercise a stale callback after every winner, including cancellation's finalizer. + timer.fire() + await asyncio.sleep(0) + result.threads_finished = all(not thread.is_alive() for thread in threads) + result.listeners = len(token.listeners) + result.cancel_count = timer.cancel_count + result.retired = timer.retired + return result + + return asyncio.run(run()) diff --git a/tests/Python/test_async_runtime.py b/tests/Python/test_async_runtime.py index 905047a8a9..aba116e32e 100644 --- a/tests/Python/test_async_runtime.py +++ b/tests/Python/test_async_runtime.py @@ -1,14 +1,10 @@ import asyncio -from collections.abc import Callable from threading import Barrier, Condition, Event, Thread import pytest -from fable_library import async_ as async_runtime from fable_library.async_ import await_event, sleep, start_with_continuations -from fable_library.async_builder import CancellationCallbackError, CancellationToken, IAsyncContext, Trampoline, singleton +from fable_library.async_builder import CancellationCallbackError, CancellationToken, IAsyncContext, Trampoline from fable_library.event import Event as FableEvent -from fable_library.mailbox_processor import MailboxProcessor -from fable_library.util import lock def test_registration_handle_and_disposal(): @@ -131,111 +127,6 @@ def dispose(): assert not token.listeners -class Timer(asyncio.TimerHandle): - def __init__(self, callback: Callable[[], None]): - super().__init__(0, callback, (), asyncio.get_running_loop()) - self.callback = callback - self.retired = False - self.cancel_count = 0 - - def cancel(self): - self.retired = True - self.cancel_count += 1 - - def fire(self): - # Deliberately fire even a retired callback to model an already queued timeout. - self.callback() - - -class Scheduler(Trampoline): - def __init__(self): - super().__init__() - self.timers: list[Timer] = [] - - def run_later(self, action: Callable[[], None], due_time: float = 0.0): - timer = Timer(action) - self.timers.append(timer) - return timer - - -@pytest.mark.parametrize("cancel_first", [True, False]) -def test_sleep_settles_once_and_cleans_up(cancel_first): - async def run(): - token, scheduler = CancellationToken(), Scheduler() - terminal: list[str] = [] - finalizers: list[str] = [] - ctx = IAsyncContext.create( - scheduler, - token, - lambda _: terminal.append("success"), - lambda _: terminal.append("error"), - lambda _: terminal.append("cancel"), - ) - work = singleton.TryFinally(async_runtime.sleep(100), lambda: finalizers.append("finally")) - work(ctx) - timer = scheduler.timers[0] - if cancel_first: - token.cancel() - timer.fire() - else: - timer.fire() - token.cancel() - timer.fire() - assert terminal == ["cancel" if cancel_first else "success"] - assert finalizers == ["finally"] - assert not token.listeners - assert timer.retired - assert timer.cancel_count == 1 - - asyncio.run(run()) - - -def test_precancelled_work_does_not_execute_or_schedule(): - async def run(): - token, scheduler = CancellationToken(True), Scheduler() - terminal: list[str] = [] - ctx = IAsyncContext.create( - scheduler, - token, - lambda _: terminal.append("success"), - lambda _: terminal.append("error"), - lambda _: terminal.append("cancel"), - ) - async_runtime.sleep(100)(ctx) - assert terminal == ["cancel"] - assert not scheduler.timers - assert not token.listeners - - asyncio.run(run()) - - -def test_cancellation_finalizer_failure_preserves_cancellation(): - async def run(): - token, scheduler = CancellationToken(), Scheduler() - terminal: list[str] = [] - finalizers: list[str] = [] - ctx = IAsyncContext.create( - scheduler, - token, - lambda _: terminal.append("success"), - lambda _: terminal.append("error"), - lambda _: terminal.append("cancel"), - ) - - def finalizer(): - finalizers.append("finally") - raise ValueError("finalizer") - - singleton.TryFinally(async_runtime.sleep(100), finalizer)(ctx) - token.cancel() - scheduler.timers[0].fire() - assert terminal == ["cancel"] - assert finalizers == ["finally"] - assert not token.listeners - - asyncio.run(run()) - - def test_concurrent_registration_and_cancellation(): token = CancellationToken() gate = Barrier(17) @@ -300,195 +191,3 @@ async def run() -> None: event.Trigger(42) assert calls == ["cancel"] assert not event.delegates - - -def test_idle_mailbox_cancellation_requires_an_event(): - # Passing a token alone does not wake Python Receive. A post triggers its cancellation check. - async def run(): - token = CancellationToken() - terminal: list[str] = [] - mailbox = MailboxProcessor(lambda _: singleton.Zero(), token) - ctx = IAsyncContext.create( - Trampoline(), - token, - lambda _: terminal.append("success"), - lambda _: terminal.append("error"), - lambda _: terminal.append("cancel"), - ) - mailbox.receive()(ctx) - token.cancel() - assert terminal == [] - assert mailbox.continuation is not None - mailbox.post(42) - assert terminal == ["cancel"] - assert mailbox.continuation is None - assert not token.listeners - - asyncio.run(run()) - - -def test_foreign_thread_sleep_cancellation_runs_on_owner_loop(): - async def run(): - token = CancellationToken() - loop = asyncio.get_running_loop() - finished = loop.create_future() - terminal: list[str] = [] - - def cancelled(_): - assert asyncio.get_running_loop() is loop - terminal.append("cancel") - finished.set_result(None) - - ctx = IAsyncContext.create( - Trampoline(), token, lambda _: terminal.append("success"), lambda _: terminal.append("error"), cancelled - ) - async_runtime.sleep(100000)(ctx) - thread = Thread(target=token.cancel) - thread.start() - await asyncio.wait_for(finished, 5) - thread.join(5) - assert terminal == ["cancel"] - assert not token.listeners - - asyncio.run(run()) - - -@pytest.mark.parametrize("winner", ["reply", "cancel", "timeout"]) -def test_concurrent_reply_cancel_timeout_settlement(winner): - async def run(): - token, scheduler = CancellationToken(), Scheduler() - key = object() - gate = Barrier(4) - winner_done = Event() - outcome: list[str] = [] - terminal: list[str] = [] - finalizers: list[str] = [] - loop = asyncio.get_running_loop() - finished = loop.create_future() - ctx = IAsyncContext.create( - scheduler, - token, - lambda _: terminal.append("success"), - lambda _: terminal.append("error"), - lambda _: terminal.append("cancel"), - ) - - def deadline(): - def callback(): - finalizers.append("finally") - choose("timeout") - - return singleton.TryFinally(async_runtime.sleep(100), callback) - - def choose(tag): - def settle(): - if outcome: - return - outcome.append(tag) - if tag == "reply": - # The winning reply retires its deadline before admitting competing signals. - token.cancel() - winner_done.set() - loop.call_soon_threadsafe(finished.set_result, None) - - lock(key, settle) - - registration = token.register(lambda: choose("cancel")) - deadline()(ctx) - timer = scheduler.timers[0] - - def compete(tag): - gate.wait(5) - if tag != winner: - assert winner_done.wait(5) - if tag == "reply": - choose("reply") - elif tag == "cancel": - token.cancel() - else: - loop.call_soon_threadsafe(timer.fire) - - threads = [Thread(target=compete, args=(tag,)) for tag in ("reply", "cancel", "timeout")] - for thread in threads: - thread.start() - gate.wait(5) - await asyncio.wait_for(finished, 5) - for thread in threads: - thread.join(5) - assert not thread.is_alive() - registration.Dispose() - token.cancel() - # Retire the deadline after any winner. Cancellation's finalizer must not settle again. - timer.fire() - await asyncio.sleep(0) - assert outcome == [winner] - assert terminal == (["success"] if winner == "timeout" else ["cancel"]) - assert finalizers == ["finally"] - assert not token.listeners - assert timer.cancel_count == 1 - - asyncio.run(run()) - - -@pytest.mark.parametrize("cancel_during_install", [True, False]) -def test_sleep_installation_releases_resources(cancel_during_install): - async def run(): - token = CancellationToken() - terminal: list[str] = [] - - class InstallingScheduler(Scheduler): - def run_later(self, action: Callable[[], None], due_time: float = 0.0): - if not cancel_during_install: - raise ValueError("scheduler") - timer = super().run_later(action, due_time) - token.cancel() - return timer - - scheduler = InstallingScheduler() - ctx = IAsyncContext.create( - scheduler, - token, - lambda _: terminal.append("success"), - lambda _: terminal.append("error"), - lambda _: terminal.append("cancel"), - ) - async_runtime.sleep(100)(ctx) - assert terminal == ["cancel" if cancel_during_install else "error"] - assert not token.listeners - if cancel_during_install: - timer = scheduler.timers[0] - timer.fire() - assert timer.cancel_count == 1 - assert terminal == ["cancel"] - - asyncio.run(run()) - - -@pytest.mark.parametrize("body_fails", [True, False]) -def test_finalizer_failure_overrides_success_or_error_once(body_fails): - async def run(): - ctx = IAsyncContext.create( - Trampoline(), - CancellationToken(), - lambda _: terminal.append("success"), - lambda error: terminal.append(str(error)), - lambda _: terminal.append("cancel"), - ) - terminal: list[str] = [] - finalizers: list[str] = [] - - def body(context): - if body_fails: - context.on_error(ValueError("body")) - else: - context.on_success(42) - - def finalizer(): - finalizers.append("finally") - raise ValueError("finalizer") - - singleton.TryFinally(body, finalizer)(ctx) - assert terminal == ["finalizer"] - assert finalizers == ["finally"] - - asyncio.run(run()) From bb67e0441ceb48b8bc76a51a0777906a993fb60f Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Sat, 10 Oct 2026 18:47:00 +0200 Subject: [PATCH 3/4] test(typescript): use a concrete timer fixture binding --- tests/Js/Main/AsyncTests.fs | 2 +- tests/Js/Main/js/async-runtime.js | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/Js/Main/AsyncTests.fs b/tests/Js/Main/AsyncTests.fs index 1348334f7c..cc97ccb24d 100644 --- a/tests/Js/Main/AsyncTests.fs +++ b/tests/Js/Main/AsyncTests.fs @@ -22,7 +22,7 @@ module private Runtime = abstract invalidHandles: int abstract fire: unit -> unit - let withTimers (action: Timer -> 'T): 'T = importMember "./js/async-runtime.js" + let withTimers (action: Timer -> unit): unit = importMember "./js/async-runtime.js" let listenerCount (token: System.Threading.CancellationToken): int = importMember "./js/async-runtime.js" let cancellationErrorCount (error: exn): int = importMember "./js/async-runtime.js" #endif diff --git a/tests/Js/Main/js/async-runtime.js b/tests/Js/Main/js/async-runtime.js index 34b8099935..871ac9224d 100644 --- a/tests/Js/Main/js/async-runtime.js +++ b/tests/Js/Main/js/async-runtime.js @@ -4,7 +4,7 @@ export function listenerCount(token) { } export function cancellationErrorCount(error) { - if (error.constructor.name !== "CancellationCallbackError" || error.name !== "AggregateException") { + if (error.constructor.name !== "CancellationCallbackError" || !Array.isArray(error.errors)) { return -1; } return error.errors.length; From a3cc801229f28ee97df8aaff4f9b311d653f0a3e Mon Sep 17 00:00:00 2001 From: Mangel Maxime Date: Mon, 12 Oct 2026 15:20:13 +0200 Subject: [PATCH 4/4] fix(js/ts/python): cancel idle mailbox Receive on its token --- src/fable-library-py/fable_library/async_.py | 38 ++++++------ .../fable_library/mailbox_processor.py | 61 +++++++++++-------- src/fable-library-ts/MailboxProcessor.ts | 22 +++++-- tests/Js/Main/AsyncTests.fs | 54 ++++++++++++++-- tests/Python/TestAsync.fs | 42 ++++++++++++- tests/Python/py/async_runtime.py | 8 +++ tests/Python/test_async_runtime.py | 25 ++++++++ 7 files changed, 195 insertions(+), 55 deletions(-) diff --git a/src/fable-library-py/fable_library/async_.py b/src/fable-library-py/fable_library/async_.py index 3f515fc847..ff1a7a03f6 100644 --- a/src/fable-library-py/fable_library/async_.py +++ b/src/fable-library-py/fable_library/async_.py @@ -80,6 +80,26 @@ def is_cancellation_requested(token: CancellationToken) -> bool: return token.is_cancelled +def call_on_owner_loop(loop: asyncio.AbstractEventLoop, action: Callable[[], None]) -> None: + # decision: foreign-thread cancellation settles on the owning loop while it is running + if loop.is_closed() or not loop.is_running(): + action() + return + try: + current_loop = asyncio.get_running_loop() + except RuntimeError: + current_loop = None + if current_loop is loop: + action() + return + try: + loop.call_soon_threadsafe(action) + except RuntimeError: + if not loop.is_closed(): + raise + action() + + def sleep(milliseconds_duetime: int | TimeSpan) -> Async[None]: def cont(ctx: IAsyncContext[None]) -> None: loop = asyncio.get_running_loop() @@ -107,23 +127,7 @@ def settle(cancelled: bool) -> None: ctx.on_success(None) def cancel() -> None: - # decision: foreign-thread cancellation settles on the owning loop while it is alive - if loop.is_closed(): - settle(True) - return - try: - current_loop = asyncio.get_running_loop() - except RuntimeError: - current_loop = None - if current_loop is loop: - settle(True) - else: - try: - loop.call_soon_threadsafe(settle, True) - except RuntimeError: - if not loop.is_closed(): - raise - settle(True) + call_on_owner_loop(loop, lambda: settle(True)) # invariant: installation and settlement cannot interleave before both resources are owned try: diff --git a/src/fable-library-py/fable_library/mailbox_processor.py b/src/fable-library-py/fable_library/mailbox_processor.py index 15ef175e32..cbdbe5f020 100644 --- a/src/fable-library-py/fable_library/mailbox_processor.py +++ b/src/fable-library-py/fable_library/mailbox_processor.py @@ -1,16 +1,19 @@ from __future__ import annotations +import asyncio from collections.abc import Callable from queue import SimpleQueue from threading import RLock from typing import Any -from .async_ import from_continuations, start_immediate +from .async_ import call_on_owner_loop, from_continuations, start_immediate from .async_builder import ( Async, CancellationToken, Continuations, + IAsyncContext, OperationCanceledError, + protected_cont, ) @@ -101,34 +104,44 @@ def receive(self) -> Async[Msg]: Returns: An asynchronous computation which will consume the first message in arrival order. No thread is blocked while - waiting for further messages. Cancellation of the token alone - does not wake an idle receive; a post triggers the cancellation check. + waiting for further messages. Cancelling the token of the + receiving computation cancels a pending receive. """ - def callback(conts: Continuations[Any]): - if self.continuation: - raise Exception("Receive can only be called once!") - - self.continuation = conts - + def cont(ctx: IAsyncContext[Msg]) -> None: + loop = asyncio.get_running_loop() + token_id: int | None = None + + def on_message(msg: Msg) -> None: + if token_id is not None: + ctx.cancel_token.remove_listener(token_id) + ctx.on_success(msg) + + continuation: Continuations[Any] = (on_message, ctx.on_error, ctx.on_cancel) + + def on_cancel() -> None: + with self.lock: + if self.continuation is not continuation: + return + self.continuation = None + call_on_owner_loop(loop, lambda: ctx.on_cancel(OperationCanceledError())) + + with self.lock: + if self.continuation: + raise Exception("Receive can only be called once!") + self.continuation = continuation + + token_id = ctx.cancel_token.add_listener(on_cancel) + with self.lock: + if self.continuation is not continuation: + return self.__process_events() - return from_continuations(callback) + return protected_cont(cont) def __process_events(self) -> None: - if not self.continuation: - return - - # Cancellation of async workflows is more tricky in Python than - # with F# so we check the cancellation token for each process. - if self.token.is_cancellation_requested: - self.continuation, cont = None, self.continuation - if cont: - cont[2](OperationCanceledError("Mailbox was cancelled")) - return - with self.lock: - if self.messages.empty(): + if not self.continuation or self.messages.empty(): return msg = self.messages.get() self.continuation, cont = None, self.continuation @@ -143,7 +156,7 @@ def start( cancellation_token: CancellationToken | None = None, ) -> MailboxProcessor[Msg]: mbox: MailboxProcessor[Msg] = MailboxProcessor(body, cancellation_token) - start_immediate(body(mbox)) + start_immediate(body(mbox), mbox.token) return mbox @@ -163,7 +176,7 @@ def post_and_async_reply[Msg, Reply]( def start_instance(mbox: MailboxProcessor[Any]) -> None: body = mbox.body(mbox) - return start_immediate(body) + return start_immediate(body, mbox.token) def start[Msg]( diff --git a/src/fable-library-ts/MailboxProcessor.ts b/src/fable-library-ts/MailboxProcessor.ts index 5ff3874435..b1771bf70a 100644 --- a/src/fable-library-ts/MailboxProcessor.ts +++ b/src/fable-library-ts/MailboxProcessor.ts @@ -1,7 +1,7 @@ import { defaultCancellationToken } from "./Async.ts"; import { fromContinuations } from "./Async.ts"; import { startImmediate } from "./Async.ts"; -import { Async } from "./AsyncBuilder.ts"; +import { Async, IAsyncContext, OperationCanceledException, protectedCont } from "./AsyncBuilder.ts"; import { Continuation, Continuations } from "./AsyncBuilder.ts"; import { CancellationToken } from "./AsyncBuilder.ts"; import { Exception } from "./Util.ts"; @@ -79,14 +79,26 @@ export function startInstance($this: MailboxProcessor) { startImmediate($this.body($this), $this.cancellationToken); } -// Token cancellation does not wake an idle Receive; delivery is driven by posts. export function receive($this: MailboxProcessor) { - return fromContinuations((conts: Continuations) => { + return protectedCont((ctx: IAsyncContext) => { if ($this.continuation) { throw new Exception("Receive can only be called once!"); } - $this.continuation = conts[0]; - __processEvents($this); + let tokenId: number | undefined; + const continuation = (message: Msg) => { + if (tokenId !== undefined) { ctx.cancelToken.removeListener(tokenId); } + ctx.onSuccess(message); + }; + $this.continuation = continuation; + tokenId = ctx.cancelToken.addListener(() => { + if ($this.continuation === continuation) { + delete $this.continuation; + ctx.onCancel(new OperationCanceledException()); + } + }); + if ($this.continuation === continuation) { + __processEvents($this); + } }); } diff --git a/tests/Js/Main/AsyncTests.fs b/tests/Js/Main/AsyncTests.fs index cc97ccb24d..362207952e 100644 --- a/tests/Js/Main/AsyncTests.fs +++ b/tests/Js/Main/AsyncTests.fs @@ -159,7 +159,6 @@ let tests = equal [| "cancel" |] (terminal.ToArray()) equal 1 finalizers -#if FABLE_COMPILER testCase "Cancellation registrations dispose pending callbacks and invoke late callbacks" <| fun () -> use source = new System.Threading.CancellationTokenSource() let calls = ResizeArray() @@ -174,7 +173,9 @@ let tests = late.Dispose() second.Dispose() equal [| 2; 0 |] (calls.ToArray()) +#if FABLE_COMPILER equal 0 (Runtime.listenerCount source.Token) +#endif testCase "Cancellation callback failure preserves remaining listeners" <| fun () -> use source = new System.Threading.CancellationTokenSource() @@ -183,11 +184,19 @@ let tests = source.Token.Register(fun () -> remaining <- remaining + 1) |> ignore source.Token.Register(fun () -> failwith "callback") |> ignore try source.Cancel() - with error -> errors <- Runtime.cancellationErrorCount error + with error -> +#if FABLE_COMPILER + errors <- Runtime.cancellationErrorCount error +#else + errors <- (error :?> AggregateException).InnerExceptions.Count +#endif equal 1 errors equal 1 remaining +#if FABLE_COMPILER equal 0 (Runtime.listenerCount source.Token) +#endif +#if FABLE_COMPILER testCase "Sleep settles once and retires its timer and listener" <| fun () -> for cancelFirst in [ true; false ] do for throws in [ true; false ] do @@ -216,18 +225,51 @@ let tests = equal 1 timer.retired equal 0 timer.invalidHandles) - testCase "Idle Receive is not woken by cancellation and still delivers a post" <| fun () -> + testCase "Idle Receive is cancelled by its token and keeps later posts queued" <| fun () -> use source = new System.Threading.CancellationTokenSource() let mailbox = new MailboxProcessor((fun _ -> async { return () }), source.Token) let terminal = ResizeArray() Async.StartWithContinuations(mailbox.Receive(), (fun value -> terminal.Add (string value)), raise, (fun _ -> terminal.Add "cancel"), source.Token) source.Cancel() - equal [||] (terminal.ToArray()) - mailbox.Post 42 - equal [| "42" |] (terminal.ToArray()) + equal [| "cancel" |] (terminal.ToArray()) equal 0 (Runtime.listenerCount source.Token) + mailbox.Post 42 + equal [| "cancel" |] (terminal.ToArray()) + Async.StartWithContinuations(mailbox.Receive(), (fun value -> terminal.Add $"next {value}"), raise, (fun _ -> terminal.Add "next cancel")) + equal [| "cancel"; "next 42" |] (terminal.ToArray()) #endif + testCaseAsync "Cancelling a waiting Receive reports cancel and keeps later posts queued" <| fun () -> async { + use source = new System.Threading.CancellationTokenSource() + let mailbox = new MailboxProcessor((fun _ -> async { return () }), source.Token) + let! outcome = + Async.FromContinuations(fun (ok, _, _) -> + Async.StartWithContinuations(mailbox.Receive(), (fun value -> ok $"value {value}"), (fun _ -> ok "error"), (fun _ -> ok "cancel"), source.Token) + source.Cancel()) + equal "cancel" outcome + mailbox.Post 42 + let! next = mailbox.Receive() + equal 42 next + } + + testCaseAsync "Cancelling the mailbox token cancels a body waiting in Receive" <| fun () -> async { + use source = new System.Threading.CancellationTokenSource() + let! outcome = + Async.FromContinuations(fun (ok, _, _) -> + MailboxProcessor.Start((fun inbox -> async { + let received = ref None + try + source.CancelAfter 50 + let! message = inbox.Receive() + received.Value <- Some message + finally + match received.Value with + | Some message -> ok $"received {message}" + | None -> ok "cancelled" }), source.Token) + |> ignore) + equal "cancelled" outcome + } + testCase "Simple async translates without exception" <| fun () -> async { return () } |> Async.StartImmediate diff --git a/tests/Python/TestAsync.fs b/tests/Python/TestAsync.fs index 3e7e53f70e..3c8563ae9e 100644 --- a/tests/Python/TestAsync.fs +++ b/tests/Python/TestAsync.fs @@ -845,15 +845,16 @@ let ``test precancelled Sleep does not execute or schedule`` () = equal 0 result.listeners [] -let ``test idle Receive observes cancellation only after a post`` () = +let ``test idle Receive is cancelled by its token and keeps later posts queued`` () = use source = new System.Threading.CancellationTokenSource() let mailbox = new MailboxProcessor((fun _ -> async { return () }), source.Token) let result = Runtime.observeMailbox.Invoke(mailbox.Receive(), mailbox, source.Token) - equal [||] result.before - equal true result.pending_before + equal [| "cancel" |] result.before + equal false result.pending_before equal [| "cancel" |] result.terminal equal false result.pending_after equal 0 result.listeners + equal [| "42" |] result.outcome [] let ``test Sleep cancellation from another thread runs on the owner loop`` () = @@ -913,3 +914,38 @@ let ``test failed Sleep installation releases its registration`` () = equal 0 result.listeners equal 0 result.scheduled #endif + +[] +let ``test cancelling a waiting Receive reports cancel and keeps later posts queued`` () = + async { + use source = new System.Threading.CancellationTokenSource() + let mailbox = new MailboxProcessor((fun _ -> async { return () }), source.Token) + let! outcome = + Async.FromContinuations(fun (ok, _, _) -> + Async.StartWithContinuations(mailbox.Receive(), (fun value -> ok $"value {value}"), (fun _ -> ok "error"), (fun _ -> ok "cancel"), source.Token) + source.Cancel()) + equal "cancel" outcome + mailbox.Post 42 + let! next = mailbox.Receive() + equal 42 next + } |> Async.RunSynchronously + +[] +let ``test cancelling the mailbox token cancels a body waiting in Receive`` () = + async { + use source = new System.Threading.CancellationTokenSource() + let! outcome = + Async.FromContinuations(fun (ok, _, _) -> + MailboxProcessor.Start((fun inbox -> async { + let received = ref None + try + source.CancelAfter 50 + let! message = inbox.Receive() + received.Value <- Some message + finally + match received.Value with + | Some message -> ok $"received {message}" + | None -> ok "cancelled" }), source.Token) + |> ignore) + equal "cancelled" outcome + } |> Async.RunSynchronously diff --git a/tests/Python/py/async_runtime.py b/tests/Python/py/async_runtime.py index 370dd50b23..52dd18d55d 100644 --- a/tests/Python/py/async_runtime.py +++ b/tests/Python/py/async_runtime.py @@ -132,6 +132,14 @@ async def run(): mailbox.post(42) result.pending_after = mailbox.continuation is not None result.listeners = len(token.listeners) + next_ctx = IAsyncContext.create( + Trampoline(), + CancellationToken(), + lambda value: result._outcome.append(str(value)), + lambda _: result._outcome.append("error"), + lambda _: result._outcome.append("cancel"), + ) + mailbox.receive()(next_ctx) return result return asyncio.run(run()) diff --git a/tests/Python/test_async_runtime.py b/tests/Python/test_async_runtime.py index aba116e32e..3016c43571 100644 --- a/tests/Python/test_async_runtime.py +++ b/tests/Python/test_async_runtime.py @@ -191,3 +191,28 @@ async def run() -> None: event.Trigger(42) assert calls == ["cancel"] assert not event.delegates + + +def test_sleep_cancelled_while_its_loop_is_stopped_settles_once(): + loop = asyncio.new_event_loop() + token = CancellationToken() + calls: list[str] = [] + + async def start() -> None: + start_with_continuations( + sleep(100_000), + lambda _: calls.append("success"), + lambda _: calls.append("error"), + lambda _: calls.append("cancel"), + token, + ) + + try: + loop.run_until_complete(start()) + cancel_thread = Thread(target=token.cancel) + cancel_thread.start() + cancel_thread.join(5) + assert calls == ["cancel"] + assert not token.listeners + finally: + loop.close()