@@ -260,7 +260,6 @@ doFork :: MonadBaseControl IO m
260260doFork action exHandler =
261261 EL.mask $ \restore ->
262262 liftBaseWith $ \runInIO -> forkIO $ do
263− -- XXX test the exception handling
264263 _ <- runInIO $ EL.catch (restore action) exHandler
265264 -- XXX restore state here?
266265 return ()
@@ -498,42 +497,48 @@ toStreamVar :: MonadParallel m => SVar m a -> Stream m a -> m ()
498497toStreamVar sv m = do
499498 liftIO $ (enqueue sv) m
500499 done <- allThreadsDone sv
501− -- XXX there may be a race here unless we are running in the consumer
502− -- thread. This is safe only when called from the consumer thread or when
503− -- no consumer is present.
500+ -- XXX This is safe only when called from the consumer thread or when no
501+ -- consumer is present. There may be a race if we are not running in the
502+ -- consumer thread.
504503 when done $ pushWorker sv
505504
506505------------------------------------------------------------------------------
507506-- Running streams concurrently
508507------------------------------------------------------------------------------
509508
510−-- Concurrency rate control. Our objective is to create more threads on demand
511−-- if the consumer is running faster than us. As soon as we encounter an
512−-- Alternative composition we create a push pull pair of threads. We use a
513−-- channel for communication between the consumer pulling from the channel and
514−-- the producer who is pushing to the channel. The producer creates more threads
515−-- if no output is seen on the channel, that is the consumer is running faster.
516−-- However this mechanism can be problematic if the initial production latency
517−-- is high, we may end up creating too many threads. So we need some way to
518−-- monitor and use the latency as well.
509+-- Concurrency rate control.
519510--
520−-- TBD We may run computations at the lower level of the composition tree
521−-- serially even if they are composed using a parallel combinator. We can use
522−-- <> in place of <| and <=> in place of <|>. If we find that a parallel
523−-- channel immediately above a computation becomes empty we can switch to
524−-- parallelizing the computation. For that we can use a state flag to fork the
525−-- rest of the computation at any point of time inside the Monad bind operation
526−-- if the consumer is running at a faster speed.
511+-- Our objective is to create more threads on demand if the consumer is running
512+-- faster than us. As soon as we encounter a concurrent composition we create a
513+-- push pull pair of threads. We use an SVar for communication between the
514+-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
515+-- The producer creates more threads if the SVar drains and becomes empty, that
516+-- is the consumer is running faster.
527517--
528−-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
529−-- If we have to dispatch in arbitrary chunksizes we will need to compose the
530−-- parallel actions using a data constructor (Free Alternative) instead so that
531−-- we can divide it in chunks of arbitrary size before dispatching. If the stream
532−-- is composed of hierarchically composed grains of different sizes then we can
533−-- always switch to a desired granularity depending on the consumer speed.
518+-- XXX Note 1: This mechanism can be problematic if the initial production
519+-- latency is high, we may end up creating too many threads. So we need some
520+-- way to monitor and use the latency as well. Having a limit on the dispatches
521+-- (programmer controlled) may also help.
534522--
535−-- TBD for pure work (when we are not in the IO monad) we can divide it into
536−-- just the number of CPUs.
523+-- TBD Note 2: We may want to run computations at the lower level of the
524+-- composition tree serially even when they are composed using a parallel
525+-- combinator. We can use 'splice' in place of 'coparallel' and 'cosplice' in
526+-- place of 'parallel'. If we find that an SVar immediately above a computation
527+-- gets drained empty we can switch to parallelizing the computation. For that
528+-- we can use a state flag to fork the rest of the computation at any point of
529+-- time inside the Monad bind operation if the consumer is running at a faster
530+-- speed.
531+--
532+-- TBD Note 3: the binary operation ('parallel') composition allows us to
533+-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
534+-- chunksizes we will need to compose the parallel actions using a data
535+-- constructor (A Free container) instead so that we can divide it in chunks of
536+-- arbitrary size before dispatching. If the stream is composed of
537+-- hierarchically composed grains of different sizes then we can always switch
538+-- to a desired granularity depending on the consumer speed.
539+--
540+-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
541+-- into just the number of CPUs.
537542
538543{-# NOINLINE withNewSVar2 #-}
539544withNewSVar2 :: MonadParallel m
@@ -543,29 +548,27 @@ withNewSVar2 style m1 m2 = Stream $ \_ stp sng yld -> do
543548 (runStream (fromStreamVar sv)) Nothing stp sng yld
544549
545550-- | Join two computations on the currently running 'SVar' queue for concurrent
546−-- execution. The 'SVarStyle' required by the current composition context is
547−-- passed as one of the parameters. If the style does not match with the style
548−-- of the current 'SVar' we create a new 'SVar' and schedule the computations
549−-- on that. The newly created SVar joins as one of the computations on the
550−-- current SVar queue.
551+-- execution. When we are using parallel composition, an SVar is passed around
552+-- as a state variable. We try to schedule a new parallel computation on the
553+-- SVar passed to us. The first time, when no SVar exists, a new SVar is
554+-- created. Subsequently, 'joinStreamVar2' may get called when a computation
555+-- already scheduled on the SVar is further evaluated. For example, when (a
556+-- `parallel` b) is evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on
557+-- the current scheduler queue.
551558--
552−-- When we are using parallel composition, an SVar is passed around as a state
553−-- variable. We try to schedule a new parallel computation on the SVar passed
554−-- to us. The first time, when no SVar exists, a new SVar is created.
555−-- Subsequently, 'joinStreamVar2' may get called when a computation already
556−-- scheduled on the SVar is further evaluated. For example, when (a \<|> b) is
557−-- evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on the current scheduler
558−-- queue. However, if the scheduling and composition style of the new
559+-- The 'SVarStyle' required by the current composition context is passed as one
560+-- of the parameters. If the scheduling and composition style of the new
559561-- computation being scheduled is different than the style of the current SVar,
560−-- then we create a new SVar and schedule it on that.
562+-- then we create a new SVar and schedule it on that. The newly created SVar
563+-- joins as one of the computations on the current SVar queue.
561564--
562−-- For example:
565+-- Cases when we need to switch to a new SVar:
563566--
564−-- * (x \<|> y) \<|> (t \<|> u) -- all of them get scheduled on the same SVar
565−-- * (x \<|> y) \<|> (t \<| u) -- @t@ and @u@ get scheduled on a new child SVar
567+-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
568+-- * (x `parallel` y) `parallel` (t `coparallel` u) -- @t@ and @u@ get scheduled on a new child SVar
566569-- because of the scheduling policy change.
567−-- * if we 'adapt' a stream of type 'AsyncT' to a stream of type
568−-- 'ParallelT', we create a new SVar at the transitioning bind.
570+-- * if we 'adapt' a stream of type 'Coparallel' to a stream of type
571+-- 'Parallel', we create a new SVar at the transitioning bind.
569572-- * When the stream is switching from disjunctive composition to conjunctive
570573-- composition and vice-versa we create a new SVar to isolate the scheduling
571574-- of the two.