2-- Module : Streamly.Internal.Data.SVar.Dispatch
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
10module Streamly.Internal.Data.SVar.Dispatch
12 -- * Latency collection
20 -- * Thread accounting
34 , sendWorkerDelayPaced
40import Control.Concurrent (takeMVar, tryReadMVar, ThreadId, threadDelay)
41import Control.Concurrent.MVar (tryPutMVar)
42import Control.Exception
43 (assert, catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
44 BlockedIndefinitelyOnSTM(..))
45import Control.Monad (when, void)
46import Control.Monad.IO.Class (MonadIO(liftIO))
47import Data.Maybe (fromJust, fromMaybe)
48import Data.IORef (IORef, modifyIORef, newIORef, readIORef, writeIORef)
49#if __GLASGOW_HASKELL__ < 804
50import Data.Semigroup ((<>))
52import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
53import Streamly.Internal.Control.ForkLifted (doFork)
54import Streamly.Internal.Data.Atomics
55 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
57import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
58import Streamly.Internal.Data.Time.Units
59 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
60 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
61import System.IO (hPutStrLn, stderr)
63import qualified Data.Heap as H
64import qualified Data.Set as S
66import Streamly.Internal.Data.SVar.Type
67import Streamly.Internal.Data.SVar.Worker
69-------------------------------------------------------------------------------
70-- Worker latency data processing
71-------------------------------------------------------------------------------
73-- Every once in a while workers update the latencies and check the yield rate.
74-- They return if we are above the expected yield rate. If we check too often
75-- it may impact performance, if we check less often we may have a stale
76-- picture. We update every minThreadDelay but we translate that into a yield
77-- count based on latency so that the checking overhead is little.
79-- XXX use a generation count to indicate that the value is updated. If the
80-- value is updated an existing worker must check it again on the next yield.
81-- Otherwise it is possible that we may keep updating it and because of the mod
82-- worker keeps skipping it.
83updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
84updateWorkerPollingInterval yinfo latency = do
85 let periodRef = workerPollingInterval yinfo
86 cnt = max 1 $ minThreadDelay `div` latency
87 period = min cnt (fromIntegral magicMaxBuffer)
89 writeIORef periodRef (fromIntegral period)
91{-# INLINE recordMinMaxLatency #-}
92recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()
93recordMinMaxLatency sv new = do
95 minLat <- readIORef (minWorkerLatency ss)
96 when (new < minLat || minLat == 0) $
97 writeIORef (minWorkerLatency ss) new
99 maxLat <- readIORef (maxWorkerLatency ss)
100 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
102recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()
103recordAvgLatency sv (count, time) = do
104 let ss = svarStats sv
105 modifyIORef (avgWorkerLatency ss) $
106 \(cnt, t) -> (cnt + count, t + time)
108-- Pour the pending latency stats into a collection bucket
109{-# INLINE collectWorkerPendingLatency #-}
110collectWorkerPendingLatency
111 :: IORef (Count, Count, NanoSecond64)
112 -> IORef (Count, Count, NanoSecond64)
113 -> IO (Count, Maybe (Count, NanoSecond64))
114collectWorkerPendingLatency cur col = do
115 (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
117 (fcnt, cnt, t) <- readIORef col
118 let totalCount = fcnt + fcount
119 latCount = cnt + count
121 writeIORef col (totalCount, latCount, latTime)
123 assert (latCount == 0 || latTime /= 0) (return ())
125 if latCount > 0 && latTime > 0
126 then Just (latCount, latTime)
128 return (totalCount, latPair)
130{-# INLINE shouldUseCollectedBatch #-}
131shouldUseCollectedBatch
137shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
138 let r = fromIntegral newLat / fromIntegral prevLat :: Double
139 in (collectedYields > fromIntegral magicMaxBuffer)
140 || (collectedTime > minThreadDelay)
141 || (prevLat > 0 && (r > 2 || r < 0.5))
144-- Returns a triple, (1) yield count since last collection, (2) the base time
145-- when we started counting, (3) average latency in the last measurement
146-- period. The former two are used for accurate measurement of the going rate
147-- whereas the average is used for future estimates e.g. how many workers
148-- should be maintained to maintain the rate.
149-- CAUTION! keep it in sync with getWorkerLatency
150collectLatency :: SVar t m a
153 -> IO (Count, AbsTime, NanoSecond64)
154collectLatency sv yinfo drain = do
155 let cur = workerPendingLatency yinfo
156 col = workerCollectedLatency yinfo
157 longTerm = svarAllTimeLatency yinfo
158 measured = workerMeasuredLatency yinfo
160 (newCount, newLatPair) <- collectWorkerPendingLatency cur col
161 (lcount, ltime) <- readIORef longTerm
162 prevLat <- readIORef measured
164 let newLcount = lcount + newCount
165 retWith lat = return (newLcount, ltime, lat)
168 Nothing -> retWith prevLat
169 Just (count, time) -> do
170 let newLat = time `div` fromIntegral count
171 when (svarInspectMode sv) $ recordMinMaxLatency sv newLat
172 -- When we have collected a significant sized batch we compute the
173 -- new latency using that batch and return the new latency,
174 -- otherwise we return the previous latency derived from the
176 if shouldUseCollectedBatch newCount time newLat prevLat || drain
178 -- XXX make this NOINLINE?
179 updateWorkerPollingInterval yinfo (max newLat prevLat)
180 when (svarInspectMode sv) $ recordAvgLatency sv (count, time)
181 writeIORef col (0, 0, 0)
182 writeIORef measured ((prevLat + newLat) `div` 2)
183 modifyIORef longTerm $ \(_, t) -> (newLcount, t)
187-------------------------------------------------------------------------------
188-- Dumping the SVar for debug/diag
189-------------------------------------------------------------------------------
191dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
192dumpSVarStats sv ss style = do
193 case yieldRateInfo sv of
196 _ <- liftIO $ collectLatency sv yinfo True
199 dispatches <- readIORef $ totalDispatches ss
200 maxWrk <- readIORef $ maxWorkers ss
201 maxOq <- readIORef $ maxOutQSize ss
202 maxHp <- readIORef $ maxHeapSize ss
203 minLat <- readIORef $ minWorkerLatency ss
204 maxLat <- readIORef $ maxWorkerLatency ss
205 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
206 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
207 Nothing -> return (0, 0, 0)
209 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
212 t <- readIORef (svarStopTime ss)
213 gl <- readIORef (svarGainedLostYields yinfo)
216 now <- getTime Monotonic
217 let interval = diffAbsTime64 now startTime
218 return (cnt, gl, interval `div` fromIntegral cnt)
220 let interval = diffAbsTime64 stopTime startTime
221 return (cnt, gl, interval `div` fromIntegral cnt)
222 else return (0, 0, 0)
225 [ "total dispatches = " <> show dispatches
226 , "max workers = " <> show maxWrk
227 , "max outQSize = " <> show maxOq
228 <> (if style == AheadVar
229 then "\nheap max size = " <> show maxHp
232 then "\nmin worker latency = " <> showNanoSecond64 minLat
235 then "\nmax worker latency = " <> showNanoSecond64 maxLat
238 then let lat = avgTime `div` fromIntegral avgCnt
239 in "\navg worker latency = " <> showNanoSecond64 lat
242 then "\nSVar latency = " <> showRelTime64 svarLat
245 then "\nSVar yield count = " <> show svarCnt
247 <> (if svarGainLossCnt > 0
248 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
252{-# NOINLINE dumpSVar #-}
253dumpSVar :: SVar t m a -> IO String
255 (oqList, oqLen) <- readIORef $ outputQueue sv
256 db <- tryReadMVar $ outputDoorBell sv
258 if svarStyle sv == AheadVar
260 (oheap, oheapSeq) <- readIORef $ outputHeap sv
261 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
263 [ "heap length = " <> show (H.size oheap)
264 , "heap seqeunce = " <> show oheapSeq
265 , "work queue length = " <> show (length wq)
266 , "work queue sequence = " <> show wqSeq
270 let style = svarStyle sv
272 if style /= ParallelVar
273 then readIORef $ needDoorBell sv
275 rthread <- readIORef $ workerThreads sv
276 workers <- readIORef $ workerCount sv
277 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
281 "Creator tid = " <> show (svarCreator sv),
282 "style = " <> show (svarStyle sv)
283 , "---------CURRENT STATE-----------"
284 , "outputQueue length computed = " <> show (length oqList)
285 , "outputQueue length maintained = " <> show oqLen
286 -- XXX print the types of events in the outputQueue, first 5
287 , "outputDoorBell = " <> show db
291 [ "needDoorBell = " <> show waiting
292 , "running threads = " <> show rthread
293 -- XXX print the status of first 5 threads
294 , "running thread count = " <> show workers
296 <> "---------STATS-----------\n"
299-- MVar diagnostics has some overhead - around 5% on AsyncT null benchmark, we
300-- can keep it on in production to debug problems quickly if and when they
301-- happen, but it may result in unexpected output when threads are left hanging
302-- until they are GCed because the consumer went away.
304{-# NOINLINE mvarExcHandler #-}
305mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
306mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
307 svInfo <- dumpSVar sv
308 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
311{-# NOINLINE stmExcHandler #-}
312stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
313stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
314 svInfo <- dumpSVar sv
315 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
318withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
319withDiagMVar sv label action =
320 if svarInspectMode sv
322 action `catches` [ Handler (mvarExcHandler sv label)
323 , Handler (stmExcHandler sv label)
327printSVar :: SVar t m a -> String -> IO ()
329 svInfo <- dumpSVar sv
330 hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo
332-------------------------------------------------------------------------------
334-------------------------------------------------------------------------------
336-- Thread tracking is needed for two reasons:
338-- 1) Killing threads on exceptions. Threads may not be left to go away by
339-- themselves because they may run for significant times before going away or
340-- worse they may be stuck in IO and never go away.
342-- 2) To know when all threads are done and the stream has ended.
344{-# NOINLINE addThread #-}
345addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
347 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
349-- This is cheaper than modifyThread because we do not have to send a
350-- outputDoorBell This can make a difference when more workers are being
352{-# INLINE delThread #-}
353delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
355 liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
357-- If present then delete else add. This takes care of out of order add and
358-- delete i.e. a delete arriving before we even added a thread.
359-- This occurs when the forked thread is done even before the 'addThread' right
360-- after the fork gets a chance to run.
361{-# INLINE modifyThread #-}
362modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
363modifyThread sv tid = do
364 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
366 then let new = S.delete tid old in (new, new)
367 else let new = S.insert tid old in (new, old)
368 when (null changed) $
371 void $ tryPutMVar (outputDoorBell sv) ()
373-- | This is safe even if we are adding more threads concurrently because if
374-- a child thread is adding another thread then anyway 'workerThreads' will
376{-# INLINE allThreadsDone #-}
377allThreadsDone :: MonadIO m => SVar t m a -> m Bool
378allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
380-------------------------------------------------------------------------------
381-- Dispatching workers
382-------------------------------------------------------------------------------
384{-# NOINLINE recordMaxWorkers #-}
385recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
386recordMaxWorkers sv = liftIO $ do
387 active <- readIORef (workerCount sv)
388 maxWrk <- readIORef (maxWorkers $ svarStats sv)
389 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
390 modifyIORef (totalDispatches $ svarStats sv) (+1)
392{-# NOINLINE pushWorker #-}
393pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
394pushWorker yieldMax sv = do
395 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
396 when (svarInspectMode sv) $ recordMaxWorkers sv
397 -- This allocation matters when significant number of workers are being
398 -- sent. We allocate it only when needed.
400 case yieldRateInfo sv of
401 Nothing -> return Nothing
402 Just _ -> liftIO $ do
404 t <- getTime Monotonic
405 lat <- newIORef (0, t)
406 return $ Just WorkerInfo
407 { workerYieldMax = yieldMax
408 , workerYieldCount = cntRef
409 , workerLatencyStart = lat
411 doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
414-- XXX we can push the workerCount modification in accountThread and use the
415-- same pushWorker for Parallel case as well.
417-- | In contrast to pushWorker which always happens only from the consumer
418-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
419-- producer side. So we need to use a thread safe modification of
420-- workerThreads. Alternatively, we can use a CreateThread event to avoid
421-- using a CAS based modification.
422{-# INLINE pushWorkerPar #-}
425 => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
426pushWorkerPar sv wloop =
427 if svarInspectMode sv
429 else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
434 {-# NOINLINE forkWithDiag #-}
436 -- We do not use workerCount in case of ParallelVar but still there is
437 -- no harm in maintaining it correctly.
438 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
440 -- This allocation matters when significant number of workers are being
441 -- sent. We allocate it only when needed. The overhead increases by 4x.
443 case yieldRateInfo sv of
444 Nothing -> return Nothing
445 Just _ -> liftIO $ do
447 t <- getTime Monotonic
448 lat <- newIORef (0, t)
449 return $ Just WorkerInfo
451 , workerYieldCount = cntRef
452 , workerLatencyStart = lat
455 doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
459-- True: can dispatch more
460-- False: cannot dispatch any more
461dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
462dispatchWorker yieldCount sv = do
463 let workerLimit = maxWorkerLimit sv
464 -- XXX in case of Ahead streams we should not send more than one worker
465 -- when the work queue is done but heap is not done.
466 done <- liftIO $ isWorkDone sv
467 -- Note, "done" may not mean that the work is actually finished if there
468 -- are workers active, because there may be a worker which has not yet
469 -- queued the leftover work.
472 qDone <- liftIO $ isQueueDone sv
473 -- This count may not be accurate as it is decremented by the workers
474 -- and we have no synchronization with that decrement.
475 active <- liftIO $ readIORef $ workerCount sv
478 -- Note that we may deadlock if the previous workers (tasks in the
479 -- stream) wait/depend on the future workers (tasks in the stream)
480 -- executing. In that case we should either configure the maxWorker
481 -- count to higher or use parallel style instead of ahead or async
483 limit <- case remainingWork sv of
484 Nothing -> return workerLimit
486 n <- liftIO $ readIORef ref
487 case yieldRateInfo sv of
488 Just _ -> return workerLimit
492 Unlimited -> Limited (fromIntegral n)
493 Limited lim -> Limited $ min lim (fromIntegral n)
495 -- XXX for ahead streams shall we take the heap yields into account
496 -- for controlling the dispatch? We should not dispatch if the heap
497 -- has already got the limit covered.
498 let dispatch = pushWorker yieldCount sv >> return True
500 Unlimited -> dispatch
501 -- Note that the use of remainingWork and workerCount is not
502 -- atomic and the counts may even have changed between reading
503 -- and using them here, so this is just approximate logic and
504 -- we cannot rely on it for correctness. We may actually
505 -- dispatch more workers than required.
506 Limited lim | lim > fromIntegral active -> dispatch
509 when (active <= 0) $ pushWorker 0 sv
513-- XXX in case of ahead style stream we need to take the heap size into account
514-- because we return the workers on the basis of that which causes a condition
515-- where we keep dispatching and they keep returning. So we must have exactly
516-- the same logic for not dispatching and for returning.
519-- True: can dispatch more
520-- False: full, no more dispatches
521dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
522dispatchWorkerPaced sv = do
523 let yinfo = fromJust $ yieldRateInfo sv
524 (svarYields, svarElapsed, wLatency) <- do
525 now <- liftIO $ getTime Monotonic
526 (yieldCount, baseTime, lat) <-
527 liftIO $ collectLatency sv yinfo False
528 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
531 then fromMaybe lat (workerBootstrapLatency yinfo)
534 return (yieldCount, elapsed, latency)
537 -- Need to measure the latency with a single worker before we can perform
541 let workerLimit = maxWorkerLimit sv
542 let targetLat = svarLatencyTarget yinfo
543 let range = svarLatencyRange yinfo
544 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
545 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
546 wLatency targetLat range
548 -- XXX we need to take yieldLimit into account here. If we are at the
549 -- end of the limit as well as the time, we should not be sleeping.
550 -- If we are not actually planning to dispatch any more workers we need
551 -- to take that in account.
554 assert (s >= 0) (return ())
555 -- XXX note that when we return from here we will block waiting
556 -- for the result from the existing worker. If that takes too
557 -- long we won't be able to send another worker until the
560 -- Sleep only if there are no active workers, otherwise we will
561 -- defer the output of those. Note we cannot use workerCount
562 -- here as it is not a reliable way to ensure there are
563 -- definitely no active workers. When workerCount is 0 we may
564 -- still have a Stop event waiting in the outputQueue.
565 done <- allThreadsDone sv
566 when done $ void $ do
567 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
568 liftIO $ threadDelay (fromIntegral us)
571 PartialWorker yields -> do
572 assert (yields > 0) (return ())
573 updateGainedLostYields yinfo yields
575 done <- allThreadsDone sv
576 when done $ void $ dispatchWorker yields sv
578 ManyWorkers netWorkers yields -> do
579 assert (netWorkers >= 1) (return ())
580 assert (yields >= 0) (return ())
581 updateGainedLostYields yinfo yields
583 let periodRef = workerPollingInterval yinfo
584 ycnt = max 1 $ yields `div` fromIntegral netWorkers
585 period = min ycnt (fromIntegral magicMaxBuffer)
587 old <- liftIO $ readIORef periodRef
588 when (period < old) $
589 liftIO $ writeIORef periodRef period
591 cnt <- liftIO $ readIORef $ workerCount sv
594 let total = netWorkers - cnt
595 batch = max 1 $ fromIntegral $
596 minThreadDelay `div` targetLat
597 -- XXX stagger the workers over a period?
598 -- XXX cannot sleep, as that would mean we cannot process
599 -- the outputs. need to try a different mechanism to
601 -- when (total > batch) $
602 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
603 dispatchN (min total batch)
608 updateGainedLostYields yinfo yields = do
609 let buf = fromIntegral $ svarRateBuffer yinfo
610 when (yields /= 0 && abs yields > buf) $ do
615 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
621 r <- dispatchWorker 0 sv
623 then dispatchN (n - 1)
626{-# NOINLINE sendWorkerWait #-}
629 => (SVar t m a -> IO ())
630 -> (SVar t m a -> m Bool)
633sendWorkerWait delay dispatch sv = do
634 -- Note that we are guaranteed to have at least one outstanding worker when
635 -- we enter this function. So if we sleep we are guaranteed to be woken up
636 -- by an outputDoorBell, when the worker exits.
639 (_, n) <- liftIO $ readIORef (outputQueue sv)
641 -- The queue may be empty temporarily if the worker has dequeued the
642 -- work item but has not enqueued the remaining part yet. For the same
643 -- reason, a worker may come back if it tries to dequeue and finds the
644 -- queue empty, even though the whole work has not finished yet.
646 -- If we find that the queue is empty, but it may be empty
647 -- temporarily, when we checked it. If that's the case we might
648 -- sleep indefinitely unless the active workers produce some
649 -- output. We may deadlock specially if the otuput from the active
650 -- workers depends on the future workers that we may never send.
651 -- So in case the queue was temporarily empty set a flag to inform
652 -- the enqueue to send us a doorbell.
654 -- Note that this is just a best effort mechanism to avoid a
655 -- deadlock. Deadlocks may still happen if for some weird reason
656 -- the consuming computation shares an MVar or some other resource
657 -- with the producing computation and gets blocked on that resource
658 -- and therefore cannot do any pushworker to add more threads to
659 -- the producer. In such cases the programmer should use a parallel
660 -- style so that all the producers are scheduled immediately and
661 -- unconditionally. We can also use a separate monitor thread to
662 -- push workers instead of pushing them from the consumer, but then
663 -- we are no longer using pull based concurrency rate adaptation.
665 -- XXX update this in the tutorial.
667 -- Having pending active workers does not mean that we are guaranteed
668 -- to be woken up if we sleep. In case of Ahead streams, there may be
669 -- queued items in the heap even though the outputQueue is empty, and
670 -- we may have active workers which are deadlocked on those items to be
671 -- processed by the consumer. We should either guarantee that any
672 -- worker, before returning, clears the heap or we send a worker to
673 -- clear it. Normally we always send a worker if no output is seen, but
674 -- if the thread limit is reached or we are using pacing then we may
675 -- not send a worker. See the concurrentApplication test in the tests,
676 -- that test case requires at least one yield from the producer to not
677 -- deadlock, if the last workers output is stuck in the heap then this
678 -- test fails. This problem can be extended to n threads when the
679 -- consumer may depend on the evaluation of next n items in the
682 -- register for the outputDoorBell before we check the queue so that if
683 -- we sleep because the queue was empty we are guaranteed to get a
684 -- doorbell on the next enqueue.
686 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
687 liftIO storeLoadBarrier
688 canDoMore <- dispatch sv
690 -- XXX test for the case when we miss sending a worker when the worker
691 -- count is more than 1500.
693 -- XXX Assert here that if the heap is not empty then there is at
694 -- least one outstanding worker. Otherwise we could be sleeping
698 then sendWorkerWait delay dispatch sv
700 liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
701 $ takeMVar (outputDoorBell sv)
702 (_, len) <- liftIO $ readIORef (outputQueue sv)
703 when (len <= 0) $ sendWorkerWait delay dispatch sv
705sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
706sendFirstWorker sv m = do
707 -- Note: We must have all the work on the queue before sending the
708 -- pushworker, otherwise the pushworker may exit before we even get a
711 liftIO $ enqueue sv (runIn, m)
712 case yieldRateInfo sv of
713 Nothing -> pushWorker 0 sv
715 if svarLatencyTarget yinfo == maxBound
716 then liftIO $ threadDelay maxBound
720sendWorkerDelayPaced :: SVar t m a -> IO ()
721sendWorkerDelayPaced _ = return ()
723sendWorkerDelay :: SVar t m a -> IO ()
725 -- XXX we need a better way to handle this than hardcoded delays. The
726 -- delays may be different for different systems.
727 -- If there is a usecase where this is required we can create a combinator
728 -- to set it as a config in the state.
731 ncpu <- getNumCapabilities
734 if (svarStyle sv == AheadVar)
738 if (svarStyle sv == AheadVar)
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