From efdad3a17216941b6f729b04b6880eb1a6555076 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Mon, 28 Sep 2026 10:42:48 +0000 Subject: [PATCH] Fix mapAsyncUnorderedParallel(Throttled) to emit in true completion order Closes #354 Root cause: task handles were posted to the internal mailbox in input/start order, and the consumer awaited each handle sequentially via mapAsync. This meant a slow-completing early item blocked delivery of an already-completed later item, even though the documented and intended behavior is completion-order emission. Fix: results (or exceptions) are now posted to the mailbox directly from each child computation as soon as it completes, using an atomic pending counter to know when the source is fully drained and all children have finished, so the end-of-stream sentinel is queued after every result. This guarantees consumers observe true completion order. Added regression tests that gate item 1 behind a signal released only after item 2 finishes its mapper, asserting the consumer receives item 2 before item 1 is released, for both mapAsyncUnorderedParallel and mapAsyncUnorderedParallelThrottled. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- RELEASE_NOTES.md | 1 + src/FSharp.Control.AsyncSeq/AsyncSeq.fs | 46 +++++++++------ .../AsyncSeqTests.fs | 56 +++++++++++++++++++ 3 files changed, 87 insertions(+), 16 deletions(-) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index f66f8ca..57d3665 100644 --- a/RELEASE_NOTES.md +++ b/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. diff --git a/src/FSharp.Control.AsyncSeq/AsyncSeq.fs b/src/FSharp.Control.AsyncSeq/AsyncSeq.fs index 6a514ce..aaa4121 100644 --- a/src/FSharp.Control.AsyncSeq/AsyncSeq.fs +++ b/src/FSharp.Control.AsyncSeq/AsyncSeq.fs @@ -1108,25 +1108,35 @@ module AsyncSeq = 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 @@ module AsyncSeq = 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 }) diff --git a/tests/FSharp.Control.AsyncSeq.Tests/AsyncSeqTests.fs b/tests/FSharp.Control.AsyncSeq.Tests/AsyncSeqTests.fs index 81ac73a..b1ca48e 100644 --- a/tests/FSharp.Control.AsyncSeq.Tests/AsyncSeqTests.fs +++ b/tests/FSharp.Control.AsyncSeq.Tests/AsyncSeqTests.fs @@ -1933,6 +1933,62 @@ let ``AsyncSeq.mapAsyncUnorderedParallel should not preserve order`` () = let allPresent = (Set.ofList resultOrder) = (Set.ofList input) Assert.IsTrue(allPresent, "All input elements should be present in results") +[] +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() + + 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) + +[] +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() + + 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) + [] let ``AsyncSeq.mapAsyncUnorderedParallelThrottled should produce all results`` () =