I'm using F# and have an AsyncSeq<Async<'t>>. Each item will take a varying amount of time to process and does I/O that's rate-limited.
I want to run all the operations in parallel and then pass them down the chain as an AsyncSeq<'t> so I can perform further manipulations on them and ultimately AsyncSeq.fold them into a final outcome.
The following AsyncSeq operations almost meet my needs:
mapAsyncParallel - does the parallelism, but it's unconstrained, (and I don't need the order preserved)iterAsyncParallelThrottled - parallel and has a max degree of parallelism but doesn't let me return results (and I don't need the order preserved)What I really need is like a mapAsyncParallelThrottled. But, to be more precise, really the operation would be entitled mapAsyncParallelThrottledUnordered.
Things I'm considering:
mapAsyncParallel but use a Semaphore within the function to constrain the parallelism myself, which is probably not going to be optimal in terms of concurrency, and due to buffering the results to reorder them.iterAsyncParallelThrottled and do some ugly folding of the results into an accumulator as they arrive guarded by a lock kinda like this - but I don't need the ordering so it won't be optimal.AsyncSeqSrc like this. I'd probably have a set of Async.StartAsTask tasks in flight and start more after each Task.WaitAny gives me something to AsyncSeqSrc.put until I reach the maxDegreeOfParallelismSurely I'm missing a simple answer and there's a better way?
Failing that, would love someone to sanity check my option 3 in either direction!
I'm open to using AsyncSeq.toAsyncEnum and then use an IAsyncEnumerable way of achieving the same outcome if that exists, though ideally without getting into TPL DataFlow or RX land if it can be avoided (I've done extensive SO searching for that without results...).
If I'm understanding your requirements then something like this will work. It effectively combines the iter unordered with a channel to allow a mapping instead.
let mapAsyncParallelBoundedUnordered boundedAmount (mapper: 't -> Async<_>) source = asyncSeq {
let! ct = Async.CancellationToken
let channel = Channel.CreateUnbounded()
let! _ =
async {
do!
source
|> AsyncSeq.iterAsyncParallelThrottled boundedAmount (fun s -> async {
let! orderChild = mapper s
do! channel.Writer.WriteAsync(orderChild, ct)
})
channel.Writer.Complete()
}
|> Async.StartChild
for item in channel.Reader.ReadAllAsync(ct) |> AsyncSeq.ofAsyncEnum do
let! toReturn = item
yield toReturn
}
Also with a little bit of variation of the above (e.g. child tasks) you can make it ordered and parallelism bounded.
let mapAsyncParallelBounded boundedAmount mapper source = asyncSeq {
let! ct = Async.CancellationToken
let channel = Channel.CreateBounded(BoundedChannelOptions(boundedAmount))
let! _ =
source
|> AsyncSeq.iterAsync (fun s -> async {
let! orderChild = mapper s |> Async.StartChild
do! channel.Writer.WriteAsync(orderChild, ct)
})
|> Async.StartChild
let! ct = Async.CancellationToken
for item in channel.Reader.ReadAllAsync(ct) |> AsyncSeq.ofAsyncEnum do
let! toReturn = item
yield toReturn
}