@@ -58,7 +58,6 @@ module Streamly.Core
58 , SVarStyle (..)
58 , SVarStyle (..)
59 , newEmptySVar
59 , newEmptySVar
60 , newStreamVar1
60 , newStreamVar1
62 , joinStreamVar2
61 , joinStreamVar2
63 , fromStreamVar
62 , fromStreamVar
64 , toStreamVar
63 , toStreamVar
@@ -147,15 +146,23 @@ data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
147-- already enqueued computations get evaluated. See 'joinStreamVar2'.
146-- already enqueued computations get evaluated. See 'joinStreamVar2'.
149data SVar m a =
148data SVar m a =
151+ svarStyle :: SVarStyle
153+ -- Shared output queue (events, length)
150− SVar { outputQueue :: IORef ([ChildEvent a], Int)
154+ , outputQueue :: IORef ([ChildEvent a], Int)
151− , doorBell :: MVar () -- signal the consumer about output
155+ , doorBell :: MVar () -- signal the consumer about output
152− , siren :: MVar () -- hooter for workers to begin work
156+ , siren :: MVar () -- hooter for workers to begin work
158+ -- Shared work queue
153 , enqueue :: Stream m a -> IO ()
159 , enqueue :: Stream m a -> IO ()
160+ , queueEmpty :: m Bool
154 , runqueue :: m ()
161 , runqueue :: m ()
163+ -- Shared, thread tracking
155 , runningThreads :: IORef (Set ThreadId)
164 , runningThreads :: IORef (Set ThreadId)
156− , queueEmpty :: m Bool
157 , activeWorkers :: IORef Int
165 , activeWorkers :: IORef Int
158− , svarStyle :: SVarStyle
161-- Slightly faster version of CAS. Gained some improvement by avoiding the use
168-- Slightly faster version of CAS. Gained some improvement by avoiding the use
@@ -302,8 +309,8 @@ doFork action exHandler =
302-- TBD Each worker can have their own queue and the consumer can empty one
309-- TBD Each worker can have their own queue and the consumer can empty one
303-- queue at a time, that way contention can be reduced.
310-- queue at a time, that way contention can be reduced.
304{-# INLINE send #-}
311{-# INLINE send #-}
305−send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
312+send :: SVar m a -> ChildEvent a -> IO ()
306−send sv msg = liftIO $ do
313+send sv msg = do
307 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
314 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
308 ((msg : es, n + 1), n)
315 ((msg : es, n + 1), n)
309 if (len <= 0) then do
316 if (len <= 0) then do
@@ -340,10 +347,13 @@ send sv msg = liftIO $ do
340 else return ()
347 else return ()
342{-# INLINE sendStop #-}
349{-# INLINE sendStop #-}
343−sendStop :: MonadIO m => SVar m a -> m ()
350+sendStop :: SVar m a -> IO ()
344−sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
351+sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
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
346−-- Note: Left associated compositions can grow this queue to a large size
355+-- expressions. Large left associated compositions can grow this to a
347{-# INLINE enqueueLIFO #-}
357{-# INLINE enqueueLIFO #-}
348enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
358enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
349enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
359enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
@@ -356,10 +366,10 @@ runqueueLIFO sv q = run
357 work <- dequeue
367 work <- dequeue
358 case work of
368 case work of
359− Nothing -> sendStop sv
369+ Nothing -> liftIO $ sendStop sv
360 Just m -> (runStream m) (Just sv) run single yield
370 Just m -> (runStream m) (Just sv) run single yield
362− sendit a = send sv (ChildYield a)
372+ sendit a = liftIO $ send sv (ChildYield a)
363 single a = sendit a >> run
373 single a = sendit a >> run
364 yield a r = sendit a >> (runStream r) (Just sv) run single yield
374 yield a r = sendit a >> (runStream r) (Just sv) run single yield
@@ -379,11 +389,11 @@ runqueueFIFO sv q = run
380 work <- dequeue
390 work <- dequeue
381 case work of
391 case work of
382− Nothing -> sendStop sv
392+ Nothing -> liftIO $ sendStop sv
383 Just m -> (runStream m) (Just sv) run single yield
393 Just m -> (runStream m) (Just sv) run single yield
385 dequeue = liftIO $ tryPopR q
395 dequeue = liftIO $ tryPopR q
386− sendit a = send sv (ChildYield a)
396+ sendit a = liftIO $ send sv (ChildYield a)
387 single a = sendit a >> run
397 single a = sendit a >> run
388 yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run
398 yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run
@@ -393,8 +403,8 @@ runOne sv m = (runStream m) (Just sv) stop single yield
396− stop = sendStop sv
406+ stop = liftIO $ sendStop sv
397− sendit a = send sv (ChildYield a)
407+ sendit a = liftIO $ send sv (ChildYield a)
398 single a = sendit a >> stop
408 single a = sendit a >> stop
399 yield a r = sendit a >> runOne sv r
409 yield a r = sendit a >> runOne sv r
@@ -442,8 +452,8 @@ allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
443{-# NOINLINE handleChildException #-}
453{-# NOINLINE handleChildException #-}
444handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
454handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
445−handleChildException sv e = do
455+handleChildException sv e = liftIO $ do
446− tid <- liftIO myThreadId
456+ tid <- myThreadId
447 send sv (ChildStop tid (Just e))
457 send sv (ChildStop tid (Just e))
449{-# NOINLINE pushWorker #-}
459{-# NOINLINE pushWorker #-}
@@ -475,9 +485,13 @@ sendWorkerWait sv = do
475 -- If there is no output pending to process and there is no worker to be
485 -- If there is no output pending to process and there is no worker to be
476 -- sent then we block, so that we do not keep looping fruitlessly.
486 -- sent then we block, so that we do not keep looping fruitlessly.
488+ -- XXX we should wait in such a way so that we can be immediately
489+ -- interrupted by a doorBell.
478 liftIO $ threadDelay 200
490 liftIO $ threadDelay 200
479 (_, n) <- liftIO $ readIORef (outputQueue sv)
491 (_, n) <- liftIO $ readIORef (outputQueue sv)
480 when (n <= 0) $ do
492 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.
481 done <- queueEmpty sv
495 done <- queueEmpty sv
482 if not done
496 if not done
@@ -620,27 +634,6 @@ newStreamVar1 style m = do
620 pushWorker sv
634 pushWorker sv
621 return sv
635 return sv
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
632− SVarStyle _ FIFO -> do
633− c <- getFifoSVar style
634− (enqueue c) m1 >> (enqueue c) m2
636− SVarStyle _ LIFO -> do
637− c <- getLifoSVar style
638− (enqueue c) m2 >> (enqueue c) m1
640− SVarStyle _ Par -> undefined
644-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
637-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
645-- be read back from the SVar using 'fromSVar'.
638-- be read back from the SVar using 'fromSVar'.
646toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
639toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
@@ -728,7 +721,7 @@ joinStreamVar2 style m1 m2 = Stream $ \svr stp sng yld ->
728 (runStream (fromStreamVar sv)) Nothing stp sng yld
721 (runStream (fromStreamVar sv)) Nothing stp sng yld
730 concurrently ma mb = Stream $ \svr stp sng yld -> do
723 concurrently ma mb = Stream $ \svr stp sng yld -> do
731− liftIO $ (enqueue (fromJust svr)) mb
724+ liftIO $ enqueue (fromJust svr) mb
732 (runStream ma) svr stp sng yld
725 (runStream ma) svr stp sng yld
734{-# INLINE joinStreamVarPar #-}
727{-# INLINE joinStreamVarPar #-}