From e68fbcc8ef59635dbdcf3dfdfbdb5b94c9becfae Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Fri, 16 Oct 2026 21:58:28 +0200 Subject: [PATCH 1/3] fix(python): complete Async.Parallel semantics --- src/fable-library-py/fable_library/async_.py | 61 ++++++++++++++++---- tests/Python/TestAsync.fs | 53 +++++++++++++++++ 2 files changed, 104 insertions(+), 10 deletions(-) diff --git a/src/fable-library-py/fable_library/async_.py b/src/fable-library-py/fable_library/async_.py index 462b951e8e..852c614fcb 100644 --- a/src/fable-library-py/fable_library/async_.py +++ b/src/fable-library-py/fable_library/async_.py @@ -2,7 +2,7 @@ import asyncio from asyncio import Future, ensure_future -from collections.abc import Awaitable, Callable, Iterable +from collections.abc import Awaitable, Callable from concurrent.futures import ThreadPoolExecutor from threading import RLock, Timer from typing import ( @@ -34,7 +34,7 @@ ) from .event import IEvent_2 from .protocols import IEnumerable_1 -from .system import InvalidOperationException__ctor_Z721C83C5, TimeoutException__ctor +from .system import ArgumentException__ctor_Z721C83C5, InvalidOperationException__ctor_Z721C83C5, TimeoutException__ctor from .task import TaskCompletionSource from .time_span import TimeSpan, to_milliseconds from .types import UNIT, Unit @@ -180,17 +180,58 @@ def binder(_: Any | Unit = UNIT) -> Async[None]: return protected_bind(computation, binder) -def parallel[T](computations: IEnumerable_1[Async[T]]) -> Async[Array[T]]: - def delayed() -> Async[Array[T]]: - tasks: Iterable[Awaitable[T]] = map(start_as_task, computations) # type: ignore[arg-type] # ty: ignore[invalid-assignment, invalid-argument-type] - all: Future[list[T]] = asyncio.gather(*tasks) +def parallel[T](computations: IEnumerable_1[Async[T]], max_degree_of_parallelism: int | None = None) -> Async[Array[T]]: + if max_degree_of_parallelism is not None and max_degree_of_parallelism < 1: + raise ArgumentException__ctor_Z721C83C5("maxDegreeOfParallelism must be positive") - def to_array(results: list[T]) -> Async[Array[T]]: - return protected_return(Array[T](results)) + def cont(ctx: IAsyncContext[Array[T]]) -> None: + works = list(computations) + token = CancellationToken() + limit = max_degree_of_parallelism or max(1, len(works)) + semaphore = asyncio.Semaphore(limit) - return protected_bind(await_task(all), to_array) + async def run_one(work: Async[T]) -> T: + async with semaphore: + try: + return await start_as_task(work, cancellation_token=token) + except asyncio.CancelledError as error: + raise TaskCanceledError() from error - return delay(delayed) + async def run_all() -> Array[T]: + # decision: retain the first failure while cancellation settles every started sibling. + registration = ctx.cancel_token.register(cancel_children) + try: + tasks = [asyncio.create_task(run_one(work)) for work in works] + first_error: Exception | None = None + for completed in asyncio.as_completed(tasks): + try: + await completed + except Exception as error: + if first_error is None: + first_error = error + cancel_children() + if first_error is not None: + raise first_error + return Array[T]([task.result() for task in tasks]) + finally: + registration.Dispose() + + def cancel_children() -> None: + try: + token.cancel() + except Exception: + pass + + def report_error(error: Exception) -> None: + if ctx.cancel_token.is_cancelled and isinstance(error, OperationCanceledError): + ctx.on_cancel(error) + else: + ctx.on_error(error) + + child_ctx = IAsyncContext.create(ctx.trampoline, ctx.cancel_token, ctx.on_success, report_error, ctx.on_cancel) + await_task(run_all())(child_ctx) + + return protected_cont(cont) def sequential[T](computations: IEnumerable_1[Async[T]]) -> Async[Array[T]]: diff --git a/tests/Python/TestAsync.fs b/tests/Python/TestAsync.fs index 6d8cd7002d..d1ac0e3676 100644 --- a/tests/Python/TestAsync.fs +++ b/tests/Python/TestAsync.fs @@ -548,6 +548,59 @@ let ``test Async.Parallel is lazy`` () = equal 3 x } |> Async.RunSynchronously +[] +let ``test Async.Parallel bounds concurrency and preserves order`` () = + async { + let gate = obj () + let mutable running = 0 + let mutable peak = 0 + let work i = async { + lock gate (fun () -> + running <- running + 1 + peak <- max peak running) + do! Async.Sleep 30 + lock gate (fun () -> running <- running - 1) + return i + } + let! results = Async.Parallel([ for i in 1..9 -> work i ], maxDegreeOfParallelism = 3) + results |> equal [| 1..9 |] + peak |> equal 3 + } |> Async.RunSynchronously + +[] +let ``test Async.Parallel cancels siblings and waits before reporting failure`` () = + async { + let mutable completed = 0 + let work i = async { + if i = 2 then + do! Async.Sleep 20 + failwith "parallel failure" + do! Async.Sleep 200 +#if FABLE_COMPILER + completed <- completed + 1 +#else + System.Threading.Interlocked.Increment(&completed) |> ignore +#endif + } + let! outcome = Async.Parallel [ for i in 0..4 -> work i ] |> Async.Catch + match outcome with + | Choice1Of2 _ -> failwith "Expected parallel failure" + | Choice2Of2 error -> error.Message |> equal "parallel failure" + completed |> equal 0 + do! Async.Sleep 250 + completed |> equal 0 + } |> Async.RunSynchronously + +[] +let ``test Async.Parallel validates degree and accepts an empty sequence`` () = + let mutable raised = false + try + Async.Parallel([ async { return 1 } ], maxDegreeOfParallelism = 0) |> ignore + with :? ArgumentException -> raised <- true + raised |> equal true + let empty = Async.RunSynchronously(Async.Parallel(([]: Async list), maxDegreeOfParallelism = 2)) + empty |> equal [||] + [] let ``test Async.Sequential works`` () = async { From 406a9504f0b886154a608f68e723be3456fdbf26 Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Fri, 16 Oct 2026 22:04:35 +0200 Subject: [PATCH 2/3] fix(python): preserve published parallel runtime behavior --- src/Fable.Transforms/Python/Replacements.fs | 3 +++ src/fable-library-py/fable_library/async_.py | 19 +++++++++++++++++-- 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/src/Fable.Transforms/Python/Replacements.fs b/src/Fable.Transforms/Python/Replacements.fs index a3877fa7c7..20e7720885 100644 --- a/src/Fable.Transforms/Python/Replacements.fs +++ b/src/Fable.Transforms/Python/Replacements.fs @@ -3693,6 +3693,9 @@ let asyncs com (ctx: Context) r t (i: CallInfo) (_: Expr option) (args: Expr lis | "Start" -> Helper.LibCall(com, "async_", "start", t, args, i.SignatureArgTypes, ?loc = r) |> Some + | "Parallel" -> + Helper.LibCall(com, "async_", "parallel_with_cancellation", t, args, i.SignatureArgTypes, ?loc = r) + |> Some // Make sure cancellationToken is called as a function and not a getter | "get_CancellationToken" -> Helper.LibCall(com, "async_", "cancellationToken", t, [], ?loc = r) |> Some // `catch` cannot be used as a function name in JS diff --git a/src/fable-library-py/fable_library/async_.py b/src/fable-library-py/fable_library/async_.py index 7602f700cb..979932a2f5 100644 --- a/src/fable-library-py/fable_library/async_.py +++ b/src/fable-library-py/fable_library/async_.py @@ -2,7 +2,7 @@ import asyncio from asyncio import Future, ensure_future -from collections.abc import Awaitable, Callable +from collections.abc import Awaitable, Callable, Iterable from concurrent.futures import ThreadPoolExecutor from threading import RLock, Timer from typing import ( @@ -180,7 +180,22 @@ def binder(_: Any | Unit = UNIT) -> Async[None]: return protected_bind(computation, binder) -def parallel[T](computations: IEnumerable_1[Async[T]], max_degree_of_parallelism: int | None = None) -> Async[Array[T]]: +def parallel[T](computations: IEnumerable_1[Async[T]]) -> Async[Array[T]]: + def delayed() -> Async[Array[T]]: + tasks: Iterable[Awaitable[T]] = map(start_as_task, computations) # type: ignore[arg-type] # ty: ignore[invalid-assignment, invalid-argument-type] + all: Future[list[T]] = asyncio.gather(*tasks) + + def to_array(results: list[T]) -> Async[Array[T]]: + return protected_return(Array[T](results)) + + return protected_bind(await_task(all), to_array) + + return delay(delayed) + + +def parallel_with_cancellation[T]( + computations: IEnumerable_1[Async[T]], max_degree_of_parallelism: int | None = None +) -> Async[Array[T]]: if max_degree_of_parallelism is not None and max_degree_of_parallelism < 1: raise ArgumentException__ctor_Z721C83C5("maxDegreeOfParallelism must be positive") From 9d24f984635911e9cdd4c19bc46063cacef1fcef Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Fri, 16 Oct 2026 22:13:22 +0200 Subject: [PATCH 3/3] fix(python): adapt async enumerables for Pyright --- src/fable-library-py/fable_library/async_.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/fable-library-py/fable_library/async_.py b/src/fable-library-py/fable_library/async_.py index 09cbb1e6b8..7fb680a815 100644 --- a/src/fable-library-py/fable_library/async_.py +++ b/src/fable-library-py/fable_library/async_.py @@ -38,6 +38,7 @@ from .task import TaskCompletionSource from .time_span import TimeSpan, to_milliseconds from .types import UNIT, Unit +from .util import to_iterable def cancellation_token() -> Async[CancellationToken]: @@ -200,7 +201,7 @@ def parallel_with_cancellation[T]( raise ArgumentException__ctor_Z721C83C5("maxDegreeOfParallelism must be positive") def cont(ctx: IAsyncContext[Array[T]]) -> None: - works = list(computations) + works = list(to_iterable(computations)) token = CancellationToken() limit = max_degree_of_parallelism or max(1, len(works)) semaphore = asyncio.Semaphore(limit) @@ -251,7 +252,7 @@ def report_error(error: Exception) -> None: def choice[T](computations: IEnumerable_1[Async[T | None]]) -> Async[T | None]: def cont(ctx: IAsyncContext[T | None]) -> None: - works = list(computations) + works = list(to_iterable(computations)) if not works: ctx.on_success(None) return