@@ -32,10 +32,10 @@ module Asyncly.Core
3232 , EndOfStream (..)
3333 , newSVar1
3434 , newSVar2
35+ , joinSVar2
3536 , streamSVar
3637
3738 -- * Concurrent Streams
38− , parallel
3939 , parAlt
4040 , parLeft
4141 )
@@ -92,19 +92,24 @@ data SVarTag = Conjunction | Disjunction deriving Eq
9292data SVarSched = LIFO | FIFO deriving Eq
9393
9494-- | 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.
9696data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
9797
98−-- | An SVar is a conduit to multiple streams running concurrently. It has an
99−-- associated runqueue that holds the streams to be picked by a pool of worker
100−-- threads. It has an associated output queue where the output stream elements
101−-- are placed by the worker threads. A doorBell is used by the worker threads
102−-- to intimate the consumer thread about availability of new results in the
103−-- output queue.
98+-- | An SVar (A Stream Var or an Sched Var) is a conduit to multiple streams
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
101+-- output queue where the output stream elements are placed by the worker
102+-- threads. A doorBell is used by the worker threads to intimate the consumer
103+-- thread about availability of new results in the output queue. More workers
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
108+-- consuming pace.
104109--
105110-- 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
107−-- enqueued work.
111+-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
112+-- already enqueued computations get evaluated. See 'joinSVar2'.
108113data SVar m a =
109114 SVar { outputQueue :: IORef [ChildEvent a]
110115 , doorBell :: MVar Bool -- wakeup mechanism for outQ
@@ -122,9 +127,9 @@ data SVar m a =
122127-- TBD use a functor instead of the bare type a?
123128
124129-- | 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
126−-- product style compositions (monadic bind multiplies streams in a ListT
127−-- fashion) and using sum style compositions like 'Semigroup', 'Monoid',
130+-- actions in monad 'm'. Streams can be composed sequentially or in parallel;
131+-- in product style compositions (monadic bind multiplies streams in a ListT
132+-- fashion) or in sum style compositions like 'Semigroup', 'Monoid',
128133-- 'Alternative' or variants of these.
129134newtype Stream m a =
130135 Stream {
@@ -138,7 +143,7 @@ newtype Stream m a =
138143-- | A monad that can perform asynchronous/concurrent IO operations.
139144type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
140145
141−-- | Yield a singleton value in a stream.
146+-- | Yield a singleton value as a stream.
142147yields :: a -> Stream m a
143148yields a = Stream $ \_ _ yld -> yld a Nothing
144149
@@ -147,7 +152,7 @@ yields a = Stream $ \_ _ yld -> yld a Nothing
147152------------------------------------------------------------------------------
148153
149154-- | '<>' 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.
151156instance Semigroup (Stream m a) where
152157 m1 <> m2 = go m1
153158 where
@@ -305,7 +310,7 @@ sendWorkerWait sv = do
305310 then (pushWorker sv) >> sendWorkerWait sv
306311 else void (liftIO $ takeMVar (doorBell sv))
307312
308−-- | An 'async' stream has finished but is still being used.
313+-- | A stream being pulled from 'SVar' has ended.
309314data EndOfStream = EndOfStream deriving Show
310315instance Exception EndOfStream
311316
@@ -457,27 +462,49 @@ newSVar2 style m1 m2 = do
457462-- TBD for pure work (when we are not in the IO monad) we can divide it into
458463-- just the number of CPUs.
459464
460−{-# NOINLINE makeAsync #-}
461−makeAsync :: MonadAsync m
465+{-# NOINLINE withNewSVar2 #-}
466+withNewSVar2 :: MonadAsync m
462467 => 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
464469 sv <- newSVar2 style m1 m2
465470 (runStream (streamSVar sv)) Nothing stp yld
466471
467−-- | Compose two streams in parallel using a scheduling policy specified by
468−-- 'SVarStyle'. Note: This is designed to scale for right associated
469−-- compositions, therefore always use a right fold for folding large or
470−-- infinite structures. For left associated structures it will first
471−-- destructure the whole structure and then start executing, consuming memory
472−-- proportional to the size of the structure, just like a left fold.
473−{-# INLINE parallel #-}
474−parallel :: MonadAsync m => SVarStyle -> Stream m a -> Stream m a -> Stream m a
475−parallel style m1 m2 = Stream $ \st stp yld -> do
472+-- | Join two computations on the currently running 'SVar' queue for concurrent
473+-- execution. The 'SVarStyle' required by the current composition context is
474+-- passed as one of the parameters. If the style does not match with the style
475+-- of the current 'SVar' we create a new 'SVar' and schedule the computations
476+-- on that. The newly created SVar joins as one of the computations on the
477+-- current SVar queue.
478+--
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.
488+--
489+-- For example:
490+--
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
498+-- of the two.
499+--
500+{-# INLINE joinSVar2 #-}
501+joinSVar2 :: MonadAsync m
502+ => SVarStyle -> Stream m a -> Stream m a -> Stream m a
503+joinSVar2 style m1 m2 = Stream $ \st stp yld -> do
476504 case st of
477− Nothing -> (runStream (makeAsync style m1 m2)) Nothing stp yld
478− 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
505+ Just sv | svarStyle sv == style ->
506+ liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
507+ _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp yld
481508
482509------------------------------------------------------------------------------
483510-- Semigroup and Monoid style compositions for parallel actions
@@ -495,15 +522,20 @@ parAhead = undefined
495522(<>|) = parAhead
496523-}
497524
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.
499527{-# INLINE parAlt #-}
500528parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
501−parAlt = parallel (SVarStyle Disjunction FIFO)
529+parAlt = joinSVar2 (SVarStyle Disjunction FIFO)
502530
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
535+-- structure.
504536{-# INLINE parLeft #-}
505537parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
506−parLeft = parallel (SVarStyle Disjunction LIFO)
538+parLeft = joinSVar2 (SVarStyle Disjunction LIFO)
507539
508540-------------------------------------------------------------------------------
509541-- Instances (only used for deriving newtype instances)