@@ -65,8 +65,9 @@ module Streamly.Core
6565 )
6666where
6767
68−import Control.Concurrent (ThreadId, myThreadId, threadDelay)
69−import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
68+import Control.Concurrent (ThreadId, myThreadId,
69+ threadDelay, getNumCapabilities)
70+import Control.Concurrent.MVar (MVar, newEmptyMVar,
7071 tryPutMVar, takeMVar)
7172import Control.Exception (SomeException (..), catch, mask)
7273import Control.Monad (when)
@@ -76,7 +77,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
7677import Control.Monad.Trans.Control (MonadBaseControl, control)
7778import Data.Atomics (casIORef, readForCAS, peekTicket
7879 ,atomicModifyIORefCAS_
79− ,writeBarrier)
80+ ,writeBarrier,storeLoadBarrier)
8081import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
8182 tryPopR, nullQ)
8283import Data.Functor (void)
@@ -88,7 +89,7 @@ import Data.IORef (IORef, modifyIORef, newIORef,
8889 , writeIORef
8990#endif
9091 )
91−import Data.Maybe (isNothing, fromJust)
92+import Data.Maybe (fromJust)
9293import Data.Semigroup (Semigroup(..))
9394import Data.Set (Set)
9495import qualified Data.Set as S
@@ -104,6 +105,7 @@ import GHC.IO (IO(..))
104105-- until they are GCed because the consumer went away.
105106
106107#ifdef DIAGNOSTICS
108+import Control.Concurrent.MVar (tryTakeMVar)
107109import Control.Exception (catches, throwIO, Handler(..),
108110 BlockedIndefinitelyOnMVar(..),
109111 BlockedIndefinitelyOnSTM(..))
@@ -181,7 +183,7 @@ data SVar m a =
181183 , Int
182184 )
183185
184− -- Shared work queue
186+ -- Shared work queue (stream, seqNo)
185187 , workQueue :: IORef ([Stream m a], Int)
186188 , enqueue :: Stream m a -> IO ()
187189 , queueEmpty :: m Bool
@@ -417,6 +419,9 @@ doFork action exHandler =
417419-- TBD Each worker can have their own queue and the consumer can empty one
418420-- queue at a time, that way contention can be reduced.
419421
422+maxOutputQLen :: Int
423+maxOutputQLen = 1500
424+
420425-- | This function is used by the producer threads to queue output for the
421426-- consumer thread to consume. Returns whether the queue has more space.
422427{-# NOINLINE send #-}
@@ -435,11 +440,13 @@ send sv msg = do
435440 -- The important point is that the consumer is guaranteed to receive a
436441 -- doorbell if something was added to the queue after it empties it.
437442 void $ tryPutMVar (doorBell sv) ()
438− return (len < 1500)
443+ return (len < maxOutputQLen)
439444
440445{-# NOINLINE sendStop #-}
441446sendStop :: SVar m a -> IO ()
442−sendStop sv = myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
447+sendStop sv = do
448+ liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
449+ myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
443450
444451-------------------------------------------------------------------------------
445452-- Async
@@ -452,21 +459,17 @@ sendStop sv = myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
452459{-# INLINE enqueueLIFO #-}
453460enqueueLIFO :: SVar m a -> IORef [Stream m a] -> Stream m a -> IO ()
454461enqueueLIFO sv q m = do
462+ atomicModifyIORefCAS_ q $ \ms -> m : ms
463+ storeLoadBarrier
455464 w <- readIORef $ waitingForWork sv
456− if w
457− then withDoorBell
458− else atomicModifyIORefCAS_ q $ \ms -> m : ms
459−
460− where
461−
462− withDoorBell = do
463− v <- atomicModifyIORefCAS q $ \ ms ->
464− case ms of
465− [] -> (m : ms, True)
466− _ -> (m : ms, False)
467− writeBarrier
468− when v $ void $ tryPutMVar (doorBell sv) ()
465+ when w $ do
466+ -- Note: the sequence of operations is important for correctness here.
467+ -- We need to set the flag to false strictly before sending the
468+ -- doorBell, otherwise the doorBell may get processed too early and
469+ -- then we may set the flag to False to later making the consumer lose
470+ -- the flag, even without receiving a doorBell.
469471 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
472+ void $ tryPutMVar (doorBell sv) ()
470473
471474runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
472475runqueueLIFO sv q = run
@@ -503,19 +506,17 @@ runqueueLIFO sv q = run
503506{-# INLINE enqueueFIFO #-}
504507enqueueFIFO :: SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> IO ()
505508enqueueFIFO sv q m = do
509+ pushL q m
510+ storeLoadBarrier
506511 w <- readIORef $ waitingForWork sv
507− if w
508− then withDoorBell
509− else pushL q m
510−
511− where
512−
513− withDoorBell = do
514− emp <- nullQ q
515− pushL q m
516− writeBarrier
517− when emp $ void $ tryPutMVar (doorBell sv) ()
512+ when w $ do
513+ -- Note: the sequence of operations is important for correctness here.
514+ -- We need to set the flag to false strictly before sending the
515+ -- doorBell, otherwise the doorBell may get processed too early and
516+ -- then we may set the flag to False to later making the consumer lose
517+ -- the flag, even without receiving a doorBell.
518518 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
519+ void $ tryPutMVar (doorBell sv) ()
519520
520521runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
521522runqueueFIFO sv q = run
@@ -615,14 +616,19 @@ runOne sv m = (runStream m) (Just sv) stop single yield
615616{-# INLINE enqueueAhead #-}
616617enqueueAhead :: SVar m a -> IORef ([Stream m a], Int) -> Stream m a -> IO ()
617618enqueueAhead sv q m = do
618− w <- readIORef $ waitingForWork sv
619619 atomicModifyIORefCAS_ q $ \ case
620620 ([], n) -> ([m], n + 1) -- increment sequence
621621 _ -> error "not empty"
622+ storeLoadBarrier
623+ w <- readIORef $ waitingForWork sv
622624 when w $ do
623− writeBarrier
624− void $ tryPutMVar (doorBell sv) ()
625+ -- Note: the sequence of operations is important for correctness here.
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.
625630 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
631+ void $ tryPutMVar (doorBell sv) ()
626632
627633-- Normally the thread that has the token should never go away. The token gets
628634-- handed over to another thread, but someone or the other has the token at any
@@ -641,6 +647,9 @@ enqueueAhead sv q m = do
641647-- priority tasks do not fill up the heap making higher priority tasks block
642648-- due to full heap. Maybe we can have a weighted space for them in the heap.
643649-- The weight is inversely proportional to the sequence number.
650+--
651+-- XXX review for livelock
652+--
644653runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
645654runqueueAhead sv q = runHeap
646655
@@ -775,12 +784,12 @@ addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
775784addThread sv tid =
776785 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
777786
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.
779789{-# INLINE delThread #-}
780790delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
781791delThread sv tid =
782792 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
783−-}
784793
785794-- If present then delete else add. This takes care of out of order add and
786795-- delete i.e. a delete arriving before we even added a thread.
@@ -846,118 +855,160 @@ pushWorkerPar sv m = do
846855#endif
847856 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
848857
849−-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
850−-- Also, if a worker blocks on the output queue we can decide if we want to
851−-- block or make it go away entirely, depending on the number of workers and
852−-- the type of the queue.
853−{-# INLINE sendWorkerWait #-}
858+{-# INLINE workDone #-}
859+workDone :: MonadIO m => SVar m a -> m Bool
860+workDone sv = do
861+ heapDone <-
862+ if (svarStyle sv == AheadVar)
863+ then do
864+ (hp, _) <- liftIO $ readIORef (outputHeap sv)
865+ return (H.size hp <= 0)
866+ else return True
867+ queueDone <- queueEmpty sv
868+ return $ queueDone && heapDone
869+
870+maxWorkerLimit :: Int
871+maxWorkerLimit = 1500
872+
873+dispatchWorker :: MonadAsync m => SVar m a -> m ()
874+dispatchWorker sv = do
875+ done <- workDone sv
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
885+ -- style.
886+ when (cnt < maxWorkerLimit) $ pushWorker sv
887+
888+{-# NOINLINE sendWorkerWait #-}
854889sendWorkerWait :: MonadAsync m => SVar m a -> m ()
855890sendWorkerWait 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.
861−
862891 -- 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.
864− -- Note that the worker count is only decremented during event processing
865− -- in fromStreamVar and therefore it is safe to read and use it without a
866− -- lock.
867−
868− -- XXX we should wait in such a way so that we can be immediately
869− -- interrupted by a doorBell. We can sleep on the doorbell and use a
870− -- monitor thread to send us a doorbell.
871− liftIO $ threadDelay 200
892+ -- we enter this function. So if we sleep we are guaranteed to be woken up
893+ -- by a doorBell, when the worker exits.
894+
895+ -- XXX we need a better way to handle this than hardcoded delays. The
896+ -- delays may be different for different systems.
897+ ncpu <- liftIO $ getNumCapabilities
898+ if ncpu <= 1
899+ then
900+ if (svarStyle sv == AheadVar)
901+ then liftIO $ threadDelay 100
902+ else liftIO $ threadDelay 25
903+ else
904+ if (svarStyle sv == AheadVar)
905+ then liftIO $ threadDelay 100
906+ else liftIO $ threadDelay 10
907+
872908 (_, n) <- liftIO $ readIORef (outputQueue sv)
873909 when (n <= 0) $ do
874910 -- The queue may be empty temporarily if the worker has dequeued the
875911 -- work item but has not enqueued the remaining part yet. For the same
876912 -- reason, a worker may come back if it tries to dequeue and finds the
877913 -- queue empty, even though the whole work has not finished yet.
878− done <- queueEmpty sv
879− if not done
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.
922+
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.
933+ --
934+ -- XXX update this in the tutorial.
935+
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.
939+
940+ liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True
941+ liftIO $ storeLoadBarrier
942+ dispatchWorker sv
943+
944+ -- XXX test for the case when we miss sending a worker when the worker
945+ -- count is more than 1500.
946+ --
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
949+ -- forever.
950+
951+ done <- workDone sv
952+ if done
880953 then do
881− cnt <- liftIO $ readIORef $ activeWorkers sv
882− -- Note that we may deadlock if the previous workers (tasks in the
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
886− -- style.
887− if (cnt < 1500)
888− then do
889− pushWorker sv
890− sendWorkerWait sv
891− else liftIO $ withDBGMVar sv "sendWorkerWait: wait for workers"
892− $ takeMVar (doorBell sv)
893− else do
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
896− -- forever.
897− --
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.
905−
906− liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True
907− liftIO $ writeBarrier
908−
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.
912− --
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.
923− --
924− -- XXX update this in the tutorial.
925− emp <- queueEmpty sv
926− when emp $
927− liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
928− $ takeMVar (doorBell sv)
954+ liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
955+ $ takeMVar (doorBell sv)
956+ (_, len) <- liftIO $ readIORef (outputQueue sv)
957+ when (len <= 0) $ sendWorkerWait sv
958+ else sendWorkerWait sv
929959
930960-- | Pull a stream from an SVar.
931961{-# NOINLINE fromStreamVar #-}
932962fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
933963fromStreamVar sv = Stream $ \_ stp sng yld -> do
934− if svarStyle sv == ParallelVar
935− then liftIO $ withDBGMVar sv "fromStreamVar: doorbell"
936− $ takeMVar (doorBell sv)
937− else do
938− res <- liftIO $ tryTakeMVar (doorBell sv)
939− when (isNothing res) $ sendWorkerWait sv
964+ (list, _) <-
965+ -- XXX we can set this in SVar
966+ if svarStyle sv == ParallelVar
967+ then do
968+ liftIO $ withDBGMVar sv "fromStreamVar: doorbell"
969+ $ takeMVar (doorBell sv)
970+ readOutputQ sv
971+ else do
972+ res@(_, len) <- readOutputQ 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.
975+ if len <= 0
976+ then blockingRead
977+ else do
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
981+ -- parallelism
982+ sendWorker
983+ return res
940984
941−#ifdef DIAGNOSTICS
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
945−#else
946− (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
947−#endif
948985 -- Reversing the output is important to guarantee that we process the
949986 -- outputs in the same order as they were generated by the constituent
950987 -- streams.
951− (runStream $ processEvents (reverse list)) Nothing stp sng yld
988+ runStream (processEvents $ reverse list) Nothing stp sng yld
952989
953990 where
954991
955− handleException e tid = do
956− liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
957− modifyThread sv tid
958− -- XXX implement kill async exception handling
959− -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
960− throwM e
992+ {-# INLINE readOutputQ #-}
993+ readOutputQ svr = liftIO $ do
994+ (list, len) <- atomicModifyIORefCAS (outputQueue svr) $
995+ \x -> (([],0), x)
996+#ifdef DIAGNOSTICS
997+ oqLen <- readIORef (maxOutQSize svr)
998+ when (len > oqLen) $ writeIORef (maxOutQSize svr) len
999+#endif
1000+ return (list, len)
1001+
1002+ sendWorker = do
1003+ cnt <- liftIO $ readIORef $ activeWorkers sv
1004+ when (cnt <= 0) $ do
1005+ done <- workDone sv
1006+ when (not done) $ pushWorker sv
1007+
1008+ {-# INLINE blockingRead #-}
1009+ blockingRead = do
1010+ sendWorkerWait sv
1011+ readOutputQ sv
9611012
9621013 allDone stp = do
9631014#ifdef DIAGNOSTICS
@@ -971,52 +1022,37 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
9711022 {-# INLINE processEvents #-}
9721023 processEvents [] = Stream $ \_ stp sng yld -> do
9731024 workersDone <- allThreadsDone sv
974− if not workersDone
975− then (runStream (fromStreamVar sv)) Nothing stp sng yld
976− else
1025+ done <-
1026+ -- XXX we can set this in SVar
9771027 if svarStyle sv == ParallelVar
978− then allDone stp
979− else do
980− -- All the workers may come back if the outputQueue is full.
981− -- So check if there is work pending and kick off a worker to
982− -- finish it.
983− -- XXX we should perhaps kickoff even before we start
984− -- processing if the active workers are zero, rather than at
985− -- the end of processing the outputQueue. But some thread stop
986− -- events may be in the outputQueue itself, so we may not be
987− -- able to determine the correct number of workers before we
988− -- process it.
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.
994− heapDone <-
995− if (svarStyle sv == AheadVar)
996− then do
997− (hp, _) <- liftIO $ readIORef (outputHeap sv)
998− return (H.size hp <= 0)
999− else return True
1000− queueDone <- queueEmpty sv
1001− if queueDone && heapDone
1002− then allDone stp
1003− else do
1004− pushWorker sv
1005− (runStream (fromStreamVar sv)) Nothing stp sng yld
1028+ then return workersDone
1029+ else
1030+ -- There may still be work pending even if there are no workers
1031+ -- pending because all the workers may return if the
1032+ -- outputQueue becomes full. In that case send off a worker to
1033+ -- kickstart the work again.
1034+ if workersDone
1035+ then do
1036+ r <- workDone sv
1037+ when (not r) $ pushWorker sv
1038+ return r
1039+ else return False
1040+
1041+ if done
1042+ then allDone stp
1043+ else runStream (fromStreamVar sv) Nothing stp sng yld
10061044
10071045 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)
1010−
1046+ let rest = processEvents es
10111047 case ev of
1012− ChildYield a -> yield a
1013− ChildStop tid e ->
1048+ ChildYield a -> yld a rest
1049+ ChildStop tid e -> do
1050+ if svarStyle sv == ParallelVar
1051+ then modifyThread sv tid
1052+ else delThread sv tid
10141053 case e of
1015− Nothing -> do
1016− let active = activeWorkers sv
1017− liftIO $ atomicModifyIORefCAS_ active $ \n -> n - 1
1018− modifyThread sv tid >> continue
1019− Just ex -> handleException ex tid
1054+ Nothing -> runStream rest Nothing stp sng yld
1055+ Just ex -> throwM ex
10201056
10211057getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
10221058getFifoSVar ctype = do