@@ -65,8 +65,9 @@ module Streamly.Core
68−import Control.Concurrent (ThreadId, myThreadId, threadDelay)
68+import Control.Concurrent (ThreadId, myThreadId,
69+ threadDelay, getNumCapabilities)
69−import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
70+import Control.Concurrent.MVar (MVar, newEmptyMVar,
70 tryPutMVar, takeMVar)
71 tryPutMVar, takeMVar)
71import Control.Exception (SomeException (..), catch, mask)
72import Control.Exception (SomeException (..), catch, mask)
72import Control.Monad (when)
73import Control.Monad (when)
@@ -76,7 +77,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
76import Control.Monad.Trans.Control (MonadBaseControl, control)
77import Control.Monad.Trans.Control (MonadBaseControl, control)
77import Data.Atomics (casIORef, readForCAS, peekTicket
78import Data.Atomics (casIORef, readForCAS, peekTicket
78 ,atomicModifyIORefCAS_
79 ,atomicModifyIORefCAS_
79− ,writeBarrier)
80+ ,writeBarrier,storeLoadBarrier)
80import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
81import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
81 tryPopR, nullQ)
82 tryPopR, nullQ)
82import Data.Functor (void)
83import Data.Functor (void)
@@ -88,7 +89,7 @@ import Data.IORef (IORef, modifyIORef, newIORef,
88 , writeIORef
89 , writeIORef
91−import Data.Maybe (isNothing, fromJust)
92+import Data.Maybe (fromJust)
92import Data.Semigroup (Semigroup(..))
93import Data.Semigroup (Semigroup(..))
93import Data.Set (Set)
94import Data.Set (Set)
94import qualified Data.Set as S
95import qualified Data.Set as S
@@ -104,6 +105,7 @@ import GHC.IO (IO(..))
104-- until they are GCed because the consumer went away.
105-- until they are GCed because the consumer went away.
106#ifdef DIAGNOSTICS
107#ifdef DIAGNOSTICS
108+import Control.Concurrent.MVar (tryTakeMVar)
107import Control.Exception (catches, throwIO, Handler(..),
109import Control.Exception (catches, throwIO, Handler(..),
108 BlockedIndefinitelyOnMVar(..),
110 BlockedIndefinitelyOnMVar(..),
109 BlockedIndefinitelyOnSTM(..))
111 BlockedIndefinitelyOnSTM(..))
@@ -181,7 +183,7 @@ data SVar m a =
184− -- Shared work queue
186+ -- Shared work queue (stream, seqNo)
185 , workQueue :: IORef ([Stream m a], Int)
187 , workQueue :: IORef ([Stream m a], Int)
186 , enqueue :: Stream m a -> IO ()
188 , enqueue :: Stream m a -> IO ()
187 , queueEmpty :: m Bool
189 , queueEmpty :: m Bool
@@ -417,6 +419,9 @@ doFork action exHandler =
417-- TBD Each worker can have their own queue and the consumer can empty one
419-- TBD Each worker can have their own queue and the consumer can empty one
418-- queue at a time, that way contention can be reduced.
420-- queue at a time, that way contention can be reduced.
420-- | This function is used by the producer threads to queue output for the
425-- | This function is used by the producer threads to queue output for the
421-- consumer thread to consume. Returns whether the queue has more space.
426-- consumer thread to consume. Returns whether the queue has more space.
422{-# NOINLINE send #-}
427{-# NOINLINE send #-}
@@ -435,11 +440,13 @@ send sv msg = do
435 -- The important point is that the consumer is guaranteed to receive a
440 -- The important point is that the consumer is guaranteed to receive a
436 -- doorbell if something was added to the queue after it empties it.
441 -- doorbell if something was added to the queue after it empties it.
437 void $ tryPutMVar (doorBell sv) ()
442 void $ tryPutMVar (doorBell sv) ()
438− return (len < 1500)
443+ return (len < maxOutputQLen)
440{-# NOINLINE sendStop #-}
445{-# NOINLINE sendStop #-}
441sendStop :: SVar m a -> IO ()
446sendStop :: SVar m a -> IO ()
448+ liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
442−sendStop sv = myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
449+ myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
444-------------------------------------------------------------------------------
451-------------------------------------------------------------------------------
@@ -452,21 +459,17 @@ sendStop sv = myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
452{-# INLINE enqueueLIFO #-}
459{-# INLINE enqueueLIFO #-}
453enqueueLIFO :: SVar m a -> IORef [Stream m a] -> Stream m a -> IO ()
460enqueueLIFO :: SVar m a -> IORef [Stream m a] -> Stream m a -> IO ()
454enqueueLIFO sv q m = do
461enqueueLIFO sv q m = do
462+ atomicModifyIORefCAS_ q $ \ms -> m : ms
455 w <- readIORef $ waitingForWork sv
464 w <- readIORef $ waitingForWork sv
456− if w
465+ when w $ do
457− then withDoorBell
466+ -- Note: the sequence of operations is important for correctness here.
458− else atomicModifyIORefCAS_ q $ \ms -> m : ms
467+ -- We need to set the flag to false strictly before sending the
459−
468+ -- doorBell, otherwise the doorBell may get processed too early and
460− where
469+ -- then we may set the flag to False to later making the consumer lose
461−
470+ -- the flag, even without receiving a doorBell.
463− v <- atomicModifyIORefCAS q $ \ ms ->
465− [] -> (m : ms, True)
466− _ -> (m : ms, False)
468− when v $ void $ tryPutMVar (doorBell sv) ()
469 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
471 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
472+ void $ tryPutMVar (doorBell sv) ()
471runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
474runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
472runqueueLIFO sv q = run
475runqueueLIFO sv q = run
@@ -503,19 +506,17 @@ runqueueLIFO sv q = run
503{-# INLINE enqueueFIFO #-}
506{-# INLINE enqueueFIFO #-}
504enqueueFIFO :: SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> IO ()
507enqueueFIFO :: SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> IO ()
505enqueueFIFO sv q m = do
508enqueueFIFO sv q m = do
506 w <- readIORef $ waitingForWork sv
511 w <- readIORef $ waitingForWork sv
507− if w
512+ when w $ do
508− then withDoorBell
513+ -- Note: the sequence of operations is important for correctness here.
509− else pushL q m
514+ -- We need to set the flag to false strictly before sending the
510−
515+ -- doorBell, otherwise the doorBell may get processed too early and
511− where
516+ -- then we may set the flag to False to later making the consumer lose
512−
517+ -- the flag, even without receiving a doorBell.
517− when emp $ void $ tryPutMVar (doorBell sv) ()
518 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
518 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
519+ void $ tryPutMVar (doorBell sv) ()
520runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
521runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
521runqueueFIFO sv q = run
522runqueueFIFO sv q = run
@@ -615,14 +616,19 @@ runOne sv m = (runStream m) (Just sv) stop single yield
615{-# INLINE enqueueAhead #-}
616{-# INLINE enqueueAhead #-}
616enqueueAhead :: SVar m a -> IORef ([Stream m a], Int) -> Stream m a -> IO ()
617enqueueAhead :: SVar m a -> IORef ([Stream m a], Int) -> Stream m a -> IO ()
617enqueueAhead sv q m = do
618enqueueAhead sv q m = do
618− w <- readIORef $ waitingForWork sv
619 atomicModifyIORefCAS_ q $ \ case
619 atomicModifyIORefCAS_ q $ \ case
620 ([], n) -> ([m], n + 1) -- increment sequence
620 ([], n) -> ([m], n + 1) -- increment sequence
621 _ -> error "not empty"
621 _ -> error "not empty"
623+ w <- readIORef $ waitingForWork sv
622 when w $ do
624 when w $ do
623− writeBarrier
625+ -- Note: the sequence of operations is important for correctness here.
624− void $ tryPutMVar (doorBell sv) ()
626+ -- We need to set the flag to false strictly before sending the
627+ -- doorBell, otherwise the doorBell may get processed too early and
628+ -- then we may set the flag to False to later making the consumer lose
629+ -- the flag, even without receiving a doorBell.
625 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
630 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
631+ void $ tryPutMVar (doorBell sv) ()
627-- Normally the thread that has the token should never go away. The token gets
633-- Normally the thread that has the token should never go away. The token gets
628-- handed over to another thread, but someone or the other has the token at any
634-- handed over to another thread, but someone or the other has the token at any
@@ -641,6 +647,9 @@ enqueueAhead sv q m = do
641-- priority tasks do not fill up the heap making higher priority tasks block
647-- priority tasks do not fill up the heap making higher priority tasks block
642-- due to full heap. Maybe we can have a weighted space for them in the heap.
648-- due to full heap. Maybe we can have a weighted space for them in the heap.
643-- The weight is inversely proportional to the sequence number.
649-- The weight is inversely proportional to the sequence number.
651+-- XXX review for livelock
644runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
653runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
645runqueueAhead sv q = runHeap
654runqueueAhead sv q = runHeap
@@ -775,12 +784,12 @@ addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
775addThread sv tid =
784addThread sv tid =
776 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
785 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
778−{-
787+-- This is cheaper than modifyThread because we do not have to send a doorBell
788+-- This can make a difference when more workers are being dispatched.
779{-# INLINE delThread #-}
789{-# INLINE delThread #-}
780delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
790delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
781delThread sv tid =
791delThread sv tid =
782 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
792 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
785-- If present then delete else add. This takes care of out of order add and
794-- If present then delete else add. This takes care of out of order add and
786-- delete i.e. a delete arriving before we even added a thread.
795-- delete i.e. a delete arriving before we even added a thread.
@@ -846,118 +855,160 @@ pushWorkerPar sv m = do
847 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
856 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
849−-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
858+{-# INLINE workDone #-}
850−-- Also, if a worker blocks on the output queue we can decide if we want to
859+workDone :: MonadIO m => SVar m a -> m Bool
851−-- block or make it go away entirely, depending on the number of workers and
860+workDone sv = do
852−-- the type of the queue.
861+ heapDone <-
862+ if (svarStyle sv == AheadVar)
864+ (hp, _) <- liftIO $ readIORef (outputHeap sv)
865+ return (H.size hp <= 0)
867+ queueDone <- queueEmpty sv
868+ return $ queueDone && heapDone
870+maxWorkerLimit :: Int
871+maxWorkerLimit = 1500
873+dispatchWorker :: MonadAsync m => SVar m a -> m ()
874+dispatchWorker sv = do
876+ when (not done) $ do
877+ -- Note that the worker count is only decremented during event
878+ -- processing in fromStreamVar and therefore it is safe to read and
879+ -- use it without a lock.
880+ cnt <- liftIO $ readIORef $ activeWorkers sv
881+ -- Note that we may deadlock if the previous workers (tasks in the
882+ -- stream) wait/depend on the future workers (tasks in the stream)
883+ -- executing. In that case we should either configure the maxWorker
884+ -- count to higher or use parallel style instead of ahead or async
886+ when (cnt < maxWorkerLimit) $ pushWorker sv
853−{-# INLINE sendWorkerWait #-}
888+{-# NOINLINE sendWorkerWait #-}
854sendWorkerWait :: MonadAsync m => SVar m a -> m ()
889sendWorkerWait :: MonadAsync m => SVar m a -> m ()
855sendWorkerWait sv = do
890sendWorkerWait sv = do
856− -- When there is no output seen we dispatch more workers to help out if
857− -- there is work pending in the work queue. But we wait a little while
858− -- and check the output again so that we are not too aggressive.
859− -- If there is no output pending to process and there is no worker to be
860− -- sent then we block, so that we do not keep looping fruitlessly.
862 -- Note that we are guaranteed to have at least one outstanding worker when
891 -- Note that we are guaranteed to have at least one outstanding worker when
863− -- we enter this function. So if we sleep we are guaranteed to be woken up.
892+ -- we enter this function. So if we sleep we are guaranteed to be woken up
864− -- Note that the worker count is only decremented during event processing
893+ -- by a doorBell, when the worker exits.
865− -- in fromStreamVar and therefore it is safe to read and use it without a
894+
866− -- lock.
895+ -- XXX we need a better way to handle this than hardcoded delays. The
867−
896+ -- delays may be different for different systems.
868− -- XXX we should wait in such a way so that we can be immediately
897+ ncpu <- liftIO $ getNumCapabilities
869− -- interrupted by a doorBell. We can sleep on the doorbell and use a
898+ if ncpu <= 1
870− -- monitor thread to send us a doorbell.
899+ then
900+ if (svarStyle sv == AheadVar)
901+ then liftIO $ threadDelay 100
902+ else liftIO $ threadDelay 25
904+ if (svarStyle sv == AheadVar)
871− liftIO $ threadDelay 200
905+ then liftIO $ threadDelay 100
906+ else liftIO $ threadDelay 10
872 (_, n) <- liftIO $ readIORef (outputQueue sv)
908 (_, n) <- liftIO $ readIORef (outputQueue sv)
873 when (n <= 0) $ do
909 when (n <= 0) $ do
874 -- The queue may be empty temporarily if the worker has dequeued the
910 -- The queue may be empty temporarily if the worker has dequeued the
875 -- work item but has not enqueued the remaining part yet. For the same
911 -- work item but has not enqueued the remaining part yet. For the same
876 -- reason, a worker may come back if it tries to dequeue and finds the
912 -- reason, a worker may come back if it tries to dequeue and finds the
877 -- queue empty, even though the whole work has not finished yet.
913 -- queue empty, even though the whole work has not finished yet.
878− done <- queueEmpty sv
914+
915+ -- If we find that the queue is empty, but it may be empty
916+ -- temporarily, when we checked it. If that's the case we might
917+ -- sleep indefinitely unless the active workers produce some
918+ -- output. We may deadlock specially if the otuput from the active
919+ -- workers depends on the future workers that we may never send.
920+ -- So in case the queue was temporarily empty set a flag to inform
921+ -- the enqueue to send us a doorbell.
923+ -- Note that this is just a best effort mechanism to avoid a
924+ -- deadlock. Deadlocks may still happen if for some weird reason
925+ -- the consuming computation shares an MVar or some other resource
926+ -- with the producing computation and gets blocked on that resource
927+ -- and therefore cannot do any pushworker to add more threads to
928+ -- the producer. In such cases the programmer should use a parallel
929+ -- style so that all the producers are scheduled immediately and
930+ -- unconditionally. We can also use a separate monitor thread to
931+ -- push workers instead of pushing them from the consumer, but then
932+ -- we are no longer using pull based concurrency rate adaptation.
934+ -- XXX update this in the tutorial.
936+ -- register for the doorBell before we check the queue so that if we
937+ -- sleep because the queue was empty we are guaranteed to get a
938+ -- doorbell on the next enqueue.
940+ liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True
941+ liftIO $ storeLoadBarrier
944+ -- XXX test for the case when we miss sending a worker when the worker
945+ -- count is more than 1500.
947+ -- XXX Assert here that if the heap is not empty then there is at
948+ -- least one outstanding worker. Otherwise we could be sleeping
879− if not done
952+ if done
954+ liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
955+ $ takeMVar (doorBell sv)
881− cnt <- liftIO $ readIORef $ activeWorkers sv
956+ (_, len) <- liftIO $ readIORef (outputQueue sv)
882− -- Note that we may deadlock if the previous workers (tasks in the
957+ when (len <= 0) $ sendWorkerWait sv
883− -- stream) wait/depend on the future workers (tasks in the stream)
884− -- executing. In that case we should either configure the maxWorker
885− -- count to higher or use parallel style instead of ahead or async
891− else liftIO $ withDBGMVar sv "sendWorkerWait: wait for workers"
892− $ takeMVar (doorBell sv)
894− -- XXX Assert here that if the heap is not empty then there is at
895− -- least one outstanding worker. Otherwise we could be sleeping
898− -- We found that the queue was empty, but it may be empty
899− -- temporarily, when we checked it. If that's the case we might
900− -- sleep indefinitely unless the active workers produce some
901− -- output. We may deadlock specially if the otuput from the other
902− -- workers depends on the future workers that we may never send.
903− -- So in case the queue was temporarily empty set a flag to inform
904− -- the enqueue to send us a doorbell.
906− liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True
907− liftIO $ writeBarrier
909− -- check again, this time we have set the waitingForWork flag so we
910− -- are guaranteed to get a doorbell in case the status changed
911− -- before we could sleep.
913− -- Note that this is just a best effort mechanism to avoid a
914− -- deadlock. Deadlocks may still happen if for some weird reason
915− -- the consuming computation shares an MVar or some other resource
916− -- with the producing computation and gets blocked on that resource
917− -- and therefore cannot do any pushworker to add more threads to
918− -- the producer. In such cases the programmer should use a parallel
919− -- style so that all the producers are scheduled immediately and
920− -- unconditionally. We can also use a separate monitor thread to
921− -- push workers instead of pushing them from the consumer, but then
922− -- we are no longer using pull based concurrency rate adaptation.
924− -- XXX update this in the tutorial.
925− emp <- queueEmpty sv
927− liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
958+ else sendWorkerWait sv
928− $ takeMVar (doorBell sv)
930-- | Pull a stream from an SVar.
960-- | Pull a stream from an SVar.
931{-# NOINLINE fromStreamVar #-}
961{-# NOINLINE fromStreamVar #-}
932fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
962fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
933fromStreamVar sv = Stream $ \_ stp sng yld -> do
963fromStreamVar sv = Stream $ \_ stp sng yld -> do
965+ -- XXX we can set this in SVar
934− if svarStyle sv == ParallelVar
966+ if svarStyle sv == ParallelVar
935− then liftIO $ withDBGMVar sv "fromStreamVar: doorbell"
968+ liftIO $ withDBGMVar sv "fromStreamVar: doorbell"
936− $ takeMVar (doorBell sv)
969+ $ takeMVar (doorBell sv)
938− res <- liftIO $ tryTakeMVar (doorBell sv)
972+ res@(_, len) <- readOutputQ sv
939− when (isNothing res) $ sendWorkerWait sv
973+ -- When there is no output seen we dispatch more workers to help
974+ -- out if there is work pending in the work queue.
978+ -- send a worker proactively, if needed, even before we start
979+ -- processing the output. This may degrade single processor
980+ -- perf but improves multi-processor, because of more
942− (list, len) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
943− oqLen <- liftIO $ readIORef (maxOutQSize sv)
944− when (len > oqLen) $ liftIO $ writeIORef (maxOutQSize sv) len
946− (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
948 -- Reversing the output is important to guarantee that we process the
985 -- Reversing the output is important to guarantee that we process the
949 -- outputs in the same order as they were generated by the constituent
986 -- outputs in the same order as they were generated by the constituent
950 -- streams.
987 -- streams.
951− (runStream $ processEvents (reverse list)) Nothing stp sng yld
988+ runStream (processEvents $ reverse list) Nothing stp sng yld
955− handleException e tid = do
992+ {-# INLINE readOutputQ #-}
956− liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
993+ readOutputQ svr = liftIO $ do
957− modifyThread sv tid
994+ (list, len) <- atomicModifyIORefCAS (outputQueue svr) $
958− -- XXX implement kill async exception handling
995+ \x -> (([],0), x)
959− -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
996+#ifdef DIAGNOSTICS
960− throwM e
997+ oqLen <- readIORef (maxOutQSize svr)
998+ when (len > oqLen) $ writeIORef (maxOutQSize svr) len
1003+ cnt <- liftIO $ readIORef $ activeWorkers sv
1004+ when (cnt <= 0) $ do
1005+ done <- workDone sv
1006+ when (not done) $ pushWorker sv
1008+ {-# INLINE blockingRead #-}
962 allDone stp = do
1013 allDone stp = do
963#ifdef DIAGNOSTICS
1014#ifdef DIAGNOSTICS
@@ -971,52 +1022,37 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
971 {-# INLINE processEvents #-}
1022 {-# INLINE processEvents #-}
972 processEvents [] = Stream $ \_ stp sng yld -> do
1023 processEvents [] = Stream $ \_ stp sng yld -> do
973 workersDone <- allThreadsDone sv
1024 workersDone <- allThreadsDone sv
974− if not workersDone
1025+ done <-
975− then (runStream (fromStreamVar sv)) Nothing stp sng yld
1026+ -- XXX we can set this in SVar
977 if svarStyle sv == ParallelVar
1027 if svarStyle sv == ParallelVar
978− then allDone stp
1028+ then return workersDone
980− -- All the workers may come back if the outputQueue is full.
1030+ -- There may still be work pending even if there are no workers
981− -- So check if there is work pending and kick off a worker to
1031+ -- pending because all the workers may return if the
982− -- finish it.
1032+ -- outputQueue becomes full. In that case send off a worker to
983− -- XXX we should perhaps kickoff even before we start
1033+ -- kickstart the work again.
984− -- processing if the active workers are zero, rather than at
1034+ if workersDone
985− -- the end of processing the outputQueue. But some thread stop
1035+ then do
986− -- events may be in the outputQueue itself, so we may not be
1036+ r <- workDone sv
987− -- able to determine the correct number of workers before we
989− -- XXX It is necessary to do this before the processing to keep
990− -- minimal concurrency always.
991− -- We should have a separate queue for Stop events so that we
992− -- can process them before the outputs and send workers
993− -- promptly when needed.
995− if (svarStyle sv == AheadVar)
997− (hp, _) <- liftIO $ readIORef (outputHeap sv)
998− return (H.size hp <= 0)
1000− queueDone <- queueEmpty sv
1001− if queueDone && heapDone
1004− pushWorker sv
1037+ when (not r) $ pushWorker sv
1005− (runStream (fromStreamVar sv)) Nothing stp sng yld
1043+ else runStream (fromStreamVar sv) Nothing stp sng yld
1007 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
1045 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
1008− let continue = (runStream (processEvents es)) Nothing stp sng yld
1009− yield a = yld a (processEvents es)
1046+ let rest = processEvents es
1011 case ev of
1047 case ev of
1012− ChildYield a -> yield a
1048+ ChildYield a -> yld a rest
1013− ChildStop tid e ->
1049+ ChildStop tid e -> do
1050+ if svarStyle sv == ParallelVar
1051+ then modifyThread sv tid
1052+ else delThread sv tid
1014 case e of
1053 case e of
1015− Nothing -> do
1054+ Nothing -> runStream rest Nothing stp sng yld
1016− let active = activeWorkers sv
1017− liftIO $ atomicModifyIORefCAS_ active $ \n -> n - 1
1018− modifyThread sv tid >> continue
1019− Just ex -> handleException ex tid
1055+ Just ex -> throwM ex
1021getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1057getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1022getFifoSVar ctype = do
1058getFifoSVar ctype = do