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)
49import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
50import Streamly.Internal.Control.ForkLifted (doFork)
51import Streamly.Internal.Data.Atomics
52 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
54import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
55import Streamly.Internal.Data.Time.Units
56 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
57 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
58import System.IO (hPutStrLn, stderr)
60import qualified Data.Heap as H
61import qualified Data.Set as S
63import Streamly.Internal.Data.SVar.Type
64import Streamly.Internal.Data.SVar.Worker
66-------------------------------------------------------------------------------
67-- Worker latency data processing
68-------------------------------------------------------------------------------
70-- Every once in a while workers update the latencies and check the yield rate.
71-- They return if we are above the expected yield rate. If we check too often
72-- it may impact performance, if we check less often we may have a stale
73-- picture. We update every minThreadDelay but we translate that into a yield
74-- count based on latency so that the checking overhead is little.
76-- XXX use a generation count to indicate that the value is updated. If the
77-- value is updated an existing worker must check it again on the next yield.
78-- Otherwise it is possible that we may keep updating it and because of the mod
79-- worker keeps skipping it.
80updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
81updateWorkerPollingInterval yinfo latency = do
82 let periodRef = workerPollingInterval yinfo
83 cnt = max 1 $ minThreadDelay `div` latency
84 period = min cnt (fromIntegral magicMaxBuffer)
86 writeIORef periodRef (fromIntegral period)
88{-# INLINE recordMinMaxLatency #-}
89recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()
90recordMinMaxLatency sv new = do
92 minLat <- readIORef (minWorkerLatency ss)
93 when (new < minLat || minLat == 0) $
94 writeIORef (minWorkerLatency ss) new
96 maxLat <- readIORef (maxWorkerLatency ss)
97 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
99recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()
100recordAvgLatency sv (count, time) = do
101 let ss = svarStats sv
102 modifyIORef (avgWorkerLatency ss) $
103 \(cnt, t) -> (cnt + count, t + time)
105-- Pour the pending latency stats into a collection bucket
106{-# INLINE collectWorkerPendingLatency #-}
107collectWorkerPendingLatency
108 :: IORef (Count, Count, NanoSecond64)
109 -> IORef (Count, Count, NanoSecond64)
110 -> IO (Count, Maybe (Count, NanoSecond64))
111collectWorkerPendingLatency cur col = do
112 (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
114 (fcnt, cnt, t) <- readIORef col
115 let totalCount = fcnt + fcount
116 latCount = cnt + count
118 writeIORef col (totalCount, latCount, latTime)
120 assert (latCount == 0 || latTime /= 0) (return ())
122 if latCount > 0 && latTime > 0
123 then Just (latCount, latTime)
125 return (totalCount, latPair)
127{-# INLINE shouldUseCollectedBatch #-}
128shouldUseCollectedBatch
134shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
135 let r = fromIntegral newLat / fromIntegral prevLat :: Double
136 in (collectedYields > fromIntegral magicMaxBuffer)
137 || (collectedTime > minThreadDelay)
138 || (prevLat > 0 && (r > 2 || r < 0.5))
141-- Returns a triple, (1) yield count since last collection, (2) the base time
142-- when we started counting, (3) average latency in the last measurement
143-- period. The former two are used for accurate measurement of the going rate
144-- whereas the average is used for future estimates e.g. how many workers
145-- should be maintained to maintain the rate.
146-- CAUTION! keep it in sync with getWorkerLatency
147collectLatency :: SVar t m a
150 -> IO (Count, AbsTime, NanoSecond64)
151collectLatency sv yinfo drain = do
152 let cur = workerPendingLatency yinfo
153 col = workerCollectedLatency yinfo
154 longTerm = svarAllTimeLatency yinfo
155 measured = workerMeasuredLatency yinfo
157 (newCount, newLatPair) <- collectWorkerPendingLatency cur col
158 (lcount, ltime) <- readIORef longTerm
159 prevLat <- readIORef measured
161 let newLcount = lcount + newCount
162 retWith lat = return (newLcount, ltime, lat)
165 Nothing -> retWith prevLat
166 Just (count, time) -> do
167 let newLat = time `div` fromIntegral count
168 when (svarInspectMode sv) $ recordMinMaxLatency sv newLat
169 -- When we have collected a significant sized batch we compute the
170 -- new latency using that batch and return the new latency,
171 -- otherwise we return the previous latency derived from the
173 if shouldUseCollectedBatch newCount time newLat prevLat || drain
175 -- XXX make this NOINLINE?
176 updateWorkerPollingInterval yinfo (max newLat prevLat)
177 when (svarInspectMode sv) $ recordAvgLatency sv (count, time)
178 writeIORef col (0, 0, 0)
179 writeIORef measured ((prevLat + newLat) `div` 2)
180 modifyIORef longTerm $ \(_, t) -> (newLcount, t)
184-------------------------------------------------------------------------------
185-- Dumping the SVar for debug/diag
186-------------------------------------------------------------------------------
188dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
189dumpSVarStats sv ss style = do
190 case yieldRateInfo sv of
193 _ <- liftIO $ collectLatency sv yinfo True
196 dispatches <- readIORef $ totalDispatches ss
197 maxWrk <- readIORef $ maxWorkers ss
198 maxOq <- readIORef $ maxOutQSize ss
199 maxHp <- readIORef $ maxHeapSize ss
200 minLat <- readIORef $ minWorkerLatency ss
201 maxLat <- readIORef $ maxWorkerLatency ss
202 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
203 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
204 Nothing -> return (0, 0, 0)
206 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
209 t <- readIORef (svarStopTime ss)
210 gl <- readIORef (svarGainedLostYields yinfo)
213 now <- getTime Monotonic
214 let interval = diffAbsTime64 now startTime
215 return (cnt, gl, interval `div` fromIntegral cnt)
217 let interval = diffAbsTime64 stopTime startTime
218 return (cnt, gl, interval `div` fromIntegral cnt)
219 else return (0, 0, 0)
222 [ "total dispatches = " <> show dispatches
223 , "max workers = " <> show maxWrk
224 , "max outQSize = " <> show maxOq
225 <> (if style == AheadVar
226 then "\nheap max size = " <> show maxHp
229 then "\nmin worker latency = " <> showNanoSecond64 minLat
232 then "\nmax worker latency = " <> showNanoSecond64 maxLat
235 then let lat = avgTime `div` fromIntegral avgCnt
236 in "\navg worker latency = " <> showNanoSecond64 lat
239 then "\nSVar latency = " <> showRelTime64 svarLat
242 then "\nSVar yield count = " <> show svarCnt
244 <> (if svarGainLossCnt > 0
245 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
249{-# NOINLINE dumpSVar #-}
250dumpSVar :: SVar t m a -> IO String
252 (oqList, oqLen) <- readIORef $ outputQueue sv
253 db <- tryReadMVar $ outputDoorBell sv
255 if svarStyle sv == AheadVar
257 (oheap, oheapSeq) <- readIORef $ outputHeap sv
258 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
260 [ "heap length = " <> show (H.size oheap)
261 , "heap seqeunce = " <> show oheapSeq
262 , "work queue length = " <> show (length wq)
263 , "work queue sequence = " <> show wqSeq
267 let style = svarStyle sv
269 if style /= ParallelVar
270 then readIORef $ needDoorBell sv
272 rthread <- readIORef $ workerThreads sv
273 workers <- readIORef $ workerCount sv
274 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
278 "Creator tid = " <> show (svarCreator sv),
279 "style = " <> show (svarStyle sv)
280 , "---------CURRENT STATE-----------"
281 , "outputQueue length computed = " <> show (length oqList)
282 , "outputQueue length maintained = " <> show oqLen
283 -- XXX print the types of events in the outputQueue, first 5
284 , "outputDoorBell = " <> show db
288 [ "needDoorBell = " <> show waiting
289 , "running threads = " <> show rthread
290 -- XXX print the status of first 5 threads
291 , "running thread count = " <> show workers
293 <> "---------STATS-----------\n"
296-- MVar diagnostics has some overhead - around 5% on AsyncT null benchmark, we
297-- can keep it on in production to debug problems quickly if and when they
298-- happen, but it may result in unexpected output when threads are left hanging
299-- until they are GCed because the consumer went away.
301{-# NOINLINE mvarExcHandler #-}
302mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
303mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
304 svInfo <- dumpSVar sv
305 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
308{-# NOINLINE stmExcHandler #-}
309stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
310stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
311 svInfo <- dumpSVar sv
312 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
315withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
316withDiagMVar sv label action =
317 if svarInspectMode sv
319 action `catches` [ Handler (mvarExcHandler sv label)
320 , Handler (stmExcHandler sv label)
324printSVar :: SVar t m a -> String -> IO ()
326 svInfo <- dumpSVar sv
327 hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo
329-------------------------------------------------------------------------------
331-------------------------------------------------------------------------------
333-- Thread tracking is needed for two reasons:
335-- 1) Killing threads on exceptions. Threads may not be left to go away by
336-- themselves because they may run for significant times before going away or
337-- worse they may be stuck in IO and never go away.
339-- 2) To know when all threads are done and the stream has ended.
341{-# NOINLINE addThread #-}
342addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
344 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
346-- This is cheaper than modifyThread because we do not have to send a
347-- outputDoorBell This can make a difference when more workers are being
349{-# INLINE delThread #-}
350delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
352 liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
354-- If present then delete else add. This takes care of out of order add and
355-- delete i.e. a delete arriving before we even added a thread.
356-- This occurs when the forked thread is done even before the 'addThread' right
357-- after the fork gets a chance to run.
358{-# INLINE modifyThread #-}
359modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
360modifyThread sv tid = do
361 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
363 then let new = S.delete tid old in (new, new)
364 else let new = S.insert tid old in (new, old)
365 when (null changed) $
368 void $ tryPutMVar (outputDoorBell sv) ()
370-- | This is safe even if we are adding more threads concurrently because if
371-- a child thread is adding another thread then anyway 'workerThreads' will
373{-# INLINE allThreadsDone #-}
374allThreadsDone :: MonadIO m => SVar t m a -> m Bool
375allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
377-------------------------------------------------------------------------------
378-- Dispatching workers
379-------------------------------------------------------------------------------
381{-# NOINLINE recordMaxWorkers #-}
382recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
383recordMaxWorkers sv = liftIO $ do
384 active <- readIORef (workerCount sv)
385 maxWrk <- readIORef (maxWorkers $ svarStats sv)
386 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
387 modifyIORef (totalDispatches $ svarStats sv) (+1)
389{-# NOINLINE pushWorker #-}
390pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
391pushWorker yieldMax sv = do
392 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
393 when (svarInspectMode sv) $ recordMaxWorkers sv
394 -- This allocation matters when significant number of workers are being
395 -- sent. We allocate it only when needed.
397 case yieldRateInfo sv of
398 Nothing -> return Nothing
399 Just _ -> liftIO $ do
401 t <- getTime Monotonic
402 lat <- newIORef (0, t)
403 return $ Just WorkerInfo
404 { workerYieldMax = yieldMax
405 , workerYieldCount = cntRef
406 , workerLatencyStart = lat
408 doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
411-- XXX we can push the workerCount modification in accountThread and use the
412-- same pushWorker for Parallel case as well.
414-- | In contrast to pushWorker which always happens only from the consumer
415-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
416-- producer side. So we need to use a thread safe modification of
417-- workerThreads. Alternatively, we can use a CreateThread event to avoid
418-- using a CAS based modification.
419{-# INLINE pushWorkerPar #-}
422 => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
423pushWorkerPar sv wloop =
424 if svarInspectMode sv
426 else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
431 {-# NOINLINE forkWithDiag #-}
433 -- We do not use workerCount in case of ParallelVar but still there is
434 -- no harm in maintaining it correctly.
435 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
437 -- This allocation matters when significant number of workers are being
438 -- sent. We allocate it only when needed. The overhead increases by 4x.
440 case yieldRateInfo sv of
441 Nothing -> return Nothing
442 Just _ -> liftIO $ do
444 t <- getTime Monotonic
445 lat <- newIORef (0, t)
446 return $ Just WorkerInfo
448 , workerYieldCount = cntRef
449 , workerLatencyStart = lat
452 doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
456-- True: can dispatch more
457-- False: cannot dispatch any more
458dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
459dispatchWorker yieldCount sv = do
460 let workerLimit = maxWorkerLimit sv
461 -- XXX in case of Ahead streams we should not send more than one worker
462 -- when the work queue is done but heap is not done.
463 done <- liftIO $ isWorkDone sv
464 -- Note, "done" may not mean that the work is actually finished if there
465 -- are workers active, because there may be a worker which has not yet
466 -- queued the leftover work.
469 qDone <- liftIO $ isQueueDone sv
470 -- This count may not be accurate as it is decremented by the workers
471 -- and we have no synchronization with that decrement.
472 active <- liftIO $ readIORef $ workerCount sv
475 -- Note that we may deadlock if the previous workers (tasks in the
476 -- stream) wait/depend on the future workers (tasks in the stream)
477 -- executing. In that case we should either configure the maxWorker
478 -- count to higher or use parallel style instead of ahead or async
480 limit <- case remainingWork sv of
481 Nothing -> return workerLimit
483 n <- liftIO $ readIORef ref
484 case yieldRateInfo sv of
485 Just _ -> return workerLimit
489 Unlimited -> Limited (fromIntegral n)
490 Limited lim -> Limited $ min lim (fromIntegral n)
492 -- XXX for ahead streams shall we take the heap yields into account
493 -- for controlling the dispatch? We should not dispatch if the heap
494 -- has already got the limit covered.
495 let dispatch = pushWorker yieldCount sv >> return True
497 Unlimited -> dispatch
498 -- Note that the use of remainingWork and workerCount is not
499 -- atomic and the counts may even have changed between reading
500 -- and using them here, so this is just approximate logic and
501 -- we cannot rely on it for correctness. We may actually
502 -- dispatch more workers than required.
503 Limited lim | lim > fromIntegral active -> dispatch
506 when (active <= 0) $ pushWorker 0 sv
510-- XXX in case of ahead style stream we need to take the heap size into account
511-- because we return the workers on the basis of that which causes a condition
512-- where we keep dispatching and they keep returning. So we must have exactly
513-- the same logic for not dispatching and for returning.
516-- True: can dispatch more
517-- False: full, no more dispatches
518dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
519dispatchWorkerPaced sv = do
520 let yinfo = fromJust $ yieldRateInfo sv
521 (svarYields, svarElapsed, wLatency) <- do
522 now <- liftIO $ getTime Monotonic
523 (yieldCount, baseTime, lat) <-
524 liftIO $ collectLatency sv yinfo False
525 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
528 then fromMaybe lat (workerBootstrapLatency yinfo)
531 return (yieldCount, elapsed, latency)
534 -- Need to measure the latency with a single worker before we can perform
538 let workerLimit = maxWorkerLimit sv
539 let targetLat = svarLatencyTarget yinfo
540 let range = svarLatencyRange yinfo
541 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
542 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
543 wLatency targetLat range
545 -- XXX we need to take yieldLimit into account here. If we are at the
546 -- end of the limit as well as the time, we should not be sleeping.
547 -- If we are not actually planning to dispatch any more workers we need
548 -- to take that in account.
551 assert (s >= 0) (return ())
552 -- XXX note that when we return from here we will block waiting
553 -- for the result from the existing worker. If that takes too
554 -- long we won't be able to send another worker until the
557 -- Sleep only if there are no active workers, otherwise we will
558 -- defer the output of those. Note we cannot use workerCount
559 -- here as it is not a reliable way to ensure there are
560 -- definitely no active workers. When workerCount is 0 we may
561 -- still have a Stop event waiting in the outputQueue.
562 done <- allThreadsDone sv
563 when done $ void $ do
564 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
565 liftIO $ threadDelay (fromIntegral us)
568 PartialWorker yields -> do
569 assert (yields > 0) (return ())
570 updateGainedLostYields yinfo yields
572 done <- allThreadsDone sv
573 when done $ void $ dispatchWorker yields sv
575 ManyWorkers netWorkers yields -> do
576 assert (netWorkers >= 1) (return ())
577 assert (yields >= 0) (return ())
578 updateGainedLostYields yinfo yields
580 let periodRef = workerPollingInterval yinfo
581 ycnt = max 1 $ yields `div` fromIntegral netWorkers
582 period = min ycnt (fromIntegral magicMaxBuffer)
584 old <- liftIO $ readIORef periodRef
585 when (period < old) $
586 liftIO $ writeIORef periodRef period
588 cnt <- liftIO $ readIORef $ workerCount sv
591 let total = netWorkers - cnt
592 batch = max 1 $ fromIntegral $
593 minThreadDelay `div` targetLat
594 -- XXX stagger the workers over a period?
595 -- XXX cannot sleep, as that would mean we cannot process
596 -- the outputs. need to try a different mechanism to
598 -- when (total > batch) $
599 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
600 dispatchN (min total batch)
605 updateGainedLostYields yinfo yields = do
606 let buf = fromIntegral $ svarRateBuffer yinfo
607 when (yields /= 0 && abs yields > buf) $ do
612 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
618 r <- dispatchWorker 0 sv
620 then dispatchN (n - 1)
623{-# NOINLINE sendWorkerWait #-}
626 => (SVar t m a -> IO ())
627 -> (SVar t m a -> m Bool)
630sendWorkerWait delay dispatch sv = do
631 -- Note that we are guaranteed to have at least one outstanding worker when
632 -- we enter this function. So if we sleep we are guaranteed to be woken up
633 -- by an outputDoorBell, when the worker exits.
636 (_, n) <- liftIO $ readIORef (outputQueue sv)
638 -- The queue may be empty temporarily if the worker has dequeued the
639 -- work item but has not enqueued the remaining part yet. For the same
640 -- reason, a worker may come back if it tries to dequeue and finds the
641 -- queue empty, even though the whole work has not finished yet.
643 -- If we find that the queue is empty, but it may be empty
644 -- temporarily, when we checked it. If that's the case we might
645 -- sleep indefinitely unless the active workers produce some
646 -- output. We may deadlock specially if the otuput from the active
647 -- workers depends on the future workers that we may never send.
648 -- So in case the queue was temporarily empty set a flag to inform
649 -- the enqueue to send us a doorbell.
651 -- Note that this is just a best effort mechanism to avoid a
652 -- deadlock. Deadlocks may still happen if for some weird reason
653 -- the consuming computation shares an MVar or some other resource
654 -- with the producing computation and gets blocked on that resource
655 -- and therefore cannot do any pushworker to add more threads to
656 -- the producer. In such cases the programmer should use a parallel
657 -- style so that all the producers are scheduled immediately and
658 -- unconditionally. We can also use a separate monitor thread to
659 -- push workers instead of pushing them from the consumer, but then
660 -- we are no longer using pull based concurrency rate adaptation.
662 -- XXX update this in the tutorial.
664 -- Having pending active workers does not mean that we are guaranteed
665 -- to be woken up if we sleep. In case of Ahead streams, there may be
666 -- queued items in the heap even though the outputQueue is empty, and
667 -- we may have active workers which are deadlocked on those items to be
668 -- processed by the consumer. We should either guarantee that any
669 -- worker, before returning, clears the heap or we send a worker to
670 -- clear it. Normally we always send a worker if no output is seen, but
671 -- if the thread limit is reached or we are using pacing then we may
672 -- not send a worker. See the concurrentApplication test in the tests,
673 -- that test case requires at least one yield from the producer to not
674 -- deadlock, if the last workers output is stuck in the heap then this
675 -- test fails. This problem can be extended to n threads when the
676 -- consumer may depend on the evaluation of next n items in the
679 -- register for the outputDoorBell before we check the queue so that if
680 -- we sleep because the queue was empty we are guaranteed to get a
681 -- doorbell on the next enqueue.
683 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
684 liftIO storeLoadBarrier
685 canDoMore <- dispatch sv
687 -- XXX test for the case when we miss sending a worker when the worker
688 -- count is more than 1500.
690 -- XXX Assert here that if the heap is not empty then there is at
691 -- least one outstanding worker. Otherwise we could be sleeping
695 then sendWorkerWait delay dispatch sv
697 liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
698 $ takeMVar (outputDoorBell sv)
699 (_, len) <- liftIO $ readIORef (outputQueue sv)
700 when (len <= 0) $ sendWorkerWait delay dispatch sv
702sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
703sendFirstWorker sv m = do
704 -- Note: We must have all the work on the queue before sending the
705 -- pushworker, otherwise the pushworker may exit before we even get a
708 liftIO $ enqueue sv (runIn, m)
709 case yieldRateInfo sv of
710 Nothing -> pushWorker 0 sv
712 if svarLatencyTarget yinfo == maxBound
713 then liftIO $ threadDelay maxBound
717sendWorkerDelayPaced :: SVar t m a -> IO ()
718sendWorkerDelayPaced _ = return ()
720sendWorkerDelay :: SVar t m a -> IO ()
722 -- XXX we need a better way to handle this than hardcoded delays. The
723 -- delays may be different for different systems.
724 -- If there is a usecase where this is required we can create a combinator
725 -- to set it as a config in the state.
728 ncpu <- getNumCapabilities
731 if (svarStyle sv == AheadVar)
735 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