@@ -19,25 +19,27 @@ module Asyncly.Core
19 ( Stream (..)
19 ( Stream (..)
20 , MonadAsync
20 , MonadAsync
21 , Streaming (..)
21 , Streaming (..)
22+ , ContextUsedAfterEOF (..)
23− , fromCallback
24+ -- * Running the Monad
25 , runStreaming
25 , runStreaming
31+ -- * Asynchronous Operations
32 , yielding
35 , yielding
34 , parallel
36 , parallel
37− , CtxType (..)
39+ , (<|)
38− , SchedPolicy (..)
40+ , async
41 , foldWith
43 , foldWith
42 , foldMapWith
44 , foldMapWith
43 , forEachWith
45 , forEachWith
@@ -51,8 +53,7 @@ import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
51 tryPutMVar, takeMVar)
53 tryPutMVar, takeMVar)
52import Control.Exception (SomeException (..), Exception)
54import Control.Exception (SomeException (..), Exception)
53import qualified Control.Exception.Lifted as EL
55import qualified Control.Exception.Lifted as EL
54−import Control.Monad (liftM, MonadPlus(..), mzero,
56+import Control.Monad (MonadPlus(..), mzero, when)
56import Control.Monad.Base (MonadBase (..), liftBaseDefault)
57import Control.Monad.Base (MonadBase (..), liftBaseDefault)
57import Control.Monad.Catch (MonadThrow, throwM)
58import Control.Monad.Catch (MonadThrow, throwM)
58import Control.Monad.Error.Class (MonadError(..))
59import Control.Monad.Error.Class (MonadError(..))
@@ -61,35 +62,23 @@ import Control.Monad.Reader.Class (MonadReader(..))
61import Control.Monad.State.Class (MonadState(..))
62import Control.Monad.State.Class (MonadState(..))
62import Control.Monad.Trans.Class (MonadTrans (lift))
63import Control.Monad.Trans.Class (MonadTrans (lift))
63import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
64import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
65+import Data.Atomics (atomicModifyIORefCAS,
66+ atomicModifyIORefCAS_)
64import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
67import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
65 tryPopR, nullQ)
68 tryPopR, nullQ)
66import Data.Functor (void)
69import Data.Functor (void)
67import Data.IORef (IORef, modifyIORef, newIORef,
70import Data.IORef (IORef, modifyIORef, newIORef,
68 readIORef)
71 readIORef)
69−import Data.Atomics (atomicModifyIORefCAS,
70− atomicModifyIORefCAS_)
71import Data.Maybe (isNothing)
72import Data.Maybe (isNothing)
72import Data.Semigroup (Semigroup(..))
73import Data.Semigroup (Semigroup(..))
73import Data.Set (Set)
74import Data.Set (Set)
74import qualified Data.Set as S
75import qualified Data.Set as S
76------------------------------------------------------------------------------
77------------------------------------------------------------------------------
77−-- Concurrency Semantics
78−------------------------------------------------------------------------------
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.
89−------------------------------------------------------------------------------
90−-- Parent child thread communication types
78+-- Parent child thread communication type
91------------------------------------------------------------------------------
79------------------------------------------------------------------------------
81+-- | Events that a child thread may send to a parent thread.
93data ChildEvent a =
82data ChildEvent a =
94 ChildYield a
83 ChildYield a
95 | ChildStop ThreadId (Maybe SomeException)
84 | ChildStop ThreadId (Maybe SomeException)
@@ -98,10 +87,24 @@ data ChildEvent a =
98-- State threaded around the monad for thread management
87-- State threaded around the monad for thread management
99------------------------------------------------------------------------------
88------------------------------------------------------------------------------
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.
101data Dimension = Conjunction | Disjunction deriving Eq
93data Dimension = Conjunction | Disjunction deriving Eq
95+-- | For fairly interleaved parallel composition the sched policy is FIFO
96+-- whereas for left biased parallel composition it is LIFO.
102data SchedPolicy = LIFO | FIFO deriving Eq
97data SchedPolicy = LIFO | FIFO deriving Eq
99+-- | Identify the type of the context used for parallel composition.
103data CtxType = CtxType Dimension SchedPolicy deriving Eq
100data CtxType = CtxType Dimension SchedPolicy deriving Eq
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 '<|>'.
105data Context m a =
108data Context m a =
106 Context { outputQueue :: IORef [ChildEvent a]
109 Context { outputQueue :: IORef [ChildEvent a]
107 , doorBell :: MVar Bool -- wakeup mechanism for outQ
110 , doorBell :: MVar Bool -- wakeup mechanism for outQ
@@ -112,10 +115,15 @@ data Context m a =
112 , ctxType :: CtxType
115 , ctxType :: CtxType
118+------------------------------------------------------------------------------
120+------------------------------------------------------------------------------
115−-- | Represents a monadic stream of values of type 'a' resulting from actions
122+-- | Represents a monadic stream of values of type 'a' constructed using
116−-- in monad 'm'. 'AsyncT' streams can be composed sequentially or in parallel
123+-- actions in monad 'm'. Streams can be composed sequentially or in parallel in
117−-- in various ways using monadic bind or variants of it and using monoidal
124+-- product style compositions (monadic bind multiplies streams in a ListT
125+-- fashion) and using sum style compositions like 'Semigroup', 'Monoid',
118−-- compositions like 'Monoid', 'Alternative' or variants of these.
126+-- 'Alternative' or variants of these.
119newtype Stream m a =
127newtype Stream m a =
121 runStream :: forall r.
129 runStream :: forall r.
@@ -125,18 +133,19 @@ newtype Stream m a =
128−-- | A monad that can do asynchronous (or parallel) IO operations.
136+-- | A monad that can perform asynchronous/concurrent IO operations.
129type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
137type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
139+-- | Yield a singleton value in a stream.
131yielding :: a -> Stream m a
140yielding :: a -> Stream m a
132yielding a = Stream $ \_ _ yld -> yld a Nothing
141yielding a = Stream $ \_ _ yld -> yld a Nothing
134------------------------------------------------------------------------------
143------------------------------------------------------------------------------
135−-- Monoid
144+-- Semigroup
136------------------------------------------------------------------------------
145------------------------------------------------------------------------------
138−-- | '<>' concatenates two 'AsyncT' streams sequentially i.e. the first stream
147+-- | '<>' concatenates two streams sequentially i.e. the first stream is
139−-- is exhausted completely before yielding any element from the second stream.
148+-- exhausted completely before yielding any element from the second stream.
140instance Semigroup (Stream m a) where
149instance Semigroup (Stream m a) where
141 m1 <> m2 = go m1
150 m1 <> m2 = go m1
@@ -146,10 +155,19 @@ instance Semigroup (Stream m a) where
146 yield a (Just r) = yld a (Just (go r))
155 yield a (Just r) = yld a (Just (go r))
147 in m Nothing stop yield
156 in m Nothing stop yield
158+------------------------------------------------------------------------------
160+------------------------------------------------------------------------------
149instance Monoid (Stream m a) where
162instance Monoid (Stream m a) where
150 mempty = Stream $ \_ stp _ -> stp
163 mempty = Stream $ \_ stp _ -> stp
151 mappend = (<>)
164 mappend = (<>)
166+------------------------------------------------------------------------------
168+------------------------------------------------------------------------------
153interleave :: Stream m a -> Stream m a -> Stream m a
171interleave :: Stream m a -> Stream m a -> Stream m a
154interleave m1 m2 = Stream $ \_ stp yld -> do
172interleave m1 m2 = Stream $ \_ stp yld -> do
155 let stop = (runStream m2) Nothing stp yld
173 let stop = (runStream m2) Nothing stp yld
@@ -158,7 +176,7 @@ interleave m1 m2 = Stream $ \_ stp yld -> do
158 (runStream m1) Nothing stop yield
176 (runStream m1) Nothing stop yield
160------------------------------------------------------------------------------
178------------------------------------------------------------------------------
161−-- Streaming threads for parallel composition
179+-- Spawning threads and collecting result in streamed fashion
162------------------------------------------------------------------------------
180------------------------------------------------------------------------------
164{-# INLINE doFork #-}
182{-# INLINE doFork #-}
@@ -285,10 +303,11 @@ sendWorkerWait ctx = do
285 then (pushWorker ctx) >> sendWorkerWait ctx
303 then (pushWorker ctx) >> sendWorkerWait ctx
286 else void (liftIO $ takeMVar (doorBell ctx))
304 else void (liftIO $ takeMVar (doorBell ctx))
306+-- | An 'async' stream has finished but is still being used.
288data ContextUsedAfterEOF = ContextUsedAfterEOF deriving Show
307data ContextUsedAfterEOF = ContextUsedAfterEOF deriving Show
289instance Exception ContextUsedAfterEOF
308instance Exception ContextUsedAfterEOF
291−-- | Pull an AsyncT stream from a context
310+-- | Pull a stream from a context
292{-# NOINLINE pullFromCtx #-}
311{-# NOINLINE pullFromCtx #-}
293pullFromCtx :: MonadAsync m => Context m a -> Stream m a
312pullFromCtx :: MonadAsync m => Context m a -> Stream m a
294pullFromCtx ctx = Stream $ \_ stp yld -> do
313pullFromCtx ctx = Stream $ \_ stp yld -> do
@@ -390,6 +409,26 @@ pushPairToCtx ctype m1 m2 = do
390 pushWorker ctx
409 pushWorker ctx
391 return ctx
410 return ctx
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
419+ CtxType _ FIFO -> do
420+ c <- getFifoCtx ctype
422+ CtxType _ LIFO -> do
423+ c <- getLifoCtx ctype
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
428+ liftIO $ (enqueue ctx) m
393{-# NOINLINE pushPullFork #-}
432{-# NOINLINE pushPullFork #-}
394pushPullFork :: MonadAsync m
433pushPullFork :: MonadAsync m
395 => CtxType -> Stream m a -> Stream m a -> Stream m a
434 => CtxType -> Stream m a -> Stream m a -> Stream m a
@@ -397,29 +436,30 @@ pushPullFork ct m1 m2 = Stream $ \_ stp yld -> do
397 ctx <- pushPairToCtx ct m1 m2
436 ctx <- pushPairToCtx ct m1 m2
398 (runStream (pullFromCtx ctx)) Nothing stp yld
437 (runStream (pullFromCtx ctx)) Nothing stp yld
400−-- Concurrency rate control. Our objective is to create more threads on
439+-- Concurrency rate control. Our objective is to create more threads on demand
401−-- demand if the consumer is running faster than us. As soon as we
440+-- if the consumer is running faster than us. As soon as we encounter an
402−-- encounter an Alternative composition we create a push pull pair of
441+-- Alternative composition we create a push pull pair of threads. We use a
403−-- threads. We use a channel for communication between the consumer that
442+-- channel for communication between the consumer pulling from the channel and
404−-- pulls from the channel and the producer that pushes to the channel. The
443+-- the producer who pushing to the channel. The producer creates more threads
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
444+-- if no output is seen on the channel, that is the consumer is running faster.
407−-- problematic if the initial production latency is high, we may end up
445+-- However this mechanism can be problematic if the initial production latency
408−-- creating too many threads. So we need some way to monitor and use the
446+-- is high, we may end up creating too many threads. So we need some way to
409−-- latency as well.
447+-- monitor and use the latency as well.
411−-- TBD For quick response we may have to increase the rate in the middle of
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
412−-- a serially running computation. For that we can use a state flag to fork
453+-- parallelizing the computation. For that we can use a state flag to fork the
413−-- the rest of the computation at any point of time inside the Monad bind
454+-- rest of the computation at any point of time inside the Monad bind operation
414−-- operation if the consumer is running at a faster speed.
455+-- if the consumer is running at a faster speed.
416-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
457-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
417-- If we have to dispatch in arbitrary chunksizes we will need to compose the
458-- 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
459+-- parallel actions using a data constructor (Free Alternative) instead so that
419−-- in chunks of arbitrary size before dispatch. When batching we can convert
460+-- we can divide it in chunks of arbitrary size before dispatch. If the stream
420−-- the structure into Alternative batches of Monoid composition. That will also
461+-- is composed of hierarchically composed grains of different sizes then we can
421−-- allow us to dispatch more work to existing threads rather than creating new
462+-- always switch to a desired granularity depending on the consumer speed.
424-- TBD for pure work (when we are not in the IO monad) we can divide it into
464-- TBD for pure work (when we are not in the IO monad) we can divide it into
425-- just the number of CPUs.
465-- just the number of CPUs.
@@ -428,8 +468,10 @@ pushPullFork ct m1 m2 = Stream $ \_ stp yld -> do
428-- residual work back to the dispatcher. It will also consume a lot of
468-- residual work back to the dispatcher. It will also consume a lot of
429-- memory due to queueing of all the work before execution starts.
469-- memory due to queueing of all the work before execution starts.
471+-- | Compose two streams in parallel using a scheduling policy specified by
431−-- Note: This is designed to scale for right associated compositions,
472+-- 'CtxType'. Note: This is designed to scale for right associated
432−-- therefore always use a right fold for folding bigger structures.
473+-- compositions, therefore always use a right fold for folding bigger
433{-# INLINE parallel #-}
475{-# INLINE parallel #-}
434parallel :: MonadAsync m => CtxType -> Stream m a -> Stream m a -> Stream m a
476parallel :: MonadAsync m => CtxType -> Stream m a -> Stream m a -> Stream m a
435parallel ct m1 m2 = Stream $ \ctx stp yld -> do
477parallel ct m1 m2 = Stream $ \ctx stp yld -> do
@@ -439,26 +481,6 @@ parallel ct m1 m2 = Stream $ \ctx stp yld -> do
439 (runStream (pushPullFork ct m1 m2)) Nothing stp yld
481 (runStream (pushPullFork ct m1 m2)) Nothing stp yld
440 Just c -> liftIO ((enqueue c) m2) >> (runStream m1) ctx stp yld
482 Just c -> liftIO ((enqueue c) m2) >> (runStream m1) ctx stp yld
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
449− CtxType _ FIFO -> do
450− c <- getFifoCtx ctype
452− CtxType _ LIFO -> do
453− c <- getLifoCtx ctype
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
458− liftIO $ (enqueue ctx) m
462------------------------------------------------------------------------------
484------------------------------------------------------------------------------
463-- Semigroup and Monoid style compositions for parallel actions
485-- Semigroup and Monoid style compositions for parallel actions
464------------------------------------------------------------------------------
486------------------------------------------------------------------------------
@@ -486,9 +508,11 @@ parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
486parLeft = parallel (CtxType Disjunction LIFO)
508parLeft = parallel (CtxType Disjunction LIFO)
488-------------------------------------------------------------------------------
510-------------------------------------------------------------------------------
511+-- Instances (only used for deriving newtype instances)
512+-------------------------------------------------------------------------------
489-- Stream type is not exposed, these instances are only for deriving instances
514-- Stream type is not exposed, these instances are only for deriving instances
490-- for the newtype wrappers based on Stream.
515-- for the newtype wrappers based on Stream.
491−-------------------------------------------------------------------------------
493-- Dummy Instances, defined to enable the definition of other instances that
517-- Dummy Instances, defined to enable the definition of other instances that
494-- require a Monad constraint. Must be defined by the newtypes.
518-- require a Monad constraint. Must be defined by the newtypes.
@@ -566,21 +590,26 @@ instance MonadState s m => MonadState s (Stream m) where
566-- Types that can behave as a Stream
590-- Types that can behave as a Stream
567------------------------------------------------------------------------------
591------------------------------------------------------------------------------
593+-- | Types that can represent a stream of elements.
569class Streaming t where
594class Streaming t where
570 toStream :: t m a -> Stream m a
595 toStream :: t m a -> Stream m a
571 fromStream :: Stream m a -> t m a
596 fromStream :: Stream m a -> t m a
598+------------------------------------------------------------------------------
599+-- Some basic stream type agnostic APIs
600+------------------------------------------------------------------------------
573-- XXX The async API is useful for exploring each stream arbitrarily when
602-- XXX The async API is useful for exploring each stream arbitrarily when
574-- zipping or merging two streams. We can use a newtype wrapper with a monad
603-- zipping or merging two streams. We can use a newtype wrapper with a monad
575-- instance that composes like regular streaming libraries to facilitate linear
604-- instance that composes like regular streaming libraries to facilitate linear
576-- composition. We will also need a yield API for that.
605-- composition. We will also need a yield API for that.
578-- | Run a computation asynchronously, triggers the computation and returns
607-- | Run a computation asynchronously, triggers the computation and returns
579−-- another computation (i.e. a promise) that when executed produces the output
608+-- another computation that when executed produces the output generated by the
580−-- from the original computation. Note that the returned action must be
609+-- original computation. Note that the returned action must be executed exactly
581−-- executed exactly once and drained completely. If not executed or not drained
610+-- once and drained completely. If not executed or not drained fully we may
582−-- fully we will may have a thread blocked forever and if executed more than
611+-- have a thread blocked forever and if executed more than once a
583−-- once a ContextUsedAfterEOF exception will be raised.
612+-- 'ContextUsedAfterEOF' exception will be raised.
585async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
614async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
586async m = do
615async m = do
@@ -589,18 +618,20 @@ async m = do
590infixr 5 <=>
619infixr 5 <=>
592−-- | Sequential interleaved composition, similar to '<>' except that it fairly
621+-- | Sequential interleaved composition, in contrast to '<>' this operator
593−-- interleaves the two 'AsyncT' streams, yielding an element from each stream
622+-- fairly interleaves the two streams instead of appending them; yielding one
594−-- alternately. Should be used only on finite streams.
623+-- element from each stream alternately. Unlike '<>' it cannot be used to fold
624+-- an infinite container of streams.
595{-# INLINE (<=>) #-}
625{-# INLINE (<=>) #-}
596(<=>) :: Streaming t => t m a -> t m a -> t m a
626(<=>) :: Streaming t => t m a -> t m a -> t m a
597m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
627m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
599−-- | Parallel composition similar to '<|>' except that it is left-biased
629+-- | Parallel interleaved composition, in contrast to '<|>' this operator
600−-- instead of being fair. Action on the left is likely to be given a chance to
630+-- "merges" streams in a left biased manner rather than fairly interleaving
601−-- execute before the action on the right. If the left action keeps yielding
631+-- them. It keeps yielding from the stream on the left as long as it can. If
602−-- results without blocking it continues running until it finishes and only
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
603−-- then the right action runs. Unlike '<|>' it can be used on infinite streams.
634+-- used to fold infinite containers of streams.
604{-# INLINE (<|) #-}
635{-# INLINE (<|) #-}
605(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
636(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
606m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
637m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
@@ -609,8 +640,8 @@ m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
609-- Running the monad
640-- Running the monad
610------------------------------------------------------------------------------
641------------------------------------------------------------------------------
612−-- | Run an 'AsyncT' computation, wait for it to finish and discard the
643+-- | Run a composed streaming computation, wait for it to finish and discard
613−-- results.
644+-- the results.
614runStreaming :: (Monad m, Streaming t) => t m a -> m ()
645runStreaming :: (Monad m, Streaming t) => t m a -> m ()
615runStreaming m = go (toStream m)
646runStreaming m = go (toStream m)
@@ -624,21 +655,6 @@ runStreaming m = go (toStream m)
624 yield _ Nothing = stop
655 yield _ Nothing = stop
625 yield _ (Just x) = go x
656 yield _ (Just x) = go x
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)
634− go m1 = (runStream m1) Nothing stop yield
638− {-# INLINE yield #-}
639− yield a Nothing = return [a]
640− yield a (Just x) = liftM (a :) (go x)
642------------------------------------------------------------------------------
658------------------------------------------------------------------------------
643-- Utilities
659-- Utilities
644------------------------------------------------------------------------------
660------------------------------------------------------------------------------
@@ -661,11 +677,3 @@ foldMapWith f g = foldr (f . g) mempty
661forEachWith :: (Monoid b, Foldable t) =>
677forEachWith :: (Monoid b, Foldable t) =>
662 (b1 -> b -> b) -> t a -> (a -> b1) -> b
678 (b1 -> b -> b) -> t a -> (a -> b1) -> b
663forEachWith f xs g = foldr (f . g) mempty xs
679forEachWith f xs g = foldr (f . g) mempty xs
665−------------------------------------------------------------------------------
666−-- Convert a callback into an 'AsyncT' computation
667−------------------------------------------------------------------------------
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)