Perf: optimise filterAsync, chooseAsync, and foldAsync with direct enumerators

- filterAsync: replace asyncSeq-builder with OptimizedFilterAsyncEnumerator,
  avoiding AsyncGenerator allocation and generator-chain dispatch per element.
- chooseAsync (non-AsyncSeqOp path): replace asyncSeq-builder with
  OptimizedChooseAsyncEnumerator for the same reason.
- foldAsync (non-AsyncSeqOp path): replace scanAsync+lastOrDefault composition
  with a direct loop, eliminating the intermediate async sequence and its
  generator machinery entirely.
- Add AsyncSeqFilterChooseFoldBenchmarks and AsyncSeqPipelineBenchmarks to
  measure the affected operations and catch future regressions.

All 317 existing tests pass.

Co-authored-by: Copilot <[email protected]>

Author: github-actions[bot]

RELEASE_NOTES.md

@@ -1,3 +1,10 @@
+### 4.9.0
+
+* Performance: `filterAsync` — replaced `asyncSeq`-builder implementation with a direct optimised enumerator, reducing allocation and generator overhead.
+* Performance: `chooseAsync` — fallback (non-`AsyncSeqOp`) path now uses a direct optimised enumerator instead of the `asyncSeq` builder.
+* Performance: `foldAsync` — fallback (non-`AsyncSeqOp`) path now uses a direct loop instead of composing `scanAsync` + `lastOrDefault`, avoiding intermediate sequence allocations.
+* Benchmarks: added `AsyncSeqFilterChooseFoldBenchmarks` and `AsyncSeqPipelineBenchmarks` benchmark classes to measure `filterAsync`, `chooseAsync`, `foldAsync`, `toArrayAsync`, and common multi-step pipelines.
+
 ### 4.8.0
 
 * Added `AsyncSeq.mapFoldAsync` — maps each element using an asynchronous folder that also threads an accumulator state, returning both the array of results and the final state; mirrors `Seq.mapFold`.

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -924,6 +924,56 @@ module AsyncSeq =
           disposed <- true
           source.Dispose()
 
+  // Optimized filterAsync enumerator that avoids computation builder overhead
+  type private OptimizedFilterAsyncEnumerator<'T>(source: IAsyncSeqEnumerator<'T>, f: 'T -> Async<bool>) =
+    let mutable disposed = false
+
+    interface IAsyncSeqEnumerator<'T> with
+      member _.MoveNext() = async {
+        let mutable result: 'T option = None
+        let mutable isDone = false
+        while not isDone do
+          let! moveResult = source.MoveNext()
+          match moveResult with
+          | None -> isDone <- true
+          | Some value ->
+            let! keep = f value
+            if keep then
+              result <- Some value
+              isDone <- true
+        return result }
+
+      member _.Dispose() =
+        if not disposed then
+          disposed <- true
+          source.Dispose()
+
+  // Optimized chooseAsync enumerator that avoids computation builder overhead
+  type private OptimizedChooseAsyncEnumerator<'T, 'U>(source: IAsyncSeqEnumerator<'T>, f: 'T -> Async<'U option>) =
+    let mutable disposed = false
+
+    interface IAsyncSeqEnumerator<'U> with
+      member _.MoveNext() = async {
+        let mutable result: 'U option = None
+        let mutable isDone = false
+        while not isDone do
+          let! moveResult = source.MoveNext()
+          match moveResult with
+          | None -> isDone <- true
+          | Some value ->
+            let! chosen = f value
+            match chosen with
+            | Some u ->
+              result <- Some u
+              isDone <- true
+            | None -> ()
+        return result }
+
+      member _.Dispose() =
+        if not disposed then
+          disposed <- true
+          source.Dispose()
+
   let mapAsync f (source : AsyncSeq<'T>) : AsyncSeq<'TResult> =
     match source with
     | :? AsyncSeqOp<'T> as source -> source.MapAsync f
@@ -1008,12 +1058,7 @@ module AsyncSeq =
     match source with
     | :? AsyncSeqOp<'T> as source -> source.ChooseAsync f
     | _ ->
-      asyncSeq {
-        for itm in source do
-          let! v = f itm
-          match v with
-          | Some v -> yield v
-          | _ -> () }
+      AsyncSeqImpl(fun () -> new OptimizedChooseAsyncEnumerator<'T, 'U>(source.GetEnumerator(), f) :> IAsyncSeqEnumerator<'U>) :> AsyncSeq<'U>
 
   let ofSeqAsync (source:seq<Async<'T>>) : AsyncSeq<'T> =
       asyncSeq {
@@ -1022,10 +1067,8 @@ module AsyncSeq =
               yield v
       }
 
-  let filterAsync f (source : AsyncSeq<'T>) = asyncSeq {
-    for v in source do
-      let! b = f v
-      if b then yield v }
+  let filterAsync f (source : AsyncSeq<'T>) : AsyncSeq<'T> =
+    AsyncSeqImpl(fun () -> new OptimizedFilterAsyncEnumerator<'T>(source.GetEnumerator(), f) :> IAsyncSeqEnumerator<'T>) :> AsyncSeq<'T>
 
   let tryLast (source : AsyncSeq<'T>) = async {
       use ie = source.GetEnumerator()
@@ -1271,7 +1314,17 @@ module AsyncSeq =
   let foldAsync f (state:'State) (source : AsyncSeq<'T>) =
     match source with
     | :? AsyncSeqOp<'T> as source -> source.FoldAsync f state
-    | _ -> source |> scanAsync f state |> lastOrDefault state
+    | _ -> async {
+        use ie = source.GetEnumerator()
+        let mutable st = state
+        let! move = ie.MoveNext()
+        let mutable b = move
+        while b.IsSome do
+          let! st' = f st b.Value
+          st <- st'
+          let! next = ie.MoveNext()
+          b <- next
+        return st }
 
   let fold f (state:'State) (source : AsyncSeq<'T>) =
     foldAsync (fun st v -> f st v |> async.Return) state source

tests/FSharp.Control.AsyncSeq.Benchmarks/AsyncSeqBenchmarks.fs

@@ -113,6 +113,72 @@ type AsyncSeqBuilderBenchmarks() =
         |> AsyncSeq.iterAsync (fun _ -> async.Return())
         |> Async.RunSynchronously
 
+/// Benchmarks for filter, choose, and fold operations (optimised direct-enumerator implementations)
+[<MemoryDiagnoser>]
+[<SimpleJob(RuntimeMoniker.Net80)>]
+type AsyncSeqFilterChooseFoldBenchmarks() =
+
+    [<Params(1000, 10000)>]
+    member val ElementCount = 0 with get, set
+
+    /// Benchmark filterAsync — all elements pass the predicate
+    [<Benchmark(Baseline = true)>]
+    member this.FilterAsyncAllPass() =
+        AsyncSeq.replicate this.ElementCount 1
+        |> AsyncSeq.filterAsync (fun _ -> async.Return true)
+        |> AsyncSeq.iterAsync (fun _ -> async.Return())
+        |> Async.RunSynchronously
+
+    /// Benchmark filterAsync — no elements pass the predicate (entire sequence scanned)
+    [<Benchmark>]
+    member this.FilterAsyncNonePass() =
+        AsyncSeq.replicate this.ElementCount 1
+        |> AsyncSeq.filterAsync (fun _ -> async.Return false)
+        |> AsyncSeq.iterAsync (fun _ -> async.Return())
+        |> Async.RunSynchronously
+
+    /// Benchmark chooseAsync — all elements selected
+    [<Benchmark>]
+    member this.ChooseAsyncAllSelected() =
+        AsyncSeq.replicate this.ElementCount 42
+        |> AsyncSeq.chooseAsync (fun x -> async.Return (Some x))
+        |> AsyncSeq.iterAsync (fun _ -> async.Return())
+        |> Async.RunSynchronously
+
+    /// Benchmark foldAsync — sum all elements
+    [<Benchmark>]
+    member this.FoldAsync() =
+        AsyncSeq.replicate this.ElementCount 1
+        |> AsyncSeq.foldAsync (fun acc x -> async.Return (acc + x)) 0
+        |> Async.RunSynchronously
+        |> ignore
+
+/// Benchmarks for multi-step pipeline composition
+[<MemoryDiagnoser>]
+[<SimpleJob(RuntimeMoniker.Net80)>]
+type AsyncSeqPipelineBenchmarks() =
+
+    [<Params(1000, 10000)>]
+    member val ElementCount = 0 with get, set
+
+    /// Benchmark map → filter → fold pipeline (exercises the three optimised combinators together)
+    [<Benchmark(Baseline = true)>]
+    member this.MapFilterFold() =
+        AsyncSeq.replicate this.ElementCount 1
+        |> AsyncSeq.mapAsync  (fun x -> async.Return (x * 2))
+        |> AsyncSeq.filterAsync (fun x -> async.Return (x > 0))
+        |> AsyncSeq.foldAsync (fun acc x -> async.Return (acc + x)) 0
+        |> Async.RunSynchronously
+        |> ignore
+
+    /// Benchmark collecting to an array
+    [<Benchmark>]
+    member this.ToArray() =
+        AsyncSeq.replicate this.ElementCount 1
+        |> AsyncSeq.toArrayAsync
+        |> Async.RunSynchronously
+        |> ignore
+
 /// Entry point for running benchmarks
 module AsyncSeqBenchmarkRunner =
 
@@ -138,15 +204,25 @@ module AsyncSeqBenchmarkRunner =
                 printfn "Running Builder Pattern Benchmarks..."
                 BenchmarkRunner.Run<AsyncSeqBuilderBenchmarks>() |> ignore
                 0
+            | Some "filter-choose-fold" ->
+                printfn "Running Filter/Choose/Fold Benchmarks..."
+                BenchmarkRunner.Run<AsyncSeqFilterChooseFoldBenchmarks>() |> ignore
+                0
+            | Some "pipeline" ->
+                printfn "Running Pipeline Composition Benchmarks..."
+                BenchmarkRunner.Run<AsyncSeqPipelineBenchmarks>() |> ignore
+                0
             | Some "all" | None ->
                 printfn "Running All Benchmarks..."
                 BenchmarkRunner.Run<AsyncSeqCoreBenchmarks>() |> ignore
                 BenchmarkRunner.Run<AsyncSeqAppendBenchmarks>() |> ignore
                 BenchmarkRunner.Run<AsyncSeqBuilderBenchmarks>() |> ignore
+                BenchmarkRunner.Run<AsyncSeqFilterChooseFoldBenchmarks>() |> ignore
+                BenchmarkRunner.Run<AsyncSeqPipelineBenchmarks>() |> ignore
                 0
             | Some suite ->
                 printfn "Unknown benchmark suite: %s" suite
-                printfn "Available suites: core, append, builder, all"
+                printfn "Available suites: core, append, builder, filter-choose-fold, pipeline, all"
                 1
 
         printfn ""