@@ -58,7 +58,6 @@ module Streamly.Core
5858 , SVarStyle (..)
5959 , newEmptySVar
6060 , newStreamVar1
61− , newStreamVar2
6261 , joinStreamVar2
6362 , fromStreamVar
6463 , toStreamVar
@@ -147,15 +146,23 @@ data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
147146-- already enqueued computations get evaluated. See 'joinStreamVar2'.
148147--
149148data SVar m a =
150− SVar { outputQueue :: IORef ([ChildEvent a], Int)
151− , doorBell :: MVar () -- signal the consumer about output
152− , siren :: MVar () -- hooter for workers to begin work
149+ SVar {
150+ -- Read only state
151+ svarStyle :: SVarStyle
152+
153+ -- Shared output queue (events, length)
154+ , outputQueue :: IORef ([ChildEvent a], Int)
155+ , doorBell :: MVar () -- signal the consumer about output
156+ , siren :: MVar () -- hooter for workers to begin work
157+
158+ -- Shared work queue
153159 , enqueue :: Stream m a -> IO ()
160+ , queueEmpty :: m Bool
154161 , runqueue :: m ()
162+
163+ -- Shared, thread tracking
155164 , runningThreads :: IORef (Set ThreadId)
156− , queueEmpty :: m Bool
157165 , activeWorkers :: IORef Int
158− , svarStyle :: SVarStyle
159166 }
160167
161168-- Slightly faster version of CAS. Gained some improvement by avoiding the use
@@ -302,8 +309,8 @@ doFork action exHandler =
302309-- TBD Each worker can have their own queue and the consumer can empty one
303310-- queue at a time, that way contention can be reduced.
304311{-# INLINE send #-}
305−send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
306−send sv msg = liftIO $ do
312+send :: SVar m a -> ChildEvent a -> IO ()
313+send sv msg = do
307314 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
308315 ((msg : es, n + 1), n)
309316 if (len <= 0) then do
@@ -340,10 +347,13 @@ send sv msg = liftIO $ do
340347 else return ()
341348
342349{-# INLINE sendStop #-}
343−sendStop :: MonadIO m => SVar m a -> m ()
344−sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
350+sendStop :: SVar m a -> IO ()
351+sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
345352
346−-- Note: Left associated compositions can grow this queue to a large size
353+-- Note: For purely right associated expressions this queue should have at most
354+-- one element. It grows to more than one when we have left associcated
355+-- expressions. Large left associated compositions can grow this to a
356+-- large size
347357{-# INLINE enqueueLIFO #-}
348358enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
349359enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
@@ -356,10 +366,10 @@ runqueueLIFO sv q = run
356366 run = do
357367 work <- dequeue
358368 case work of
359− Nothing -> sendStop sv
369+ Nothing -> liftIO $ sendStop sv
360370 Just m -> (runStream m) (Just sv) run single yield
361371
362− sendit a = send sv (ChildYield a)
372+ sendit a = liftIO $ send sv (ChildYield a)
363373 single a = sendit a >> run
364374 yield a r = sendit a >> (runStream r) (Just sv) run single yield
365375
@@ -379,11 +389,11 @@ runqueueFIFO sv q = run
379389 run = do
380390 work <- dequeue
381391 case work of
382− Nothing -> sendStop sv
392+ Nothing -> liftIO $ sendStop sv
383393 Just m -> (runStream m) (Just sv) run single yield
384394
385395 dequeue = liftIO $ tryPopR q
386− sendit a = send sv (ChildYield a)
396+ sendit a = liftIO $ send sv (ChildYield a)
387397 single a = sendit a >> run
388398 yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run
389399
@@ -393,8 +403,8 @@ runOne sv m = (runStream m) (Just sv) stop single yield
393403
394404 where
395405
396− stop = sendStop sv
397− sendit a = send sv (ChildYield a)
406+ stop = liftIO $ sendStop sv
407+ sendit a = liftIO $ send sv (ChildYield a)
398408 single a = sendit a >> stop
399409 yield a r = sendit a >> runOne sv r
400410
@@ -442,8 +452,8 @@ allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
442452
443453{-# NOINLINE handleChildException #-}
444454handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
445−handleChildException sv e = do
446− tid <- liftIO myThreadId
455+handleChildException sv e = liftIO $ do
456+ tid <- myThreadId
447457 send sv (ChildStop tid (Just e))
448458
449459{-# NOINLINE pushWorker #-}
@@ -475,9 +485,13 @@ sendWorkerWait sv = do
475485 -- If there is no output pending to process and there is no worker to be
476486 -- sent then we block, so that we do not keep looping fruitlessly.
477487
488+ -- XXX we should wait in such a way so that we can be immediately
489+ -- interrupted by a doorBell.
478490 liftIO $ threadDelay 200
479491 (_, n) <- liftIO $ readIORef (outputQueue sv)
480492 when (n <= 0) $ do
493+ -- XXX the queue may be empty temporarily if the worker has dequeued
494+ -- the work item but has not enqueued the remaining part yet.
481495 done <- queueEmpty sv
482496 if not done
483497 then do
@@ -620,27 +634,6 @@ newStreamVar1 style m = do
620634 pushWorker sv
621635 return sv
622636
623−-- | Create a new SVar and enqueue two stream computations on it.
624−newStreamVar2 :: MonadAsync m
625− => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
626−newStreamVar2 style m1 m2 = do
627− -- Note: We must have all the work on the queue before sending the
628− -- pushworker, otherwise the pushworker may exit before we even get a
629− -- chance to push.
630− sv <- liftIO $
631− case style of
632− SVarStyle _ FIFO -> do
633− c <- getFifoSVar style
634− (enqueue c) m1 >> (enqueue c) m2
635− return c
636− SVarStyle _ LIFO -> do
637− c <- getLifoSVar style
638− (enqueue c) m2 >> (enqueue c) m1
639− return c
640− SVarStyle _ Par -> undefined
641− pushWorker sv
642− return sv
643−
644637-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
645638-- be read back from the SVar using 'fromSVar'.
646639toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
@@ -728,7 +721,7 @@ joinStreamVar2 style m1 m2 = Stream $ \svr stp sng yld ->
728721 (runStream (fromStreamVar sv)) Nothing stp sng yld
729722 where
730723 concurrently ma mb = Stream $ \svr stp sng yld -> do
731− liftIO $ (enqueue (fromJust svr)) mb
724+ liftIO $ enqueue (fromJust svr) mb
732725 (runStream ma) svr stp sng yld
733726
734727{-# INLINE joinStreamVarPar #-}