diff --git a/src/Fable.Transforms/Python/Replacements.fs b/src/Fable.Transforms/Python/Replacements.fs index 663097b10f..d41f37f3cf 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 | "StartChild" -> Helper.LibCall(com, "async_", "start_child_linked", t, args, i.SignatureArgTypes, ?loc = r) |> Some diff --git a/src/fable-library-py/fable_library/async_.py b/src/fable-library-py/fable_library/async_.py index 034137e0cb..7fb680a815 100644 --- a/src/fable-library-py/fable_library/async_.py +++ b/src/fable-library-py/fable_library/async_.py @@ -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 @@ -194,6 +194,62 @@ def to_array(results: list[T]) -> Async[Array[T]]: 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") + + def cont(ctx: IAsyncContext[Array[T]]) -> None: + works = list(to_iterable(computations)) + token = CancellationToken() + limit = max_degree_of_parallelism or max(1, len(works)) + semaphore = asyncio.Semaphore(limit) + + 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 + + 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 choice[T](computations: IEnumerable_1[Async[T | None]]) -> Async[T | None]: def cont(ctx: IAsyncContext[T | None]) -> None: works = list(to_iterable(computations)) diff --git a/tests/Python/TestAsync.fs b/tests/Python/TestAsync.fs index 054999bf81..38547d19ba 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 {