@@ -30,10 +30,12 @@ module Asyncly.Core
3030 , SVarTag (..)
3131 , SVarStyle (..)
3232 , EndOfStream (..)
33− , newSVar1
34− , newSVar2
35− , joinSVar2
36− , streamSVar
33+ , newEmptySVar
34+ , newStreamVar1
35+ , newStreamVar2
36+ , joinStreamVar2
37+ , fromStreamVar
38+ , toStreamVar
3739
3840 -- * Concurrent Streams
3941 , parAlt
@@ -95,21 +97,30 @@ data SVarSched = LIFO | FIFO deriving Eq
9597-- be scheduled on the same SVar.
9698data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
9799
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.
100+-- | An SVar or a Stream Var is a conduit to the output from multiple streams
101+-- running concurrently and asynchronously. An SVar can be thought of as an
102+-- asynchronous IO handle. We can write any number of streams to an SVar in a
103+-- non-blocking manner and then read them back at any time at any pace. The
104+-- SVar would run the streams asynchronously and accumulate results. An SVar
105+-- may not really execute the stream completely and accumulate all the results.
106+-- However, it ensures that the reader can read the results at whatever paces
107+-- it wants to read. The SVar monitors and adapts to the consumer's pace.
108+--
109+-- An SVar is a mini scheduler, it has an associated runqueue that holds the
110+-- stream tasks to be picked and run by a pool of worker threads. It has an
111+-- associated output queue where the output stream elements are placed by the
112+-- worker threads. A doorBell is used by the worker threads to intimate the
113+-- consumer thread about availability of new results in the output queue. More
114+-- workers are added to the SVar by 'fromStreamVar' on demand if the output
115+-- produced is not keeping pace with the consumer. On bounded SVars, workers
116+-- block on the output queue to provide throttling of the producer when the
117+-- consumer is not pulling fast enough. The number of workers may even get
118+-- reduced depending on the consuming pace.
109119--
110120-- New work is enqueued either at the time of creation of the SVar or as a
111121-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
112−-- already enqueued computations get evaluated. See 'joinSVar2'.
122+-- already enqueued computations get evaluated. See 'joinStreamVar2'.
123+--
113124data SVar m a =
114125 SVar { outputQueue :: IORef [ChildEvent a]
115126 , doorBell :: MVar Bool -- wakeup mechanism for outQ
@@ -315,9 +326,9 @@ data EndOfStream = EndOfStream deriving Show
315326instance Exception EndOfStream
316327
317328-- | Pull a stream from an SVar.
318−{-# NOINLINE streamSVar #-}
319−streamSVar :: MonadAsync m => SVar m a -> Stream m a
320−streamSVar sv = Stream $ \_ stp yld -> do
329+{-# NOINLINE fromStreamVar #-}
330+fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
331+fromStreamVar sv = Stream $ \_ stp yld -> do
321332 -- XXX if reading the IORef is costly we can use a flag in the SVar to
322333 -- indicate we are done.
323334 done <- allThreadsDone sv
@@ -343,7 +354,7 @@ streamSVar sv = Stream $ \_ stp yld -> do
343354 processEvents [] = Stream $ \_ stp yld -> do
344355 done <- allThreadsDone sv
345356 if not done
346− then (runStream (streamSVar sv)) Nothing stp yld
357+ then (runStream (fromStreamVar sv)) Nothing stp yld
347358 else stp
348359
349360 processEvents (ev : es) = Stream $ \_ stp yld -> do
@@ -392,9 +403,9 @@ getLifoSVar ctype = do
392403 }
393404 in return sv
394405
395−-- | Create a new SVar and enqueue one stream computation on it.
396−newSVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
397−newSVar1 style m = do
406+-- | Create a new empty SVar.
407+newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
408+newEmptySVar style = do
398409 sv <- liftIO $
399410 case style of
400411 SVarStyle _ FIFO -> do
@@ -403,6 +414,12 @@ newSVar1 style m = do
403414 SVarStyle _ LIFO -> do
404415 c <- getLifoSVar style
405416 return c
417+ return sv
418+
419+-- | Create a new SVar and enqueue one stream computation on it.
420+newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
421+newStreamVar1 style m = do
422+ sv <- newEmptySVar style
406423 -- Note: We must have all the work on the queue before sending the
407424 -- pushworker, otherwise the pushworker may exit before we even get a
408425 -- chance to push.
@@ -411,9 +428,9 @@ newSVar1 style m = do
411428 return sv
412429
413430-- | Create a new SVar and enqueue two stream computations on it.
414−newSVar2 :: MonadAsync m
431+newStreamVar2 :: MonadAsync m
415432 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
416−newSVar2 style m1 m2 = do
433+newStreamVar2 style m1 m2 = do
417434 -- Note: We must have all the work on the queue before sending the
418435 -- pushworker, otherwise the pushworker may exit before we even get a
419436 -- chance to push.
@@ -430,6 +447,17 @@ newSVar2 style m1 m2 = do
430447 pushWorker sv
431448 return sv
432449
450+-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
451+-- be read back from the SVar using 'fromSVar'.
452+toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
453+toStreamVar sv m = do
454+ liftIO $ (enqueue sv) m
455+ done <- allThreadsDone sv
456+ -- XXX there may be a race here unless we are running in the consumer
457+ -- thread. This is safe only when called from the consumer thread or when
458+ -- no consumer is present.
459+ when done $ pushWorker sv
460+
433461------------------------------------------------------------------------------
434462-- Running streams concurrently
435463------------------------------------------------------------------------------
@@ -466,8 +494,8 @@ newSVar2 style m1 m2 = do
466494withNewSVar2 :: MonadAsync m
467495 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
468496withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
469− sv <- newSVar2 style m1 m2
470− (runStream (streamSVar sv)) Nothing stp yld
497+ sv <- newStreamVar2 style m1 m2
498+ (runStream (fromStreamVar sv)) Nothing stp yld
471499
472500-- | Join two computations on the currently running 'SVar' queue for concurrent
473501-- execution. The 'SVarStyle' required by the current composition context is
@@ -479,9 +507,9 @@ withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
479507-- When we are using parallel composition, an SVar is passed around as a state
480508-- variable. We try to schedule a new parallel computation on the SVar passed
481509-- to us. The first time, when no SVar exists, a new SVar is created.
482−-- Subsequently, 'joinSVar2' may get called when a computation already
510+-- Subsequently, 'joinStreamVar2' may get called when a computation already
483511-- 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
512+-- evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on the current scheduler
485513-- queue. However, if the scheduling and composition style of the new
486514-- computation being scheduled is different than the style of the current SVar,
487515-- then we create a new SVar and schedule it on that.
@@ -497,10 +525,10 @@ withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
497525-- composition and vice-versa we create a new SVar to isolate the scheduling
498526-- of the two.
499527--
500−{-# INLINE joinSVar2 #-}
501−joinSVar2 :: MonadAsync m
528+{-# INLINE joinStreamVar2 #-}
529+joinStreamVar2 :: MonadAsync m
502530 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
503−joinSVar2 style m1 m2 = Stream $ \st stp yld -> do
531+joinStreamVar2 style m1 m2 = Stream $ \st stp yld -> do
504532 case st of
505533 Just sv | svarStyle sv == style ->
506534 liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
@@ -526,7 +554,7 @@ parAhead = undefined
526554-- cannot fold infinite number of streams using this operation.
527555{-# INLINE parAlt #-}
528556parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
529−parAlt = joinSVar2 (SVarStyle Disjunction FIFO)
557+parAlt = joinStreamVar2 (SVarStyle Disjunction FIFO)
530558
531559-- | Same as '<|'. Since this schedules the left side computation first you can
532560-- right fold an infinite container using this operator. However a left fold
@@ -535,7 +563,7 @@ parAlt = joinSVar2 (SVarStyle Disjunction FIFO)
535563-- structure.
536564{-# INLINE parLeft #-}
537565parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
538−parLeft = joinSVar2 (SVarStyle Disjunction LIFO)
566+parLeft = joinStreamVar2 (SVarStyle Disjunction LIFO)
539567
540568-------------------------------------------------------------------------------
541569-- Instances (only used for deriving newtype instances)