Async.ParallelIgnore

Author: eulerfx

src/FSharpx.Async/Async.fs

@@ -61,6 +61,50 @@ module AsyncExtensions =
         let! c = c
         return a,b,c }
 
+      /// Creates an async computation which runs the provided sequence of computations and completes
+      /// when all computations in the sequence complete. Up to parallelism computations will
+      /// be in-flight at any given point in time. Error or cancellation of any computation in
+      /// the sequence causes the resulting computation to error or cancel, respectively.
+      static member ParallelIgnore (parallelism:int) (xs:seq<Async<_>>) = async {
+                
+        let sm = new SemaphoreSlim(parallelism)
+        let cde = new CountdownEvent(1)
+        let tcs = new TaskCompletionSource<unit>()
+    
+        let inline ok _ =
+          sm.Release() |> ignore
+          if (cde.Signal()) then
+            tcs.SetResult(())
+                
+        let inline err (ex:exn) =
+          tcs.SetException ex
+          sm.Release() |> ignore
+                
+        let inline cnc (ex:OperationCanceledException) =      
+          tcs.SetCanceled()
+          sm.Release() |> ignore
+
+        try
+
+          for computation in xs do
+            sm.Wait()
+            cde.AddCount(1)            
+            // the following decreases throughput 3x but avoids blocking
+            // do! sm.WaitAsync() |> Async.AwaitTask
+            Async.StartWithContinuations(computation, ok, err, cnc)                    
+      
+          if (cde.Signal()) then
+            tcs.SetResult(())
+
+          do! tcs.Task |> Async.AwaitTask
+
+        finally
+      
+          cde.Dispose()    
+          sm.Dispose()
+                          
+        }
+
       /// An async computation which does nothing.
       static member inline unit = AsyncOps.unit
 

tests/FSharpx.Async.Tests/AsyncTest.fs

@@ -1,5 +1,46 @@
-
 module FSharpx.Async.Tests.AsyncTest
 
+open System
+open System.Threading
+open System.Threading.Tasks
+open NUnit.Framework
+open FSharpx.Control
+
+[<Test>]
+let ``Async.ParallelIgnore should run argument computations``() =  
+  let bag = System.Collections.Concurrent.ConcurrentBag<_>()  
+  let s = Seq.init 10 id |> Set.ofSeq    
+  s 
+  |> Seq.map (fun i -> bag.Add i ; Async.unit)
+  |> Async.ParallelIgnore 1
+  |> Async.RunSynchronously
+  Assert.True((s = (bag |> Set.ofSeq)))
+
+[<Test>]
+let ``Async.ParallelIgnore should fail upon first failure``() =
+  let s =
+    [
+      async { return failwith "catch me if you can" }
+    ]
+  Assert.Throws<AggregateException>(fun() ->
+    s
+    |> Async.ParallelIgnore 1
+    |> Async.RunSynchronously
+  )
+  |> ignore
 
-// TODO:
+[<Test>]
+let ``Async.ParallelIgnore should cancel upon first cancellation``() =
+  let tcs = new TaskCompletionSource<unit>()
+  let s =
+    [
+      tcs.Task |> Async.AwaitTask
+    ]
+  tcs.SetCanceled()
+  Assert.Throws<OperationCanceledException>(fun() ->
+    s
+    |> Async.ParallelIgnore 1
+    |> Async.RunSynchronously
+  )
+  |> ignore
+