@@ -0,0 +1,673 @@
2+-- Module : Streamly.Internal.Data.Stream.Async.Channel.Dispatcher
3+-- Copyright : (c) 2017 Composewell Technologies
4+-- License : BSD-3-Clause
5+-- Maintainer : streamly@composewell.com
6+-- Stability : experimental
10+module Streamly.Internal.Data.Stream.Async.Channel.Dispatcher
12+ -- * Latency collection
20+ -- * Thread accounting
29+ , dispatchWorkerPaced
33+ , sendWorkerDelayPaced
37+import Control.Monad.Trans.Control (MonadBaseControl)
38+import Control.Concurrent (takeMVar, tryReadMVar, ThreadId, threadDelay)
39+import Control.Concurrent.MVar (tryPutMVar)
40+import Control.Exception
41+ (assert, catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
42+ BlockedIndefinitelyOnSTM(..))
43+import Control.Monad (when, void)
44+import Control.Monad.IO.Class (MonadIO(liftIO))
45+import Data.Maybe (fromJust, fromMaybe)
46+import Data.IORef (IORef, modifyIORef, newIORef, readIORef, writeIORef)
47+#if __GLASGOW_HASKELL__ < 804
48+import Data.Semigroup ((<>))
50+import Streamly.Internal.Control.ForkLifted (doFork)
51+import Streamly.Internal.Data.Atomics
52+ (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
54+import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
55+import Streamly.Internal.Data.Time.Units
56+ (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
57+ fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
58+import System.IO (hPutStrLn, stderr)
60+-- import qualified Data.Heap as H
61+import qualified Data.Set as S
63+import Streamly.Internal.Data.Stream.Async.Channel.Type
64+import Streamly.Internal.Data.Stream.Async.Channel.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.
80+updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
81+updateWorkerPollingInterval 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 #-}
89+recordMinMaxLatency :: Channel m a -> NanoSecond64 -> IO ()
90+recordMinMaxLatency sv new = do
91+ let ss = svarStats sv
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
99+recordAvgLatency :: Channel m a -> (Count, NanoSecond64) -> IO ()
100+recordAvgLatency 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 #-}
107+collectWorkerPendingLatency
108+ :: IORef (Count, Count, NanoSecond64)
109+ -> IORef (Count, Count, NanoSecond64)
110+ -> IO (Count, Maybe (Count, NanoSecond64))
111+collectWorkerPendingLatency 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 #-}
128+shouldUseCollectedBatch
134+shouldUseCollectedBatch 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
147+collectLatency :: Channel m a
150+ -> IO (Count, AbsTime, NanoSecond64)
151+collectLatency 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)
182+ else retWith prevLat
184+-------------------------------------------------------------------------------
185+-- Dumping the SVar for debug/diag
186+-------------------------------------------------------------------------------
188+dumpSVarStats :: Channel m a -> SVarStats -> IO String
189+dumpSVarStats sv ss = do
190+ case yieldRateInfo sv of
191+ Nothing -> return ()
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
226+ then "\nmin worker latency = " <> showNanoSecond64 minLat
229+ then "\nmax worker latency = " <> showNanoSecond64 maxLat
232+ then let lat = avgTime `div` fromIntegral avgCnt
233+ in "\navg worker latency = " <> showNanoSecond64 lat
236+ then "\nSVar latency = " <> showRelTime64 svarLat
239+ then "\nSVar yield count = " <> show svarCnt
241+ <> (if svarGainLossCnt > 0
242+ then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
246+{-# NOINLINE dumpSVar #-}
247+dumpSVar :: Channel m a -> IO String
249+ (oqList, oqLen) <- readIORef $ outputQueue sv
250+ db <- tryReadMVar $ outputDoorBell sv
252+ waiting <- readIORef $ needDoorBell sv
253+ rthread <- readIORef $ workerThreads sv
254+ workers <- readIORef $ workerCount sv
255+ stats <- dumpSVarStats sv (svarStats sv)
259+ "Creator tid = " <> show (svarCreator sv)
260+ , "---------CURRENT STATE-----------"
261+ , "outputQueue length computed = " <> show (length oqList)
262+ , "outputQueue length maintained = " <> show oqLen
263+ -- XXX print the types of events in the outputQueue, first 5
264+ , "outputDoorBell = " <> show db
267+ [ "needDoorBell = " <> show waiting
268+ , "running threads = " <> show rthread
269+ -- XXX print the status of first 5 threads
270+ , "running thread count = " <> show workers
272+ <> "---------STATS-----------\n"
275+-- MVar diagnostics has some overhead - around 5% on AsyncT null benchmark, we
276+-- can keep it on in production to debug problems quickly if and when they
277+-- happen, but it may result in unexpected output when threads are left hanging
278+-- until they are GCed because the consumer went away.
280+{-# NOINLINE mvarExcHandler #-}
281+mvarExcHandler :: Channel m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
282+mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
283+ svInfo <- dumpSVar sv
284+ hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
287+{-# NOINLINE stmExcHandler #-}
288+stmExcHandler :: Channel m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
289+stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
290+ svInfo <- dumpSVar sv
291+ hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
294+withDiagMVar :: Channel m a -> String -> IO () -> IO ()
295+withDiagMVar sv label action =
296+ if svarInspectMode sv
298+ action `catches` [ Handler (mvarExcHandler sv label)
299+ , Handler (stmExcHandler sv label)
303+printSVar :: Channel m a -> String -> IO ()
304+printSVar sv how = do
305+ svInfo <- dumpSVar sv
306+ hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo
308+-------------------------------------------------------------------------------
310+-------------------------------------------------------------------------------
312+-- Thread tracking is needed for two reasons:
314+-- 1) Killing threads on exceptions. Threads may not be left to go away by
315+-- themselves because they may run for significant times before going away or
316+-- worse they may be stuck in IO and never go away.
318+-- 2) To know when all threads are done and the stream has ended.
320+{-# NOINLINE addThread #-}
321+addThread :: MonadIO m => Channel m a -> ThreadId -> m ()
323+ liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
325+-- This is cheaper than modifyThread because we do not have to send a
326+-- outputDoorBell This can make a difference when more workers are being
328+{-# INLINE delThread #-}
329+delThread :: MonadIO m => Channel m a -> ThreadId -> m ()
331+ liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
333+-- If present then delete else add. This takes care of out of order add and
334+-- delete i.e. a delete arriving before we even added a thread.
335+-- This occurs when the forked thread is done even before the 'addThread' right
336+-- after the fork gets a chance to run.
337+{-# INLINE modifyThread #-}
338+modifyThread :: MonadIO m => Channel m a -> ThreadId -> m ()
339+modifyThread sv tid = do
340+ changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
342+ then let new = S.delete tid old in (new, new)
343+ else let new = S.insert tid old in (new, old)
344+ when (null changed) $
347+ void $ tryPutMVar (outputDoorBell sv) ()
349+-- | This is safe even if we are adding more threads concurrently because if
350+-- a child thread is adding another thread then anyway 'workerThreads' will
352+{-# INLINE allThreadsDone #-}
353+allThreadsDone :: MonadIO m => Channel m a -> m Bool
354+allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
356+-------------------------------------------------------------------------------
357+-- Dispatching workers
358+-------------------------------------------------------------------------------
360+{-# NOINLINE recordMaxWorkers #-}
361+recordMaxWorkers :: MonadIO m => Channel m a -> m ()
362+recordMaxWorkers sv = liftIO $ do
363+ active <- readIORef (workerCount sv)
364+ maxWrk <- readIORef (maxWorkers $ svarStats sv)
365+ when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
366+ modifyIORef (totalDispatches $ svarStats sv) (+1)
368+{-# NOINLINE pushWorker #-}
369+pushWorker :: (MonadIO m, MonadBaseControl IO m) => Count -> Channel m a -> m ()
370+pushWorker yieldMax sv = do
371+ liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
372+ when (svarInspectMode sv) $ recordMaxWorkers sv
373+ -- This allocation matters when significant number of workers are being
374+ -- sent. We allocate it only when needed.
376+ case yieldRateInfo sv of
377+ Nothing -> return Nothing
378+ Just _ -> liftIO $ do
379+ cntRef <- newIORef 0
380+ t <- getTime Monotonic
381+ lat <- newIORef (0, t)
382+ return $ Just WorkerInfo
383+ { workerYieldMax = yieldMax
384+ , workerYieldCount = cntRef
385+ , workerLatencyStart = lat
387+ doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
391+-- True: can dispatch more
392+-- False: cannot dispatch any more
393+dispatchWorker :: (MonadIO m, MonadBaseControl IO m) => Count -> Channel m a -> m Bool
394+dispatchWorker yieldCount sv = do
395+ let workerLimit = maxWorkerLimit sv
396+ -- XXX in case of Ahead streams we should not send more than one worker
397+ -- when the work queue is done but heap is not done.
398+ done <- liftIO $ isWorkDone sv
399+ -- Note, "done" may not mean that the work is actually finished if there
400+ -- are workers active, because there may be a worker which has not yet
401+ -- queued the leftover work.
404+ qDone <- liftIO $ isQueueDone sv
405+ -- This count may not be accurate as it is decremented by the workers
406+ -- and we have no synchronization with that decrement.
407+ active <- liftIO $ readIORef $ workerCount sv
410+ -- Note that we may deadlock if the previous workers (tasks in the
411+ -- stream) wait/depend on the future workers (tasks in the stream)
412+ -- executing. In that case we should either configure the maxWorker
413+ -- count to higher or use parallel style instead of ahead or async
415+ limit <- case remainingWork sv of
416+ Nothing -> return workerLimit
418+ n <- liftIO $ readIORef ref
419+ case yieldRateInfo sv of
420+ Just _ -> return workerLimit
424+ Unlimited -> Limited (fromIntegral n)
425+ Limited lim -> Limited $ min lim (fromIntegral n)
427+ -- XXX for ahead streams shall we take the heap yields into account
428+ -- for controlling the dispatch? We should not dispatch if the heap
429+ -- has already got the limit covered.
430+ let dispatch = pushWorker yieldCount sv >> return True
432+ Unlimited -> dispatch
433+ -- Note that the use of remainingWork and workerCount is not
434+ -- atomic and the counts may even have changed between reading
435+ -- and using them here, so this is just approximate logic and
436+ -- we cannot rely on it for correctness. We may actually
437+ -- dispatch more workers than required.
438+ Limited lim | lim > fromIntegral active -> dispatch
441+ when (active <= 0) $ pushWorker 0 sv
445+-- XXX in case of ahead style stream we need to take the heap size into account
446+-- because we return the workers on the basis of that which causes a condition
447+-- where we keep dispatching and they keep returning. So we must have exactly
448+-- the same logic for not dispatching and for returning.
451+-- True: can dispatch more
452+-- False: full, no more dispatches
453+dispatchWorkerPaced :: (MonadIO m, MonadBaseControl IO m) =>
454+ Channel m a -> m Bool
455+dispatchWorkerPaced sv = do
456+ let yinfo = fromJust $ yieldRateInfo sv
457+ (svarYields, svarElapsed, wLatency) <- do
458+ now <- liftIO $ getTime Monotonic
459+ (yieldCount, baseTime, lat) <-
460+ liftIO $ collectLatency sv yinfo False
461+ let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
464+ then fromMaybe lat (workerBootstrapLatency yinfo)
467+ return (yieldCount, elapsed, latency)
470+ -- Need to measure the latency with a single worker before we can perform
474+ let workerLimit = maxWorkerLimit sv
475+ let targetLat = svarLatencyTarget yinfo
476+ let range = svarLatencyRange yinfo
477+ gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
478+ let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
479+ wLatency targetLat range
481+ -- XXX we need to take yieldLimit into account here. If we are at the
482+ -- end of the limit as well as the time, we should not be sleeping.
483+ -- If we are not actually planning to dispatch any more workers we need
484+ -- to take that in account.
487+ assert (s >= 0) (return ())
488+ -- XXX note that when we return from here we will block waiting
489+ -- for the result from the existing worker. If that takes too
490+ -- long we won't be able to send another worker until the
493+ -- Sleep only if there are no active workers, otherwise we will
494+ -- defer the output of those. Note we cannot use workerCount
495+ -- here as it is not a reliable way to ensure there are
496+ -- definitely no active workers. When workerCount is 0 we may
497+ -- still have a Stop event waiting in the outputQueue.
498+ done <- allThreadsDone sv
499+ when done $ void $ do
500+ let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
501+ liftIO $ threadDelay (fromIntegral us)
504+ PartialWorker yields -> do
505+ assert (yields > 0) (return ())
506+ updateGainedLostYields yinfo yields
508+ done <- allThreadsDone sv
509+ when done $ void $ dispatchWorker yields sv
511+ ManyWorkers netWorkers yields -> do
512+ assert (netWorkers >= 1) (return ())
513+ assert (yields >= 0) (return ())
514+ updateGainedLostYields yinfo yields
516+ let periodRef = workerPollingInterval yinfo
517+ ycnt = max 1 $ yields `div` fromIntegral netWorkers
518+ period = min ycnt (fromIntegral magicMaxBuffer)
520+ old <- liftIO $ readIORef periodRef
521+ when (period < old) $
522+ liftIO $ writeIORef periodRef period
524+ cnt <- liftIO $ readIORef $ workerCount sv
527+ let total = netWorkers - cnt
528+ batch = max 1 $ fromIntegral $
529+ minThreadDelay `div` targetLat
530+ -- XXX stagger the workers over a period?
531+ -- XXX cannot sleep, as that would mean we cannot process
532+ -- the outputs. need to try a different mechanism to
534+ -- when (total > batch) $
535+ -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
536+ dispatchN (min total batch)
541+ updateGainedLostYields yinfo yields = do
542+ let buf = fromIntegral $ svarRateBuffer yinfo
543+ when (yields /= 0 && abs yields > buf) $ do
548+ liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
554+ r <- dispatchWorker 0 sv
556+ then dispatchN (n - 1)
559+{-# NOINLINE sendWorkerWait #-}
562+ => (Channel m a -> IO ())
563+ -> (Channel m a -> m Bool)
566+sendWorkerWait delay dispatch sv = do
567+ -- Note that we are guaranteed to have at least one outstanding worker when
568+ -- we enter this function. So if we sleep we are guaranteed to be woken up
569+ -- by an outputDoorBell, when the worker exits.
572+ (_, n) <- liftIO $ readIORef (outputQueue sv)
574+ -- The queue may be empty temporarily if the worker has dequeued the
575+ -- work item but has not enqueued the remaining part yet. For the same
576+ -- reason, a worker may come back if it tries to dequeue and finds the
577+ -- queue empty, even though the whole work has not finished yet.
579+ -- If we find that the queue is empty, but it may be empty
580+ -- temporarily, when we checked it. If that's the case we might
581+ -- sleep indefinitely unless the active workers produce some
582+ -- output. We may deadlock specially if the otuput from the active
583+ -- workers depends on the future workers that we may never send.
584+ -- So in case the queue was temporarily empty set a flag to inform
585+ -- the enqueue to send us a doorbell.
587+ -- Note that this is just a best effort mechanism to avoid a
588+ -- deadlock. Deadlocks may still happen if for some weird reason
589+ -- the consuming computation shares an MVar or some other resource
590+ -- with the producing computation and gets blocked on that resource
591+ -- and therefore cannot do any pushworker to add more threads to
592+ -- the producer. In such cases the programmer should use a parallel
593+ -- style so that all the producers are scheduled immediately and
594+ -- unconditionally. We can also use a separate monitor thread to
595+ -- push workers instead of pushing them from the consumer, but then
596+ -- we are no longer using pull based concurrency rate adaptation.
598+ -- XXX update this in the tutorial.
600+ -- Having pending active workers does not mean that we are guaranteed
601+ -- to be woken up if we sleep. In case of Ahead streams, there may be
602+ -- queued items in the heap even though the outputQueue is empty, and
603+ -- we may have active workers which are deadlocked on those items to be
604+ -- processed by the consumer. We should either guarantee that any
605+ -- worker, before returning, clears the heap or we send a worker to
606+ -- clear it. Normally we always send a worker if no output is seen, but
607+ -- if the thread limit is reached or we are using pacing then we may
608+ -- not send a worker. See the concurrentApplication test in the tests,
609+ -- that test case requires at least one yield from the producer to not
610+ -- deadlock, if the last workers output is stuck in the heap then this
611+ -- test fails. This problem can be extended to n threads when the
612+ -- consumer may depend on the evaluation of next n items in the
615+ -- register for the outputDoorBell before we check the queue so that if
616+ -- we sleep because the queue was empty we are guaranteed to get a
617+ -- doorbell on the next enqueue.
619+ liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
620+ liftIO storeLoadBarrier
621+ canDoMore <- dispatch sv
623+ -- XXX test for the case when we miss sending a worker when the worker
624+ -- count is more than 1500.
626+ -- XXX Assert here that if the heap is not empty then there is at
627+ -- least one outstanding worker. Otherwise we could be sleeping
631+ then sendWorkerWait delay dispatch sv
633+ liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
634+ $ takeMVar (outputDoorBell sv)
635+ (_, len) <- liftIO $ readIORef (outputQueue sv)
636+ when (len <= 0) $ sendWorkerWait delay dispatch sv
638+-- | Start the evaluation of the channel's work queue by kicking off a worker.
639+-- Note: Work queue must not be empty otherwise the worker will exit without
641+startChannel :: (MonadIO m, MonadBaseControl IO m) =>
643+startChannel chan = do
644+ case yieldRateInfo chan of
645+ Nothing -> pushWorker 0 chan
647+ if svarLatencyTarget yinfo == maxBound
648+ then liftIO $ threadDelay maxBound
649+ else pushWorker 1 chan
651+sendWorkerDelayPaced :: Channel m a -> IO ()
652+sendWorkerDelayPaced _ = return ()
654+sendWorkerDelay :: Channel m a -> IO ()
655+sendWorkerDelay _sv =
656+ -- XXX we need a better way to handle this than hardcoded delays. The
657+ -- delays may be different for different systems.
658+ -- If there is a usecase where this is required we can create a combinator
659+ -- to set it as a config in the state.
662+ ncpu <- getNumCapabilities
665+ if (svarStyle sv == AheadVar)
666+ then threadDelay 100
669+ if (svarStyle sv == AheadVar)
670+ then threadDelay 100