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
61import Control.Concurrent
62 (ThreadId, myThreadId, threadDelay, getNumCapabilities)
63import Control.Concurrent.MVar
64 (MVar, newEmptyMVar, tryPutMVar, takeMVar)
65import Control.Exception (SomeException(..), catch, mask)
66import Control.Monad (when)
67import Control.Monad.Catch (MonadThrow)
68import Control.Monad.IO.Class (MonadIO(..))
69import Control.Monad.Trans.Control (MonadBaseControl, control)
71 (casIORef, readForCAS, peekTicket, atomicModifyIORefCAS_,
72 writeBarrier, storeLoadBarrier)
73import Data.Concurrent.Queue.MichaelScott
74 (LinkedQueue, pushL, tryPopR)
75import Data.Functor (void)
76import Data.Heap (Heap, Entry(..))
78 (IORef, modifyIORef, newIORef, readIORef, atomicModifyIORef)
79import Data.Maybe (fromJust)
81import GHC.Conc (ThreadId(..))
85import qualified Data.Heap as H
86import qualified Data.Set as S
88-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
89-- can keep it on in production to debug problems quickly if and when they
90-- happen, but it may result in unexpected output when threads are left hanging
91-- until they are GCed because the consumer went away.
94import Control.Concurrent.MVar (tryTakeMVar)
95import Control.Exception
96 (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
97 BlockedIndefinitelyOnSTM(..))
98import Data.IORef (writeIORef)
99import System.IO (hPutStrLn, stderr)
102------------------------------------------------------------------------------
103-- Parent child thread communication type
104------------------------------------------------------------------------------
106-- | Events that a child thread may send to a parent thread.
109 | ChildStop ThreadId (Maybe SomeException)
111-- | Sorting out-of-turn outputs in a heap for Ahead style streams
112data AheadHeapEntry (t :: (* -> *) -> * -> *) m a =
114 | AheadEntryStream (t m a)
116------------------------------------------------------------------------------
117-- State threaded around the monad for thread management
118------------------------------------------------------------------------------
120-- | Identify the type of the SVar. Two computations using the same style can
121-- be scheduled on the same SVar.
123 AsyncVar -- depth first concurrent
124 | WAsyncVar -- breadth first concurrent
125 | ParallelVar -- all parallel
126 | AheadVar -- Concurrent look ahead
129-- | An SVar or a Stream Var is a conduit to the output from multiple streams
130-- running concurrently and asynchronously. An SVar can be thought of as an
131-- asynchronous IO handle. We can write any number of streams to an SVar in a
132-- non-blocking manner and then read them back at any time at any pace. The
133-- SVar would run the streams asynchronously and accumulate results. An SVar
134-- may not really execute the stream completely and accumulate all the results.
135-- However, it ensures that the reader can read the results at whatever paces
136-- it wants to read. The SVar monitors and adapts to the consumer's pace.
138-- An SVar is a mini scheduler, it has an associated workLoop that holds the
139-- stream tasks to be picked and run by a pool of worker threads. It has an
140-- associated output queue where the output stream elements are placed by the
141-- worker threads. A outputDoorBell is used by the worker threads to intimate the
142-- consumer thread about availability of new results in the output queue. More
143-- workers are added to the SVar by 'fromStreamVar' on demand if the output
144-- produced is not keeping pace with the consumer. On bounded SVars, workers
145-- block on the output queue to provide throttling of the producer when the
146-- consumer is not pulling fast enough. The number of workers may even get
147-- reduced depending on the consuming pace.
149-- New work is enqueued either at the time of creation of the SVar or as a
150-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
151-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
153-- XXX can we use forall t m.
157 svarStyle :: SVarStyle
159 -- Shared output queue (events, length)
160 , outputQueue :: IORef ([ChildEvent a], 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)
195defaultMaxThreads, defaultMaxBuffer :: Int
196defaultMaxThreads = 1500
197defaultMaxBuffer = 1500
199defState :: State t m a
201 { streamVar = Nothing
202 , threadsHigh = defaultMaxThreads
203 , bufferHigh = defaultMaxBuffer
206-- We can optimize this so that we clear it only if it is a Just value, it
207-- results in slightly better perf for zip/zipM but the performance of scan
208-- worsens a lot, it does not fuse.
209rstState :: State t m a -> State t m b
210rstState st = st {streamVar = Nothing}
213{-# NOINLINE dumpSVar #-}
214dumpSVar :: SVar t m a -> IO String
217 (oqList, oqLen) <- readIORef $ outputQueue sv
218 db <- tryTakeMVar $ outputDoorBell sv
220 if svarStyle sv == AheadVar
222 (oheap, oheapSeq) <- readIORef $ outputHeap sv
223 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
224 maxHp <- readIORef $ maxHeapSize sv
226 [ "heap length = " ++ show (H.size oheap)
227 , "heap seqeunce = " ++ show oheapSeq
228 , "work queue length = " ++ show (length wq)
229 , "work queue sequence = " ++ show wqSeq
230 , "heap max size = " ++ show maxHp
234 waiting <- readIORef $ needDoorBell sv
235 rthread <- readIORef $ workerThreads sv
236 workers <- readIORef $ workerCount sv
237 maxWrk <- readIORef $ maxWorkers sv
238 dispatches <- readIORef $ totalDispatches sv
239 maxOq <- readIORef $ maxOutQSize sv
242 [ "tid = " ++ show tid
243 , "style = " ++ show (svarStyle sv)
244 , "outputQueue length computed = " ++ show (length oqList)
245 , "outputQueue length maintained = " ++ show oqLen
246 , "output outputDoorBell = " ++ show db
247 , "total dispatches = " ++ show dispatches
248 , "max workers = " ++ show maxWrk
249 , "max outQSize = " ++ show maxOq
251 ++ aheadDump ++ unlines
252 [ "needDoorBell = " ++ show waiting
253 , "running threads = " ++ show rthread
254 , "running thread count = " ++ show workers
257{-# NOINLINE mvarExcHandler #-}
258mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
259mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
260 svInfo <- dumpSVar sv
261 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo
264{-# NOINLINE stmExcHandler #-}
265stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
266stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
267 svInfo <- dumpSVar sv
268 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo
271withDBGMVar :: SVar t m a -> String -> IO () -> IO ()
272withDBGMVar sv label action =
273 action `catches` [ Handler (mvarExcHandler sv label)
274 , Handler (stmExcHandler sv label)
277withDBGMVar :: SVar t m a -> String -> IO () -> IO ()
278withDBGMVar _ _ action = action
281-- Slightly faster version of CAS. Gained some improvement by avoiding the use
282-- of "evaluate" because we know we do not have exceptions in fn.
283{-# INLINE atomicModifyIORefCAS #-}
284atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
285atomicModifyIORefCAS ref fn = do
286 tkt <- readForCAS ref
292 loop _ 0 = atomicModifyIORef ref fn
294 let (new, result) = fn $ peekTicket old
295 (success, tkt) <- casIORef ref old new
298 else loop tkt (tries - 1)
300------------------------------------------------------------------------------
301-- Spawning threads and collecting result in streamed fashion
302------------------------------------------------------------------------------
304-- | A monad that can perform concurrent or parallel IO operations. Streams
305-- that can be composed concurrently require the underlying monad to be
309type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
311-- Stolen from the async package. The perf improvement is modest, 2% on a
312-- thread heavy benchmark (parallel composition using noop computations).
313-- A version of forkIO that does not include the outer exception
314-- handler: saves a bit of time when we will be installing our own
316{-# INLINE rawForkIO #-}
317rawForkIO :: IO () -> IO ThreadId
318rawForkIO action = IO $ \ s ->
319 case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
322doFork :: MonadBaseControl IO m
324 -> (SomeException -> IO ())
326doFork action exHandler =
327 control $ \runInIO ->
328 mask $ \restore -> do
329 tid <- rawForkIO $ catch (restore $ void $ runInIO action)
333-- XXX exception safety of all atomic/MVar operations
335-- TBD Each worker can have their own queue and the consumer can empty one
336-- queue at a time, that way contention can be reduced.
338-- | This function is used by the producer threads to queue output for the
339-- consumer thread to consume. Returns whether the queue has more space.
341send :: Int -> SVar t m a -> ChildEvent a -> IO Bool
342send maxOutputQLen sv msg = do
343 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
344 ((msg : es, n + 1), n)
346 -- The wake up must happen only after the store has finished otherwise
347 -- we can have lost wakeup problems.
349 -- Since multiple workers can try this at the same time, it is possible
350 -- that we may put a spurious MVar after the consumer has already seen
351 -- the output. But that's harmless, at worst it may cause the consumer
352 -- to read the queue again and find it empty.
353 -- The important point is that the consumer is guaranteed to receive a
354 -- doorbell if something was added to the queue after it empties it.
355 void $ tryPutMVar (outputDoorBell sv) ()
356 return (len < maxOutputQLen || maxOutputQLen < 0)
358{-# NOINLINE sendStop #-}
359sendStop :: SVar t m a -> IO ()
361 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
362 myThreadId >>= \tid -> void $ send (-1) sv (ChildStop tid Nothing)
364-------------------------------------------------------------------------------
366-------------------------------------------------------------------------------
368-- Note: For purely right associated expressions this queue should have at most
369-- one element. It grows to more than one when we have left associcated
370-- expressions. Large left associated compositions can grow this to a
372{-# INLINE enqueueLIFO #-}
373enqueueLIFO :: SVar t m a -> IORef [t m a] -> t m a -> IO ()
374enqueueLIFO sv q m = do
375 atomicModifyIORefCAS_ q $ \ms -> m : ms
377 w <- readIORef $ needDoorBell sv
379 -- Note: the sequence of operations is important for correctness here.
380 -- We need to set the flag to false strictly before sending the
381 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
382 -- then we may set the flag to False to later making the consumer lose
383 -- the flag, even without receiving a outputDoorBell.
384 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
385 void $ tryPutMVar (outputDoorBell sv) ()
387{-# INLINE workLoopLIFO #-}
388workLoopLIFO :: MonadIO m
389 => (State t m a -> IORef [t m a] -> t m a -> m () -> m ())
390 -> State t m a -> IORef [t m a] -> m ()
391workLoopLIFO f st q = run
395 sv = fromJust $ streamVar st
399 Nothing -> liftIO $ sendStop sv
400 Just m -> f st q m run
402 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
404 x : xs -> (xs, Just x)
406-------------------------------------------------------------------------------
408-------------------------------------------------------------------------------
410-- XXX we can use the Ahead style sequence/heap mechanism to make the best
411-- effort to always try to finish the streams on the left side of an expression
412-- first as long as possible.
414{-# INLINE enqueueFIFO #-}
415enqueueFIFO :: SVar t m a -> LinkedQueue (t m a) -> t m a -> IO ()
416enqueueFIFO sv q m = do
419 w <- readIORef $ needDoorBell sv
421 -- Note: the sequence of operations is important for correctness here.
422 -- We need to set the flag to false strictly before sending the
423 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
424 -- then we may set the flag to False to later making the consumer lose
425 -- the flag, even without receiving a outputDoorBell.
426 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
427 void $ tryPutMVar (outputDoorBell sv) ()
429{-# INLINE workLoopFIFO #-}
430workLoopFIFO :: MonadIO m
431 => (State t m a -> LinkedQueue (t m a) -> t m a -> m () -> m ())
432 -> State t m a -> LinkedQueue (t m a) -> m ()
433workLoopFIFO f st q = run
437 sv = fromJust $ streamVar st
439 work <- liftIO $ tryPopR q
441 Nothing -> liftIO $ sendStop sv
442 Just m -> f st q m run
444-------------------------------------------------------------------------------
446-------------------------------------------------------------------------------
448-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
449-- but always serve them in the same order as they appear in the stream. To
450-- implement lookahead streams efficiently we assign a sequence number to each
451-- task when the task is picked up for execution. When the task finishes, the
452-- output is tagged with the same sequence number and we rearrange the outputs
453-- in sequence based on that number.
455-- To explain the mechanism imagine that the current task at the head of the
456-- stream has a "token" to yield to the outputQueue. The ownership of the token
457-- is determined by the current sequence number is maintained in outputHeap.
458-- Sequence number is assigned when a task is queued. When a thread dequeues a
459-- task it picks up the sequence number as well and when the output is ready it
460-- uses the sequence number to queue the output to the outputQueue.
462-- The thread with current sequence number sends the output directly to the
463-- outputQueue. Other threads push the output to the outputHeap. When the task
464-- being queued on the heap is a stream of many elements we evaluate only the
465-- first element and keep the rest of the unevaluated computation in the heap.
466-- When such a task gets the "token" for outputQueue it evaluates and directly
467-- yields all the elements to the outputQueue without checking for the
470-- Note that no two outputs in the heap can have the same sequence numbers and
471-- therefore we do not need a stable heap. We have also separated the buffer
472-- for the current task (outputQueue) and the pending tasks (outputHeap) so
473-- that the pending tasks cannot interfere with the current task. Note that for
474-- a single task just the outputQueue is enough and for the case of many
475-- threads just a heap is good enough. However we balance between these two
476-- cases, so that both are efficient.
478-- For bigger streams it may make sense to have separate buffers for each
479-- stream. However, for singleton streams this may become inefficient. However,
480-- if we do not have separate buffers, then the streams that come later in
481-- sequence may hog the buffer, hindering the streams that are ahead. For this
482-- reason we have a single element buffer limitation for the streams being
483-- executed in advance.
485-- This scheme works pretty efficiently with less than 40% extra overhead
486-- compared to the Async streams where we do not have any kind of sequencing of
487-- the outputs. It is especially devised so that we are most efficient when we
488-- have short tasks and need just a single thread. Also when a thread yields
489-- many items it can hold lockfree access to the outputQueue and do it
492-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
493-- that they do not hog the resources and hinder the progress of the threads in
496-- Left associated ahead expressions are expensive. We start a new SVar for
497-- each left associative expression. The queue is used only for right
498-- associated expression, we queue the right expression and execute the left.
499-- Thererefore the queue never has more than on item in it.
500{-# INLINE enqueueAhead #-}
501enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
502enqueueAhead sv q m = do
503 atomicModifyIORefCAS_ q $ \ case
504 ([], n) -> ([m], n + 1) -- increment sequence
505 _ -> error "not empty"
507 w <- readIORef $ needDoorBell sv
509 -- Note: the sequence of operations is important for correctness here.
510 -- We need to set the flag to false strictly before sending the
511 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
512 -- then we may set the flag to False to later making the consumer lose
513 -- the flag, even without receiving a outputDoorBell.
514 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
515 void $ tryPutMVar (outputDoorBell sv) ()
517-- Normally the thread that has the token should never go away. The token gets
518-- handed over to another thread, but someone or the other has the token at any
519-- point of time. But if the task that has the token finds that the outputQueue
520-- is full, in that case it can go away without even handing over the token to
521-- another thread. In that case it sets the nextSequence number in the heap its
522-- own sequence number before going away. To handle this case, any task that
523-- does not have the token tries to dequeue from the heap first before
524-- dequeuing from the work queue. If it finds that the task at the top of the
525-- heap is the one that owns the current sequence number then it grabs the
526-- token and starts with that.
528-- XXX instead of queueing just the head element and the remaining computation
529-- on the heap, evaluate as many as we can and place them on the heap. But we
530-- need to give higher priority to the lower sequence numbers so that lower
531-- priority tasks do not fill up the heap making higher priority tasks block
532-- due to full heap. Maybe we can have a weighted space for them in the heap.
533-- The weight is inversely proportional to the sequence number.
535-- XXX review for livelock
540{-# INLINE queueEmptyAhead #-}
541queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
542queueEmptyAhead q = liftIO $ do
543 (xs, _) <- readIORef q
546{-# INLINE dequeueAhead #-}
547dequeueAhead :: MonadIO m
548 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
549dequeueAhead q = liftIO $ do
550 atomicModifyIORefCAS q $ \case
551 ([], n) -> (([], n), Nothing)
552 (x : [], n) -> (([], n), Just (x, n))
553 _ -> error "more than one item on queue"
555{-# INLINE dequeueFromHeap #-}
557 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
558 -> IO (Maybe (Entry Int (AheadHeapEntry t m a)))
559dequeueFromHeap hpRef = do
560 atomicModifyIORefCAS hpRef $ \hp@(h, snum) -> do
563 Nothing -> (hp, Nothing)
564 Just (ent@(Entry seqNo _ev), hp') ->
566 then ((hp', seqNo), Just ent)
569-------------------------------------------------------------------------------
571-------------------------------------------------------------------------------
573-- XXX To be implemented. Use a linked queue like WAsync and put back the
574-- remaining computation at the back of the queue instead of the heap, and
575-- increment the sequence number.
577-- Thread tracking is needed for two reasons:
579-- 1) Killing threads on exceptions. Threads may not be left to go away by
580-- themselves because they may run for significant times before going away or
581-- worse they may be stuck in IO and never go away.
583-- 2) To know when all threads are done and the stream has ended.
585{-# NOINLINE addThread #-}
586addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
588 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
590-- This is cheaper than modifyThread because we do not have to send a
591-- outputDoorBell This can make a difference when more workers are being
593{-# INLINE delThread #-}
594delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
596 liftIO $ modifyIORef (workerThreads sv) $ (\s -> S.delete tid s)
598-- If present then delete else add. This takes care of out of order add and
599-- delete i.e. a delete arriving before we even added a thread.
600-- This occurs when the forked thread is done even before the 'addThread' right
601-- after the fork gets a chance to run.
602{-# INLINE modifyThread #-}
603modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
604modifyThread sv tid = do
605 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
606 if (S.member tid old)
607 then let new = (S.delete tid old) in (new, new)
608 else let new = (S.insert tid old) in (new, old)
612 void $ tryPutMVar (outputDoorBell sv) ()
615-- | This is safe even if we are adding more threads concurrently because if
616-- a child thread is adding another thread then anyway 'workerThreads' will
618{-# INLINE allThreadsDone #-}
619allThreadsDone :: MonadIO m => SVar t m a -> m Bool
620allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
622{-# NOINLINE handleChildException #-}
623handleChildException :: SVar t m a -> SomeException -> IO ()
624handleChildException sv e = do
626 void $ send (-1) sv (ChildStop tid (Just e))
629recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
630recordMaxWorkers sv = liftIO $ do
631 active <- readIORef (workerCount sv)
632 maxWrk <- readIORef (maxWorkers sv)
633 when (active > maxWrk) $ writeIORef (maxWorkers sv) active
634 modifyIORef (totalDispatches sv) (+1)
637{-# NOINLINE pushWorker #-}
638pushWorker :: MonadAsync m => SVar t m a -> m ()
640 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
644 doFork (workLoop sv) (handleChildException sv) >>= addThread sv
646-- XXX we can push the workerCount modification in accountThread and use the
647-- same pushWorker for Parallel case as well.
649-- | In contrast to pushWorker which always happens only from the consumer
650-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
651-- producer side. So we need to use a thread safe modification of
652-- workerThreads. Alternatively, we can use a CreateThread event to avoid
653-- using a CAS based modification.
654{-# NOINLINE pushWorkerPar #-}
655pushWorkerPar :: MonadAsync m => SVar t m a -> m () -> m ()
656pushWorkerPar sv wloop = do
657 -- We do not use workerCount in case of ParallelVar but still there is no
658 -- harm in maintaining it correctly.
660 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
663 doFork wloop (handleChildException sv) >>= modifyThread sv
665dispatchWorker :: MonadAsync m => Int -> SVar t m a -> m ()
666dispatchWorker maxWorkerLimit sv = do
667 done <- liftIO $ isWorkDone sv
669 -- Note that the worker count is only decremented during event
670 -- processing in fromStreamVar and therefore it is safe to read and
671 -- use it without a lock.
672 cnt <- liftIO $ readIORef $ workerCount sv
673 -- Note that we may deadlock if the previous workers (tasks in the
674 -- stream) wait/depend on the future workers (tasks in the stream)
675 -- executing. In that case we should either configure the maxWorker
676 -- count to higher or use parallel style instead of ahead or async
678 when (cnt < maxWorkerLimit || maxWorkerLimit < 0) $ pushWorker sv
680{-# NOINLINE sendWorkerWait #-}
681sendWorkerWait :: MonadAsync m => Int -> SVar t m a -> m ()
682sendWorkerWait maxWorkerLimit sv = do
683 -- Note that we are guaranteed to have at least one outstanding worker when
684 -- we enter this function. So if we sleep we are guaranteed to be woken up
685 -- by a outputDoorBell, when the worker exits.
687 -- XXX we need a better way to handle this than hardcoded delays. The
688 -- delays may be different for different systems.
689 ncpu <- liftIO $ getNumCapabilities
692 if (svarStyle sv == AheadVar)
693 then liftIO $ threadDelay 100
694 else liftIO $ threadDelay 25
696 if (svarStyle sv == AheadVar)
697 then liftIO $ threadDelay 100
698 else liftIO $ threadDelay 10
700 (_, n) <- liftIO $ readIORef (outputQueue sv)
702 -- The queue may be empty temporarily if the worker has dequeued the
703 -- work item but has not enqueued the remaining part yet. For the same
704 -- reason, a worker may come back if it tries to dequeue and finds the
705 -- queue empty, even though the whole work has not finished yet.
707 -- If we find that the queue is empty, but it may be empty
708 -- temporarily, when we checked it. If that's the case we might
709 -- sleep indefinitely unless the active workers produce some
710 -- output. We may deadlock specially if the otuput from the active
711 -- workers depends on the future workers that we may never send.
712 -- So in case the queue was temporarily empty set a flag to inform
713 -- the enqueue to send us a doorbell.
715 -- Note that this is just a best effort mechanism to avoid a
716 -- deadlock. Deadlocks may still happen if for some weird reason
717 -- the consuming computation shares an MVar or some other resource
718 -- with the producing computation and gets blocked on that resource
719 -- and therefore cannot do any pushworker to add more threads to
720 -- the producer. In such cases the programmer should use a parallel
721 -- style so that all the producers are scheduled immediately and
722 -- unconditionally. We can also use a separate monitor thread to
723 -- push workers instead of pushing them from the consumer, but then
724 -- we are no longer using pull based concurrency rate adaptation.
726 -- XXX update this in the tutorial.
728 -- register for the outputDoorBell before we check the queue so that if we
729 -- sleep because the queue was empty we are guaranteed to get a
730 -- doorbell on the next enqueue.
732 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
733 liftIO $ storeLoadBarrier
734 dispatchWorker maxWorkerLimit sv
736 -- XXX test for the case when we miss sending a worker when the worker
737 -- count is more than 1500.
739 -- XXX Assert here that if the heap is not empty then there is at
740 -- least one outstanding worker. Otherwise we could be sleeping
743 done <- liftIO $ isWorkDone sv
746 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
747 $ takeMVar (outputDoorBell sv)
748 (_, len) <- liftIO $ readIORef (outputQueue sv)
749 when (len <= 0) $ sendWorkerWait maxWorkerLimit sv
750 else sendWorkerWait maxWorkerLimit sv
752{-# INLINE readOutputQRaw #-}
753readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
754readOutputQRaw sv = do
755 (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
757 oqLen <- readIORef (maxOutQSize sv)
758 when (len > oqLen) $ writeIORef (maxOutQSize sv) len
762readOutputQBounded :: MonadAsync m => Int -> SVar t m a -> m [ChildEvent a]
763readOutputQBounded n sv = do
764 (list, len) <- liftIO $ readOutputQRaw sv
765 -- When there is no output seen we dispatch more workers to help
766 -- out if there is work pending in the work queue.
770 -- send a worker proactively, if needed, even before we start
771 -- processing the output. This may degrade single processor
772 -- perf but improves multi-processor, because of more
780 cnt <- liftIO $ readIORef $ workerCount sv
782 done <- liftIO $ isWorkDone sv
783 when (not done) $ pushWorker sv
785 {-# INLINE blockingRead #-}
788 liftIO $ (readOutputQRaw sv >>= return . fst)
790postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
791postProcessBounded sv = do
792 workersDone <- allThreadsDone sv
793 -- There may still be work pending even if there are no workers
794 -- pending because all the workers may return if the
795 -- outputQueue becomes full. In that case send off a worker to
796 -- kickstart the work again.
799 r <- liftIO $ isWorkDone sv
800 when (not r) $ pushWorker sv
804getAheadSVar :: MonadAsync m
807 -> IORef ([t m a], Int)
808 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
811getAheadSVar st f = do
812 outQ <- newIORef ([], 0)
813 outH <- newIORef (H.empty, 0)
814 outQMv <- newEmptyMVar
816 wfw <- newIORef False
817 running <- newIORef S.empty
818 q <- newIORef ([], -1)
828 SVar { outputQueue = outQ
829 , outputDoorBell = outQMv
830 , readOutputQ = readOutputQBounded (threadsHigh st) sv
831 , postProcess = postProcessBounded sv
832 , workerThreads = running
833 -- , workLoop = workLoopAhead sv q outH
834 , workLoop = f st{streamVar = Just sv} q outH
835 , enqueue = enqueueAhead sv q
836 , isWorkDone = isWorkDoneAhead q outH
838 , svarStyle = AheadVar
839 , workerCount = active
840 , accountThread = delThread sv
844 , totalDispatches = disp
845 , maxWorkers = maxWrk
846 , maxOutQSize = maxOq
847 , maxHeapSize = maxHs
848 , maxWorkQSize = maxWq
855 {-# INLINE isWorkDoneAhead #-}
856 isWorkDoneAhead q ref = do
858 (hp, _) <- readIORef ref
859 return (H.size hp <= 0)
860 queueDone <- checkEmpty q
861 return $ queueDone && heapDone
864 (xs, _) <- readIORef q
867getParallelSVar :: MonadIO m => IO (SVar t m a)
869 outQ <- newIORef ([], 0)
870 outQMv <- newEmptyMVar
872 running <- newIORef S.empty
881 SVar { outputQueue = outQ
882 , outputDoorBell = outQMv
883 , readOutputQ = readOutputQPar sv
884 , postProcess = allThreadsDone sv
885 , workerThreads = running
886 , workLoop = undefined
887 , enqueue = undefined
888 , isWorkDone = undefined
889 , needDoorBell = undefined
890 , svarStyle = ParallelVar
891 , workerCount = active
892 , accountThread = modifyThread sv
894 , aheadWorkQueue = undefined
895 , outputHeap = undefined
896 , totalDispatches = disp
897 , maxWorkers = maxWrk
898 , maxOutQSize = maxOq
899 , maxHeapSize = maxHs
900 , maxWorkQSize = maxWq
907 readOutputQPar sv = liftIO $ do
908 withDBGMVar sv "readOutputQPar: doorbell" $ takeMVar (outputDoorBell sv)
909 readOutputQRaw sv >>= return . fst
911sendWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
913 -- Note: We must have all the work on the queue before sending the
914 -- pushworker, otherwise the pushworker may exit before we even get a
916 liftIO $ enqueue sv m
920{-# INLINABLE newAheadVar #-}
921newAheadVar :: MonadAsync m
925 -> IORef ([t m a], Int)
926 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
929newAheadVar st m wloop = do
930 sv <- liftIO $ getAheadSVar st wloop
933{-# INLINABLE newParallelVar #-}
934newParallelVar :: MonadAsync m => m (SVar t m a)
935newParallelVar = liftIO $ getParallelSVar
937-- XXX this errors out for Parallel/Ahead SVars
938-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
939-- be read back from the SVar using 'fromSVar'.
940toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
942 liftIO $ (enqueue sv) m
943 done <- allThreadsDone sv
944 -- XXX This is safe only when called from the consumer thread or when no
945 -- consumer is present. There may be a race if we are not running in the
947 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