@@ -76,16 +76,14 @@ import Prelude hiding (take, drop)
7676-- Concurrency Semantics
7777------------------------------------------------------------------------------
7878--
79−-- Asyncly is essentially a concurrent list transformer. To concatenate the
80−-- lists it provides three different ways of composing them. The Monoid
81−-- instance concatenates lists in a non-concurrent in-order fashion. The
82−-- Alternative instance concatenates list in a concurrent manner potentially
83−-- running each action in parallel and therefore the order of the resulting
84−-- items is not deterministic. Computations in Alternative composition may or
85−-- may not run in parallel. Thirdly, the <||> operator provides a way to
86−-- compose computations that are guaranteed to run in parallel. This provides a
87−-- way to run infinite loops or long running computations without blocking
88−-- or starving other computations.
79+-- Asyncly is essentially a concurrent list transformer. To fold the lists
80+-- (AsyncT m a) it provides many different ways of composing them. The Monoid
81+-- instance folds lists in a sequential fashion. The Alternative instance folds
82+-- a list in a concurrent manner running each action in parallel and therefore
83+-- the order of results in the fold is not deterministic, but the results are
84+-- fairly interleaved. There are other ways to bind and compose. The
85+-- combinations of different types of binds and monoidal compositions can be
86+-- used to achieve the desired results.
8987
9088------------------------------------------------------------------------------
9189-- Parent child thread communication types
@@ -108,6 +106,10 @@ data Context m a =
108106 , queueEmpty :: m Bool
109107 }
110108
109+-- | Represents a monadic stream of values of type 'a' resulting from actions
110+-- in monad 'm'. 'AsyncT' streams can be composed sequentially or in parallel
111+-- in various ways using monadic bind or variants of it and using monoidal
112+-- compositions like 'Monoid', 'Alternative' or variants of these.
111113newtype AsyncT m a =
112114 AsyncT {
113115 runAsyncT :: forall r.
@@ -117,13 +119,14 @@ newtype AsyncT m a =
117119 -> m r
118120 }
119121
122+-- | A monad that can do asynchronous (or parallel) IO operations.
120123type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
121124
122125------------------------------------------------------------------------------
123126-- Monoid
124127------------------------------------------------------------------------------
125128
126−-- | Appends the results of two AsyncT computations in order.
129+-- | '<>' concatenates two 'AsyncT' streams sequentially.
127130instance Semigroup (AsyncT m a) where
128131 m1 <> m2 = go m1
129132 where
@@ -133,12 +136,11 @@ instance Semigroup (AsyncT m a) where
133136 yield a c (Just r) = yld a c (Just (go r))
134137 in m ctx stop yield
135138
136−-- | Appends the results of two AsyncT computations in order.
137139instance Monoid (AsyncT m a) where
138140 mempty = AsyncT $ \_ stp _ -> stp
139141 mappend = (<>)
140142
141−-- | Interleaves the results of two 'AsyncT' streams
143+-- | Same as '<=>'.
142144interleave :: AsyncT m a -> AsyncT m a -> AsyncT m a
143145interleave m1 m2 = AsyncT $ \ctx stp yld -> do
144146 let stop = (runAsyncT m2) ctx stp yld
@@ -148,6 +150,8 @@ interleave m1 m2 = AsyncT $ \ctx stp yld -> do
148150
149151infixr 5 <=>
150152
153+-- | Sequential compostion, similar to '<>' except that it interleaves the two
154+-- 'AsyncT' streams fairly.
151155(<=>) :: AsyncT m a -> AsyncT m a -> AsyncT m a
152156(<=>) = interleave
153157
@@ -177,21 +181,6 @@ instance Monad m => Applicative (AsyncT m) where
177181-- Monad
178182------------------------------------------------------------------------------
179183
180−-- We do not use bind for parallelism. That is, we do not start each iteration
181−-- of the list in parallel. That will introduce too much uncontrolled
182−-- parallelism. Instead we control parallelism using <|>, it is used for
183−-- actions that 'may' be run in parallel. The library decides whether they will
184−-- be started in parallel. This puts the parallel semantics in the hands of the
185−-- user and operational semantics in the hands of the library. Ideally we would
186−-- want parallelism to be completely hidden from the user, for example we can
187−-- automatically decide where to start an action in parallel so that it is most
188−-- efficient. We can decide whether a particular bind should fork a parallel
189−-- thread or not based on the level of the bind in the tree. Maybe at some
190−-- point we shall be able to do that?
191−
192−-- XXX Using a common bindWith function is 15% slower than duplicating the
193−-- code. In case that performance is important we can duplicate it.
194−--
195184-- | A thread context is valid only until the next bind. Upon a bind we
196185-- reset the context to Nothing.
197186{-# INLINE bindWith #-}
@@ -209,6 +198,8 @@ bindWith k m f = go m
209198 yield a _ (Just r) = run $ f a `k` (go r)
210199 in g Nothing stp yield
211200
201+-- | '>>=' runs a monadic action for each element in the stream, sequentially,
202+-- in a depth first manner.
212203instance Monad m => Monad (AsyncT m) where
213204 return = pure
214205
@@ -218,18 +209,34 @@ instance Monad m => Monad (AsyncT m) where
218209 yield a _ (Just r) = run $ f a <> (r >>= f)
219210 in m Nothing stp yield
220211
221−-- | Serial, interleaved bind
212+------------------------------------------------------------------------------
213+-- Alternative ways to bind
214+------------------------------------------------------------------------------
215+
216+-- | Run a monadic action for each element in the stream, sequentially, in a
217+-- breadth first manner i.e. fairly interleaved.
222218(>->) :: AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
223219(>->) = bindWith (<=>)
224220
225−-- | Parallel, depth first bind
221+-- | Run a monadic action for each element in the stream, in parallel, with
222+-- a depth first bias.
226223(>>|) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
227224(>>|) = bindWith (<|)
228225
229−-- | Parallel, breadth first bind
226+-- | Run a monadic action for each element in the stream, in parallel, with a
227+-- breadth first bias i.e. fairly interleaved.
230228(>|>) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
231229(>|>) = bindWith (<|>)
232230
231+-- TBD Ideally we would want parallelism to be completely hidden from the user,
232+-- for example we can automatically decide where to start an action in parallel
233+-- so that it is most efficient. We can decide whether a particular bind should
234+-- fork a parallel thread or not based on the level of the bind in the tree.
235+-- This is equivalent to dynamically deciding (maybe based user policies
236+-- applied using combinators) which type of bind or monoidal composition we
237+-- should use. We can have another bind operator (maybe the standard >>= should
238+-- be used for that) for this type of behavior.
239+
233240------------------------------------------------------------------------------
234241-- Alternative
235242------------------------------------------------------------------------------
@@ -473,30 +480,41 @@ parallel m1 m2 fifo = AsyncT $ \ctx stp yld -> do
473480 Nothing -> (runAsyncT (pullFork m1 m2 fifo)) Nothing stp yld
474481 Just c -> liftIO ((enqueue c) m2) >> (runAsyncT m1) ctx stp yld
475482
483+-- | `empty` represents an action that takes non-zero time to complete. Since
484+-- all actions take non-zero time, an `Alternative` composition ('<|>') is a
485+-- monoidal composition executing all actions in parallel, it is similar to
486+-- '<>' except that it runs all the actions in parallel and interleaves their
487+-- results fairly.
476488instance MonadAsync m => Alternative (AsyncT m) where
477489 empty = mempty
478490
479491 {-# INLINE (<|>) #-}
480492 m1 <|> m2 = parallel m1 m2 True
481493
482−-- | Just like '<>' except that it can execute the action on the right in
483−-- parallel ahead of time. Returns the results in serial order like '<>' from
484−-- left to right.
494+------------------------------------------------------------------------------
495+-- Other monoidal compositions for parallel actions
496+------------------------------------------------------------------------------
497+
498+-- | Same as '<>|'.
485499parAhead :: AsyncT m a -> AsyncT m a -> AsyncT m a
486500parAhead = undefined
487501
502+-- | Sequential composition similar to '<>' except that it can execute the
503+-- action on the right in parallel ahead of time. Returns the results in
504+-- sequential order like '<>' from left to right.
488505(<>|) :: AsyncT m a -> AsyncT m a -> AsyncT m a
489506(<>|) = parAhead
490507
491−-- | Left biased parallel execution. Actions may run in parallel but not
492−-- necessarily, the action on the left is executed before the one on the right
493−-- when parallelism is not needed. This combinator is useful when fairness is
494−-- not required.
508+-- | Same as '<|'.
495509{-# INLINE parLeft #-}
496510parLeft :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
497511parLeft m1 m2 = parallel m1 m2 False
498512
499−-- | Same as 'parLeft'.
513+-- | Parallel composition similar to `<|>` except that it is left-biased
514+-- instead of being fair. Action on the left is likely to be given a chance to
515+-- execute before the action on the right. If the left action keeps yielding
516+-- results without blocking it continues running until it finishes and only
517+-- then the right action runs.
500518{-# INLINE (<|) #-}
501519(<|) :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
502520(<|) = parLeft
@@ -594,7 +612,7 @@ instance MonadState s m => MonadState s (AsyncT m) where
594612-- Running the monad
595613------------------------------------------------------------------------------
596614
597−-- | Run an 'AsyncT m' computation, wait for it to finish and discard the
615+-- | Run an 'AsyncT' computation, wait for it to finish and discard the
598616-- results.
599617runAsyncly :: MonadAsync m => AsyncT m a -> m ()
600618runAsyncly m = run Nothing m
@@ -609,8 +627,7 @@ runAsyncly m = run Nothing m
609627
610628 run ct mx = (runAsyncT mx) ct stop yield
611629
612−-- | Run an 'AsyncT m' computation and collect the results generated by each
613−-- thread of the computation in a list.
630+-- | Collect the results of an 'AsyncT' stream into a list.
614631{-# INLINABLE toList #-}
615632toList :: MonadAsync m => AsyncT m a -> m [a]
616633toList m = run Nothing m
@@ -630,6 +647,7 @@ toList m = run Nothing m
630647-- Transformation
631648------------------------------------------------------------------------------
632649
650+-- | Take first 'n' elements from the stream and discard the rest.
633651take :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
634652take n m = AsyncT $ \ctx stp yld -> do
635653 let yield a c Nothing = yld a c Nothing
@@ -638,6 +656,7 @@ take n m = AsyncT $ \ctx stp yld -> do
638656 then stp
639657 else (runAsyncT m) ctx stp yield
640658
659+-- | Discard first 'n' elements from the stream and take the rest.
641660drop :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
642661drop n m = AsyncT $ \ctx stp yld -> do
643662 let yield _ _ Nothing = stp
@@ -662,7 +681,7 @@ foldMapWith :: (Monoid b, Foldable t) =>
662681 (b1 -> b -> b) -> (a -> b1) -> t a -> b
663682foldMapWith f g = foldr (f . g) mempty
664683
665−-- | Fold a 'Foldable' container using a function that is a compostioin of the
684+-- | Fold a 'Foldable' container using a function that is a composition of the
666685-- first and the third argument.
667686{-# INLINABLE forEachWith #-}
668687forEachWith :: (Monoid b, Foldable t) =>
@@ -673,5 +692,6 @@ forEachWith f xs g = foldr (f . g) mempty xs
673692-- Convert a callback into an 'AsyncT' computation
674693------------------------------------------------------------------------------
675694
695+-- | Convert a callback into an 'AsyncT' stream.
676696fromCallback :: (forall r. (a -> m r) -> m r) -> AsyncT m a
677697fromCallback k = AsyncT $ \ctx _ yld -> k (\a -> yld a ctx Nothing)