@@ -30,10 +30,12 @@ module Asyncly.Core
30 , SVarTag (..)
30 , SVarTag (..)
31 , SVarStyle (..)
31 , SVarStyle (..)
32 , EndOfStream (..)
32 , EndOfStream (..)
33− , newSVar1
33+ , newEmptySVar
34− , newSVar2
34+ , newStreamVar1
35− , joinSVar2
35+ , newStreamVar2
36− , streamSVar
36+ , joinStreamVar2
38 -- * Concurrent Streams
40 -- * Concurrent Streams
@@ -95,21 +97,30 @@ data SVarSched = LIFO | FIFO deriving Eq
95-- be scheduled on the same SVar.
97-- be scheduled on the same SVar.
96data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
98data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
98−-- | An SVar (A Stream Var or an Sched Var) is a conduit to multiple streams
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.
99−-- running concurrently. It has an associated runqueue that holds the streams
109+-- An SVar is a mini scheduler, it has an associated runqueue that holds the
100−-- to be picked and run by a pool of worker threads. It has an associated
110+-- stream tasks to be picked and run by a pool of worker threads. It has an
101−-- output queue where the output stream elements are placed by the worker
111+-- associated output queue where the output stream elements are placed by the
102−-- threads. A doorBell is used by the worker threads to intimate the consumer
112+-- worker threads. A doorBell is used by the worker threads to intimate the
103−-- thread about availability of new results in the output queue. More workers
113+-- consumer thread about availability of new results in the output queue. More
104−-- are added to the SVar by 'streamSVar' on demand if the output produced is
114+-- workers are added to the SVar by 'fromStreamVar' on demand if the output
105−-- not keeping pace with the consumer. On bounded SVars, workers block on the
115+-- produced is not keeping pace with the consumer. On bounded SVars, workers
106−-- output queue to provide throttling when the consumer is not pulling fast
116+-- block on the output queue to provide throttling of the producer when the
107−-- enough. The number of workers may even get reduced depending on the
117+-- consumer is not pulling fast enough. The number of workers may even get
108−-- consuming pace.
118+-- reduced depending on the consuming pace.
110-- New work is enqueued either at the time of creation of the SVar or as a
120-- New work is enqueued either at the time of creation of the SVar or as a
111-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
121-- 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'.
113data SVar m a =
124data SVar m a =
114 SVar { outputQueue :: IORef [ChildEvent a]
125 SVar { outputQueue :: IORef [ChildEvent a]
115 , doorBell :: MVar Bool -- wakeup mechanism for outQ
126 , doorBell :: MVar Bool -- wakeup mechanism for outQ
@@ -315,9 +326,9 @@ data EndOfStream = EndOfStream deriving Show
315instance Exception EndOfStream
326instance Exception EndOfStream
317-- | Pull a stream from an SVar.
328-- | Pull a stream from an SVar.
318−{-# NOINLINE streamSVar #-}
329+{-# NOINLINE fromStreamVar #-}
319−streamSVar :: MonadAsync m => SVar m a -> Stream m a
330+fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
320−streamSVar sv = Stream $ \_ stp yld -> do
331+fromStreamVar sv = Stream $ \_ stp yld -> do
321 -- XXX if reading the IORef is costly we can use a flag in the SVar to
332 -- XXX if reading the IORef is costly we can use a flag in the SVar to
322 -- indicate we are done.
333 -- indicate we are done.
323 done <- allThreadsDone sv
334 done <- allThreadsDone sv
@@ -343,7 +354,7 @@ streamSVar sv = Stream $ \_ stp yld -> do
343 processEvents [] = Stream $ \_ stp yld -> do
354 processEvents [] = Stream $ \_ stp yld -> do
344 done <- allThreadsDone sv
355 done <- allThreadsDone sv
345 if not done
356 if not done
346− then (runStream (streamSVar sv)) Nothing stp yld
357+ then (runStream (fromStreamVar sv)) Nothing stp yld
349 processEvents (ev : es) = Stream $ \_ stp yld -> do
360 processEvents (ev : es) = Stream $ \_ stp yld -> do
@@ -392,9 +403,9 @@ getLifoSVar ctype = do
393 in return sv
404 in return sv
395−-- | Create a new SVar and enqueue one stream computation on it.
406+-- | Create a new empty SVar.
396−newSVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
407+newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
397−newSVar1 style m = do
408+newEmptySVar style = do
398 sv <- liftIO $
409 sv <- liftIO $
399 case style of
410 case style of
400 SVarStyle _ FIFO -> do
411 SVarStyle _ FIFO -> do
@@ -403,6 +414,12 @@ newSVar1 style m = do
403 SVarStyle _ LIFO -> do
414 SVarStyle _ LIFO -> do
404 c <- getLifoSVar style
415 c <- getLifoSVar style
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
406 -- Note: We must have all the work on the queue before sending the
423 -- Note: We must have all the work on the queue before sending the
407 -- pushworker, otherwise the pushworker may exit before we even get a
424 -- pushworker, otherwise the pushworker may exit before we even get a
408 -- chance to push.
425 -- chance to push.
@@ -411,9 +428,9 @@ newSVar1 style m = do
411 return sv
428 return sv
413-- | Create a new SVar and enqueue two stream computations on it.
430-- | Create a new SVar and enqueue two stream computations on it.
414−newSVar2 :: MonadAsync m
431+newStreamVar2 :: MonadAsync m
415 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
432 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
416−newSVar2 style m1 m2 = do
433+newStreamVar2 style m1 m2 = do
417 -- Note: We must have all the work on the queue before sending the
434 -- Note: We must have all the work on the queue before sending the
418 -- pushworker, otherwise the pushworker may exit before we even get a
435 -- pushworker, otherwise the pushworker may exit before we even get a
419 -- chance to push.
436 -- chance to push.
@@ -430,6 +447,17 @@ newSVar2 style m1 m2 = do
430 pushWorker sv
447 pushWorker sv
431 return sv
448 return sv
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
433------------------------------------------------------------------------------
461------------------------------------------------------------------------------
434-- Running streams concurrently
462-- Running streams concurrently
435------------------------------------------------------------------------------
463------------------------------------------------------------------------------
@@ -466,8 +494,8 @@ newSVar2 style m1 m2 = do
466withNewSVar2 :: MonadAsync m
494withNewSVar2 :: MonadAsync m
467 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
495 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
468withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
496withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
469− sv <- newSVar2 style m1 m2
497+ sv <- newStreamVar2 style m1 m2
470− (runStream (streamSVar sv)) Nothing stp yld
498+ (runStream (fromStreamVar sv)) Nothing stp yld
472-- | Join two computations on the currently running 'SVar' queue for concurrent
500-- | Join two computations on the currently running 'SVar' queue for concurrent
473-- execution. The 'SVarStyle' required by the current composition context is
501-- execution. The 'SVarStyle' required by the current composition context is
@@ -479,9 +507,9 @@ withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
479-- When we are using parallel composition, an SVar is passed around as a state
507-- 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
508-- 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.
509-- 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
483-- scheduled on the SVar is further evaluated. For example, when (a \<|> b) is
511-- 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
485-- queue. However, if the scheduling and composition style of the new
513-- queue. However, if the scheduling and composition style of the new
486-- computation being scheduled is different than the style of the current SVar,
514-- computation being scheduled is different than the style of the current SVar,
487-- then we create a new SVar and schedule it on that.
515-- then we create a new SVar and schedule it on that.
@@ -497,10 +525,10 @@ withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
497-- composition and vice-versa we create a new SVar to isolate the scheduling
525-- composition and vice-versa we create a new SVar to isolate the scheduling
498-- of the two.
526-- of the two.
500−{-# INLINE joinSVar2 #-}
528+{-# INLINE joinStreamVar2 #-}
501−joinSVar2 :: MonadAsync m
529+joinStreamVar2 :: MonadAsync m
502 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
530 => 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
504 case st of
532 case st of
505 Just sv | svarStyle sv == style ->
533 Just sv | svarStyle sv == style ->
506 liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
534 liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
@@ -526,7 +554,7 @@ parAhead = undefined
526-- cannot fold infinite number of streams using this operation.
554-- cannot fold infinite number of streams using this operation.
527{-# INLINE parAlt #-}
555{-# INLINE parAlt #-}
528parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
556parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
529−parAlt = joinSVar2 (SVarStyle Disjunction FIFO)
557+parAlt = joinStreamVar2 (SVarStyle Disjunction FIFO)
531-- | Same as '<|'. Since this schedules the left side computation first you can
559-- | 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
560-- right fold an infinite container using this operator. However a left fold
@@ -535,7 +563,7 @@ parAlt = joinSVar2 (SVarStyle Disjunction FIFO)
535-- structure.
563-- structure.
536{-# INLINE parLeft #-}
564{-# INLINE parLeft #-}
537parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
565parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
538−parLeft = joinSVar2 (SVarStyle Disjunction LIFO)
566+parLeft = joinStreamVar2 (SVarStyle Disjunction LIFO)
540-------------------------------------------------------------------------------
568-------------------------------------------------------------------------------
541-- Instances (only used for deriving newtype instances)
569-- Instances (only used for deriving newtype instances)