2{-# LANGUAGE KindSignatures #-}
3{-# LANGUAGE ConstraintKinds #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE FlexibleInstances #-}
6{-# LANGUAGE LambdaCase #-}
7{-# LANGUAGE MagicHash #-}
8{-# LANGUAGE MultiParamTypeClasses #-}
9{-# LANGUAGE UnboxedTuples #-}
12-- Module : Streamly.SVar
13-- Copyright : (c) 2017 Harendra Kumar
16-- Maintainer : harendra.kumar@gmail.com
17-- Stability : experimental
37 , atomicModifyIORefCAS
60import Control.Concurrent
61 (ThreadId, myThreadId, threadDelay, getNumCapabilities)
62import Control.Concurrent.MVar
63 (MVar, newEmptyMVar, tryPutMVar, takeMVar)
64import Control.Exception (SomeException(..), catch, mask)
65import Control.Monad (when)
66import Control.Monad.Catch (MonadThrow)
67import Control.Monad.IO.Class (MonadIO(..))
68import Control.Monad.Trans.Control (MonadBaseControl, control)
70 (casIORef, readForCAS, peekTicket, atomicModifyIORefCAS_,
71 writeBarrier, storeLoadBarrier)
72import Data.Concurrent.Queue.MichaelScott
73 (LinkedQueue, pushL, tryPopR)
74import Data.Functor (void)
75import Data.Heap (Heap, Entry(..))
77 (IORef, modifyIORef, newIORef, readIORef, atomicModifyIORef)
78import Data.Maybe (fromJust)
80import GHC.Conc (ThreadId(..))
84import qualified Data.Heap as H
85import qualified Data.Set as S
87-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
88-- can keep it on in production to debug problems quickly if and when they
89-- happen, but it may result in unexpected output when threads are left hanging
90-- until they are GCed because the consumer went away.
93import Control.Concurrent.MVar (tryTakeMVar)
94import Control.Exception
95 (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
96 BlockedIndefinitelyOnSTM(..))
97import Data.IORef (writeIORef)
98import System.IO (hPutStrLn, stderr)
101------------------------------------------------------------------------------
102-- Parent child thread communication type
103------------------------------------------------------------------------------
105-- | Events that a child thread may send to a parent thread.
108 | ChildStop ThreadId (Maybe SomeException)
110-- | Sorting out-of-turn outputs in a heap for Ahead style streams
111data AheadHeapEntry (t :: (* -> *) -> * -> *) m a =
113 | AheadEntryStream (t m a)
115------------------------------------------------------------------------------
116-- State threaded around the monad for thread management
117------------------------------------------------------------------------------
119-- | Identify the type of the SVar. Two computations using the same style can
120-- be scheduled on the same SVar.
122 AsyncVar -- depth first concurrent
123 | WAsyncVar -- breadth first concurrent
124 | ParallelVar -- all parallel
125 | AheadVar -- Concurrent look ahead
128-- | An SVar or a Stream Var is a conduit to the output from multiple streams
129-- running concurrently and asynchronously. An SVar can be thought of as an
130-- asynchronous IO handle. We can write any number of streams to an SVar in a
131-- non-blocking manner and then read them back at any time at any pace. The
132-- SVar would run the streams asynchronously and accumulate results. An SVar
133-- may not really execute the stream completely and accumulate all the results.
134-- However, it ensures that the reader can read the results at whatever paces
135-- it wants to read. The SVar monitors and adapts to the consumer's pace.
137-- An SVar is a mini scheduler, it has an associated workLoop that holds the
138-- stream tasks to be picked and run by a pool of worker threads. It has an
139-- associated output queue where the output stream elements are placed by the
140-- worker threads. A outputDoorBell is used by the worker threads to intimate the
141-- consumer thread about availability of new results in the output queue. More
142-- workers are added to the SVar by 'fromStreamVar' on demand if the output
143-- produced is not keeping pace with the consumer. On bounded SVars, workers
144-- block on the output queue to provide throttling of the producer when the
145-- consumer is not pulling fast enough. The number of workers may even get
146-- reduced depending on the consuming pace.
148-- New work is enqueued either at the time of creation of the SVar or as a
149-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
150-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
152-- XXX can we use forall t m.
156 svarStyle :: SVarStyle
158 -- Shared output queue (events, length)
159 , outputQueue :: IORef ([ChildEvent a], Int)
160 , maxYieldLimit :: Maybe (IORef Int)
161 , outputDoorBell :: MVar () -- signal the consumer about output
162 , readOutputQ :: m [ChildEvent a]
163 , postProcess :: m Bool
165 -- Used only by bounded SVar types
166 , enqueue :: t m a -> IO ()
167 , isWorkDone :: IO Bool
168 , needDoorBell :: IORef Bool
171 -- Shared, thread tracking
172 , workerThreads :: IORef (Set ThreadId)
173 , workerCount :: IORef Int
174 , accountThread :: ThreadId -> m ()
176 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a))
179 -- Shared work queue (stream, seqNo)
180 , aheadWorkQueue :: IORef ([t m a], Int)
181 , totalDispatches :: IORef Int
182 , maxWorkers :: IORef Int
183 , maxOutQSize :: IORef Int
184 , maxHeapSize :: IORef Int
185 , maxWorkQSize :: IORef Int
189data State t m a = State
190 { streamVar :: Maybe (SVar t m a)
191 , yieldLimit :: Maybe Int
196defaultMaxThreads, defaultMaxBuffer :: Int
197defaultMaxThreads = 1500
198defaultMaxBuffer = 1500
200defState :: State t m a
202 { streamVar = Nothing
203 , yieldLimit = Nothing
204 , threadsHigh = defaultMaxThreads
205 , bufferHigh = defaultMaxBuffer
208-- XXX if perf gets affected we can have all the Nothing params in a single
209-- structure so that we reset is fast. We can also use rewrite rules such that
210-- reset occurs only in concurrent streams to reduce the impact on serial
212-- We can optimize this so that we clear it only if it is a Just value, it
213-- results in slightly better perf for zip/zipM but the performance of scan
214-- worsens a lot, it does not fuse.
215rstState :: State t m a -> State t m b
217 { streamVar = Nothing
218 , yieldLimit = Nothing
222{-# NOINLINE dumpSVar #-}
223dumpSVar :: SVar t m a -> IO String
226 (oqList, oqLen) <- readIORef $ outputQueue sv
227 db <- tryTakeMVar $ outputDoorBell sv
229 if svarStyle sv == AheadVar
231 (oheap, oheapSeq) <- readIORef $ outputHeap sv
232 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
233 maxHp <- readIORef $ maxHeapSize sv
235 [ "heap length = " ++ show (H.size oheap)
236 , "heap seqeunce = " ++ show oheapSeq
237 , "work queue length = " ++ show (length wq)
238 , "work queue sequence = " ++ show wqSeq
239 , "heap max size = " ++ show maxHp
243 waiting <- readIORef $ needDoorBell sv
244 rthread <- readIORef $ workerThreads sv
245 workers <- readIORef $ workerCount sv
246 maxWrk <- readIORef $ maxWorkers sv
247 dispatches <- readIORef $ totalDispatches sv
248 maxOq <- readIORef $ maxOutQSize sv
251 [ "tid = " ++ show tid
252 , "style = " ++ show (svarStyle sv)
253 , "outputQueue length computed = " ++ show (length oqList)
254 , "outputQueue length maintained = " ++ show oqLen
255 , "output outputDoorBell = " ++ show db
256 , "total dispatches = " ++ show dispatches
257 , "max workers = " ++ show maxWrk
258 , "max outQSize = " ++ show maxOq
260 ++ aheadDump ++ unlines
261 [ "needDoorBell = " ++ show waiting
262 , "running threads = " ++ show rthread
263 , "running thread count = " ++ show workers
266{-# NOINLINE mvarExcHandler #-}
267mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
268mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
269 svInfo <- dumpSVar sv
270 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo
273{-# NOINLINE stmExcHandler #-}
274stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
275stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
276 svInfo <- dumpSVar sv
277 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo
280withDBGMVar :: SVar t m a -> String -> IO () -> IO ()
281withDBGMVar sv label action =
282 action `catches` [ Handler (mvarExcHandler sv label)
283 , Handler (stmExcHandler sv label)
286withDBGMVar :: SVar t m a -> String -> IO () -> IO ()
287withDBGMVar _ _ action = action
290-- Slightly faster version of CAS. Gained some improvement by avoiding the use
291-- of "evaluate" because we know we do not have exceptions in fn.
292{-# INLINE atomicModifyIORefCAS #-}
293atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
294atomicModifyIORefCAS ref fn = do
295 tkt <- readForCAS ref
301 loop _ 0 = atomicModifyIORef ref fn
303 let (new, result) = fn $ peekTicket old
304 (success, tkt) <- casIORef ref old new
307 else loop tkt (tries - 1)
309------------------------------------------------------------------------------
310-- Spawning threads and collecting result in streamed fashion
311------------------------------------------------------------------------------
313-- | A monad that can perform concurrent or parallel IO operations. Streams
314-- that can be composed concurrently require the underlying monad to be
318type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
320-- Stolen from the async package. The perf improvement is modest, 2% on a
321-- thread heavy benchmark (parallel composition using noop computations).
322-- A version of forkIO that does not include the outer exception
323-- handler: saves a bit of time when we will be installing our own
325{-# INLINE rawForkIO #-}
326rawForkIO :: IO () -> IO ThreadId
327rawForkIO action = IO $ \ s ->
328 case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
331doFork :: MonadBaseControl IO m
333 -> (SomeException -> IO ())
335doFork action exHandler =
336 control $ \runInIO ->
337 mask $ \restore -> do
338 tid <- rawForkIO $ catch (restore $ void $ runInIO action)
342-- XXX exception safety of all atomic/MVar operations
344-- TBD Each worker can have their own queue and the consumer can empty one
345-- queue at a time, that way contention can be reduced.
347-- | This function is used by the producer threads to queue output for the
348-- consumer thread to consume. Returns whether the queue has more space.
349send :: Int -> SVar t m a -> ChildEvent a -> IO Bool
350send maxOutputQLen sv msg = do
351 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
352 ((msg : es, n + 1), n)
354 -- The wake up must happen only after the store has finished otherwise
355 -- we can have lost wakeup problems.
357 -- Since multiple workers can try this at the same time, it is possible
358 -- that we may put a spurious MVar after the consumer has already seen
359 -- the output. But that's harmless, at worst it may cause the consumer
360 -- to read the queue again and find it empty.
361 -- The important point is that the consumer is guaranteed to receive a
362 -- doorbell if something was added to the queue after it empties it.
363 void $ tryPutMVar (outputDoorBell sv) ()
364 return (len < maxOutputQLen || maxOutputQLen < 0)
366{-# NOINLINE sendYield #-}
367sendYield :: Int -> SVar t m a -> ChildEvent a -> IO Bool
368sendYield maxOutputQLen sv msg = do
369 ylimit <- case maxYieldLimit sv of
370 Nothing -> return True
371 Just ref -> atomicModifyIORefCAS ref $ \x -> (x - 1, x > 1)
372 r <- send maxOutputQLen sv msg
375{-# NOINLINE sendStop #-}
376sendStop :: SVar t m a -> IO ()
378 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
379 myThreadId >>= \tid -> void $ send (-1) sv (ChildStop tid Nothing)
381-------------------------------------------------------------------------------
383-------------------------------------------------------------------------------
385-- Note: For purely right associated expressions this queue should have at most
386-- one element. It grows to more than one when we have left associcated
387-- expressions. Large left associated compositions can grow this to a
389{-# INLINE enqueueLIFO #-}
390enqueueLIFO :: SVar t m a -> IORef [t m a] -> t m a -> IO ()
391enqueueLIFO sv q m = do
392 atomicModifyIORefCAS_ q $ \ms -> m : ms
394 w <- readIORef $ needDoorBell sv
396 -- Note: the sequence of operations is important for correctness here.
397 -- We need to set the flag to false strictly before sending the
398 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
399 -- then we may set the flag to False to later making the consumer lose
400 -- the flag, even without receiving a outputDoorBell.
401 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
402 void $ tryPutMVar (outputDoorBell sv) ()
404{-# INLINE workLoopLIFO #-}
405workLoopLIFO :: MonadIO m
406 => (State t m a -> IORef [t m a] -> t m a -> m () -> m ())
407 -> State t m a -> IORef [t m a] -> m ()
408workLoopLIFO f st q = run
412 sv = fromJust $ streamVar st
416 Nothing -> liftIO $ sendStop sv
417 Just m -> f st q m run
419 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
421 x : xs -> (xs, Just x)
423-------------------------------------------------------------------------------
425-------------------------------------------------------------------------------
427-- XXX we can use the Ahead style sequence/heap mechanism to make the best
428-- effort to always try to finish the streams on the left side of an expression
429-- first as long as possible.
431{-# INLINE enqueueFIFO #-}
432enqueueFIFO :: SVar t m a -> LinkedQueue (t m a) -> t m a -> IO ()
433enqueueFIFO sv q m = do
436 w <- readIORef $ needDoorBell sv
438 -- Note: the sequence of operations is important for correctness here.
439 -- We need to set the flag to false strictly before sending the
440 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
441 -- then we may set the flag to False to later making the consumer lose
442 -- the flag, even without receiving a outputDoorBell.
443 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
444 void $ tryPutMVar (outputDoorBell sv) ()
446{-# INLINE workLoopFIFO #-}
447workLoopFIFO :: MonadIO m
448 => (State t m a -> LinkedQueue (t m a) -> t m a -> m () -> m ())
449 -> State t m a -> LinkedQueue (t m a) -> m ()
450workLoopFIFO f st q = run
454 sv = fromJust $ streamVar st
456 work <- liftIO $ tryPopR q
458 Nothing -> liftIO $ sendStop sv
459 Just m -> f st q m run
461-------------------------------------------------------------------------------
463-------------------------------------------------------------------------------
465-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
466-- but always serve them in the same order as they appear in the stream. To
467-- implement lookahead streams efficiently we assign a sequence number to each
468-- task when the task is picked up for execution. When the task finishes, the
469-- output is tagged with the same sequence number and we rearrange the outputs
470-- in sequence based on that number.
472-- To explain the mechanism imagine that the current task at the head of the
473-- stream has a "token" to yield to the outputQueue. The ownership of the token
474-- is determined by the current sequence number is maintained in outputHeap.
475-- Sequence number is assigned when a task is queued. When a thread dequeues a
476-- task it picks up the sequence number as well and when the output is ready it
477-- uses the sequence number to queue the output to the outputQueue.
479-- The thread with current sequence number sends the output directly to the
480-- outputQueue. Other threads push the output to the outputHeap. When the task
481-- being queued on the heap is a stream of many elements we evaluate only the
482-- first element and keep the rest of the unevaluated computation in the heap.
483-- When such a task gets the "token" for outputQueue it evaluates and directly
484-- yields all the elements to the outputQueue without checking for the
487-- Note that no two outputs in the heap can have the same sequence numbers and
488-- therefore we do not need a stable heap. We have also separated the buffer
489-- for the current task (outputQueue) and the pending tasks (outputHeap) so
490-- that the pending tasks cannot interfere with the current task. Note that for
491-- a single task just the outputQueue is enough and for the case of many
492-- threads just a heap is good enough. However we balance between these two
493-- cases, so that both are efficient.
495-- For bigger streams it may make sense to have separate buffers for each
496-- stream. However, for singleton streams this may become inefficient. However,
497-- if we do not have separate buffers, then the streams that come later in
498-- sequence may hog the buffer, hindering the streams that are ahead. For this
499-- reason we have a single element buffer limitation for the streams being
500-- executed in advance.
502-- This scheme works pretty efficiently with less than 40% extra overhead
503-- compared to the Async streams where we do not have any kind of sequencing of
504-- the outputs. It is especially devised so that we are most efficient when we
505-- have short tasks and need just a single thread. Also when a thread yields
506-- many items it can hold lockfree access to the outputQueue and do it
509-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
510-- that they do not hog the resources and hinder the progress of the threads in
513-- Left associated ahead expressions are expensive. We start a new SVar for
514-- each left associative expression. The queue is used only for right
515-- associated expression, we queue the right expression and execute the left.
516-- Thererefore the queue never has more than on item in it.
517{-# INLINE enqueueAhead #-}
518enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
519enqueueAhead sv q m = do
520 atomicModifyIORefCAS_ q $ \ case
521 ([], n) -> ([m], n + 1) -- increment sequence
522 _ -> error "not empty"
524 w <- readIORef $ needDoorBell sv
526 -- Note: the sequence of operations is important for correctness here.
527 -- We need to set the flag to false strictly before sending the
528 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
529 -- then we may set the flag to False to later making the consumer lose
530 -- the flag, even without receiving a outputDoorBell.
531 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
532 void $ tryPutMVar (outputDoorBell sv) ()
534-- Normally the thread that has the token should never go away. The token gets
535-- handed over to another thread, but someone or the other has the token at any
536-- point of time. But if the task that has the token finds that the outputQueue
537-- is full, in that case it can go away without even handing over the token to
538-- another thread. In that case it sets the nextSequence number in the heap its
539-- own sequence number before going away. To handle this case, any task that
540-- does not have the token tries to dequeue from the heap first before
541-- dequeuing from the work queue. If it finds that the task at the top of the
542-- heap is the one that owns the current sequence number then it grabs the
543-- token and starts with that.
545-- XXX instead of queueing just the head element and the remaining computation
546-- on the heap, evaluate as many as we can and place them on the heap. But we
547-- need to give higher priority to the lower sequence numbers so that lower
548-- priority tasks do not fill up the heap making higher priority tasks block
549-- due to full heap. Maybe we can have a weighted space for them in the heap.
550-- The weight is inversely proportional to the sequence number.
552-- XXX review for livelock
554{-# INLINE queueEmptyAhead #-}
555queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
556queueEmptyAhead q = liftIO $ do
557 (xs, _) <- readIORef q
560{-# INLINE dequeueAhead #-}
561dequeueAhead :: MonadIO m
562 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
563dequeueAhead q = liftIO $ do
564 atomicModifyIORefCAS q $ \case
565 ([], n) -> (([], n), Nothing)
566 (x : [], n) -> (([], n), Just (x, n))
567 _ -> error "more than one item on queue"
569{-# INLINE dequeueFromHeap #-}
571 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
572 -> IO (Maybe (Entry Int (AheadHeapEntry t m a)))
573dequeueFromHeap hpRef = do
574 atomicModifyIORefCAS hpRef $ \hp@(h, snum) -> do
577 Nothing -> (hp, Nothing)
578 Just (ent@(Entry seqNo _ev), hp') ->
580 then ((hp', seqNo), Just ent)
583-------------------------------------------------------------------------------
585-------------------------------------------------------------------------------
587-- XXX To be implemented. Use a linked queue like WAsync and put back the
588-- remaining computation at the back of the queue instead of the heap, and
589-- increment the sequence number.
591-- Thread tracking is needed for two reasons:
593-- 1) Killing threads on exceptions. Threads may not be left to go away by
594-- themselves because they may run for significant times before going away or
595-- worse they may be stuck in IO and never go away.
597-- 2) To know when all threads are done and the stream has ended.
599{-# NOINLINE addThread #-}
600addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
602 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
604-- This is cheaper than modifyThread because we do not have to send a
605-- outputDoorBell This can make a difference when more workers are being
607{-# INLINE delThread #-}
608delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
610 liftIO $ modifyIORef (workerThreads sv) $ (\s -> S.delete tid s)
612-- If present then delete else add. This takes care of out of order add and
613-- delete i.e. a delete arriving before we even added a thread.
614-- This occurs when the forked thread is done even before the 'addThread' right
615-- after the fork gets a chance to run.
616{-# INLINE modifyThread #-}
617modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
618modifyThread sv tid = do
619 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
620 if (S.member tid old)
621 then let new = (S.delete tid old) in (new, new)
622 else let new = (S.insert tid old) in (new, old)
626 void $ tryPutMVar (outputDoorBell sv) ()
629-- | This is safe even if we are adding more threads concurrently because if
630-- a child thread is adding another thread then anyway 'workerThreads' will
632{-# INLINE allThreadsDone #-}
633allThreadsDone :: MonadIO m => SVar t m a -> m Bool
634allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
636{-# NOINLINE handleChildException #-}
637handleChildException :: SVar t m a -> SomeException -> IO ()
638handleChildException sv e = do
640 void $ send (-1) sv (ChildStop tid (Just e))
643recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
644recordMaxWorkers sv = liftIO $ do
645 active <- readIORef (workerCount sv)
646 maxWrk <- readIORef (maxWorkers sv)
647 when (active > maxWrk) $ writeIORef (maxWorkers sv) active
648 modifyIORef (totalDispatches sv) (+1)
651{-# NOINLINE pushWorker #-}
652pushWorker :: MonadAsync m => SVar t m a -> m ()
654 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
658 doFork (workLoop sv) (handleChildException sv) >>= addThread sv
660-- XXX we can push the workerCount modification in accountThread and use the
661-- same pushWorker for Parallel case as well.
663-- | In contrast to pushWorker which always happens only from the consumer
664-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
665-- producer side. So we need to use a thread safe modification of
666-- workerThreads. Alternatively, we can use a CreateThread event to avoid
667-- using a CAS based modification.
668{-# NOINLINE pushWorkerPar #-}
669pushWorkerPar :: MonadAsync m => SVar t m a -> m () -> m ()
670pushWorkerPar sv wloop = do
671 -- We do not use workerCount in case of ParallelVar but still there is no
672 -- harm in maintaining it correctly.
674 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
677 doFork wloop (handleChildException sv) >>= modifyThread sv
679dispatchWorker :: MonadAsync m => Int -> SVar t m a -> m ()
680dispatchWorker maxWorkerLimit sv = do
681 done <- liftIO $ isWorkDone sv
683 -- Note that the worker count is only decremented during event
684 -- processing in fromStreamVar and therefore it is safe to read and
685 -- use it without a lock.
686 cnt <- liftIO $ readIORef $ workerCount sv
687 -- Note that we may deadlock if the previous workers (tasks in the
688 -- stream) wait/depend on the future workers (tasks in the stream)
689 -- executing. In that case we should either configure the maxWorker
690 -- count to higher or use parallel style instead of ahead or async
692 limit <- case maxYieldLimit sv of
693 Nothing -> return maxWorkerLimit
695 lim <- liftIO $ readIORef x
697 if maxWorkerLimit > 0
698 then min maxWorkerLimit lim
700 when (cnt < limit || limit < 0) $ pushWorker sv
702{-# NOINLINE sendWorkerWait #-}
703sendWorkerWait :: MonadAsync m => Int -> SVar t m a -> m ()
704sendWorkerWait maxWorkerLimit sv = do
705 -- Note that we are guaranteed to have at least one outstanding worker when
706 -- we enter this function. So if we sleep we are guaranteed to be woken up
707 -- by a outputDoorBell, when the worker exits.
709 -- XXX we need a better way to handle this than hardcoded delays. The
710 -- delays may be different for different systems.
711 ncpu <- liftIO $ getNumCapabilities
714 if (svarStyle sv == AheadVar)
715 then liftIO $ threadDelay 100
716 else liftIO $ threadDelay 25
718 if (svarStyle sv == AheadVar)
719 then liftIO $ threadDelay 100
720 else liftIO $ threadDelay 10
722 (_, n) <- liftIO $ readIORef (outputQueue sv)
724 -- The queue may be empty temporarily if the worker has dequeued the
725 -- work item but has not enqueued the remaining part yet. For the same
726 -- reason, a worker may come back if it tries to dequeue and finds the
727 -- queue empty, even though the whole work has not finished yet.
729 -- If we find that the queue is empty, but it may be empty
730 -- temporarily, when we checked it. If that's the case we might
731 -- sleep indefinitely unless the active workers produce some
732 -- output. We may deadlock specially if the otuput from the active
733 -- workers depends on the future workers that we may never send.
734 -- So in case the queue was temporarily empty set a flag to inform
735 -- the enqueue to send us a doorbell.
737 -- Note that this is just a best effort mechanism to avoid a
738 -- deadlock. Deadlocks may still happen if for some weird reason
739 -- the consuming computation shares an MVar or some other resource
740 -- with the producing computation and gets blocked on that resource
741 -- and therefore cannot do any pushworker to add more threads to
742 -- the producer. In such cases the programmer should use a parallel
743 -- style so that all the producers are scheduled immediately and
744 -- unconditionally. We can also use a separate monitor thread to
745 -- push workers instead of pushing them from the consumer, but then
746 -- we are no longer using pull based concurrency rate adaptation.
748 -- XXX update this in the tutorial.
750 -- register for the outputDoorBell before we check the queue so that if we
751 -- sleep because the queue was empty we are guaranteed to get a
752 -- doorbell on the next enqueue.
754 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
755 liftIO $ storeLoadBarrier
756 dispatchWorker maxWorkerLimit sv
758 -- XXX test for the case when we miss sending a worker when the worker
759 -- count is more than 1500.
761 -- XXX Assert here that if the heap is not empty then there is at
762 -- least one outstanding worker. Otherwise we could be sleeping
765 done <- liftIO $ isWorkDone sv
768 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
769 $ takeMVar (outputDoorBell sv)
770 (_, len) <- liftIO $ readIORef (outputQueue sv)
771 when (len <= 0) $ sendWorkerWait maxWorkerLimit sv
772 else sendWorkerWait maxWorkerLimit sv
774{-# INLINE readOutputQRaw #-}
775readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
776readOutputQRaw sv = do
777 (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
779 oqLen <- readIORef (maxOutQSize sv)
780 when (len > oqLen) $ writeIORef (maxOutQSize sv) len
784readOutputQBounded :: MonadAsync m => Int -> SVar t m a -> m [ChildEvent a]
785readOutputQBounded n sv = do
786 (list, len) <- liftIO $ readOutputQRaw sv
787 -- When there is no output seen we dispatch more workers to help
788 -- out if there is work pending in the work queue.
792 -- send a worker proactively, if needed, even before we start
793 -- processing the output. This may degrade single processor
794 -- perf but improves multi-processor, because of more
802 cnt <- liftIO $ readIORef $ workerCount sv
804 done <- liftIO $ isWorkDone sv
805 when (not done) $ pushWorker sv
807 {-# INLINE blockingRead #-}
810 liftIO $ (readOutputQRaw sv >>= return . fst)
812postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
813postProcessBounded sv = do
814 workersDone <- allThreadsDone sv
815 -- There may still be work pending even if there are no workers
816 -- pending because all the workers may return if the
817 -- outputQueue becomes full. In that case send off a worker to
818 -- kickstart the work again.
821 r <- liftIO $ isWorkDone sv
822 when (not r) $ pushWorker sv
826getAheadSVar :: MonadAsync m
829 -> IORef ([t m a], Int)
830 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
833getAheadSVar st f = do
834 outQ <- newIORef ([], 0)
835 outH <- newIORef (H.empty, 0)
836 outQMv <- newEmptyMVar
838 wfw <- newIORef False
839 running <- newIORef S.empty
840 q <- newIORef ([], -1)
841 yl <- case yieldLimit st of
842 Nothing -> return Nothing
843 Just x -> Just <$> newIORef x
853 SVar { outputQueue = outQ
855 , outputDoorBell = outQMv
856 , readOutputQ = readOutputQBounded (threadsHigh st) sv
857 , postProcess = postProcessBounded sv
858 , workerThreads = running
859 -- , workLoop = workLoopAhead sv q outH
860 , workLoop = f st{streamVar = Just sv} q outH
861 , enqueue = enqueueAhead sv q
862 , isWorkDone = isWorkDoneAhead q outH
864 , svarStyle = AheadVar
865 , workerCount = active
866 , accountThread = delThread sv
870 , totalDispatches = disp
871 , maxWorkers = maxWrk
872 , maxOutQSize = maxOq
873 , maxHeapSize = maxHs
874 , maxWorkQSize = maxWq
881 {-# INLINE isWorkDoneAhead #-}
882 isWorkDoneAhead q ref = do
884 (hp, _) <- readIORef ref
885 return (H.size hp <= 0)
886 queueDone <- checkEmpty q
887 return $ queueDone && heapDone
890 (xs, _) <- readIORef q
893getParallelSVar :: MonadIO m => IO (SVar t m a)
895 outQ <- newIORef ([], 0)
896 outQMv <- newEmptyMVar
898 running <- newIORef S.empty
907 SVar { outputQueue = outQ
908 , maxYieldLimit = Nothing
909 , outputDoorBell = outQMv
910 , readOutputQ = readOutputQPar sv
911 , postProcess = allThreadsDone sv
912 , workerThreads = running
913 , workLoop = undefined
914 , enqueue = undefined
915 , isWorkDone = undefined
916 , needDoorBell = undefined
917 , svarStyle = ParallelVar
918 , workerCount = active
919 , accountThread = modifyThread sv
921 , aheadWorkQueue = undefined
922 , outputHeap = undefined
923 , totalDispatches = disp
924 , maxWorkers = maxWrk
925 , maxOutQSize = maxOq
926 , maxHeapSize = maxHs
927 , maxWorkQSize = maxWq
934 readOutputQPar sv = liftIO $ do
935 withDBGMVar sv "readOutputQPar: doorbell" $ takeMVar (outputDoorBell sv)
936 readOutputQRaw sv >>= return . fst
938sendWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
940 -- Note: We must have all the work on the queue before sending the
941 -- pushworker, otherwise the pushworker may exit before we even get a
943 liftIO $ enqueue sv m
947{-# INLINABLE newAheadVar #-}
948newAheadVar :: MonadAsync m
952 -> IORef ([t m a], Int)
953 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
956newAheadVar st m wloop = do
957 sv <- liftIO $ getAheadSVar st wloop
960{-# INLINABLE newParallelVar #-}
961newParallelVar :: MonadAsync m => m (SVar t m a)
962newParallelVar = liftIO $ getParallelSVar
964-- XXX this errors out for Parallel/Ahead SVars
965-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
966-- be read back from the SVar using 'fromSVar'.
967toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
969 liftIO $ (enqueue sv) m
970 done <- allThreadsDone sv
971 -- XXX This is safe only when called from the consumer thread or when no
972 -- consumer is present. There may be a race if we are not running in the
974 when done $ pushWorker sv
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list