Merge pull request #47 from eulerfx/groupby

AsyncSeq.groupBy, AsyncSeqSrc, docs

Author: eulerfx

FSharp.Control.AsyncSeq.sln

@@ -39,6 +39,7 @@ EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "library", "library", "{B4BB88FA-DB8C-4651-9869-A8FEA79F044E}"
 	ProjectSection(SolutionItems) = preProject
 		docs\content\library\AsyncSeq.fsx = docs\content\library\AsyncSeq.fsx
+		docs\content\library\AsyncSeqExamples.fsx = docs\content\library\AsyncSeqExamples.fsx
 	EndProjectSection
 EndProject
 Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "FSharp.Control.AsyncSeq.Profile7", "src\FSharp.Control.AsyncSeq.Profile7\FSharp.Control.AsyncSeq.Profile7.fsproj", "{114DCF3C-D7CC-4E53-AFF0-AFBFC0C8B966}"

docs/content/index.md

@@ -22,7 +22,9 @@ This includes additional brief samples on using most of the functions.
 
 [Terminology](terminology.html) a reference for some of the terminology around F# async.
 
-[FSharp.Control.AsyncSeq](library/AsyncSeq.html) contains narrative and code samples explaining asynchronous sequences.
+[AsyncSeq](library/AsyncSeq.html) contains narrative and code samples explaining asynchronous sequences.
+
+[AsyncSeq Examples](library/AsyncSeqExamples.html) contains examples.
 
 Contributing and copyright
 --------------------------

docs/content/library/AsyncSeqExamples.fsx

@@ -0,0 +1,160 @@
+(**
+
+# F# AsyncSeq Examples
+
+*)
+
+#r "../../../bin/FSharp.Control.AsyncSeq.dll"
+open System
+open FSharp.Control
+
+
+(**
+
+### Group By
+
+Suppose we would like to consume a stream of events `AsyncSeq<Event>` and perform an operation on each event. 
+The operation on each event is of type `Event -> Async<unit>`. This can be done as follows:
+
+*)
+
+
+type Event = {
+  entityId : int64
+  data : string 
+}
+
+let stream : AsyncSeq<Event> =
+  failwith "undefined"
+
+let action (e:Event) : Async<unit> =
+  failwith "undefined"
+
+stream 
+|> AsyncSeq.iterAsync action
+
+
+(**
+
+The above workflow will read an event from the stream, perform an operation and then read the next event.
+While the read operation and the operation on the event are *asynchronous*, the stream is processed *sequentially*.
+It may be desirable to parallelize the processing of the stream. Suppose that events correspond to some entity, 
+such as a shopping cart. Events belonging to some shopping cart must be processed in a sequential order, however they
+are independent from events belonging to other shopping carts. Therefore, events belonging to distinct shopping carts
+can be processed in parallel. Using `AsyncSeq.groupBy`, we can partition the stream into a fixed set of sub-streams 
+and then process the sub-streams in parallel using `AsyncSeq.mapAsyncParallel`:
+
+*)
+
+stream
+|> AsyncSeq.groupBy (fun e -> int e.entityId % 4)
+|> AsyncSeq.mapAsyncParallel (snd >> AsyncSeq.iterAsync action)
+|> AsyncSeq.iter ignore
+
+(**
+
+`AsyncSeq.groupBy` partitions the input sequence into sub-sequences based on a key returned by a projection function. 
+The resulting sub-sequences emit elements when the source sequence emits an element corresponding to the key of the 
+sub-sequence. Elements of the resulting sequence are pairs of keys and sub-sequences, in this case `int * AsyncSeq<Event>`. Since by definition, these sub-sequences are independent, they can be processed in parallel. In fact, the sub-sequences *must* be processed in parallel, because it isn't possible to complete the processing of a sub-sequence until all elements of the source sequence are exhausted.
+
+To continue improving the efficiency of our workflow, we can make use of batching. Specifically, we can read the incoming
+events in batches and we can perform actions on entire batches of events.
+
+*)
+
+let batchStream : AsyncSeq<Event[]> =
+  failwith "undefined"
+
+let batchAction (es:Event[]) : Async<unit> =
+  failwith "undefined"
+
+
+(**
+
+Ordering is still important. For example, the batch action could write events into a full-text search index. We would like the full-text search index to be sequentially consistent. As such, the events need to be applied in the order they were emitted. The following workflow has the desired properties:
+
+*)
+
+batchStream
+|> AsyncSeq.concatSeq // flatten the sequence of event arrays
+|> AsyncSeq.groupBy (fun e -> int e.entityId % 4) // partition into 4 groups
+|> AsyncSeq.mapAsyncParallel (snd 
+  >> AsyncSeq.bufferByCountAndTime 500 1000 // buffer sub-sequences
+  >> AsyncSeq.iterAsync batchAction) // perform the batch operation
+|> AsyncSeq.iter ignore
+
+
+(**
+
+The above workflow:
+
+1. Reads events in batches.
+2. Flattens the batches.
+3. Partitions the events into mutually exclusive sub-sequences.
+4. Buffers elements of each sub-sequence by time and space.
+5. Processes the sub-sequences in parallel, but individual sub-sequences sequentially.
+
+*)
+
+
+
+(**
+
+### Merge
+
+`AsyncSeq.merge` non-deterministically merges two async sequences into one. It is non-deterministic in the sense that the resulting sequence emits elements 
+whenever *either* input sequence emits a value. Since it isn't always known which will emit a value first, if at all, the operation is non-deterministic. This operation
+is in contrast to `AsyncSeq.zip` which also takes two async sequences and returns a single async sequence, but as opposed to emitting an element when *either* input
+sequence produces a value, it emits an element when *both* sequences emit a value.
+
+Suppose we would like to trigger an operation whenever a change occurs. We can represent changes as an `AsyncSeq`. To gain intuition for this, consider the [Consul](https://www.consul.io/)
+configuration management system. It stores configuration information in a tree-like structure. For this purpose of this discussion, it can be thought of as a key-value store
+exposed via HTTP. In addition, `Consul` supports change notifications using HTTP long-polling - when an HTTP GET request is made to retrieve the value of a key, 
+if the request specified a modify-index, `Consul` won't respond to the request until a change has occurred *since* the modify-index. We can represent this operation using 
+the type `Key * ModifyIndex -> Async<Value * ModifyIndex>`. Next, we can take this operation and turn it into an `AsyncSeq` of changes as follows:
+*)
+
+type Key = string
+
+type Value = string
+
+type ModifyIndex = int64
+
+let longPollKey (key:Key, mi:ModifyIndex) : Async<Value * ModifyIndex> =
+  failwith "undefined"
+
+let changes (key:Key, mi:ModifyIndex) : AsyncSeq<Value> =
+  AsyncSeq.unfoldAsync 
+    (fun (mi:ModifyIndex) -> async {
+      let! value,mi = longPollKey (key, mi)
+      return Some (value,mi) })
+    mi
+
+(**
+
+The function `changes` produces an async sequence which emits elements whenever the value corresponding to the key changes. Suppose also that we would like to trigger an operation
+whenever the key changes or based on a fixed interval. We can represent a fixed interval as an async sequence as follows:
+
+*)
+
+let intervalMs (periodMs:int) = asyncSeq {
+  yield DateTime.UtcNow
+  while true do
+    do! Async.Sleep periodMs
+    yield DateTime.UtcNow }
+
+(**
+
+Putting it all together:
+
+*)
+
+let changesOrInterval : AsyncSeq<Choice<Value, DateTime>> =
+  AsyncSeq.mergeChoice (changes ("myKey", 0L)) (intervalMs (1000 * 60))
+
+
+(**
+
+We can now consume this async sequence and use it to trigger downstream operations, such as updating the configuration of a running program, in flight.
+
+*)

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -25,6 +25,12 @@ type IAsyncEnumerable<'T> =
 type AsyncSeq<'T> = IAsyncEnumerable<'T>
 //    abstract GetEnumerator : unit -> IAsyncEnumerator<'T>
 
+type AsyncSeqSrc<'a> = private { tail : AsyncSeqSrcNode<'a> ref }
+
+and private AsyncSeqSrcNode<'a> =
+  val tcs : TaskCompletionSource<('a * AsyncSeqSrcNode<'a>) option>
+  new (tcs) = { tcs = tcs }
+
 [<AutoOpen>]
 module internal Utils = 
     module internal Choice =
@@ -79,6 +85,51 @@ module internal Utils =
             elif i = 1 then return (Choice2Of2 (b.Result, a)) 
             else return! failwith (sprintf "unreachable, i = %d" i) }
 
+
+    type internal MbReq<'a> =
+      | Put of 'a
+      | Take of AsyncReplyChannel<'a>
+
+    /// An unbounded FIFO mailbox.
+    type Mb<'a> internal () =
+
+      let agent = MailboxProcessor.Start (fun agent ->
+        let rec receive () = async {
+          return!
+            agent.Scan(function
+              | Put a -> Some (send a)
+              | _ -> None ) }
+        and send (a:'a) = async {
+          return!
+            agent.Scan(function
+              | Take rep -> Some (rep.Reply a ; receive ())
+              | _ -> None) }
+        receive ())
+
+      member __.Put (a:'a) =
+        agent.Post (Put a)
+
+      member __.Take =
+        agent.PostAndAsyncReply (Take)
+
+      interface IDisposable with
+        member __.Dispose () = (agent :> IDisposable).Dispose()
+
+
+    /// Operations on unbounded FIFO mailboxes.
+    module Mb =
+  
+      /// Creates a new unbounded mailbox.
+      let create () = new Mb<'a> ()
+
+      /// Puts a message into a mailbox, no waiting.
+      let put (a:'a) (mb:Mb<'a>) = mb.Put a
+
+      /// Creates an async computation that completes when a message is available in a mailbox.
+      let take (mb:Mb<'a>) = mb.Take    
+     
+
+
 /// Module with helper functions for working with asynchronous sequences
 module AsyncSeq = 
 
@@ -549,6 +600,22 @@ module AsyncSeq =
       i := i.Value + 1L
       yield v }
 
+  let mapAsyncParallel (f:'a -> Async<'b>) (s:AsyncSeq<'a>) = asyncSeq {
+    use mb = Mb.create ()
+    do! s |> iterAsync (fun a -> async {
+      let! b = Async.StartChild (f a)
+      mb |> Mb.put (Some b) })
+    mb.Put None
+    let rec loop () = asyncSeq {
+      let! b = Mb.take mb
+      match b with
+      | None -> ()
+      | Some b -> 
+        let! b = b
+        yield b
+        yield! loop () }
+    yield! loop () }
+
   let chooseAsync f (source : AsyncSeq<'T>) : AsyncSeq<'R> = asyncSeq {
     for itm in source do
       let! v = f itm
@@ -621,6 +688,26 @@ module AsyncSeq =
           let! moven = ie.MoveNext()
           b := moven }
 
+  let pickAsync (f:'T -> Async<'U option>) (source:AsyncSeq<'T>) = async { 
+      use ie = source.GetEnumerator() 
+      let! v = ie.MoveNext()
+      let b = ref v
+      let res = ref None
+      while b.Value.IsSome && not res.Value.IsSome do
+          let! fv = f b.Value.Value
+          match fv with 
+          | None -> 
+              let! moven = ie.MoveNext()
+              b := moven
+          | Some _ as r -> 
+              res := r
+      match res.Value with
+      | Some _ -> return res.Value.Value
+      | None -> return raise(KeyNotFoundException()) }
+
+  let pick f (source:AsyncSeq<'T>) =
+    pickAsync (f >> async.Return) source
+
   let tryPickAsync f (source : AsyncSeq<'T>) = async { 
       use ie = source.GetEnumerator() 
       let! v = ie.MoveNext()
@@ -1200,6 +1287,79 @@ module AsyncSeq =
        }
 
 
+  module AsyncSeqSrcImpl =
+
+    let private createNode () =
+      new AsyncSeqSrcNode<_>(new TaskCompletionSource<_>())
+    
+    let create () : AsyncSeqSrc<'a> =
+      { tail = ref (createNode ()) }
+      
+    let put (a:'a) (s:AsyncSeqSrc<'a>) =      
+      let newTail = createNode ()
+      let tail = Interlocked.Exchange(s.tail, newTail)
+      tail.tcs.SetResult(Some(a, newTail))
+      
+    let close (s:AsyncSeqSrc<'a>) : unit =
+      s.tail.Value.tcs.SetResult(None)
+
+    let error (ex:exn) (s:AsyncSeqSrc<'a>) : unit =
+      s.tail.Value.tcs.SetException(ex)
+
+    let rec private toAsyncSeqImpl (s:AsyncSeqSrcNode<'a>) : AsyncSeq<'a> = 
+      asyncSeq {
+        let! next = s.tcs.Task |> Async.AwaitTask
+        match next with
+        | None -> ()
+        | Some (a,tl) ->
+          yield a
+          yield! toAsyncSeqImpl tl }
+
+    let toAsyncSeq (s:AsyncSeqSrc<'a>) : AsyncSeq<'a> =
+      toAsyncSeqImpl s.tail.Value
+      
+
+  
+  type private Group<'k, 'a> = { key : 'k ; src : AsyncSeqSrc<'a> }
+    
+  let groupByAsync (p:'a -> Async<'k>) (s:AsyncSeq<'a>) : AsyncSeq<'k * AsyncSeq<'a>> = asyncSeq {
+    let groups = Collections.Generic.Dictionary<'k, Group<'k, 'a>>()
+    let close group =
+      groups.Remove(group.key) |> ignore
+      AsyncSeqSrcImpl.close group.src
+    let closeGroups () =
+      groups.Values |> Seq.toArray |> Array.iter close
+    use enum = s.GetEnumerator()
+    let rec go () = asyncSeq {
+      try            
+        let! next = enum.MoveNext ()
+        match next with
+        | None -> closeGroups ()
+        | Some a ->
+          let! key = p a
+          let mutable group = Unchecked.defaultof<_>
+          if groups.TryGetValue(key, &group) then
+            AsyncSeqSrcImpl.put a group.src
+            yield! go ()
+          else
+            let src = AsyncSeqSrcImpl.create ()
+            let subSeq = src |> AsyncSeqSrcImpl.toAsyncSeq
+            AsyncSeqSrcImpl.put a src
+            let group = { key = key ; src = src }
+            groups.Add(key, group)
+            yield key,subSeq
+            yield! go ()
+      with ex ->
+        closeGroups ()
+        raise ex }
+    yield! go () }
+
+  let groupBy (p:'a -> 'k) (s:AsyncSeq<'a>) : AsyncSeq<'k * AsyncSeq<'a>> =
+    groupByAsync (p >> async.Return) s
+
+
+
+
 [<AutoOpen>]
 module AsyncSeqExtensions = 
   let asyncSeq = new AsyncSeq.AsyncSeqBuilder()
@@ -1209,6 +1369,14 @@ module AsyncSeqExtensions =
     member x.For (seq:AsyncSeq<'T>, action:'T -> Async<unit>) = 
       seq |> AsyncSeq.iterAsync action 
 
+module AsyncSeqSrc =
+    
+  let create () = AsyncSeq.AsyncSeqSrcImpl.create ()
+  let put a s = AsyncSeq.AsyncSeqSrcImpl.put a s
+  let close s = AsyncSeq.AsyncSeqSrcImpl.close s
+  let toAsyncSeq s = AsyncSeq.AsyncSeqSrcImpl.toAsyncSeq s
+  let error e s = AsyncSeq.AsyncSeqSrcImpl.error e s
+
 module Seq = 
 
   let ofAsyncSeq (source : AsyncSeq<'T>) =