@@ -19,25 +19,27 @@ module Asyncly.Core
1919 ( Stream (..)
2020 , MonadAsync
2121 , Streaming (..)
22+ , ContextUsedAfterEOF (..)
2223
23− , fromCallback
24−
24+ -- * Running the Monad
2525 , runStreaming
26− , toList
27− , async
2826
27+ -- * Interleaving
28+ , interleave
2929 , (<=>)
30− , (<|)
3130
31+ -- * Asynchronous Operations
32+ , CtxType (..)
33+ , SchedPolicy (..)
34+ , Dimension (..)
3235 , yielding
33− , interleave
3436 , parallel
3537 , parAlt
3638 , parLeft
37− , CtxType (..)
38− , SchedPolicy (..)
39− , Dimension (..)
39+ , (<|)
40+ , async
4041
42+ -- * Fold Utilities
4143 , foldWith
4244 , foldMapWith
4345 , forEachWith
@@ -51,8 +53,7 @@ import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
5153 tryPutMVar, takeMVar)
5254import Control.Exception (SomeException (..), Exception)
5355import qualified Control.Exception.Lifted as EL
54−import Control.Monad (liftM, MonadPlus(..), mzero,
55− when)
56+import Control.Monad (MonadPlus(..), mzero, when)
5657import Control.Monad.Base (MonadBase (..), liftBaseDefault)
5758import Control.Monad.Catch (MonadThrow, throwM)
5859import Control.Monad.Error.Class (MonadError(..))
@@ -61,35 +62,23 @@ import Control.Monad.Reader.Class (MonadReader(..))
6162import Control.Monad.State.Class (MonadState(..))
6263import Control.Monad.Trans.Class (MonadTrans (lift))
6364import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
65+import Data.Atomics (atomicModifyIORefCAS,
66+ atomicModifyIORefCAS_)
6467import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
6568 tryPopR, nullQ)
6669import Data.Functor (void)
6770import Data.IORef (IORef, modifyIORef, newIORef,
6871 readIORef)
69−import Data.Atomics (atomicModifyIORefCAS,
70− atomicModifyIORefCAS_)
7172import Data.Maybe (isNothing)
7273import Data.Semigroup (Semigroup(..))
7374import Data.Set (Set)
7475import qualified Data.Set as S
7576
7677------------------------------------------------------------------------------
77−-- Concurrency Semantics
78−------------------------------------------------------------------------------
79−--
80−-- Asyncly is essentially a concurrent list transformer. To fold the lists
81−-- (AsyncT m a) it provides many different ways of composing them. The Monoid
82−-- instance folds lists in a sequential fashion. The Alternative instance folds
83−-- a list in a concurrent manner running each action in parallel and therefore
84−-- the order of results in the fold is not deterministic, but the results are
85−-- fairly interleaved. There are other ways to bind and compose. The
86−-- combinations of different types of binds and monoidal compositions can be
87−-- used to achieve the desired results.
88−
89−------------------------------------------------------------------------------
90−-- Parent child thread communication types
78+-- Parent child thread communication type
9179------------------------------------------------------------------------------
9280
81+-- | Events that a child thread may send to a parent thread.
9382data ChildEvent a =
9483 ChildYield a
9584 | ChildStop ThreadId (Maybe SomeException)
@@ -98,10 +87,24 @@ data ChildEvent a =
9887-- State threaded around the monad for thread management
9988------------------------------------------------------------------------------
10089
90+-- | Conjunction is used for monadic/product style composition. Disjunction is
91+-- used for fold/sum style composition. We need to distiguish the two types of
92+-- contexts so that the scheduing of the two is independent.
10193data Dimension = Conjunction | Disjunction deriving Eq
94+
95+-- | For fairly interleaved parallel composition the sched policy is FIFO
96+-- whereas for left biased parallel composition it is LIFO.
10297data SchedPolicy = LIFO | FIFO deriving Eq
98+
99+-- | Identify the type of the context used for parallel composition.
103100data CtxType = CtxType Dimension SchedPolicy deriving Eq
104101
102+-- | A Context represents a parallel composition. It has a runqueue which holds
103+-- the tasks that are to be picked by a pool of worker threads. It has an
104+-- output queue where the results are placed by the worker threads. A
105+-- doorBell is used by the worker threads to intimate the consumer thread
106+-- about availability of new results. New work is usually enqueued as a result
107+-- of executing the parallel combinators i.e. '<|' and '<|>'.
105108data Context m a =
106109 Context { outputQueue :: IORef [ChildEvent a]
107110 , doorBell :: MVar Bool -- wakeup mechanism for outQ
@@ -112,10 +115,15 @@ data Context m a =
112115 , ctxType :: CtxType
113116 }
114117
115−-- | Represents a monadic stream of values of type 'a' resulting from actions
116−-- in monad 'm'. 'AsyncT' streams can be composed sequentially or in parallel
117−-- in various ways using monadic bind or variants of it and using monoidal
118−-- compositions like 'Monoid', 'Alternative' or variants of these.
118+------------------------------------------------------------------------------
119+-- The stream type
120+------------------------------------------------------------------------------
121+
122+-- | Represents a monadic stream of values of type 'a' constructed using
123+-- actions in monad 'm'. Streams can be composed sequentially or in parallel in
124+-- product style compositions (monadic bind multiplies streams in a ListT
125+-- fashion) and using sum style compositions like 'Semigroup', 'Monoid',
126+-- 'Alternative' or variants of these.
119127newtype Stream m a =
120128 Stream {
121129 runStream :: forall r.
@@ -125,18 +133,19 @@ newtype Stream m a =
125133 -> m r
126134 }
127135
128−-- | A monad that can do asynchronous (or parallel) IO operations.
136+-- | A monad that can perform asynchronous/concurrent IO operations.
129137type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
130138
139+-- | Yield a singleton value in a stream.
131140yielding :: a -> Stream m a
132141yielding a = Stream $ \_ _ yld -> yld a Nothing
133142
134143------------------------------------------------------------------------------
135−-- Monoid
144+-- Semigroup
136145------------------------------------------------------------------------------
137146
138−-- | '<>' concatenates two 'AsyncT' streams sequentially i.e. the first stream
139−-- is exhausted completely before yielding any element from the second stream.
147+-- | '<>' concatenates two streams sequentially i.e. the first stream is
148+-- exhausted completely before yielding any element from the second stream.
140149instance Semigroup (Stream m a) where
141150 m1 <> m2 = go m1
142151 where
@@ -146,10 +155,19 @@ instance Semigroup (Stream m a) where
146155 yield a (Just r) = yld a (Just (go r))
147156 in m Nothing stop yield
148157
158+------------------------------------------------------------------------------
159+-- Monoid
160+------------------------------------------------------------------------------
161+
149162instance Monoid (Stream m a) where
150163 mempty = Stream $ \_ stp _ -> stp
151164 mappend = (<>)
152165
166+------------------------------------------------------------------------------
167+-- Interleave
168+------------------------------------------------------------------------------
169+
170+-- | Same as '<=>'.
153171interleave :: Stream m a -> Stream m a -> Stream m a
154172interleave m1 m2 = Stream $ \_ stp yld -> do
155173 let stop = (runStream m2) Nothing stp yld
@@ -158,7 +176,7 @@ interleave m1 m2 = Stream $ \_ stp yld -> do
158176 (runStream m1) Nothing stop yield
159177
160178------------------------------------------------------------------------------
161−-- Streaming threads for parallel composition
179+-- Spawning threads and collecting result in streamed fashion
162180------------------------------------------------------------------------------
163181
164182{-# INLINE doFork #-}
@@ -285,10 +303,11 @@ sendWorkerWait ctx = do
285303 then (pushWorker ctx) >> sendWorkerWait ctx
286304 else void (liftIO $ takeMVar (doorBell ctx))
287305
306+-- | An 'async' stream has finished but is still being used.
288307data ContextUsedAfterEOF = ContextUsedAfterEOF deriving Show
289308instance Exception ContextUsedAfterEOF
290309
291−-- | Pull an AsyncT stream from a context
310+-- | Pull a stream from a context
292311{-# NOINLINE pullFromCtx #-}
293312pullFromCtx :: MonadAsync m => Context m a -> Stream m a
294313pullFromCtx ctx = Stream $ \_ stp yld -> do
@@ -390,6 +409,26 @@ pushPairToCtx ctype m1 m2 = do
390409 pushWorker ctx
391410 return ctx
392411
412+-- | The computation specified in the argument is pushed to a new thread and
413+-- the context is returned. The returned context can be used to pull the output
414+-- of the computation.
415+pushOneToCtx :: MonadAsync m => CtxType -> Stream m a -> m (Context m a)
416+pushOneToCtx ctype m = do
417+ ctx <- liftIO $
418+ case ctype of
419+ CtxType _ FIFO -> do
420+ c <- getFifoCtx ctype
421+ return c
422+ CtxType _ LIFO -> do
423+ c <- getLifoCtx ctype
424+ return c
425+ -- Note: We must have all the work on the queue before sending the
426+ -- pushworker, otherwise the pushworker may exit before we even get a
427+ -- chance to push.
428+ liftIO $ (enqueue ctx) m
429+ pushWorker ctx
430+ return ctx
431+
393432{-# NOINLINE pushPullFork #-}
394433pushPullFork :: MonadAsync m
395434 => CtxType -> Stream m a -> Stream m a -> Stream m a
@@ -397,29 +436,30 @@ pushPullFork ct m1 m2 = Stream $ \_ stp yld -> do
397436 ctx <- pushPairToCtx ct m1 m2
398437 (runStream (pullFromCtx ctx)) Nothing stp yld
399438
400−-- Concurrency rate control. Our objective is to create more threads on
401−-- demand if the consumer is running faster than us. As soon as we
402−-- encounter an Alternative composition we create a push pull pair of
403−-- threads. We use a channel for communication between the consumer that
404−-- pulls from the channel and the producer that pushes to the channel. The
405−-- producer creates more threads if the channel becomes empty at times,
406−-- that is the consumer is running faster. However this mechanism can be
407−-- problematic if the initial production latency is high, we may end up
408−-- creating too many threads. So we need some way to monitor and use the
409−-- latency as well.
439+-- Concurrency rate control. Our objective is to create more threads on demand
440+-- if the consumer is running faster than us. As soon as we encounter an
441+-- Alternative composition we create a push pull pair of threads. We use a
442+-- channel for communication between the consumer pulling from the channel and
443+-- the producer who pushing to the channel. The producer creates more threads
444+-- if no output is seen on the channel, that is the consumer is running faster.
445+-- However this mechanism can be problematic if the initial production latency
446+-- is high, we may end up creating too many threads. So we need some way to
447+-- monitor and use the latency as well.
410448--
411−-- TBD For quick response we may have to increase the rate in the middle of
412−-- a serially running computation. For that we can use a state flag to fork
413−-- the rest of the computation at any point of time inside the Monad bind
414−-- operation if the consumer is running at a faster speed.
449+-- TBD We may run computations at the lower level of the composition tree
450+-- serially even if they are composed using a parallel combinator. We can use
451+-- <> in place of <| and <=> in place of <|>. If we find that a parallel
452+-- channel immediately above a computation becomes empty we can switch to
453+-- parallelizing the computation. For that we can use a state flag to fork the
454+-- rest of the computation at any point of time inside the Monad bind operation
455+-- if the consumer is running at a faster speed.
415456--
416457-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
417458-- If we have to dispatch in arbitrary chunksizes we will need to compose the
418−-- parallel actions using a data constructor instead so that we can divide it
419−-- in chunks of arbitrary size before dispatch. When batching we can convert
420−-- the structure into Alternative batches of Monoid composition. That will also
421−-- allow us to dispatch more work to existing threads rather than creating new
422−-- threads always.
459+-- parallel actions using a data constructor (Free Alternative) instead so that
460+-- we can divide it in chunks of arbitrary size before dispatch. If the stream
461+-- is composed of hierarchically composed grains of different sizes then we can
462+-- always switch to a desired granularity depending on the consumer speed.
423463--
424464-- TBD for pure work (when we are not in the IO monad) we can divide it into
425465-- just the number of CPUs.
@@ -428,8 +468,10 @@ pushPullFork ct m1 m2 = Stream $ \_ stp yld -> do
428468-- residual work back to the dispatcher. It will also consume a lot of
429469-- memory due to queueing of all the work before execution starts.
430470
431−-- Note: This is designed to scale for right associated compositions,
432−-- therefore always use a right fold for folding bigger structures.
471+-- | Compose two streams in parallel using a scheduling policy specified by
472+-- 'CtxType'. Note: This is designed to scale for right associated
473+-- compositions, therefore always use a right fold for folding bigger
474+-- structures.
433475{-# INLINE parallel #-}
434476parallel :: MonadAsync m => CtxType -> Stream m a -> Stream m a -> Stream m a
435477parallel ct m1 m2 = Stream $ \ctx stp yld -> do
@@ -439,26 +481,6 @@ parallel ct m1 m2 = Stream $ \ctx stp yld -> do
439481 (runStream (pushPullFork ct m1 m2)) Nothing stp yld
440482 Just c -> liftIO ((enqueue c) m2) >> (runStream m1) ctx stp yld
441483
442−-- | The computation specified in the argument is pushed to a new thread and
443−-- the context is returned. The returned context can be used to pull the output
444−-- of the computation.
445−pushOneToCtx :: MonadAsync m => CtxType -> Stream m a -> m (Context m a)
446−pushOneToCtx ctype m = do
447− ctx <- liftIO $
448− case ctype of
449− CtxType _ FIFO -> do
450− c <- getFifoCtx ctype
451− return c
452− CtxType _ LIFO -> do
453− c <- getLifoCtx ctype
454− return c
455− -- Note: We must have all the work on the queue before sending the
456− -- pushworker, otherwise the pushworker may exit before we even get a
457− -- chance to push.
458− liftIO $ (enqueue ctx) m
459− pushWorker ctx
460− return ctx
461−
462484------------------------------------------------------------------------------
463485-- Semigroup and Monoid style compositions for parallel actions
464486------------------------------------------------------------------------------
@@ -486,9 +508,11 @@ parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
486508parLeft = parallel (CtxType Disjunction LIFO)
487509
488510-------------------------------------------------------------------------------
511+-- Instances (only used for deriving newtype instances)
512+-------------------------------------------------------------------------------
513+
489514-- Stream type is not exposed, these instances are only for deriving instances
490515-- for the newtype wrappers based on Stream.
491−-------------------------------------------------------------------------------
492516
493517-- Dummy Instances, defined to enable the definition of other instances that
494518-- require a Monad constraint. Must be defined by the newtypes.
@@ -566,21 +590,26 @@ instance MonadState s m => MonadState s (Stream m) where
566590-- Types that can behave as a Stream
567591------------------------------------------------------------------------------
568592
593+-- | Types that can represent a stream of elements.
569594class Streaming t where
570595 toStream :: t m a -> Stream m a
571596 fromStream :: Stream m a -> t m a
572597
598+------------------------------------------------------------------------------
599+-- Some basic stream type agnostic APIs
600+------------------------------------------------------------------------------
601+
573602-- XXX The async API is useful for exploring each stream arbitrarily when
574603-- zipping or merging two streams. We can use a newtype wrapper with a monad
575604-- instance that composes like regular streaming libraries to facilitate linear
576605-- composition. We will also need a yield API for that.
577606
578607-- | Run a computation asynchronously, triggers the computation and returns
579−-- another computation (i.e. a promise) that when executed produces the output
580−-- from the original computation. Note that the returned action must be
581−-- executed exactly once and drained completely. If not executed or not drained
582−-- fully we will may have a thread blocked forever and if executed more than
583−-- once a ContextUsedAfterEOF exception will be raised.
608+-- another computation that when executed produces the output generated by the
609+-- original computation. Note that the returned action must be executed exactly
610+-- once and drained completely. If not executed or not drained fully we may
611+-- have a thread blocked forever and if executed more than once a
612+-- 'ContextUsedAfterEOF' exception will be raised.
584613
585614async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
586615async m = do
@@ -589,18 +618,20 @@ async m = do
589618
590619infixr 5 <=>
591620
592−-- | Sequential interleaved composition, similar to '<>' except that it fairly
593−-- interleaves the two 'AsyncT' streams, yielding an element from each stream
594−-- alternately. Should be used only on finite streams.
621+-- | Sequential interleaved composition, in contrast to '<>' this operator
622+-- fairly interleaves the two streams instead of appending them; yielding one
623+-- element from each stream alternately. Unlike '<>' it cannot be used to fold
624+-- an infinite container of streams.
595625{-# INLINE (<=>) #-}
596626(<=>) :: Streaming t => t m a -> t m a -> t m a
597627m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
598628
599−-- | Parallel composition similar to '<|>' except that it is left-biased
600−-- instead of being fair. Action on the left is likely to be given a chance to
601−-- execute before the action on the right. If the left action keeps yielding
602−-- results without blocking it continues running until it finishes and only
603−-- then the right action runs. Unlike '<|>' it can be used on infinite streams.
629+-- | Parallel interleaved composition, in contrast to '<|>' this operator
630+-- "merges" streams in a left biased manner rather than fairly interleaving
631+-- them. It keeps yielding from the stream on the left as long as it can. If
632+-- the left stream blocks or cannot keep up with the pace of the consumer it
633+-- can yield from the stream on the right in parallel. Unlike '<|>' it can be
634+-- used to fold infinite containers of streams.
604635{-# INLINE (<|) #-}
605636(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
606637m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
@@ -609,8 +640,8 @@ m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
609640-- Running the monad
610641------------------------------------------------------------------------------
611642
612−-- | Run an 'AsyncT' computation, wait for it to finish and discard the
613−-- results.
643+-- | Run a composed streaming computation, wait for it to finish and discard
644+-- the results.
614645runStreaming :: (Monad m, Streaming t) => t m a -> m ()
615646runStreaming m = go (toStream m)
616647
@@ -624,21 +655,6 @@ runStreaming m = go (toStream m)
624655 yield _ Nothing = stop
625656 yield _ (Just x) = go x
626657
627−-- | Collect the results of an 'AsyncT' stream into a list.
628−{-# INLINABLE toList #-}
629−toList :: (Monad m, Streaming t) => t m a -> m [a]
630−toList m = go (toStream m)
631−
632− where
633−
634− go m1 = (runStream m1) Nothing stop yield
635−
636− stop = return []
637−
638− {-# INLINE yield #-}
639− yield a Nothing = return [a]
640− yield a (Just x) = liftM (a :) (go x)
641−
642658------------------------------------------------------------------------------
643659-- Utilities
644660------------------------------------------------------------------------------
@@ -661,11 +677,3 @@ foldMapWith f g = foldr (f . g) mempty
661677forEachWith :: (Monoid b, Foldable t) =>
662678 (b1 -> b -> b) -> t a -> (a -> b1) -> b
663679forEachWith f xs g = foldr (f . g) mempty xs
664−
665−------------------------------------------------------------------------------
666−-- Convert a callback into an 'AsyncT' computation
667−------------------------------------------------------------------------------
668−
669−-- | Convert a callback into an 'AsyncT' stream.
670−fromCallback :: (forall r. (a -> m r) -> m r) -> Stream m a
671−fromCallback k = Stream $ \_ _ yld -> k (\a -> yld a Nothing)