Uh oh!
There was an error while loading. Please reload this page.
feat(Task, Async): parallelLimit, sequential, startAsyncImmediate - #20294
feat(Task, Async): parallelLimit, sequential, startAsyncImmediate#20294bartelink wants to merge 20 commits into
Conversation
✅ No release notes required |
c7f4834 to
0279f13CompareThere was a problem hiding this comment.
Pull request overview
Adds bounded-parallelism helpers to FSharp.Core’s Async and Task modules (parallelLimit / parallelDoLimit), along with unit tests, surface-area baselines, and release notes, to support controlled concurrency for async/task workflows.
Changes:
- Add
Async.parallelLimit/Async.parallelDoLimitwrappers overAsync.ParallelwithmaxDegreeOfParallelism. - Add
Task.parallelLimit/Task.parallelDoLimitimplemented viaSemaphoreSlim+Task.WhenAll, flowing a providedCancellationTokento task factories. - Add unit tests + surface area baseline updates + release note entry for the new APIs.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/FSharp.Core.UnitTests/FSharp.Core/Microsoft.FSharp.Control/TaskModuleFunctions.fs | Adds unit tests for Task.parallelLimit / Task.parallelDoLimit (results + concurrency cap). |
| tests/FSharp.Core.UnitTests/FSharp.Core/Microsoft.FSharp.Control/AsyncModuleFunctions.fs | Adds unit tests for Async.parallelLimit / Async.parallelDoLimit. |
| tests/FSharp.Core.UnitTests/FSharp.Core.SurfaceArea.netstandard21.release.bsl | Surface-area baseline update for new AsyncModule/TaskModule members (netstandard2.1, release). |
| tests/FSharp.Core.UnitTests/FSharp.Core.SurfaceArea.netstandard21.debug.bsl | Surface-area baseline update (netstandard2.1, debug). |
| tests/FSharp.Core.UnitTests/FSharp.Core.SurfaceArea.netstandard20.release.bsl | Surface-area baseline update (netstandard2.0, release). |
| tests/FSharp.Core.UnitTests/FSharp.Core.SurfaceArea.netstandard20.debug.bsl | Surface-area baseline update (netstandard2.0, debug). |
| src/FSharp.Core/tasks.fsi | Public API + docs for Task.parallelLimit / Task.parallelDoLimit. |
| src/FSharp.Core/tasks.fs | Implementation of Task.parallelLimit / Task.parallelDoLimit. |
| src/FSharp.Core/async.fsi | Public API + docs for Async.parallelLimit / Async.parallelDoLimit. |
| src/FSharp.Core/async.fs | Implementation of Async.parallelLimit / Async.parallelDoLimit. |
| docs/release-notes/.FSharp.Core/11.0.100.md | Release note entry for the new bounded-parallelism helpers. |
Suppressed comments (1)
src/FSharp.Core/tasks.fs:820
SemaphoreSlimdisposal is wired up via aContinueWith, but if enumeratingcomputationsthrows while building theTask.WhenAllinput array, the semaphore is leaked (the continuation is never attached). Usinguse sem = ...and dropping the continuation avoids the leak and simplifies the implementation.
allTask.ContinueWith(
(fun (_: Task<'T[]>) -> sem.Dispose()),
CancellationToken.None,
TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default
💡 Add a code-review agent skill for context-aware, tailored reviews. Learn more in the docs.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
| [| | ||
| for f in computations -> | ||
| backgroundTask { | ||
| do! sem.WaitAsync ct |
There was a problem hiding this comment.
(while this is true, the complexity ramps significantly too; Task.WhenAll internally doesn't get into this sort of thing either. My feeling is to keep it simple for now; if anyone want's to extend it, we'd likely want more extensive test scenarios on the consumption end to go with that at the same time. On the other hand if anyone wants to chuck an agent at it, I'm not entirely averse to going into the rabbit hole at this time...)
Uh oh!
There was an error while loading. Please reload this page.
🔍 Tooling Safety Check — Affects-Test-Tooling
|
@T-Gro if you can chuck an AI or human review on this I'll attend to it @T-Gro@TheAngryByrd In my musings in fsharp/fslang-suggestions#685 (comment) I noted that parallelLimit and friends should probably make an internal linked CT and cancel siblings where one of the computations faults in order to match the @TheAngryByrd@T-Gro Any thoughts on whether we should add Async.StartChild(computation, externalCt) per fsharp/fslang-suggestions#685 into the mix? @TheAngryByrd Are there any other common things that might belong in the modules that get heavy usage, e.g.
|
66b3900 to
95690f9Compare28f7eee to
67fbceeCompare# Conflicts: # src/FSharp.Core/async.fsi # src/FSharp.Core/tasks.fs # src/FSharp.Core/tasks.fsi # tests/FSharp.Core.UnitTests/FSharp.Core.SurfaceArea.netstandard20.debug.bsl # tests/FSharp.Core.UnitTests/FSharp.Core.SurfaceArea.netstandard20.release.bsl # tests/FSharp.Core.UnitTests/FSharp.Core.SurfaceArea.netstandard21.debug.bsl # tests/FSharp.Core.UnitTests/FSharp.Core.SurfaceArea.netstandard21.release.bsl # tests/FSharp.Core.UnitTests/FSharp.Core/Microsoft.FSharp.Control/AsyncModuleFunctions.fs # tests/FSharp.Core.UnitTests/FSharp.Core/Microsoft.FSharp.Control/TaskModuleFunctions.fs
| with e when interceptNonCancellationExn (fun () -> innerCts.Cancel()) e -> | ||
| // We'll never get here as the filter always returns false | ||
| return Unchecked.defaultof<_> |
There was a problem hiding this comment.
I can't say I like this way of using when clause to do logic.
Could it not do the Cts.Cancel() in the finally (by storing the Task instance locally perhaps?)
There was a problem hiding this comment.
I don't either, but I need to trap/intercept both the CT -> Task throwing (i.e. the f CT bit), as well as the actual Task outcome
It seems that would mean 2x try/with ? my main aim here is to avoid doing a raise e as I don't want to alter the call stack, or wrap it (and reraise doesnt work in a task {)
The exception type filtering is not important.
I could make a task { and then ContinueWith OnlyOnFaulted to represent the side-effect instead, but it feels like it'd get messy too. Want me to try that, or any other ideas?
There was a problem hiding this comment.
Yeah, a completed flag + finally is what I meant — no with, so the exception is untouched, and it covers both the f ct throw and the task fault:
let mutablecompleted=falsetrylet!r= f innerCt
completed <-truereturn r
finallyifnot completed then innerCts.Cancel()(Sem release stays in the finally too.)
There was a problem hiding this comment.
Far too elegant and clean for my convoluted mind! Thanks, will do!
T-Gro
left a comment
There was a problem hiding this comment.
🤖🕵️ AI review — verify independently.
| /// do! Async.Sleep 500 // Or any other async activity | ||
| /// let! v1 = completor1 | ||
| /// let! v2 = completor2 | ||
| /// and! v2 = completor2 |
There was a problem hiding this comment.
🤖🕵️ Doc example doesn't compile — async {} defines no MergeSources/Bind2, so and! is a hard FS3343 (verified via dotnet fsi). Same at L382. Revert both and! → let!.
async{let!v1= completor1
and! v2 = completor2 // error FS3343...}There was a problem hiding this comment.
Good catch, thanks.
Note the actual thing bothering me about the example is that if you needed to do things, you would still do the last one inline and only do one StartChild. While I'd never have used two StartChild calls in an example in the first instance, I can imagine not showing that explicitly as a possibility is a loss to some. Let me know if you think it should be edited down to one or whether having a fresh minimal example is useful (but assuming no)
but... should there be a MergeSources/Bind in the box? Is there a tracking issue?
| match! Async.catch sut with | ||
| | Error (:? InvalidOperationException as e) -> | ||
| Assert.Equal("boom1", e.Message) |
There was a problem hiding this comment.
🤖🕵️ Flaky (~15% — measured boom2 winning 44/300). Async.parallelLimit → Async.Parallel surfaces the first-observed exception; both siblings are released simultaneously, so the winner is a race. Assert the invariant, not which one wins:
match! Async.catch sut with| Error (:? AggregateException)-> failwith "should be a single exception, not an AggregateException"| Error (:? InvalidOperationException)| Error (:? ArgumentException)->()// either sibling may win| x -> failwith $"unexpected %A{x}"There was a problem hiding this comment.
Done (also applied to clone for Task on assumption that same applies there).
xlmdoc for task version left it open
/// <p>Where multiple computations Fault, a single exception is propagated.</p>
xmldoc for Async.parallel[Do]Limit says only:
/// <remarks>While the result order matches the input order, the relative start and completion order of computations is arbitrary.</remarks>
Async.Parallel remarks says, among other things:
/// If any child computation raises an exception, then the overall computation will trigger an exception, and cancel the others.
Any suggestions for better wordings appreciated
| return! | ||
| Task.WhenAll | ||
| [| | ||
| for f in computations -> |
There was a problem hiding this comment.
🤖🕵️ A throwing input seq orphans already-started tasks; use sem/use innerCts then dispose out from under them → unobserved ObjectDisposedException (measured 2/2 against the built dll; process-fatal under ThrowUnobservedTaskExceptions). Materialize before starting any work (as Async.Parallel does):
letcomputations= Array.ofSeq computations // enumerate before any backgroundTask startsTrigger:
seq{yieldfun _ ->task{do! Task.Delay 500;return1}yieldfun _ ->task{do! Task.Delay 500;return2}
failwith "boom"}// -> 2 unobserved ObjectDisposedException|> Task.parallelLimit 4 CancellationToken.NoneThere was a problem hiding this comment.
Done (and special cased 0/1 tasks to avoid spinning up cts/sem/backgroundTask etc)
| return! | ||
| Task.WhenAll | ||
| [| | ||
| for f in computations -> |
There was a problem hiding this comment.
backgroundTask under a non-null SynchronizationContext / non-default TaskScheduler.Current starts via Task.Run (tasks.fs:256-270). Since [| for f in computations -> backgroundTask {…} |] builds all n up front, every item queues a work item at creation — the semaphore never gets to gate. So maxDop bounds concurrency, not scheduling.
100k trivial factories, maxDop = 2, captured sync context:
| work items | peak queued | |
|---|---|---|
sequential | 0 | 0 |
parallelLimit 2 | ~200k (≈2·n) | ~63k |
Async.parallelLimit doesn't do this — Async.Parallel's bounded branch (async.fs:1667) runs maxDop worker loops, O(maxDop) live. Worth at least a doc note; the worker-loop shape also closes the "materialize before starting" thread.
worker-loop sketch
letitems= computations |> Seq.toArray // materialize once (also fixes the enum-throw leak)let mutablei=-1letworker()= backgroundTask {let mutablej= Interlocked.Increment &i
while j < items.Length dolet!r= items.[j] innerCt // → results.[j]
j <- Interlocked.Increment &i }
Task.WhenAll [|for_in1.. min maxDop items.Length -> worker ()|]There was a problem hiding this comment.
@T-Gro Done-ish, but stuff is choking and I'v spent too long in a circle not to ask someone!
| req when maxDegreeOfParallelism = 1 || req.Length = 1 -> sequential ct reqerrors as:
tasks.fs(827,47): error FS0043: The type 'bool' does not support the operator '||'
I can work around that by doing:
| [| _ |] as req -> sequential ct req | req when maxDegreeOfParallelism = 1 -> sequential ct reqInterlocked.Increment &i gives me
FSharp.Core/tasks.fs(837,67): error FS0001: Type mismatch. Expecting a 'byref<'a>' but given a ''a' The types ''a' and 'byref<'a>' cannot be unified.
FSharp.Core/tasks.fs(837,57): error FS0041: A unique overload for method 'Increment' could not be determined based on type information prior to this program point. A type annotation may be needed.Known type of argument: byref<'a>Candidates: - Interlocked.Increment(location: byref) : int64 - Interlocked.Increment(location: byref) : int - Interlocked.Increment(location: byref) : uint32 - Interlocked.Increment(location: byref) : uint64Assigning the result to a temp that's not mutable doesnt help
(using a ref, moving stuff out of line, explicit markers on the mutable and many other things are not helping)the
&&inindex < req.Length && not innerCts.IsCancellationRequestedgives metasks.fs(839,50): error FS0043: The type 'bool' does not support the operator '&&'
trying to move the expression out of the
whileexpr into the assignments doesnt help(I've definitely met clones of this before in TaskSeq, which also has fsi/fs file pairs?)
(Wondering if this is a fixed fantomas issue and the fantomas in this repo is behind, or whether it should be logged)
- Fantomas accepts
items.[j] innerCtbut take out the.and it rendersitems[j]innerCt
# This is the 1st commit message: review: correct handling under exception in enumeration # The commit message #2 will be skipped: # f
Part of the helpers/signatures proposed fsharp/fslang-suggestions#1467
Resolvesfsprojects/FSharp.Control.TaskSeq#143
Adds
parallelLimitandparallelDoLimitfunctions formodule Asyncandmodule TaskLimitvsThrottlednaming is/was discussed in the linked issues; any discussion on naming should happen therecc @TheAngryByrd This replicates functionality in FsToolkit.ErrorHandling so reviews would be much appreciated (and/or feel free to @ in anyone relevant)
cc @xperiandri
Task.parallelLimithas same sig asCancelableTask.whenAllThrottledin #20128 (I intend to take the enhancement and review comments from there into account too)Checklist:
Task.sequentialDo: CT -> seq<CT->Task<unit>>(same logic without result collector in mix)Task.startAsyncImmediate: CT -> Async'T> ->Task<'T>feat(Task, Async): parallelLimit, parallelDoLimit bartelink/fsharp#2 (comment)TaskEx: Async.startImmediateAsTask fsprojects/FSharp.Control.TaskSeq#142Async/Task interop/cancellation/limits fsharp/fslang-suggestions#1467