@@ -76,16 +76,14 @@ import Prelude hiding (take, drop)
76-- Concurrency Semantics
76-- Concurrency Semantics
77------------------------------------------------------------------------------
77------------------------------------------------------------------------------
79−-- Asyncly is essentially a concurrent list transformer. To concatenate the
79+-- Asyncly is essentially a concurrent list transformer. To fold the lists
80−-- lists it provides three different ways of composing them. The Monoid
80+-- (AsyncT m a) it provides many different ways of composing them. The Monoid
81−-- instance concatenates lists in a non-concurrent in-order fashion. The
81+-- instance folds lists in a sequential fashion. The Alternative instance folds
82−-- Alternative instance concatenates list in a concurrent manner potentially
83−-- running each action in parallel and therefore the order of the resulting
82+-- a list in a concurrent manner running each action in parallel and therefore
84−-- items is not deterministic. Computations in Alternative composition may or
83+-- the order of results in the fold is not deterministic, but the results are
85−-- may not run in parallel. Thirdly, the <||> operator provides a way to
84+-- fairly interleaved. There are other ways to bind and compose. The
86−-- compose computations that are guaranteed to run in parallel. This provides a
85+-- combinations of different types of binds and monoidal compositions can be
87−-- way to run infinite loops or long running computations without blocking
86+-- used to achieve the desired results.
88−-- or starving other computations.
90------------------------------------------------------------------------------
88------------------------------------------------------------------------------
91-- Parent child thread communication types
89-- Parent child thread communication types
@@ -108,6 +106,10 @@ data Context m a =
108 , queueEmpty :: m Bool
106 , queueEmpty :: m Bool
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.
111newtype AsyncT m a =
113newtype AsyncT m a =
113 runAsyncT :: forall r.
115 runAsyncT :: forall r.
@@ -117,13 +119,14 @@ newtype AsyncT m a =
122+-- | A monad that can do asynchronous (or parallel) IO operations.
120type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
123type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
122------------------------------------------------------------------------------
125------------------------------------------------------------------------------
124------------------------------------------------------------------------------
127------------------------------------------------------------------------------
126−-- | Appends the results of two AsyncT computations in order.
129+-- | '<>' concatenates two 'AsyncT' streams sequentially.
127instance Semigroup (AsyncT m a) where
130instance Semigroup (AsyncT m a) where
128 m1 <> m2 = go m1
131 m1 <> m2 = go m1
@@ -133,12 +136,11 @@ instance Semigroup (AsyncT m a) where
133 yield a c (Just r) = yld a c (Just (go r))
136 yield a c (Just r) = yld a c (Just (go r))
134 in m ctx stop yield
137 in m ctx stop yield
136−-- | Appends the results of two AsyncT computations in order.
137instance Monoid (AsyncT m a) where
139instance Monoid (AsyncT m a) where
138 mempty = AsyncT $ \_ stp _ -> stp
140 mempty = AsyncT $ \_ stp _ -> stp
139 mappend = (<>)
141 mappend = (<>)
141−-- | Interleaves the results of two 'AsyncT' streams
143+-- | Same as '<=>'.
142interleave :: AsyncT m a -> AsyncT m a -> AsyncT m a
144interleave :: AsyncT m a -> AsyncT m a -> AsyncT m a
143interleave m1 m2 = AsyncT $ \ctx stp yld -> do
145interleave m1 m2 = AsyncT $ \ctx stp yld -> do
144 let stop = (runAsyncT m2) ctx stp yld
146 let stop = (runAsyncT m2) ctx stp yld
@@ -148,6 +150,8 @@ interleave m1 m2 = AsyncT $ \ctx stp yld -> do
149infixr 5 <=>
151infixr 5 <=>
153+-- | Sequential compostion, similar to '<>' except that it interleaves the two
154+-- 'AsyncT' streams fairly.
151(<=>) :: AsyncT m a -> AsyncT m a -> AsyncT m a
155(<=>) :: AsyncT m a -> AsyncT m a -> AsyncT m a
152(<=>) = interleave
156(<=>) = interleave
@@ -177,21 +181,6 @@ instance Monad m => Applicative (AsyncT m) where
178------------------------------------------------------------------------------
182------------------------------------------------------------------------------
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?
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.
195-- | A thread context is valid only until the next bind. Upon a bind we
184-- | A thread context is valid only until the next bind. Upon a bind we
196-- reset the context to Nothing.
185-- reset the context to Nothing.
197{-# INLINE bindWith #-}
186{-# INLINE bindWith #-}
@@ -209,6 +198,8 @@ bindWith k m f = go m
209 yield a _ (Just r) = run $ f a `k` (go r)
198 yield a _ (Just r) = run $ f a `k` (go r)
210 in g Nothing stp yield
199 in g Nothing stp yield
201+-- | '>>=' runs a monadic action for each element in the stream, sequentially,
202+-- in a depth first manner.
212instance Monad m => Monad (AsyncT m) where
203instance Monad m => Monad (AsyncT m) where
213 return = pure
204 return = pure
@@ -218,18 +209,34 @@ instance Monad m => Monad (AsyncT m) where
218 yield a _ (Just r) = run $ f a <> (r >>= f)
209 yield a _ (Just r) = run $ f a <> (r >>= f)
219 in m Nothing stp yield
210 in m Nothing stp yield
221−-- | Serial, interleaved bind
212+------------------------------------------------------------------------------
213+-- Alternative ways to bind
214+------------------------------------------------------------------------------
216+-- | Run a monadic action for each element in the stream, sequentially, in a
217+-- breadth first manner i.e. fairly interleaved.
222(>->) :: AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
218(>->) :: AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
223(>->) = bindWith (<=>)
219(>->) = bindWith (<=>)
221+-- | Run a monadic action for each element in the stream, in parallel, with
225−-- | Parallel, depth first bind
222+-- a depth first bias.
226(>>|) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
223(>>|) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
227(>>|) = bindWith (<|)
224(>>|) = bindWith (<|)
226+-- | Run a monadic action for each element in the stream, in parallel, with a
229−-- | Parallel, breadth first bind
227+-- breadth first bias i.e. fairly interleaved.
230(>|>) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
228(>|>) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
231(>|>) = bindWith (<|>)
229(>|>) = bindWith (<|>)
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.
233------------------------------------------------------------------------------
240------------------------------------------------------------------------------
234-- Alternative
241-- Alternative
235------------------------------------------------------------------------------
242------------------------------------------------------------------------------
@@ -473,30 +480,41 @@ parallel m1 m2 fifo = AsyncT $ \ctx stp yld -> do
473 Nothing -> (runAsyncT (pullFork m1 m2 fifo)) Nothing stp yld
480 Nothing -> (runAsyncT (pullFork m1 m2 fifo)) Nothing stp yld
474 Just c -> liftIO ((enqueue c) m2) >> (runAsyncT m1) ctx stp yld
481 Just c -> liftIO ((enqueue c) m2) >> (runAsyncT m1) ctx stp yld
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
476instance MonadAsync m => Alternative (AsyncT m) where
488instance MonadAsync m => Alternative (AsyncT m) where
477 empty = mempty
489 empty = mempty
479 {-# INLINE (<|>) #-}
491 {-# INLINE (<|>) #-}
480 m1 <|> m2 = parallel m1 m2 True
492 m1 <|> m2 = parallel m1 m2 True
482−-- | Just like '<>' except that it can execute the action on the right in
494+------------------------------------------------------------------------------
483−-- parallel ahead of time. Returns the results in serial order like '<>' from
495+-- Other monoidal compositions for parallel actions
484−-- left to right.
496+------------------------------------------------------------------------------
485parAhead :: AsyncT m a -> AsyncT m a -> AsyncT m a
499parAhead :: AsyncT m a -> AsyncT m a -> AsyncT m a
486parAhead = undefined
500parAhead = undefined
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.
488(<>|) :: AsyncT m a -> AsyncT m a -> AsyncT m a
505(<>|) :: AsyncT m a -> AsyncT m a -> AsyncT m a
489(<>|) = parAhead
506(<>|) = parAhead
491−-- | Left biased parallel execution. Actions may run in parallel but not
508+-- | Same as '<|'.
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
495{-# INLINE parLeft #-}
509{-# INLINE parLeft #-}
496parLeft :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
510parLeft :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
497parLeft m1 m2 = parallel m1 m2 False
511parLeft m1 m2 = parallel m1 m2 False
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.
500{-# INLINE (<|) #-}
518{-# INLINE (<|) #-}
501(<|) :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
519(<|) :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
502(<|) = parLeft
520(<|) = parLeft
@@ -594,7 +612,7 @@ instance MonadState s m => MonadState s (AsyncT m) where
594-- Running the monad
612-- Running the monad
595------------------------------------------------------------------------------
613------------------------------------------------------------------------------
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
598-- results.
616-- results.
599runAsyncly :: MonadAsync m => AsyncT m a -> m ()
617runAsyncly :: MonadAsync m => AsyncT m a -> m ()
600runAsyncly m = run Nothing m
618runAsyncly m = run Nothing m
@@ -609,8 +627,7 @@ runAsyncly m = run Nothing m
610 run ct mx = (runAsyncT mx) ct stop yield
628 run ct mx = (runAsyncT mx) ct stop yield
612−-- | Run an 'AsyncT m' computation and collect the results generated by each
630+-- | Collect the results of an 'AsyncT' stream into a list.
613−-- thread of the computation in a list.
614{-# INLINABLE toList #-}
631{-# INLINABLE toList #-}
615toList :: MonadAsync m => AsyncT m a -> m [a]
632toList :: MonadAsync m => AsyncT m a -> m [a]
616toList m = run Nothing m
633toList m = run Nothing m
@@ -630,6 +647,7 @@ toList m = run Nothing m
630-- Transformation
647-- Transformation
631------------------------------------------------------------------------------
648------------------------------------------------------------------------------
650+-- | Take first 'n' elements from the stream and discard the rest.
633take :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
651take :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
634take n m = AsyncT $ \ctx stp yld -> do
652take n m = AsyncT $ \ctx stp yld -> do
635 let yield a c Nothing = yld a c Nothing
653 let yield a c Nothing = yld a c Nothing
@@ -638,6 +656,7 @@ take n m = AsyncT $ \ctx stp yld -> do
639 else (runAsyncT m) ctx stp yield
657 else (runAsyncT m) ctx stp yield
659+-- | Discard first 'n' elements from the stream and take the rest.
641drop :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
660drop :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
642drop n m = AsyncT $ \ctx stp yld -> do
661drop n m = AsyncT $ \ctx stp yld -> do
643 let yield _ _ Nothing = stp
662 let yield _ _ Nothing = stp
@@ -662,7 +681,7 @@ foldMapWith :: (Monoid b, Foldable t) =>
662 (b1 -> b -> b) -> (a -> b1) -> t a -> b
681 (b1 -> b -> b) -> (a -> b1) -> t a -> b
663foldMapWith f g = foldr (f . g) mempty
682foldMapWith f g = foldr (f . g) mempty
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
666-- first and the third argument.
685-- first and the third argument.
667{-# INLINABLE forEachWith #-}
686{-# INLINABLE forEachWith #-}
668forEachWith :: (Monoid b, Foldable t) =>
687forEachWith :: (Monoid b, Foldable t) =>
@@ -673,5 +692,6 @@ forEachWith f xs g = foldr (f . g) mempty xs
673-- Convert a callback into an 'AsyncT' computation
692-- Convert a callback into an 'AsyncT' computation
674------------------------------------------------------------------------------
693------------------------------------------------------------------------------
695+-- | Convert a callback into an 'AsyncT' stream.
676fromCallback :: (forall r. (a -> m r) -> m r) -> AsyncT m a
696fromCallback :: (forall r. (a -> m r) -> m r) -> AsyncT m a
677fromCallback k = AsyncT $ \ctx _ yld -> k (\a -> yld a ctx Nothing)
697fromCallback k = AsyncT $ \ctx _ yld -> k (\a -> yld a ctx Nothing)