[repo-assist] Fix mapAsyncUnorderedParallel(Throttled) to emit in true completion order

RELEASE_NOTES.md

@@ -1,5 +1,6 @@
 ### Unreleased
 
+* Bug fix: `AsyncSeq.mapAsyncUnorderedParallel` and `AsyncSeq.mapAsyncUnorderedParallelThrottled` now emit results in true completion order rather than input/start order. Previously, task handles were posted to the internal mailbox in input order and awaited sequentially, so a slow-completing early item could block delivery of an already-completed later item. Results (and exceptions) are now posted to the mailbox directly from each child as soon as it finishes, so a fast-finishing later item is no longer blocked by a slower earlier one. (#354)
 * Test coverage: Added tests for `AsyncSeq.zipWithIndexAsync` (obsolete alias for `mapiAsync`) and `AsyncSeq.ofIQueryable`, which previously had no dedicated test coverage. No functional changes.
 * Test coverage: Added tests for previously-untested public API functions `AsyncSeq.tryFirst`, `AsyncSeq.firstOrDefault`, `AsyncSeq.zipWithParallel`, `AsyncSeq.combineLatestWithAsync`, and `AsyncSeq.toObservable`. No functional changes.
 * Test coverage: Added tests for `AsyncSeq.distinctUntilChanged` (default-equality variant), `AsyncSeq.takeWhile`, and `AsyncSeq.skipWhile` (sync-predicate variants), which previously had no direct tests. No functional changes.

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -1108,25 +1108,35 @@
       replicateUntilNoneAsync (Task.chooseTask (err |> Task.taskFault) (async.Delay mb.Receive))
       |> mapAsync id }
 
+  // Note: results are posted to the mailbox directly from each child computation as soon as
+  // it completes (rather than posting a handle to be awaited later), so that the consumer
+  // observes true completion order instead of input/start order. A pending counter (starting
+  // at 1 to represent the enumeration of the source itself) is decremented each time a child
+  // completes or the source is fully enumerated; the "no more results" sentinel is posted only
+  // once it reaches zero, guaranteeing it is queued after every child's result.
   let mapAsyncUnorderedParallel (f:'a -> Async<'b>) (s:AsyncSeq<'a>) : AsyncSeq<'b> = asyncSeq {
     use mb = MailboxProcessor.Start (fun _ -> async.Return())
+    let! ct = Async.CancellationToken
+    let pending = ref 1
+    let complete () =
+      if Interlocked.Decrement pending = 0 then
+        mb.Post None
     let! err =
       s
       |> iterAsync (fun a -> async {
-        let! b = Async.StartChild (async {
+        Interlocked.Increment pending |> ignore
+        Async.Start (async {
           try
             let! result = f a
-            return Choice1Of2 result
+            mb.Post (Some (Choice1Of2 result))
           with ex ->
-            return Choice2Of2 ex
-        })
-        mb.Post (Some b) })
-      |> Async.map (fun _ -> mb.Post None)
+            mb.Post (Some (Choice2Of2 ex))
+          complete () }, ct) })
+      |> Async.map complete
       |> Async.StartChildAsTask
     yield!
       replicateUntilNoneAsync (Task.chooseTask (err |> Task.taskFault) (async.Delay mb.Receive))
-      |> mapAsync (fun childAsync -> async {
-        let! result = childAsync
+      |> mapAsync (fun result -> async {
         match result with
         | Choice1Of2 value -> return value
         | Choice2Of2 ex -> return raise ex })
@@ -1135,26 +1145,30 @@
   let mapAsyncUnorderedParallelThrottled (parallelism:int) (f:'a -> Async<'b>) (s:AsyncSeq<'a>) : AsyncSeq<'b> = asyncSeq {
     use mb = MailboxProcessor.Start (fun _ -> async.Return())
     use sm = new SemaphoreSlim(parallelism)
+    let! ct = Async.CancellationToken
+    let pending = ref 1
+    let complete () =
+      if Interlocked.Decrement pending = 0 then
+        mb.Post None
     let! err =
       s
       |> iterAsync (fun a -> async {
         do! sm.WaitAsync () |> Async.awaitTaskUnitCancellationAsError
-        let! b = Async.StartChild (async {
+        Interlocked.Increment pending |> ignore
+        Async.Start (async {
           try
             let! result = f a
             sm.Release() |> ignore
-            return Choice1Of2 result
+            mb.Post (Some (Choice1Of2 result))
           with ex ->
             sm.Release() |> ignore
-            return Choice2Of2 ex
-        })
-        mb.Post (Some b) })
-      |> Async.map (fun _ -> mb.Post None)
+            mb.Post (Some (Choice2Of2 ex))
+          complete () }, ct) })
+      |> Async.map complete
       |> Async.StartChildAsTask
     yield!
       replicateUntilNoneAsync (Task.chooseTask (err |> Task.taskFault) (async.Delay mb.Receive))
-      |> mapAsync (fun childAsync -> async {
-        let! result = childAsync
+      |> mapAsync (fun result -> async {
         match result with
         | Choice1Of2 value -> return value
         | Choice2Of2 ex -> return raise ex })
@@ -2902,7 +2916,7 @@
 
   [<CompilerMessage("The result of groupBy must be consumed with a parallel combinator such as AsyncSeq.mapAsyncParallel. Sequential consumption will deadlock because sub-sequence completion depends on other sub-sequences being consumed concurrently.", 9999)>]
   let groupBy (p:'a -> 'k) (s:AsyncSeq<'a>) : AsyncSeq<'k * AsyncSeq<'a>> =
     groupByAsync (p >> async.Return) s
   #endif
   #endif
 

tests/FSharp.Control.AsyncSeq.Tests/AsyncSeqTests.fs

@@ -1933,6 +1933,62 @@
   let allPresent = (Set.ofList resultOrder) = (Set.ofList input)
   Assert.IsTrue(allPresent, "All input elements should be present in results")
 
+[<Test>]
+let ``AsyncSeq.mapAsyncUnorderedParallel should emit in completion order not input order`` () =
+  // Regression test for https://github.com/fsprojects/FSharp.Control.AsyncSeq/issues/354
+  // Item 1 is gated behind a signal that is only released after item 2 has finished its
+  // mapper. The consumer must receive item 2 before item 1 is released.
+  use releaseFirst = new SemaphoreSlim(0)
+  use secondFinished = new SemaphoreSlim(0)
+  let emitted = System.Collections.Concurrent.ConcurrentQueue<int>()
+
+  let pipeline =
+    [1; 2]
+    |> AsyncSeq.ofSeq
+    |> AsyncSeq.mapAsyncUnorderedParallel (fun x -> async {
+      if x = 1 then
+        do! releaseFirst.WaitAsync() |> Async.AwaitTask
+      else
+        secondFinished.Release() |> ignore
+      return x
+    })
+    |> AsyncSeq.iter (fun x -> emitted.Enqueue(x))
+
+  let task = Async.StartAsTask pipeline
+  Assert.IsTrue(secondFinished.Wait(5000), "Second mapper did not complete in time")
+  // Give the consumer a moment to observe item 2's emission before item 1 is released.
+  Async.Sleep(200) |> Async.RunSynchronously
+  Assert.AreEqual([2], List.ofSeq emitted, "Item 2 should be emitted before item 1 is released")
+  releaseFirst.Release() |> ignore
+  Assert.IsTrue(task.Wait(5000), "Pipeline did not finish in time")
+  Assert.AreEqual([2; 1], List.ofSeq emitted)
+
+[<Test>]
+let ``AsyncSeq.mapAsyncUnorderedParallelThrottled should emit in completion order not input order`` () =
+  use releaseFirst = new SemaphoreSlim(0)
+  use secondFinished = new SemaphoreSlim(0)
+  let emitted = System.Collections.Concurrent.ConcurrentQueue<int>()
+
+  let pipeline =
+    [1; 2]
+    |> AsyncSeq.ofSeq
+    |> AsyncSeq.mapAsyncUnorderedParallelThrottled 2 (fun x -> async {
+      if x = 1 then
+        do! releaseFirst.WaitAsync() |> Async.AwaitTask
+      else
+        secondFinished.Release() |> ignore
+      return x
+    })
+    |> AsyncSeq.iter (fun x -> emitted.Enqueue(x))
+
+  let task = Async.StartAsTask pipeline
+  Assert.IsTrue(secondFinished.Wait(5000), "Second mapper did not complete in time")
+  Async.Sleep(200) |> Async.RunSynchronously
+  Assert.AreEqual([2], List.ofSeq emitted, "Item 2 should be emitted before item 1 is released")
+  releaseFirst.Release() |> ignore
+  Assert.IsTrue(task.Wait(5000), "Pipeline did not finish in time")
+  Assert.AreEqual([2; 1], List.ofSeq emitted)
+
 
 [<Test>]
 let ``AsyncSeq.mapAsyncUnorderedParallelThrottled should produce all results`` () =
@@ -2096,7 +2152,7 @@
       let actual =
         ls
         |> AsyncSeq.ofSeq
         |> AsyncSeq.groupBy p
         |> AsyncSeq.mapAsyncParallel (snd >> AsyncSeq.toListAsync)
       Assert.AreEqual(expected, actual)
 
@@ -2105,7 +2161,7 @@
   let expected = asyncSeq { raise (exn("test")) }
   let actual =
     asyncSeq { raise (exn("test")) }
     |> AsyncSeq.groupBy (fun i -> i % 3)
     |> AsyncSeq.mapAsyncParallel (snd >> AsyncSeq.toListAsync)
   Assert.AreEqual(expected, actual)
 
@@ -4886,7 +4942,7 @@
 let ``AsyncSeq.groupByAsync groups elements by async projection`` () =
   let result =
     AsyncSeq.ofSeq [1..6]
     |> AsyncSeq.groupByAsync (fun x -> async { return x % 2 })
     |> AsyncSeq.mapAsyncParallel (fun (key, grp) -> async {
         let! items = AsyncSeq.toArrayAsync grp
         return key, Array.sort items })
@@ -4899,7 +4955,7 @@
 let ``AsyncSeq.groupByAsync on empty sequence returns empty`` () =
   let result =
     AsyncSeq.empty<int>
     |> AsyncSeq.groupByAsync (fun x -> async { return x % 2 })
     |> AsyncSeq.toArrayAsync
     |> Async.RunSynchronously
   Assert.AreEqual([||], result)
@@ -4908,7 +4964,7 @@
 let ``AsyncSeq.groupByAsync with all-same key produces single group`` () =
   let result =
     AsyncSeq.ofSeq [1; 2; 3]
     |> AsyncSeq.groupByAsync (fun _ -> async { return "same" })
     |> AsyncSeq.mapAsyncParallel (fun (key, grp) -> async {
         let! items = AsyncSeq.toArrayAsync grp
         return key, Array.sort items })