fix(python): complete Async.Parallel semantics

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

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))

tests/Python/TestAsync.fs

@@ -548,6 +548,59 @@ let ``test Async.Parallel is lazy`` () =
         equal 3 x
     } |> Async.RunSynchronously
 
+[<Fact>]
+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
+
+[<Fact>]
+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<int>
+#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
+
+[<Fact>]
+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<int> list), maxDegreeOfParallelism = 2))
+    empty |> equal [||]
+
 [<Fact>]
 let ``test Async.Sequential works`` () =
     async {