Merge pull request #20 from eulerfx/master

Async.bindChoice, AsyncSeq.zipWithAsync, AsyncSeq.merge, etc

Author: eulerfx

RELEASE_NOTES.md

@@ -1,5 +1,6 @@
 ### 1.11.1 - 26.03.2015
 * BUGFIX: AsyncSeq.toBlockingSeq does not hung forever if an exception is thrown and reraise it outside - https://github.com/fsprojects/FSharpx.Async/pull/21
+* Added Async.bindChoice, Async.ParallelIgnore, AsyncSeq.zipWithAsync, AsyncSeq.zappAsync, AsyncSeq.threadStateAsync, AsyncSeq.merge, AsyncSeq.traverseOptionAsync, AsyncSeq.traverseChoiceAsync
 
 ### 1.11.0 - 27.02.2015
 * Added Async.map, Async.bind, Async.unit

src/FSharpx.Async/Async.fs

@@ -5,6 +5,7 @@
 namespace FSharpx.Control
 open System
 open System.Threading
+open System.Threading.Tasks
 
 // ----------------------------------------------------------------------------
 
@@ -60,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
 
@@ -70,4 +115,47 @@ module AsyncExtensions =
       /// Creates an async computation which binds the result of the specified 
       /// async computation to the specified function. The computation produced 
       /// by the specified function is returned.
-      static member inline bind f a = async.Bind(a, f)
+      static member inline bind f a = async.Bind(a, f)
+
+      /// Maps over an async computation which produces a choice value
+      /// using a function which maps over Choice1Of2 and itself returns a choice. 
+      /// A value of Choice2Of2 is treated like an error and passed through.
+      static member mapChoice (f:'a -> Choice<'b, 'e>) (a:Async<Choice<'a, 'e>>) : Async<Choice<'b, 'e>> = 
+        a |> Async.map (function 
+          | Choice1Of2 a' -> f a'
+          | Choice2Of2 e -> Choice2Of2 e)
+
+      /// Binds an async computation producing a choice value to another async
+      /// computation producing a choice such that a Choice2Of2 value is passed through.
+      static member bindChoice (f:'a -> Async<Choice<'b, 'e>>) (a:Async<Choice<'a, 'e>>) : Async<Choice<'b, 'e>> = 
+        a |> Async.bind (function
+          | Choice1Of2 a' -> f a'
+          | Choice2Of2 e ->  Choice2Of2 e |> async.Return)
+
+      /// Binds an async computation producing a choice value to another async
+      /// computation producing a choice such that a Choice2Of2 value is passed through.
+      static member bindChoices (f:'a -> Async<Choice<'b, 'e2>>) (a:Async<Choice<'a, 'e1>>) : Async<Choice<'b, Choice<'e1, 'e2>>> = 
+        a |> Async.bind (function
+          | Choice1Of2 a' -> f a' |> Async.map (function Choice1Of2 b -> Choice1Of2 b | Choice2Of2 e2 -> Choice2Of2 (Choice2Of2 e2))
+          | Choice2Of2 e1 -> Choice2Of2 (Choice1Of2 e1) |> async.Return)
+
+      /// Creates a computation which produces a tuple consiting of the value produces by the first
+      /// argument computation to complete and a handle to the other computation. The second computation
+      /// to complete is memoized.
+      static member internal chooseBoth (a:Async<'a>) (b:Async<'a>) : Async<'a * Async<'a>> =
+        Async.FromContinuations <| fun (ok,err,cnc) ->
+          let state = ref 0            
+          let tcs = TaskCompletionSource<'a>()            
+          let inline ok a =
+            if (Interlocked.CompareExchange(state, 1, 0) = 0) then
+              ok (a, tcs.Task |> Async.AwaitTask)
+            else
+              tcs.SetResult a
+          let inline err (ex:exn) =
+            if (Interlocked.CompareExchange(state, 1, 0) = 0) then err ex
+            else tcs.SetException ex
+          let inline cnc ex =
+            if (Interlocked.CompareExchange(state, 1, 0) = 0) then cnc ex
+            else tcs.SetCanceled()
+          Async.StartWithContinuations(a, ok, err, cnc)
+          Async.StartWithContinuations(b, ok, err, cnc)

src/FSharpx.Async/AsyncSeq.fs

@@ -7,6 +7,7 @@ namespace FSharpx.Control
 open System
 open System.Threading
 open System.IO
+open FSharpx.Control.Utils
 
 // ----------------------------------------------------------------------------
 
@@ -34,6 +35,17 @@ module AsyncSeq =
   /// Creates an asynchronous sequence that generates a single element and then ends
   let singleton (v:'T) : AsyncSeq<'T> = 
     async { return Cons(v, empty) }
+    
+  /// Generates an async sequence using the specified generator function.
+  let rec unfoldAsync (f:'State -> Async<('T * 'State) option>) (s:'State) : AsyncSeq<'T> = 
+    f s
+    |> Async.map (function
+      | Some (a,s) -> Cons(a, unfoldAsync f s)
+      | None -> Nil)
+
+  /// Creates an async sequence which repeats the specified value indefinitely.
+  let rec replicate (v:'T) : AsyncSeq<'T> =    
+    Cons(v, async.Delay <| fun() -> replicate v) |> async.Return    
 
   /// Yields all elements of the first asynchronous sequence and then 
   /// all elements of the second asynchronous sequence.
@@ -413,8 +425,19 @@ module AsyncSeq =
 
   // --------------------------------------------------------------------------
 
+  /// Threads a state through the mapping over an async sequence using an async function.
+  let rec threadStateAsync (f:'s -> 'a -> Async<'b * 's>) (st:'s) (s:AsyncSeq<'a>) : AsyncSeq<'b> = asyncSeq {
+    let! s = s
+    match s with
+    | Nil -> ()
+    | Cons(a,tl) ->       
+      let! b,st' = f st a
+      yield b
+      yield! threadStateAsync f st' tl }
+
   /// Combines two asynchronous sequences into a sequence of pairs. 
   /// The values from sequences are retrieved in parallel. 
+  /// The resulting sequence stops when either of the argument sequences stop.
   let rec zip (input1 : AsyncSeq<'T1>) (input2 : AsyncSeq<'T2>) : AsyncSeq<_> = async {
     let! ft = input1 |> Async.StartChild
     let! s = input2
@@ -424,6 +447,65 @@ module AsyncSeq =
         return Cons( (hf, hs), zip tf ts)
     | _ -> return Nil }
 
+  /// Combines two asynchronous sequences using the specified function.
+  /// The values from sequences are retrieved in parallel.
+  /// The resulting sequence stops when either of the argument sequences stop.
+  let rec zipWithAsync (z:'a -> 'b -> Async<'c>) (a:AsyncSeq<'a>) (b:AsyncSeq<'b>) : AsyncSeq<'c> = async {
+    let! a,b = Async.Parallel(a, b)    
+    match a,b with 
+    | Cons(a, atl), Cons(b, btl) ->
+      let! c = z a b
+      return Cons(c, zipWithAsync z atl btl)
+    | _ -> return Nil }
+
+  /// Combines two asynchronous sequences using the specified function.
+  /// The values from sequences are retrieved in parallel.
+  /// The resulting sequence stops when either of the argument sequences stop.
+  let inline zipWith (z:'a -> 'b -> 'c) (a:AsyncSeq<'a>) (b:AsyncSeq<'b>) : AsyncSeq<'c> =
+    zipWithAsync (fun a b -> z a b |> async.Return) a b
+
+  /// Combines two asynchronous sequences using the specified function to which it also passes the index.
+  /// The values from sequences are retrieved in parallel.
+  /// The resulting sequence stops when either of the argument sequences stop.
+  let inline zipWithIndexAsync (f:int -> 'a -> Async<'b>) (s:AsyncSeq<'a>) : AsyncSeq<'b> =
+    threadStateAsync (fun i a -> f i a |> Async.map (fun b -> b,i + 1)) 0 s        
+
+  /// Feeds an async sequence of values into an async sequence of async functions.
+  let inline zappAsync (fs:AsyncSeq<'a -> Async<'b>>) (s:AsyncSeq<'a>) : AsyncSeq<'b> =
+    zipWithAsync ((|>)) s fs
+
+  /// Feeds an async sequence of values into an async sequence of functions.
+  let inline zapp (fs:AsyncSeq<'a -> 'b>) (s:AsyncSeq<'a>) : AsyncSeq<'b> =
+    zipWith ((|>)) s fs
+
+  /// Traverses an async sequence an applies to specified function such that if None is returned the traversal short-circuits
+  /// and None is returned as the result. Otherwise, the entire sequence is traversed and the result returned as Some.
+  let rec traverseOptionAsync (f:'a -> Async<'b option>) (s:AsyncSeq<'a>) : Async<AsyncSeq<'b> option> = async {
+    let! s = s
+    match s with
+    | Nil -> return Some (Nil |> async.Return)
+    | Cons(a,tl) ->
+      let! b = f a
+      match b with
+      | Some b -> 
+        return! traverseOptionAsync f tl |> Async.map (Option.map (fun tl -> Cons(b, tl) |> async.Return))
+      | None -> 
+        return None }
+
+  /// Traverses an async sequence an applies to specified function such that if Choice2Of2 is returned the traversal short-circuits
+  /// and Choice2Of2 is returned as the result. Otherwise, the entire sequence is traversed and the result returned as Choice1Of2.
+  let rec traverseChoiceAsync (f:'a -> Async<Choice<'b, 'e>>) (s:AsyncSeq<'a>) : Async<Choice<AsyncSeq<'b>, 'e>> = async {
+    let! s = s
+    match s with
+    | Nil -> return Choice1Of2 (Nil |> async.Return)
+    | Cons(a,tl) ->
+      let! b = f a
+      match b with
+      | Choice1Of2 b -> 
+        return! traverseChoiceAsync f tl |> Async.map (Choice.mapl (fun tl -> Cons(b, tl) |> async.Return))
+      | Choice2Of2 e -> 
+        return Choice2Of2 e }
+
   /// Returns elements from an asynchronous sequence while the specified 
   /// predicate holds. The predicate is evaluated asynchronously.
   let rec takeWhileAsync p (input : AsyncSeq<'T>) : AsyncSeq<_> = async {
@@ -492,15 +574,6 @@ module AsyncSeq =
     |> fold (fun arr a -> a::arr) []
     |> Async.map List.rev
 
-  /// Generates an async sequence using the specified generator function.
-  let rec unfoldAsync (f:'State -> Async<('T * 'State) option>) (s:'State) : AsyncSeq<'T> = asyncSeq {        
-    let! r = f s
-    match r with
-    | Some (a,s) -> 
-      yield a
-      yield! unfoldAsync f s
-    | None -> () }
-
   /// Flattens an AsyncSeq of sequences.
   let rec concatSeq (input:AsyncSeq<#seq<'T>>) : AsyncSeq<'T> = asyncSeq {
     let! v = input
@@ -529,7 +602,6 @@ module AsyncSeq =
 
     left
 
-
   /// Buffer items from the async sequence into buffers of a specified size.
   /// The last buffer returned may be less than the specified buffer size.
   let rec bufferByCount (bufferSize:int) (s:AsyncSeq<'T>) : AsyncSeq<'T[]> = 
@@ -552,7 +624,24 @@ module AsyncSeq =
             return! loop tl            
       }
       return! loop s
-    }    
+    }
+
+  /// Merges two async sequences into an async sequence non-deterministically.
+  let rec merge (a:AsyncSeq<'a>) (b:AsyncSeq<'a>) : AsyncSeq<'a> = async {
+    let! one,other = Async.chooseBoth a b
+    match one with
+    | Nil -> return! other
+    | Cons(hd,tl) ->
+      return Cons(hd, merge tl other) }
+
+  /// Merges all specified async sequences into an async sequence non-deterministically.
+  let rec mergeAll (ss:AsyncSeq<'a> list) : AsyncSeq<'a> =
+    match ss with
+    | [] -> empty
+    | [s] -> s
+    | [a;b] -> merge a b
+    | hd::tl -> merge hd (mergeAll tl)
+    
 
 
 [<AutoOpen>]

src/FSharpx.Async/Utils.fs

@@ -1,3 +1,15 @@
-module FSharpx.Control.Utils
+module internal FSharpx.Control.Utils
 
-let inline konst a _ = a
+let inline konst a _ = a
+
+module Choice =
+  
+  /// Maps over the left result type.
+  let mapl (f:'a -> 'b) = function
+    | Choice1Of2 a -> f a |> Choice1Of2
+    | Choice2Of2 e -> Choice2Of2 e
+
+  /// Maps over the right result type.
+  let mapr (f:'b -> 'c) = function
+    | Choice1Of2 a -> Choice1Of2 a
+    | Choice2Of2 e -> f e |> Choice2Of2