@@ -32,10 +32,10 @@ module Asyncly.Core
32 , EndOfStream (..)
32 , EndOfStream (..)
33 , newSVar1
33 , newSVar1
34 , newSVar2
34 , newSVar2
35 , streamSVar
36 , streamSVar
37 -- * Concurrent Streams
38 -- * Concurrent Streams
@@ -92,19 +92,24 @@ data SVarTag = Conjunction | Disjunction deriving Eq
92data SVarSched = LIFO | FIFO deriving Eq
92data SVarSched = LIFO | FIFO deriving Eq
94-- | Identify the type of the SVar. Two computations using the same style can
94-- | Identify the type of the SVar. Two computations using the same style can
95−-- be bunched on the same SVar.
95+-- be scheduled on the same SVar.
96data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
96data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
98−-- | An SVar is a conduit to multiple streams running concurrently. It has an
98+-- | An SVar (A Stream Var or an Sched Var) is a conduit to multiple streams
99−-- associated runqueue that holds the streams to be picked by a pool of worker
99+-- running concurrently. It has an associated runqueue that holds the streams
100+-- to be picked and run by a pool of worker threads. It has an associated
100−-- threads. It has an associated output queue where the output stream elements
101+-- output queue where the output stream elements are placed by the worker
101−-- are placed by the worker threads. A doorBell is used by the worker threads
102+-- threads. A doorBell is used by the worker threads to intimate the consumer
102−-- to intimate the consumer thread about availability of new results in the
103+-- thread about availability of new results in the output queue. More workers
103−-- output queue.
104+-- are added to the SVar by 'streamSVar' on demand if the output produced is
105+-- not keeping pace with the consumer. On bounded SVars, workers block on the
106+-- output queue to provide throttling when the consumer is not pulling fast
107+-- enough. The number of workers may even get reduced depending on the
105-- New work is enqueued either at the time of creation of the SVar or as a
110-- New work is enqueued either at the time of creation of the SVar or as a
106−-- result of executing the parallel combinators i.e. '<|' and '<|>' by already
111+-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
107−-- enqueued work.
112+-- already enqueued computations get evaluated. See 'joinSVar2'.
108data SVar m a =
113data SVar m a =
109 SVar { outputQueue :: IORef [ChildEvent a]
114 SVar { outputQueue :: IORef [ChildEvent a]
110 , doorBell :: MVar Bool -- wakeup mechanism for outQ
115 , doorBell :: MVar Bool -- wakeup mechanism for outQ
@@ -122,9 +127,9 @@ data SVar m a =
122-- TBD use a functor instead of the bare type a?
127-- TBD use a functor instead of the bare type a?
124-- | Represents a monadic stream of values of type 'a' constructed using
129-- | Represents a monadic stream of values of type 'a' constructed using
125−-- actions in monad 'm'. Streams can be composed sequentially or in parallel in
130+-- actions in monad 'm'. Streams can be composed sequentially or in parallel;
126−-- product style compositions (monadic bind multiplies streams in a ListT
131+-- in product style compositions (monadic bind multiplies streams in a ListT
127−-- fashion) and using sum style compositions like 'Semigroup', 'Monoid',
132+-- fashion) or in sum style compositions like 'Semigroup', 'Monoid',
128-- 'Alternative' or variants of these.
133-- 'Alternative' or variants of these.
129newtype Stream m a =
134newtype Stream m a =
@@ -138,7 +143,7 @@ newtype Stream m a =
138-- | A monad that can perform asynchronous/concurrent IO operations.
143-- | A monad that can perform asynchronous/concurrent IO operations.
139type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
144type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
141−-- | Yield a singleton value in a stream.
146+-- | Yield a singleton value as a stream.
142yields :: a -> Stream m a
147yields :: a -> Stream m a
143yields a = Stream $ \_ _ yld -> yld a Nothing
148yields a = Stream $ \_ _ yld -> yld a Nothing
@@ -147,7 +152,7 @@ yields a = Stream $ \_ _ yld -> yld a Nothing
147------------------------------------------------------------------------------
152------------------------------------------------------------------------------
149-- | '<>' concatenates two streams sequentially i.e. the first stream is
154-- | '<>' concatenates two streams sequentially i.e. the first stream is
150−-- exhausted completely before yields any element from the second stream.
155+-- exhausted completely before yielding any element from the second stream.
151instance Semigroup (Stream m a) where
156instance Semigroup (Stream m a) where
152 m1 <> m2 = go m1
157 m1 <> m2 = go m1
@@ -305,7 +310,7 @@ sendWorkerWait sv = do
305 then (pushWorker sv) >> sendWorkerWait sv
310 then (pushWorker sv) >> sendWorkerWait sv
306 else void (liftIO $ takeMVar (doorBell sv))
311 else void (liftIO $ takeMVar (doorBell sv))
308−-- | An 'async' stream has finished but is still being used.
313+-- | A stream being pulled from 'SVar' has ended.
309data EndOfStream = EndOfStream deriving Show
314data EndOfStream = EndOfStream deriving Show
310instance Exception EndOfStream
315instance Exception EndOfStream
@@ -457,27 +462,49 @@ newSVar2 style m1 m2 = do
457-- TBD for pure work (when we are not in the IO monad) we can divide it into
462-- TBD for pure work (when we are not in the IO monad) we can divide it into
458-- just the number of CPUs.
463-- just the number of CPUs.
460−{-# NOINLINE makeAsync #-}
465+{-# NOINLINE withNewSVar2 #-}
461−makeAsync :: MonadAsync m
466+withNewSVar2 :: MonadAsync m
462 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
467 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
463−makeAsync style m1 m2 = Stream $ \_ stp yld -> do
468+withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
464 sv <- newSVar2 style m1 m2
469 sv <- newSVar2 style m1 m2
465 (runStream (streamSVar sv)) Nothing stp yld
470 (runStream (streamSVar sv)) Nothing stp yld
467−-- | Compose two streams in parallel using a scheduling policy specified by
472+-- | Join two computations on the currently running 'SVar' queue for concurrent
468−-- 'SVarStyle'. Note: This is designed to scale for right associated
473+-- execution. The 'SVarStyle' required by the current composition context is
469−-- compositions, therefore always use a right fold for folding large or
474+-- passed as one of the parameters. If the style does not match with the style
470−-- infinite structures. For left associated structures it will first
475+-- of the current 'SVar' we create a new 'SVar' and schedule the computations
471−-- destructure the whole structure and then start executing, consuming memory
476+-- on that. The newly created SVar joins as one of the computations on the
472−-- proportional to the size of the structure, just like a left fold.
477+-- current SVar queue.
479+-- When we are using parallel composition, an SVar is passed around as a state
480+-- variable. We try to schedule a new parallel computation on the SVar passed
481+-- to us. The first time, when no SVar exists, a new SVar is created.
482+-- Subsequently, 'joinSVar2' may get called when a computation already
483+-- scheduled on the SVar is further evaluated. For example, when (a \<|> b) is
484+-- evaluated it calls a 'joinSVar2' to put 'a' and 'b' on the current scheduler
485+-- queue. However, if the scheduling and composition style of the new
486+-- computation being scheduled is different than the style of the current SVar,
487+-- then we create a new SVar and schedule it on that.
491+-- * (x \<|> y) \<|> (t \<|> u) -- all of them get scheduled on the same SVar
492+-- * (x \<|> y) \<|> (t \<| u) -- @t@ and @u@ get scheduled on a new child SVar
493+-- because of the scheduling policy change.
494+-- * if we 'adapt' a stream of type 'AsyncT' to a stream of type
495+-- 'ParallelT', we create a new SVar at the transitioning bind.
496+-- * When the stream is switching from disjunctive composition to conjunctive
497+-- composition and vice-versa we create a new SVar to isolate the scheduling
473−{-# INLINE parallel #-}
500+{-# INLINE joinSVar2 #-}
501+joinSVar2 :: MonadAsync m
474−parallel :: MonadAsync m => SVarStyle -> Stream m a -> Stream m a -> Stream m a
502+ => SVarStyle -> Stream m a -> Stream m a -> Stream m a
475−parallel style m1 m2 = Stream $ \st stp yld -> do
503+joinSVar2 style m1 m2 = Stream $ \st stp yld -> do
476 case st of
504 case st of
477− Nothing -> (runStream (makeAsync style m1 m2)) Nothing stp yld
478− Just sv | svarStyle sv /= style ->
505+ Just sv | svarStyle sv == style ->
479− (runStream (makeAsync style m1 m2)) Nothing stp yld
480− Just sv -> liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
506+ liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
507+ _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp yld
482------------------------------------------------------------------------------
509------------------------------------------------------------------------------
483-- Semigroup and Monoid style compositions for parallel actions
510-- Semigroup and Monoid style compositions for parallel actions
@@ -495,15 +522,20 @@ parAhead = undefined
495(<>|) = parAhead
522(<>|) = parAhead
498−-- | Same as '<|>'.
525+-- | Same as '<|>'. Since this schedules all the composed streams fairly you
526+-- cannot fold infinite number of streams using this operation.
499{-# INLINE parAlt #-}
527{-# INLINE parAlt #-}
500parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
528parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
501−parAlt = parallel (SVarStyle Disjunction FIFO)
529+parAlt = joinSVar2 (SVarStyle Disjunction FIFO)
503−-- | Same as '<|'.
531+-- | Same as '<|'. Since this schedules the left side computation first you can
532+-- right fold an infinite container using this operator. However a left fold
533+-- will not work well as it first unpeels the whole structure before scheduling
534+-- a computation requiring an amount of memory proportional to the size of the
504{-# INLINE parLeft #-}
536{-# INLINE parLeft #-}
505parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
537parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
506−parLeft = parallel (SVarStyle Disjunction LIFO)
538+parLeft = joinSVar2 (SVarStyle Disjunction LIFO)
508-------------------------------------------------------------------------------
540-------------------------------------------------------------------------------
509-- Instances (only used for deriving newtype instances)
541-- Instances (only used for deriving newtype instances)