2{-# LANGUAGE KindSignatures #-}
3{-# LANGUAGE ConstraintKinds #-}
4{-# LANGUAGE ExistentialQuantification #-}
5{-# LANGUAGE FlexibleContexts #-}
6{-# LANGUAGE FlexibleInstances #-}
7{-# LANGUAGE GeneralizedNewtypeDeriving #-}
8{-# LANGUAGE LambdaCase #-}
9{-# LANGUAGE MagicHash #-}
10{-# LANGUAGE MultiParamTypeClasses #-}
11{-# LANGUAGE RankNTypes #-}
12{-# LANGUAGE ScopedTypeVariables #-}
13{-# LANGUAGE UnboxedTuples #-}
16-- Module : Streamly.SVar
17-- Copyright : (c) 2017 Harendra Kumar
20-- Maintainer : harendra.kumar@gmail.com
21-- Stability : experimental
30 -- State threaded around the stream
48 , cleanupSVarFromWorker
56 , atomicModifyIORefCAS
74 , HeapDequeueResult(..)
91 , decrementYieldLimitPost
108import Control.Concurrent
109 (ThreadId, myThreadId, threadDelay, throwTo)
110import Control.Concurrent.MVar
111 (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar, tryReadMVar)
112import Control.Exception
113 (SomeException(..), catch, mask, assert, Exception, catches,
114 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
115 BlockedIndefinitelyOnSTM(..))
116import Control.Monad (when)
117import Control.Monad.Catch (MonadThrow)
118import Control.Monad.IO.Class (MonadIO(..))
119import Control.Monad.Trans.Control (MonadBaseControl, control, StM)
121 (casIORef, readForCAS, peekTicket, atomicModifyIORefCAS_,
122 writeBarrier, storeLoadBarrier)
123import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
124import Data.Functor (void)
125import Data.Heap (Heap, Entry(..))
126import Data.Int (Int64)
128 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
129import Data.List ((\\))
130import Data.Maybe (fromJust)
131import Data.Semigroup ((<>))
133import GHC.Conc (ThreadId(..))
135import GHC.IO (IO(..))
136import System.Clock (TimeSpec, Clock(Monotonic), getTime, toNanoSecs)
137import System.IO (hPutStrLn, stderr)
138import Text.Printf (printf)
140import qualified Data.Heap as H
141import qualified Data.Set as S
143-- Always use signed arithmetic to avoid inadvertant overflows of signed values
144-- on conversion when comparing unsigned quantities with signed.
145newtype NanoSecs = NanoSecs Int64
157newtype Count = Count Int64
169------------------------------------------------------------------------------
170-- Parent child thread communication type
171------------------------------------------------------------------------------
173data ThreadAbort = ThreadAbort deriving Show
175instance Exception ThreadAbort
177-- | Events that a child thread may send to a parent thread.
180 | ChildStop ThreadId (Maybe SomeException)
182-- | Sorting out-of-turn outputs in a heap for Ahead style streams
183data AheadHeapEntry (t :: (* -> *) -> * -> *) m a =
186 | AheadEntryStream (t m a)
188------------------------------------------------------------------------------
189-- State threaded around the monad for thread management
190------------------------------------------------------------------------------
192-- | Identify the type of the SVar. Two computations using the same style can
193-- be scheduled on the same SVar.
195 AsyncVar -- depth first concurrent
196 | WAsyncVar -- breadth first concurrent
197 | ParallelVar -- all parallel
198 | AheadVar -- Concurrent look ahead
201-- | An SVar or a Stream Var is a conduit to the output from multiple streams
202-- running concurrently and asynchronously. An SVar can be thought of as an
203-- asynchronous IO handle. We can write any number of streams to an SVar in a
204-- non-blocking manner and then read them back at any time at any pace. The
205-- SVar would run the streams asynchronously and accumulate results. An SVar
206-- may not really execute the stream completely and accumulate all the results.
207-- However, it ensures that the reader can read the results at whatever paces
208-- it wants to read. The SVar monitors and adapts to the consumer's pace.
210-- An SVar is a mini scheduler, it has an associated workLoop that holds the
211-- stream tasks to be picked and run by a pool of worker threads. It has an
212-- associated output queue where the output stream elements are placed by the
213-- worker threads. A outputDoorBell is used by the worker threads to intimate the
214-- consumer thread about availability of new results in the output queue. More
215-- workers are added to the SVar by 'fromStreamVar' on demand if the output
216-- produced is not keeping pace with the consumer. On bounded SVars, workers
217-- block on the output queue to provide throttling of the producer when the
218-- consumer is not pulling fast enough. The number of workers may even get
219-- reduced depending on the consuming pace.
221-- New work is enqueued either at the time of creation of the SVar or as a
222-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
223-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
225-- We measure the individual worker latencies to estimate the number of workers
226-- needed or the amount of time we have to sleep between dispatches to achieve
227-- a particular rate when controlled pace mode it used.
228data WorkerInfo = WorkerInfo
229 { workerYieldMax :: Count -- 0 means unlimited
230 -- total number of yields by the worker till now
231 , workerYieldCount :: IORef Count
232 -- yieldCount at start, timestamp
233 , workerLatencyStart :: IORef (Count, TimeSpec)
237-- | Specifies the stream yield rate in yields per second (@Hertz@).
238-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
239-- allow only as many yields as we have accumulated as per 'rateGoal' since the
240-- start of time. If the consumer or the producer is slower or faster, the
241-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
242-- between the two by increasing or decreasing the pull rate from the producer.
243-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
244-- much as 'rateBuffer'.
246-- 'rateLow' puts a bound on how low the instantaneous rate can go when
247-- recovering the rate gap. In other words, it determines the maximum yield
248-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
249-- rate can go when recovering the rate gap. In other words, it determines the
250-- minimum yield latency. We reduce the latency by increasing concurrency,
251-- therefore we can say that it puts an upper bound on concurrency.
253-- If the 'rateGoal' is 0 or negative the stream never yields a value.
254-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
258 { rateLow :: Double -- ^ The lower rate limit
259 , rateGoal :: Double -- ^ The target rate we want to achieve
260 , rateHigh :: Double -- ^ The upper rate limit
261 , rateBuffer :: Int -- ^ Maximum slack from the goal
264data LatencyRange = LatencyRange
265 { minLatency :: NanoSecs
266 , maxLatency :: NanoSecs
270data YieldRateInfo = YieldRateInfo
271 { svarLatencyTarget :: NanoSecs
272 , svarLatencyRange :: LatencyRange
273 , svarRateBuffer :: Int
274 , svarGainedLostYields :: IORef Count
276 -- Actual latency/througput as seen from the consumer side, we count the
277 -- yields and the time it took to generates those yields. This is used to
278 -- increase or decrease the number of workers needed to achieve the desired
279 -- rate. The idle time of workers is adjusted in this, so that we only
280 -- account for the rate when the consumer actually demands data.
281 -- XXX interval latency is enough, we can move this under diagnostics build
282 , svarAllTimeLatency :: IORef (Count, TimeSpec)
284 -- XXX Worker latency specified by the user to be used before the first
285 -- actual measurement arrives. Not yet implemented
286 , workerBootstrapLatency :: Maybe NanoSecs
288 -- After how many yields the worker should update the latency information.
289 -- If the latency is high, this count is kept lower and vice-versa. XXX If
290 -- the latency suddenly becomes too high this count may remain too high for
291 -- long time, in such cases the consumer can change it.
292 -- 0 means no latency computation
293 -- XXX this is derivable from workerMeasuredLatency, can be removed.
294 , workerPollingInterval :: IORef Count
296 -- This is in progress latency stats maintained by the workers which we
297 -- empty into workerCollectedLatency stats at certain intervals - whenever
298 -- we process the stream elements yielded in this period.
299 -- (yieldCount, timeTaken)
300 , workerPendingLatency :: IORef (Count, NanoSecs)
302 -- This is the second level stat which is an accmulation from
303 -- workerPendingLatency stats. We keep accumulating latencies in this
304 -- bucket until we have stats for a sufficient period and then we reset it
305 -- to start collecting for the next period and retain the computed average
306 -- latency for the last period in workerMeasuredLatency.
307 -- (yieldCount, timeTaken)
308 , workerCollectedLatency :: IORef (Count, NanoSecs)
310 -- Latency as measured by workers, aggregated for the last period.
311 , workerMeasuredLatency :: IORef NanoSecs
314data SVarStats = SVarStats {
315 totalDispatches :: IORef Int
316 , maxWorkers :: IORef Int
317 , maxOutQSize :: IORef Int
318 , maxHeapSize :: IORef Int
319 , maxWorkQSize :: IORef Int
320 , avgWorkerLatency :: IORef (Count, NanoSecs)
321 , minWorkerLatency :: IORef NanoSecs
322 , maxWorkerLatency :: IORef NanoSecs
323 , svarStopTime :: IORef (Maybe TimeSpec)
326data Limit = Unlimited | Limited Word deriving Show
328data SVar t m a = SVar
331 svarStyle :: SVarStyle
332 , svarMrun :: RunInIO m
334 -- Shared output queue (events, length)
335 -- XXX For better efficiency we can try a preallocated array type (perhaps
336 -- something like a vector) that allows an O(1) append. That way we will
337 -- avoid constructing and reversing the list. Possibly we can also avoid
338 -- the GC copying overhead. When the size increases we should be able to
339 -- allocate the array in chunks.
340 , outputQueue :: IORef ([ChildEvent a], Int)
341 , outputDoorBell :: MVar () -- signal the consumer about output
342 , readOutputQ :: m [ChildEvent a]
343 , postProcess :: m Bool
345 -- Combined/aggregate parameters
346 , maxWorkerLimit :: Limit
347 , maxBufferLimit :: Limit
348 , remainingWork :: Maybe (IORef Count)
349 , yieldRateInfo :: Maybe YieldRateInfo
351 -- Used only by bounded SVar types
352 , enqueue :: t m a -> IO ()
353 , isWorkDone :: IO Bool
354 , isQueueDone :: IO Bool
355 , needDoorBell :: IORef Bool
356 , workLoop :: Maybe WorkerInfo -> m ()
358 -- Shared, thread tracking
359 , workerThreads :: IORef (Set ThreadId)
360 , workerCount :: IORef Int
361 , accountThread :: ThreadId -> m ()
362 , workerStopMVar :: MVar ()
364 , svarStats :: SVarStats
365 -- to track garbage collection of SVar
366 , svarRef :: Maybe (IORef ())
368 -- Only for diagnostics
369 , svarInspectMode :: Bool
370 , svarCreator :: ThreadId
371 , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))
373 -- Shared work queue (stream, seqNo)
374 , aheadWorkQueue :: IORef ([t m a], Int)
377-------------------------------------------------------------------------------
378-- State for concurrency control
379-------------------------------------------------------------------------------
381-- XXX we can put the resettable fields in a oneShotConfig field and others in
382-- a persistentConfig field. That way reset would be fast and scalable
383-- irrespective of the number of fields.
385-- XXX make all these Limited types and use phantom types to distinguish them
386data State t m a = State
387 { -- one shot configuration, automatically reset for each API call
388 streamVar :: Maybe (SVar t m a)
389 , _yieldLimit :: Maybe Count
391 -- persistent configuration, state that remains valid until changed by
392 -- an explicit setting via a combinator.
393 , _threadsHigh :: Limit
394 , _bufferHigh :: Limit
395 -- XXX these two can be collapsed into a single type
396 , _streamLatency :: Maybe NanoSecs -- bootstrap latency
397 , _maxStreamRate :: Maybe Rate
398 , _inspectMode :: Bool
401-------------------------------------------------------------------------------
402-- State defaults and reset
403-------------------------------------------------------------------------------
405-- A magical value for the buffer size arrived at by running the smallest
406-- possible task and measuring the optimal value of the buffer for that. This
407-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
409magicMaxBuffer :: Word
412defaultMaxThreads, defaultMaxBuffer :: Limit
413defaultMaxThreads = Limited magicMaxBuffer
414defaultMaxBuffer = Limited magicMaxBuffer
416-- The fields prefixed by an _ are not to be accessed or updated directly but
417-- via smart accessor APIs.
418defState :: State t m a
420 { streamVar = Nothing
421 , _yieldLimit = Nothing
422 , _threadsHigh = defaultMaxThreads
423 , _bufferHigh = defaultMaxBuffer
424 , _maxStreamRate = Nothing
425 , _streamLatency = Nothing
426 , _inspectMode = False
429-- XXX if perf gets affected we can have all the Nothing params in a single
430-- structure so that we reset is fast. We can also use rewrite rules such that
431-- reset occurs only in concurrent streams to reduce the impact on serial
433-- We can optimize this so that we clear it only if it is a Just value, it
434-- results in slightly better perf for zip/zipM but the performance of scan
435-- worsens a lot, it does not fuse.
437-- This has a side effect of clearing the SVar and yieldLimit, therefore it
438-- should not be used to convert from the same type to the same type, unless
439-- you want to clear the SVar. For clearing the SVar you should be using the
440-- appropriate unStream functions instead.
442-- | Adapt the stream state from one type to another.
443adaptState :: State t m a -> State t m b
445 { streamVar = Nothing
446 , _yieldLimit = Nothing
449-------------------------------------------------------------------------------
450-- Smart get/set routines for State
451-------------------------------------------------------------------------------
453-- Use get/set routines instead of directly accessing the State fields
454setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
455setYieldLimit lim st =
462 else Just (fromIntegral n)
465getYieldLimit :: State t m a -> Maybe Count
466getYieldLimit = _yieldLimit
468setMaxThreads :: Int -> State t m a -> State t m a
474 then defaultMaxThreads
475 else Limited (fromIntegral n)
478getMaxThreads :: State t m a -> Limit
479getMaxThreads = _threadsHigh
481setMaxBuffer :: Int -> State t m a -> State t m a
487 then defaultMaxBuffer
488 else Limited (fromIntegral n)
491getMaxBuffer :: State t m a -> Limit
492getMaxBuffer = _bufferHigh
494setStreamRate :: Maybe Rate -> State t m a -> State t m a
495setStreamRate r st = st { _maxStreamRate = r }
497getStreamRate :: State t m a -> Maybe Rate
498getStreamRate = _maxStreamRate
500setStreamLatency :: Int -> State t m a -> State t m a
501setStreamLatency n st =
502 st { _streamLatency =
505 else Just (fromIntegral n)
508getStreamLatency :: State t m a -> Maybe NanoSecs
509getStreamLatency = _streamLatency
511setInspectMode :: State t m a -> State t m a
512setInspectMode st = st { _inspectMode = True }
514getInspectMode :: State t m a -> Bool
515getInspectMode = _inspectMode
517-------------------------------------------------------------------------------
519-------------------------------------------------------------------------------
521cleanupSVar :: SVar t m a -> IO ()
523 workers <- readIORef (workerThreads sv)
524 Prelude.mapM_ (`throwTo` ThreadAbort)
527cleanupSVarFromWorker :: SVar t m a -> IO ()
528cleanupSVarFromWorker sv = do
529 workers <- readIORef (workerThreads sv)
531 mapM_ (`throwTo` ThreadAbort)
532 (S.toList workers \\ [self])
534-------------------------------------------------------------------------------
535-- Dumping the SVar for debug/diag
536-------------------------------------------------------------------------------
538-- | Convert a number of seconds to a string. The string will consist
539-- of four decimal places, followed by a short description of the time
541secs :: Double -> String
543 | k < 0 = '-' : secs (-k)
544 | k >= 1 = k `with` "s"
545 | k >= 1e-3 = (k*1e3) `with` "ms"
546#ifdef mingw32_HOST_OS
547 | k >= 1e-6 = (k*1e6) `with` "us"
549 | k >= 1e-6 = (k*1e6) `with` "μs"
551 | k >= 1e-9 = (k*1e9) `with` "ns"
552 | k >= 1e-12 = (k*1e12) `with` "ps"
553 | k >= 1e-15 = (k*1e15) `with` "fs"
554 | k >= 1e-18 = (k*1e18) `with` "as"
555 | otherwise = printf "%g s" k
556 where with (t :: Double) (u :: String)
557 | t >= 1e9 = printf "%.4g %s" t u
558 | t >= 1e3 = printf "%.0f %s" t u
559 | t >= 1e2 = printf "%.1f %s" t u
560 | t >= 1e1 = printf "%.2f %s" t u
561 | otherwise = printf "%.3f %s" t u
563-- XXX Code duplicated from collectLatency
564drainLatency :: SVar t m a -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
565drainLatency sv yinfo = do
566 let cur = workerPendingLatency yinfo
567 col = workerCollectedLatency yinfo
568 longTerm = svarAllTimeLatency yinfo
569 measured = workerMeasuredLatency yinfo
571 (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
572 (colCount, colTime) <- readIORef col
573 (lcount, ltime) <- readIORef longTerm
574 prev <- readIORef measured
576 let pendingCount = colCount + count
577 pendingTime = colTime + time
579 lcount' = lcount + pendingCount
580 notUpdated = (lcount', ltime, prev)
582 if (pendingCount > 0)
584 let new = pendingTime `div` (fromIntegral pendingCount)
585 when (svarInspectMode sv) $ do
586 let ss = svarStats sv
587 minLat <- readIORef (minWorkerLatency ss)
588 when (new < minLat || minLat == 0) $
589 writeIORef (minWorkerLatency ss) new
591 maxLat <- readIORef (maxWorkerLatency ss)
592 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
593 modifyIORef (avgWorkerLatency ss) $
594 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
595 -- To avoid minor fluctuations update in batches
596 writeIORef col (0, 0)
597 writeIORef measured new
598 modifyIORef longTerm $ \(_, t) -> (lcount', t)
599 return (lcount', ltime, new)
600 else return notUpdated
602dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
603dumpSVarStats sv ss style = do
604 case yieldRateInfo sv of
607 _ <- liftIO $ drainLatency sv yinfo
610 dispatches <- readIORef $ totalDispatches ss
611 maxWrk <- readIORef $ maxWorkers ss
612 maxOq <- readIORef $ maxOutQSize ss
613 maxHp <- readIORef $ maxHeapSize ss
614 minLat <- readIORef $ minWorkerLatency ss
615 maxLat <- readIORef $ maxWorkerLatency ss
616 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
617 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
618 Nothing -> return (0, 0, 0)
620 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
623 t <- readIORef (svarStopTime ss)
624 gl <- readIORef (svarGainedLostYields yinfo)
627 now <- getTime Monotonic
628 let interval = toNanoSecs (now - startTime)
629 return (cnt, gl, interval `div` fromIntegral cnt)
631 let interval = toNanoSecs (stopTime - startTime)
632 return (cnt, gl, interval `div` fromIntegral cnt)
633 else return (0, 0, 0)
636 [ "total dispatches = " <> show dispatches
637 , "max workers = " <> show maxWrk
638 , "max outQSize = " <> show maxOq
639 <> (if style == AheadVar
640 then "\nheap max size = " <> show maxHp
643 then "\nmin worker latency = "
644 <> secs (fromIntegral minLat * 1e-9)
647 then "\nmax worker latency = "
648 <> secs (fromIntegral maxLat * 1e-9)
651 then let lat = avgTime `div` fromIntegral avgCnt
652 in "\navg worker latency = "
653 <> secs (fromIntegral lat * 1e-9)
656 then "\nSVar latency = "
657 <> secs (fromIntegral svarLat * 1e-9)
660 then "\nSVar yield count = " <> show svarCnt
662 <> (if svarGainLossCnt > 0
663 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
667{-# NOINLINE dumpSVar #-}
668dumpSVar :: SVar t m a -> IO String
670 (oqList, oqLen) <- readIORef $ outputQueue sv
671 db <- tryReadMVar $ outputDoorBell sv
673 if svarStyle sv == AheadVar
675 (oheap, oheapSeq) <- readIORef $ outputHeap sv
676 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
678 [ "heap length = " <> show (H.size oheap)
679 , "heap seqeunce = " <> show oheapSeq
680 , "work queue length = " <> show (length wq)
681 , "work queue sequence = " <> show wqSeq
685 let style = svarStyle sv
687 if style /= ParallelVar
688 then readIORef $ needDoorBell sv
690 rthread <- readIORef $ workerThreads sv
691 workers <- readIORef $ workerCount sv
692 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
696 "Creator tid = " <> show (svarCreator sv),
697 "style = " <> show (svarStyle sv)
698 , "---------CURRENT STATE-----------"
699 , "outputQueue length computed = " <> show (length oqList)
700 , "outputQueue length maintained = " <> show oqLen
701 -- XXX print the types of events in the outputQueue, first 5
702 , "outputDoorBell = " <> show db
706 [ "needDoorBell = " <> show waiting
707 , "running threads = " <> show rthread
708 -- XXX print the status of first 5 threads
709 , "running thread count = " <> show workers
711 <> "---------STATS-----------\n"
714-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
715-- can keep it on in production to debug problems quickly if and when they
716-- happen, but it may result in unexpected output when threads are left hanging
717-- until they are GCed because the consumer went away.
719{-# NOINLINE mvarExcHandler #-}
720mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
721mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
722 svInfo <- dumpSVar sv
723 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
726{-# NOINLINE stmExcHandler #-}
727stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
728stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
729 svInfo <- dumpSVar sv
730 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
733withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
734withDiagMVar sv label action =
735 if svarInspectMode sv
737 action `catches` [ Handler (mvarExcHandler sv label)
738 , Handler (stmExcHandler sv label)
742-------------------------------------------------------------------------------
744-------------------------------------------------------------------------------
746-- Slightly faster version of CAS. Gained some improvement by avoiding the use
747-- of "evaluate" because we know we do not have exceptions in fn.
748{-# INLINE atomicModifyIORefCAS #-}
749atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
750atomicModifyIORefCAS ref fn = do
751 tkt <- readForCAS ref
757 loop _ 0 = atomicModifyIORef ref fn
759 let (new, result) = fn $ peekTicket old
760 (success, tkt) <- casIORef ref old new
763 else loop tkt (tries - 1)
765{-# INLINE ringDoorBell #-}
766ringDoorBell :: SVar t m a -> IO ()
769 w <- readIORef $ needDoorBell sv
771 -- Note: the sequence of operations is important for correctness here.
772 -- We need to set the flag to false strictly before sending the
773 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
774 -- then we may set the flag to False to later making the consumer lose
775 -- the flag, even without receiving a outputDoorBell.
776 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
777 void $ tryPutMVar (outputDoorBell sv) ()
780------------------------------------------------------------------------------
781-- Spawning threads and collecting result in streamed fashion
782------------------------------------------------------------------------------
784-- | A monad that can perform concurrent or parallel IO operations. Streams
785-- that can be composed concurrently require the underlying monad to be
789type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
791-- When we run computations concurrently, we completely isolate the state of
792-- the concurrent computations from the parent computation. The invariant is
793-- that we should never be running two concurrent computations in the same
794-- thread without using the runInIO function. Also, we should never be running
795-- a concurrent computation in the parent thread, otherwise it may affect the
796-- state of the parent which is against the defined semantics of concurrent
798newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }
800captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
801captureMonadState = control $ \run -> run (return $ RunInIO run)
803-- Stolen from the async package. The perf improvement is modest, 2% on a
804-- thread heavy benchmark (parallel composition using noop computations).
805-- A version of forkIO that does not include the outer exception
806-- handler: saves a bit of time when we will be installing our own
808{-# INLINE rawForkIO #-}
809rawForkIO :: IO () -> IO ThreadId
810rawForkIO action = IO $ \ s ->
811 case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)
814doFork :: MonadBaseControl IO m
817 -> (SomeException -> IO ())
819doFork action (RunInIO mrun) exHandler =
821 mask $ \restore -> do
822 tid <- rawForkIO $ catch (restore $ void $ mrun action)
826-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
827-- faster, along with the fields in sv required by send?
828-- XXX make it noinline
830-- XXX we may want to employ an increment and decrement in batches when the
831-- througput is high or when the cost of synchronization is high. For example
832-- if the application is distributed then inc/dec of a shared variable may be
835-- Note that we need it to be an Int type so that we have the ability to undo a
836-- decrement that takes below zero.
837{-# INLINE decrementYieldLimit #-}
838decrementYieldLimit :: SVar t m a -> IO Bool
839decrementYieldLimit sv =
840 case remainingWork sv of
841 Nothing -> return True
843 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
846-- decrementYieldLimit returns False when the old limit is 0. This one returns
847-- False when the old limit is 1.
848{-# INLINE decrementYieldLimitPost #-}
849decrementYieldLimitPost :: SVar t m a -> IO Bool
850decrementYieldLimitPost sv =
851 case remainingWork sv of
852 Nothing -> return True
854 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
857{-# INLINE incrementYieldLimit #-}
858incrementYieldLimit :: SVar t m a -> IO ()
859incrementYieldLimit sv =
860 case remainingWork sv of
862 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
864-- XXX exception safety of all atomic/MVar operations
866-- TBD Each worker can have their own queue and the consumer can empty one
867-- queue at a time, that way contention can be reduced.
869-- XXX Only yields should be counted in the buffer limit and not the Stop
872-- | This function is used by the producer threads to queue output for the
873-- consumer thread to consume. Returns whether the queue has more space.
874send :: SVar t m a -> ChildEvent a -> IO Bool
876 -- XXX can the access to outputQueue and maxBufferLimit be made faster
878 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
879 ((msg : es, n + 1), n)
881 -- The wake up must happen only after the store has finished otherwise
882 -- we can have lost wakeup problems.
884 -- Since multiple workers can try this at the same time, it is possible
885 -- that we may put a spurious MVar after the consumer has already seen
886 -- the output. But that's harmless, at worst it may cause the consumer
887 -- to read the queue again and find it empty.
888 -- The important point is that the consumer is guaranteed to receive a
889 -- doorbell if something was added to the queue after it empties it.
890 void $ tryPutMVar (outputDoorBell sv) ()
892 -- XXX we should reserve the buffer when we pick up the work from the
893 -- queue, instead of checking it here when it is too late.
894 let limit = maxBufferLimit sv
896 Unlimited -> return True
898 active <- readIORef (workerCount sv)
899 return $ len < (fromIntegral lim - active)
901workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecs))
902workerCollectLatency winfo = do
903 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
904 cnt1 <- readIORef (workerYieldCount winfo)
905 let cnt = cnt1 - cnt0
909 t1 <- getTime Monotonic
910 let period = fromInteger $ toNanoSecs (t1 - t0)
911 writeIORef (workerLatencyStart winfo) (cnt1, t1)
912 return $ Just (cnt, period)
915-- XXX There are a number of gotchas in measuring latencies.
916-- 1) We measure latencies only when a worker yields a value
917-- 2) It is possible that a stream calls the stop continuation, in which case
918-- the worker would not yield a value and we would not account that worker in
919-- latencies. Even though this case should ideally be accounted we do not
920-- account it because we cannot or do not distinguish it from the case
922-- 3) It is possible that a worker returns without yielding anything because it
923-- never got a chance to pick up work.
925-- We can fix this if we measure the latencies by counting the work items
926-- picked rather than based on the outputs yielded.
927workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
928workerUpdateLatency yinfo winfo = do
929 r <- workerCollectLatency winfo
931 Just (cnt, period) -> do
932 let ref = workerPendingLatency yinfo
933 atomicModifyIORefCAS_ ref $ \(n, t) -> (n + cnt, t + period)
936updateYieldCount :: WorkerInfo -> IO Count
937updateYieldCount winfo = do
938 cnt <- readIORef (workerYieldCount winfo)
940 writeIORef (workerYieldCount winfo) cnt1
943isBeyondMaxYield :: Count -> WorkerInfo -> Bool
944isBeyondMaxYield cnt winfo =
945 let ymax = workerYieldMax winfo
946 in ymax /= 0 && cnt >= ymax
948-- XXX we should do rate control periodically based on the total yields rather
949-- than based on the worker local yields as other workers may have yielded more
950-- and we should stop based on the aggregate yields. However, latency update
951-- period can be based on individual worker yields.
952{-# NOINLINE checkRatePeriodic #-}
953checkRatePeriodic :: SVar t m a
958checkRatePeriodic sv yinfo winfo ycnt = do
959 i <- readIORef (workerPollingInterval yinfo)
960 -- XXX use generation count to check if the interval has been updated
961 if i /= 0 && (ycnt `mod` i) == 0
963 workerUpdateLatency yinfo winfo
964 -- XXX not required for parallel streams
965 isBeyondMaxRate sv yinfo
968-- CAUTION! this also updates the yield count and therefore should be called
969-- only when we are actually yielding an element.
970{-# NOINLINE workerRateControl #-}
971workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
972workerRateControl sv yinfo winfo = do
973 cnt <- updateYieldCount winfo
974 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
975 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
977-- XXX we should do rate control here but not latency update in case of ahead
978-- streams. latency update must be done when we yield directly to outputQueue
979-- or when we yield to heap.
980{-# INLINE sendYield #-}
981sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
982sendYield sv mwinfo msg = do
987 case yieldRateInfo sv of
988 Nothing -> return True
989 Just yinfo -> workerRateControl sv yinfo winfo
990 Nothing -> return True
991 return $ r && rateLimitOk
993{-# INLINE workerStopUpdate #-}
994workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
995workerStopUpdate winfo info = do
996 i <- readIORef (workerPollingInterval info)
997 when (i /= 0) $ workerUpdateLatency info winfo
999{-# INLINABLE sendStop #-}
1000sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
1001sendStop sv mwinfo = do
1002 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
1003 case (mwinfo, yieldRateInfo sv) of
1004 (Just winfo, Just info) ->
1005 workerStopUpdate winfo info
1008 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
1010-------------------------------------------------------------------------------
1012-------------------------------------------------------------------------------
1014-- Note: For purely right associated expressions this queue should have at most
1015-- one element. It grows to more than one when we have left associcated
1016-- expressions. Large left associated compositions can grow this to a
1018{-# INLINE enqueueLIFO #-}
1019enqueueLIFO :: SVar t m a -> IORef [t m a] -> t m a -> IO ()
1020enqueueLIFO sv q m = do
1021 atomicModifyIORefCAS_ q $ \ms -> m : ms
1024-------------------------------------------------------------------------------
1026-------------------------------------------------------------------------------
1028-- XXX we can use the Ahead style sequence/heap mechanism to make the best
1029-- effort to always try to finish the streams on the left side of an expression
1030-- first as long as possible.
1032{-# INLINE enqueueFIFO #-}
1033enqueueFIFO :: SVar t m a -> LinkedQueue (t m a) -> t m a -> IO ()
1034enqueueFIFO sv q m = do
1038-------------------------------------------------------------------------------
1040-------------------------------------------------------------------------------
1042-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
1043-- but always serve them in the same order as they appear in the stream. To
1044-- implement lookahead streams efficiently we assign a sequence number to each
1045-- task when the task is picked up for execution. When the task finishes, the
1046-- output is tagged with the same sequence number and we rearrange the outputs
1047-- in sequence based on that number.
1049-- To explain the mechanism imagine that the current task at the head of the
1050-- stream has a "token" to yield to the outputQueue. The ownership of the token
1051-- is determined by the current sequence number is maintained in outputHeap.
1052-- Sequence number is assigned when a task is queued. When a thread dequeues a
1053-- task it picks up the sequence number as well and when the output is ready it
1054-- uses the sequence number to queue the output to the outputQueue.
1056-- The thread with current sequence number sends the output directly to the
1057-- outputQueue. Other threads push the output to the outputHeap. When the task
1058-- being queued on the heap is a stream of many elements we evaluate only the
1059-- first element and keep the rest of the unevaluated computation in the heap.
1060-- When such a task gets the "token" for outputQueue it evaluates and directly
1061-- yields all the elements to the outputQueue without checking for the
1064-- Note that no two outputs in the heap can have the same sequence numbers and
1065-- therefore we do not need a stable heap. We have also separated the buffer
1066-- for the current task (outputQueue) and the pending tasks (outputHeap) so
1067-- that the pending tasks cannot interfere with the current task. Note that for
1068-- a single task just the outputQueue is enough and for the case of many
1069-- threads just a heap is good enough. However we balance between these two
1070-- cases, so that both are efficient.
1072-- For bigger streams it may make sense to have separate buffers for each
1073-- stream. However, for singleton streams this may become inefficient. However,
1074-- if we do not have separate buffers, then the streams that come later in
1075-- sequence may hog the buffer, hindering the streams that are ahead. For this
1076-- reason we have a single element buffer limitation for the streams being
1077-- executed in advance.
1079-- This scheme works pretty efficiently with less than 40% extra overhead
1080-- compared to the Async streams where we do not have any kind of sequencing of
1081-- the outputs. It is especially devised so that we are most efficient when we
1082-- have short tasks and need just a single thread. Also when a thread yields
1083-- many items it can hold lockfree access to the outputQueue and do it
1086-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
1087-- that they do not hog the resources and hinder the progress of the threads in
1090-- XXX Left associated ahead expressions are expensive. We start a new SVar for
1091-- each left associative expression. The queue is used only for right
1092-- associated expression, we queue the right expression and execute the left.
1093-- Thererefore the queue never has more than one item in it.
1095-- XXX we can fix this. When we queue more than one item on the queue we can
1096-- mark the previously queued item as not-runnable. The not-runnable item is
1097-- not dequeued until the already running one has finished and at that time we
1098-- would also know the exact sequence number of the already queued item.
1100-- we can even run the already queued items but they will have to be sorted in
1101-- layers in the heap. We can use a list of heaps for that.
1102{-# INLINE enqueueAhead #-}
1103enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1104enqueueAhead sv q m = do
1105 atomicModifyIORefCAS_ q $ \ case
1106 ([], n) -> ([m], n + 1) -- increment sequence
1107 _ -> error "not empty"
1110-- enqueue without incrementing the sequence number
1111{-# INLINE reEnqueueAhead #-}
1112reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1113reEnqueueAhead sv q m = do
1114 atomicModifyIORefCAS_ q $ \ case
1115 ([], n) -> ([m], n) -- DO NOT increment sequence
1116 _ -> error "not empty"
1119-- Normally the thread that has the token should never go away. The token gets
1120-- handed over to another thread, but someone or the other has the token at any
1121-- point of time. But if the task that has the token finds that the outputQueue
1122-- is full, in that case it can go away without even handing over the token to
1123-- another thread. In that case it sets the nextSequence number in the heap its
1124-- own sequence number before going away. To handle this case, any task that
1125-- does not have the token tries to dequeue from the heap first before
1126-- dequeuing from the work queue. If it finds that the task at the top of the
1127-- heap is the one that owns the current sequence number then it grabs the
1128-- token and starts with that.
1130-- XXX instead of queueing just the head element and the remaining computation
1131-- on the heap, evaluate as many as we can and place them on the heap. But we
1132-- need to give higher priority to the lower sequence numbers so that lower
1133-- priority tasks do not fill up the heap making higher priority tasks block
1134-- due to full heap. Maybe we can have a weighted space for them in the heap.
1135-- The weight is inversely proportional to the sequence number.
1137-- XXX review for livelock
1139{-# INLINE queueEmptyAhead #-}
1140queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
1141queueEmptyAhead q = liftIO $ do
1142 (xs, _) <- readIORef q
1145{-# INLINE dequeueAhead #-}
1146dequeueAhead :: MonadIO m
1147 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
1148dequeueAhead q = liftIO $
1149 atomicModifyIORefCAS q $ \case
1150 ([], n) -> (([], n), Nothing)
1151 (x : [], n) -> (([], n), Just (x, n))
1152 _ -> error "more than one item on queue"
1154-------------------------------------------------------------------------------
1156-------------------------------------------------------------------------------
1158withIORef :: IORef a -> (a -> IO b) -> IO b
1159withIORef ref f = readIORef ref >>= f
1161atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
1162atomicModifyIORef_ ref f =
1163 atomicModifyIORef ref $ \x -> (f x, ())
1165data HeapDequeueResult t m a =
1168 | Ready (Entry Int (AheadHeapEntry t m a))
1170{-# INLINE dequeueFromHeap #-}
1172 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1173 -> IO (HeapDequeueResult t m a)
1174dequeueFromHeap hpVar =
1175 atomicModifyIORef hpVar $ \pair@(hp, snum) ->
1177 Nothing -> (pair, Clearing)
1181 Just (ent@(Entry seqNo _ev), hp') ->
1183 then ((hp', Nothing), Ready ent)
1184 else assert (seqNo >= n) (pair, Waiting n)
1185 Nothing -> (pair, Waiting n)
1187{-# INLINE dequeueFromHeapSeq #-}
1189 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1191 -> IO (HeapDequeueResult t m a)
1192dequeueFromHeapSeq hpVar i =
1193 atomicModifyIORef hpVar $ \(hp, snum) ->
1198 Just (ent@(Entry seqNo _ev), hp') ->
1200 then ((hp', Nothing), Ready ent)
1201 else assert (seqNo >= i) ((hp, Just i), Waiting i)
1202 Nothing -> ((hp, Just i), Waiting i)
1203 Just _ -> error "dequeueFromHeapSeq: unreachable"
1205heapIsSane :: Maybe Int -> Int -> Bool
1206heapIsSane snum seqNo =
1209 Just n -> seqNo >= n
1211{-# INLINE requeueOnHeapTop #-}
1213 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1214 -> Entry Int (AheadHeapEntry t m a)
1217requeueOnHeapTop hpVar ent seqNo =
1218 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1219 assert (heapIsSane snum seqNo) (H.insert ent hp, Just seqNo)
1221{-# INLINE updateHeapSeq #-}
1223 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1226updateHeapSeq hpVar seqNo =
1227 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1228 assert (heapIsSane snum seqNo) (hp, Just seqNo)
1230-------------------------------------------------------------------------------
1232-------------------------------------------------------------------------------
1234-- XXX To be implemented. Use a linked queue like WAsync and put back the
1235-- remaining computation at the back of the queue instead of the heap, and
1236-- increment the sequence number.
1238-- Thread tracking is needed for two reasons:
1240-- 1) Killing threads on exceptions. Threads may not be left to go away by
1241-- themselves because they may run for significant times before going away or
1242-- worse they may be stuck in IO and never go away.
1244-- 2) To know when all threads are done and the stream has ended.
1246{-# NOINLINE addThread #-}
1247addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1249 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
1251-- This is cheaper than modifyThread because we do not have to send a
1252-- outputDoorBell This can make a difference when more workers are being
1254{-# INLINE delThread #-}
1255delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1257 liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
1259-- If present then delete else add. This takes care of out of order add and
1260-- delete i.e. a delete arriving before we even added a thread.
1261-- This occurs when the forked thread is done even before the 'addThread' right
1262-- after the fork gets a chance to run.
1263{-# INLINE modifyThread #-}
1264modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1265modifyThread sv tid = do
1266 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1268 then let new = S.delete tid old in (new, new)
1269 else let new = S.insert tid old in (new, old)
1270 when (null changed) $
1273 void $ tryPutMVar (outputDoorBell sv) ()
1275-- | This is safe even if we are adding more threads concurrently because if
1276-- a child thread is adding another thread then anyway 'workerThreads' will
1278{-# INLINE allThreadsDone #-}
1279allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1280allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1282{-# NOINLINE handleChildException #-}
1283handleChildException :: SVar t m a -> SomeException -> IO ()
1284handleChildException sv e = do
1286 void $ send sv (ChildStop tid (Just e))
1288{-# NOINLINE recordMaxWorkers #-}
1289recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1290recordMaxWorkers sv = liftIO $ do
1291 active <- readIORef (workerCount sv)
1292 maxWrk <- readIORef (maxWorkers $ svarStats sv)
1293 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1294 modifyIORef (totalDispatches $ svarStats sv) (+1)
1296{-# NOINLINE pushWorker #-}
1297pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1298pushWorker yieldMax sv = do
1299 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1300 when (svarInspectMode sv) $ recordMaxWorkers sv
1301 -- This allocation matters when significant number of workers are being
1302 -- sent. We allocate it only when needed.
1304 case yieldRateInfo sv of
1305 Nothing -> return Nothing
1306 Just _ -> liftIO $ do
1307 cntRef <- newIORef 0
1308 t <- getTime Monotonic
1309 lat <- newIORef (0, t)
1310 return $ Just WorkerInfo
1311 { workerYieldMax = yieldMax
1312 , workerYieldCount = cntRef
1313 , workerLatencyStart = lat
1315 doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
1318-- XXX we can push the workerCount modification in accountThread and use the
1319-- same pushWorker for Parallel case as well.
1321-- | In contrast to pushWorker which always happens only from the consumer
1322-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1323-- producer side. So we need to use a thread safe modification of
1324-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1325-- using a CAS based modification.
1326{-# INLINE pushWorkerPar #-}
1329 => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1330pushWorkerPar sv wloop =
1331 if svarInspectMode sv
1333 else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
1338 {-# NOINLINE forkWithDiag #-}
1340 -- We do not use workerCount in case of ParallelVar but still there is
1341 -- no harm in maintaining it correctly.
1342 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1344 -- This allocation matters when significant number of workers are being
1345 -- sent. We allocate it only when needed. The overhead increases by 4x.
1347 case yieldRateInfo sv of
1348 Nothing -> return Nothing
1349 Just _ -> liftIO $ do
1350 cntRef <- newIORef 0
1351 t <- getTime Monotonic
1352 lat <- newIORef (0, t)
1353 return $ Just WorkerInfo
1354 { workerYieldMax = 0
1355 , workerYieldCount = cntRef
1356 , workerLatencyStart = lat
1359 doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
1363-- True: can dispatch more
1364-- False: cannot dispatch any more
1365dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1366dispatchWorker yieldCount sv = do
1367 let workerLimit = maxWorkerLimit sv
1368 -- XXX in case of Ahead streams we should not send more than one worker
1369 -- when the work queue is done but heap is not done.
1370 done <- liftIO $ isWorkDone sv
1371 -- Note, "done" may not mean that the work is actually finished if there
1372 -- are workers active, because there may be a worker which has not yet
1373 -- queued the leftover work.
1376 qDone <- liftIO $ isQueueDone sv
1377 -- Note that the worker count is only decremented during event
1378 -- processing in fromStreamVar and therefore it is safe to read and
1379 -- use it without a lock.
1380 active <- liftIO $ readIORef $ workerCount sv
1383 -- Note that we may deadlock if the previous workers (tasks in the
1384 -- stream) wait/depend on the future workers (tasks in the stream)
1385 -- executing. In that case we should either configure the maxWorker
1386 -- count to higher or use parallel style instead of ahead or async
1388 limit <- case remainingWork sv of
1389 Nothing -> return workerLimit
1391 n <- liftIO $ readIORef ref
1394 Unlimited -> Limited (fromIntegral n)
1395 Limited lim -> Limited $ min lim (fromIntegral n)
1397 -- XXX for ahead streams shall we take the heap yields into account for
1398 -- controlling the dispatch? We should not dispatch if the heap has
1399 -- already got the limit covered.
1400 let dispatch = pushWorker yieldCount sv >> return True
1402 Unlimited -> dispatch
1403 -- Note that the use of remainingWork and workerCount is not
1404 -- atomic and the counts may even have changed between reading and
1405 -- using them here, so this is just approximate logic and we cannot
1406 -- rely on it for correctness. We may actually dispatch more
1407 -- workers than required.
1408 Limited lim | lim > 0 -> dispatch
1411 when (active <= 0) $ pushWorker 0 sv
1415-- | This is a magic number and it is overloaded, and used at several places to
1418-- 1. If we have to sleep to slowdown this is the minimum period that we
1419-- accumulate before we sleep. Also, workers do not stop until this much
1420-- sleep time is accumulated.
1421-- 3. Collected latencies are computed and transferred to measured latency
1422-- after a minimum of this period.
1423minThreadDelay :: NanoSecs
1424minThreadDelay = 10^(6 :: Int)
1426-- | Another magic number! When we have to start more workers to cover up a
1427-- number of yields that we are lagging by then we cannot start one worker for
1428-- each yield because that may be a very big number and if the latency of the
1429-- workers is low these number of yields could be very high. We assume that we
1430-- run each extra worker for at least this much time.
1431rateRecoveryTime :: NanoSecs
1432rateRecoveryTime = 1000000
1434nanoToMicroSecs :: NanoSecs -> Int
1435nanoToMicroSecs s = fromIntegral s `div` 1000
1437-- We either block, or send one worker with limited yield count or one or more
1438-- workers with unlimited yield count.
1440 = BlockWait NanoSecs
1441 | PartialWorker Count
1442 | ManyWorkers Int Count
1445-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1455estimateWorkers workerLimit svarYields gainLossYields
1456 svarElapsed wLatency targetLat range =
1457 -- XXX we can have a maxEfficiency combinator as well which runs the
1458 -- producer at the maximal efficiency i.e. the number of workers are chosen
1459 -- such that the latency is minimum or within a range. Or we can call it
1460 -- maxWorkerLatency.
1463 -- How many workers do we need to acheive the required rate?
1465 -- When the workers are IO bound we can increase the throughput by
1466 -- increasing the number of workers as long as the IO device has enough
1467 -- capacity to process all the requests concurrently. If the IO
1468 -- bandwidth is saturated increasing the workers won't help. Also, if
1469 -- the CPU utilization in processing all these requests exceeds the CPU
1470 -- bandwidth, then increasing the number of workers won't help.
1472 -- When the workers are purely CPU bound, increasing the workers beyond
1473 -- the number of CPUs won't help.
1475 -- TODO - measure the CPU and IO requirements of the workers. Have a
1476 -- way to specify the max bandwidth of the underlying IO mechanism and
1477 -- use that to determine the max rate of workers, and also take the CPU
1478 -- bandwidth into account. We can also discover the IO bandwidth if we
1479 -- know that we are not CPU bound, then how much steady state rate are
1480 -- we able to acheive. Design tests for CPU bound and IO bound cases.
1482 -- Calculate how many yields are we ahead or behind to match the exact
1483 -- required rate. Based on that we increase or decrease the effective
1486 -- When the worker latency is lower than required latency we begin with
1487 -- a yield and then wait rather than first waiting and then yielding.
1488 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1489 effectiveYields = svarYields + gainLossYields
1490 deltaYields = fromIntegral targetYields - effectiveYields
1492 -- We recover the deficit by running at a higher/lower rate for a
1493 -- certain amount of time. To keep the effective rate in reasonable
1494 -- limits we use rateRecoveryTime, minLatency and maxLatency.
1495 in if deltaYields > 0
1497 let deltaYieldsFreq :: Double
1499 fromIntegral deltaYields /
1500 fromIntegral rateRecoveryTime
1501 yieldsFreq = 1.0 / fromIntegral targetLat
1502 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1503 requiredLat = NanoSecs $ round $ 1.0 / totalYieldsFreq
1504 adjustedLat = min (max requiredLat (minLatency range))
1506 in assert (adjustedLat > 0) $
1507 if wLatency <= adjustedLat
1508 then PartialWorker deltaYields
1509 else let workers = withLimit $ wLatency `div` adjustedLat
1510 limited = min workers (fromIntegral deltaYields)
1511 in ManyWorkers (fromIntegral limited) deltaYields
1513 let expectedDuration = fromIntegral effectiveYields * targetLat
1514 sleepTime = expectedDuration - svarElapsed
1515 maxSleepTime = maxLatency range - wLatency
1516 s = min sleepTime maxSleepTime
1517 in assert (sleepTime >= 0) $
1518 -- if s is less than 0 it means our maxSleepTime is less
1519 -- than the worker latency.
1520 if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
1525 Limited x -> min n (fromIntegral x)
1527-- | Get the worker latency without resetting workerPendingLatency
1528-- Returns (total yield count, base time, measured latency)
1529-- CAUTION! keep it in sync with collectLatency
1530getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1531getWorkerLatency yinfo = do
1532 let cur = workerPendingLatency yinfo
1533 col = workerCollectedLatency yinfo
1534 longTerm = svarAllTimeLatency yinfo
1535 measured = workerMeasuredLatency yinfo
1537 (count, time) <- readIORef cur
1538 (colCount, colTime) <- readIORef col
1539 (lcount, ltime) <- readIORef longTerm
1540 prev <- readIORef measured
1542 let pendingCount = colCount + count
1543 pendingTime = colTime + time
1546 then let lat = pendingTime `div` fromIntegral pendingCount
1547 -- XXX Give more weight to new?
1548 in (lat + prev) `div` 2
1550 return (lcount + pendingCount, ltime, new)
1552isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1553isBeyondMaxRate sv yinfo = do
1554 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1555 now <- getTime Monotonic
1556 let duration = fromInteger $ toNanoSecs $ now - tstamp
1557 let targetLat = svarLatencyTarget yinfo
1558 gainLoss <- readIORef (svarGainedLostYields yinfo)
1559 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1560 wLatency targetLat (svarLatencyRange yinfo)
1561 cnt <- readIORef $ workerCount sv
1562 return $ case work of
1563 -- XXX set the worker's maxYields or polling interval based on yields
1564 PartialWorker _yields -> cnt > 1
1565 ManyWorkers n _ -> cnt > n
1568-- Every once in a while workers update the latencies and check the yield rate.
1569-- They return if we are above the expected yield rate. If we check too often
1570-- it may impact performance, if we check less often we may have a stale
1571-- picture. We update every minThreadDelay but we translate that into a yield
1572-- count based on latency so that the checking overhead is little.
1574-- XXX use a generation count to indicate that the value is updated. If the
1575-- value is updated an existing worker must check it again on the next yield.
1576-- Otherwise it is possible that we may keep updating it and because of the mod
1577-- worker keeps skipping it.
1578updateWorkerPollingInterval :: YieldRateInfo -> NanoSecs -> IO ()
1579updateWorkerPollingInterval yinfo latency = do
1580 let periodRef = workerPollingInterval yinfo
1581 cnt = max 1 $ minThreadDelay `div` latency
1582 period = min cnt (fromIntegral magicMaxBuffer)
1584 writeIORef periodRef (fromIntegral period)
1586-- Returns a triple, (1) yield count since last collection, (2) the base time
1587-- when we started counting, (3) average latency in the last measurement
1588-- period. The former two are used for accurate measurement of the going rate
1589-- whereas the average is used for future estimates e.g. how many workers
1590-- should be maintained to maintain the rate.
1591-- CAUTION! keep it in sync with getWorkerLatency
1592collectLatency :: SVar t m a -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1593collectLatency sv yinfo = do
1594 let cur = workerPendingLatency yinfo
1595 col = workerCollectedLatency yinfo
1596 longTerm = svarAllTimeLatency yinfo
1597 measured = workerMeasuredLatency yinfo
1599 (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
1600 (colCount, colTime) <- readIORef col
1601 (lcount, ltime) <- readIORef longTerm
1602 prev <- readIORef measured
1604 let pendingCount = colCount + count
1605 pendingTime = colTime + time
1607 lcount' = lcount + pendingCount
1608 tripleWith lat = (lcount', ltime, lat)
1612 let new = pendingTime `div` (fromIntegral pendingCount)
1613 when (svarInspectMode sv) $ do
1614 let ss = svarStats sv
1615 minLat <- readIORef (minWorkerLatency ss)
1616 when (new < minLat || minLat == 0) $
1617 writeIORef (minWorkerLatency ss) new
1619 maxLat <- readIORef (maxWorkerLatency ss)
1620 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
1621 -- When we have collected a significant sized batch we compute the new
1622 -- latency using that batch and return the new latency, otherwise we
1623 -- return the previous latency derived from the previous batch.
1624 if (pendingCount > fromIntegral magicMaxBuffer)
1625 || (pendingTime > minThreadDelay)
1626 || (let r = fromIntegral new / fromIntegral prev :: Double
1627 in prev > 0 && (r > 2 || r < 0.5))
1630 when (svarInspectMode sv) $ do
1631 let ss = svarStats sv
1632 modifyIORef (avgWorkerLatency ss) $
1633 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
1634 updateWorkerPollingInterval yinfo (max new prev)
1635 writeIORef col (0, 0)
1636 writeIORef measured ((prev + new) `div` 2)
1637 modifyIORef longTerm $ \(_, t) -> (lcount', t)
1638 return $ tripleWith new
1640 writeIORef col (pendingCount, pendingTime)
1641 return $ tripleWith prev
1642 else return $ tripleWith prev
1644-- XXX in case of ahead style stream we need to take the heap size into account
1645-- because we return the workers on the basis of that which causes a condition
1646-- where we keep dispatching and they keep returning. So we must have exactly
1647-- the same logic for not dispatching and for returning.
1650-- True: can dispatch more
1651-- False: full, no more dispatches
1652dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1653dispatchWorkerPaced sv = do
1654 let yinfo = fromJust $ yieldRateInfo sv
1655 (svarYields, svarElapsed, wLatency) <- do
1656 now <- liftIO $ getTime Monotonic
1657 (yieldCount, baseTime, lat) <-
1658 liftIO $ collectLatency sv yinfo
1659 let elapsed = fromInteger $ toNanoSecs $ now - baseTime
1663 case workerBootstrapLatency yinfo of
1668 return (yieldCount, elapsed, latency)
1671 -- Need to measure the latency with a single worker before we can perform
1675 let workerLimit = maxWorkerLimit sv
1676 let targetLat = svarLatencyTarget yinfo
1677 let range = svarLatencyRange yinfo
1678 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1679 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
1680 wLatency targetLat range
1682 -- XXX we need to take yieldLimit into account here. If we are at the
1683 -- end of the limit as well as the time, we should not be sleeping.
1684 -- If we are not actually planning to dispatch any more workers we need
1685 -- to take that in account.
1688 assert (s >= 0) (return ())
1689 -- XXX note that when we return from here we will block waiting
1690 -- for the result from the existing worker. If that takes too
1691 -- long we won't be able to send another worker until the
1694 -- Sleep only if there are no active workers, otherwise we will
1695 -- defer the output of those. Note we cannot use workerCount
1696 -- here as it is not a reliable way to ensure there are
1697 -- definitely no active workers. When workerCount is 0 we may
1698 -- still have a Stop event waiting in the outputQueue.
1699 done <- allThreadsDone sv
1700 when done $ void $ do
1701 liftIO $ threadDelay $ nanoToMicroSecs s
1704 PartialWorker yields -> do
1705 assert (yields > 0) (return ())
1706 updateGainedLostYields yinfo yields
1708 done <- allThreadsDone sv
1709 when done $ void $ dispatchWorker yields sv
1711 ManyWorkers netWorkers yields -> do
1712 assert (netWorkers >= 1) (return ())
1713 assert (yields >= 0) (return ())
1714 updateGainedLostYields yinfo yields
1716 let periodRef = workerPollingInterval yinfo
1717 ycnt = max 1 $ yields `div` fromIntegral netWorkers
1718 period = min ycnt (fromIntegral magicMaxBuffer)
1720 old <- liftIO $ readIORef periodRef
1721 when (period < old) $
1722 liftIO $ writeIORef periodRef period
1724 cnt <- liftIO $ readIORef $ workerCount sv
1727 let total = netWorkers - cnt
1728 batch = max 1 $ fromIntegral $
1729 minThreadDelay `div` targetLat
1730 -- XXX stagger the workers over a period?
1731 -- XXX cannot sleep, as that would mean we cannot process the
1732 -- outputs. need to try a different mechanism to stagger.
1733 -- when (total > batch) $
1734 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1735 dispatchN (min total batch)
1740 updateGainedLostYields yinfo yields = do
1741 let buf = fromIntegral $ svarRateBuffer yinfo
1742 when (yields /= 0 && abs yields > buf) $ do
1747 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1753 r <- dispatchWorker 0 sv
1755 then dispatchN (n - 1)
1758sendWorkerDelayPaced :: SVar t m a -> IO ()
1759sendWorkerDelayPaced _ = return ()
1761sendWorkerDelay :: SVar t m a -> IO ()
1762sendWorkerDelay _sv =
1763 -- XXX we need a better way to handle this than hardcoded delays. The
1764 -- delays may be different for different systems.
1765 -- If there is a usecase where this is required we can create a combinator
1766 -- to set it as a config in the state.
1769 ncpu <- getNumCapabilities
1772 if (svarStyle sv == AheadVar)
1773 then threadDelay 100
1776 if (svarStyle sv == AheadVar)
1777 then threadDelay 100
1782{-# NOINLINE sendWorkerWait #-}
1785 => (SVar t m a -> IO ())
1786 -> (SVar t m a -> m Bool)
1789sendWorkerWait delay dispatch sv = do
1790 -- Note that we are guaranteed to have at least one outstanding worker when
1791 -- we enter this function. So if we sleep we are guaranteed to be woken up
1792 -- by an outputDoorBell, when the worker exits.
1795 (_, n) <- liftIO $ readIORef (outputQueue sv)
1797 -- The queue may be empty temporarily if the worker has dequeued the
1798 -- work item but has not enqueued the remaining part yet. For the same
1799 -- reason, a worker may come back if it tries to dequeue and finds the
1800 -- queue empty, even though the whole work has not finished yet.
1802 -- If we find that the queue is empty, but it may be empty
1803 -- temporarily, when we checked it. If that's the case we might
1804 -- sleep indefinitely unless the active workers produce some
1805 -- output. We may deadlock specially if the otuput from the active
1806 -- workers depends on the future workers that we may never send.
1807 -- So in case the queue was temporarily empty set a flag to inform
1808 -- the enqueue to send us a doorbell.
1810 -- Note that this is just a best effort mechanism to avoid a
1811 -- deadlock. Deadlocks may still happen if for some weird reason
1812 -- the consuming computation shares an MVar or some other resource
1813 -- with the producing computation and gets blocked on that resource
1814 -- and therefore cannot do any pushworker to add more threads to
1815 -- the producer. In such cases the programmer should use a parallel
1816 -- style so that all the producers are scheduled immediately and
1817 -- unconditionally. We can also use a separate monitor thread to
1818 -- push workers instead of pushing them from the consumer, but then
1819 -- we are no longer using pull based concurrency rate adaptation.
1821 -- XXX update this in the tutorial.
1823 -- Having pending active workers does not mean that we are guaranteed
1824 -- to be woken up if we sleep. In case of Ahead streams, there may be
1825 -- queued items in the heap even though the outputQueue is empty, and
1826 -- we may have active workers which are deadlocked on those items to be
1827 -- processed by the consumer. We should either guarantee that any
1828 -- worker, before returning, clears the heap or we send a worker to clear
1829 -- it. Normally we always send a worker if no output is seen, but if
1830 -- the thread limit is reached or we are using pacing then we may not
1831 -- send a worker. See the concurrentApplication test in the tests, that
1832 -- test case requires at least one yield from the producer to not
1833 -- deadlock, if the last workers output is stuck in the heap then this
1834 -- test fails. This problem can be extended to n threads when the
1835 -- consumer may depend on the evaluation of next n items in the
1838 -- register for the outputDoorBell before we check the queue so that if we
1839 -- sleep because the queue was empty we are guaranteed to get a
1840 -- doorbell on the next enqueue.
1842 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
1843 liftIO storeLoadBarrier
1844 canDoMore <- dispatch sv
1846 -- XXX test for the case when we miss sending a worker when the worker
1847 -- count is more than 1500.
1849 -- XXX Assert here that if the heap is not empty then there is at
1850 -- least one outstanding worker. Otherwise we could be sleeping
1854 then sendWorkerWait delay dispatch sv
1856 liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
1857 $ takeMVar (outputDoorBell sv)
1858 (_, len) <- liftIO $ readIORef (outputQueue sv)
1859 when (len <= 0) $ sendWorkerWait delay dispatch sv
1861{-# INLINE readOutputQRaw #-}
1862readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
1863readOutputQRaw sv = do
1864 (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
1865 when (svarInspectMode sv) $ do
1866 let ref = maxOutQSize $ svarStats sv
1867 oqLen <- readIORef ref
1868 when (len > oqLen) $ writeIORef ref len
1871readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1872readOutputQBounded sv = do
1873 (list, len) <- liftIO $ readOutputQRaw sv
1874 -- When there is no output seen we dispatch more workers to help
1875 -- out if there is work pending in the work queue.
1879 -- send a worker proactively, if needed, even before we start
1880 -- processing the output. This may degrade single processor
1881 -- perf but improves multi-processor, because of more
1889 cnt <- liftIO $ readIORef $ workerCount sv
1890 when (cnt <= 0) $ do
1891 done <- liftIO $ isWorkDone sv
1892 when (not done) (pushWorker 0 sv)
1894 {-# INLINE blockingRead #-}
1896 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
1897 liftIO (fst `fmap` readOutputQRaw sv)
1899readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1900readOutputQPaced sv = do
1901 (list, len) <- liftIO $ readOutputQRaw sv
1905 -- XXX send a worker proactively, if needed, even before we start
1906 -- processing the output.
1907 void $ dispatchWorkerPaced sv
1912 {-# INLINE blockingRead #-}
1914 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
1915 liftIO (fst `fmap` readOutputQRaw sv)
1917postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
1918postProcessBounded sv = do
1919 workersDone <- allThreadsDone sv
1920 -- There may still be work pending even if there are no workers pending
1921 -- because all the workers may return if the outputQueue becomes full. In
1922 -- that case send off a worker to kickstart the work again.
1924 -- Note that isWorkDone can only be safely checked if all workers are done.
1925 -- When some workers are in progress they may have decremented the yield
1926 -- Limit and later ending up incrementing it again. If we look at the yield
1927 -- limit in that window we may falsely say that it is 0 and therefore we
1931 r <- liftIO $ isWorkDone sv
1932 -- Note that we need to guarantee a worker, therefore we cannot just
1933 -- use dispatchWorker which may or may not send a worker.
1934 when (not r) (pushWorker 0 sv)
1935 -- XXX do we need to dispatch many here?
1936 -- void $ dispatchWorker sv
1940postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
1941postProcessPaced sv = do
1942 workersDone <- allThreadsDone sv
1943 -- XXX If during consumption we figure out we are getting delayed then we
1944 -- should trigger dispatch there as well. We should try to check on the
1945 -- workers after consuming every n item from the buffer?
1948 r <- liftIO $ isWorkDone sv
1950 void $ dispatchWorkerPaced sv
1951 -- Note that we need to guarantee a worker since the work is not
1952 -- finished, therefore we cannot just rely on dispatchWorkerPaced
1953 -- which may or may not send a worker.
1954 noWorker <- allThreadsDone sv
1955 when noWorker $ pushWorker 0 sv
1959getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
1960getYieldRateInfo st = do
1961 -- convert rate in Hertz to latency in Nanoseconds
1962 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
1963 case getStreamRate st of
1964 Just (Rate low goal high buf) ->
1965 let l = rateToLatency goal
1966 minl = rateToLatency high
1967 maxl = rateToLatency low
1968 in mkYieldRateInfo l (LatencyRange minl maxl) buf
1969 Nothing -> return Nothing
1973 mkYieldRateInfo latency latRange buf = do
1974 measured <- newIORef 0
1975 wcur <- newIORef (0,0)
1976 wcol <- newIORef (0,0)
1977 now <- getTime Monotonic
1978 wlong <- newIORef (0,now)
1979 period <- newIORef 1
1980 gainLoss <- newIORef (Count 0)
1982 return $ Just YieldRateInfo
1983 { svarLatencyTarget = latency
1984 , svarLatencyRange = latRange
1985 , svarRateBuffer = buf
1986 , svarGainedLostYields = gainLoss
1987 , workerBootstrapLatency = getStreamLatency st
1988 , workerPollingInterval = period
1989 , workerMeasuredLatency = measured
1990 , workerPendingLatency = wcur
1991 , workerCollectedLatency = wcol
1992 , svarAllTimeLatency = wlong
1995newSVarStats :: IO SVarStats
1998 maxWrk <- newIORef 0
2002 avgLat <- newIORef (0, NanoSecs 0)
2003 maxLat <- newIORef (NanoSecs 0)
2004 minLat <- newIORef (NanoSecs 0)
2005 stpTime <- newIORef Nothing
2008 { totalDispatches = disp
2009 , maxWorkers = maxWrk
2010 , maxOutQSize = maxOq
2011 , maxHeapSize = maxHs
2012 , maxWorkQSize = maxWq
2013 , avgWorkerLatency = avgLat
2014 , minWorkerLatency = minLat
2015 , maxWorkerLatency = maxLat
2016 , svarStopTime = stpTime
2019getAheadSVar :: MonadAsync m
2021 -> ( IORef ([t m a], Int)
2022 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2029getAheadSVar st f mrun = do
2030 outQ <- newIORef ([], 0)
2031 -- the second component of the tuple is "Nothing" when heap is being
2032 -- cleared, "Just n" when we are expecting sequence number n to arrive
2033 -- before we can start clearing the heap.
2034 outH <- newIORef (H.empty, Just 0)
2035 outQMv <- newEmptyMVar
2036 active <- newIORef 0
2037 wfw <- newIORef False
2038 running <- newIORef S.empty
2039 -- Sequence number is incremented whenever something is queued, therefore,
2040 -- first sequence number would be 0
2041 q <- newIORef ([], -1)
2042 stopMVar <- newMVar ()
2043 yl <- case getYieldLimit st of
2044 Nothing -> return Nothing
2045 Just x -> Just <$> newIORef x
2046 rateInfo <- getYieldRateInfo st
2048 stats <- newSVarStats
2051 let getSVar sv readOutput postProc = SVar
2052 { outputQueue = outQ
2053 , remainingWork = yl
2054 , maxBufferLimit = getMaxBuffer st
2055 , maxWorkerLimit = getMaxThreads st
2056 , yieldRateInfo = rateInfo
2057 , outputDoorBell = outQMv
2058 , readOutputQ = readOutput sv
2059 , postProcess = postProc sv
2060 , workerThreads = running
2061 , workLoop = f q outH st{streamVar = Just sv} sv
2062 , enqueue = enqueueAhead sv q
2063 , isWorkDone = isWorkDoneAhead sv q outH
2064 , isQueueDone = isQueueDoneAhead sv q
2065 , needDoorBell = wfw
2066 , svarStyle = AheadVar
2068 , workerCount = active
2069 , accountThread = delThread sv
2070 , workerStopMVar = stopMVar
2072 , svarInspectMode = getInspectMode st
2074 , aheadWorkQueue = q
2080 case getStreamRate st of
2081 Nothing -> getSVar sv readOutputQBounded postProcessBounded
2082 Just _ -> getSVar sv readOutputQPaced postProcessPaced
2087 {-# INLINE isQueueDoneAhead #-}
2088 isQueueDoneAhead sv q = do
2089 queueDone <- checkEmpty q
2091 case remainingWork sv of
2095 Nothing -> return False
2096 -- XXX note that yieldsDone can only be authoritative only when there
2097 -- are no workers running. If there are active workers they can
2098 -- later increment the yield count and therefore change the result.
2099 return $ yieldsDone || queueDone
2101 {-# INLINE isWorkDoneAhead #-}
2102 isWorkDoneAhead sv q ref = do
2104 (hp, _) <- readIORef ref
2105 return (H.size hp <= 0)
2106 queueDone <- isQueueDoneAhead sv q
2107 return $ heapDone && queueDone
2110 (xs, _) <- readIORef q
2113getParallelSVar :: MonadIO m => State t m a -> RunInIO m -> IO (SVar t m a)
2114getParallelSVar st mrun = do
2115 outQ <- newIORef ([], 0)
2116 outQMv <- newEmptyMVar
2117 active <- newIORef 0
2118 running <- newIORef S.empty
2119 yl <- case getYieldLimit st of
2120 Nothing -> return Nothing
2121 Just x -> Just <$> newIORef x
2122 rateInfo <- getYieldRateInfo st
2124 stats <- newSVarStats
2128 SVar { outputQueue = outQ
2129 , remainingWork = yl
2130 , maxBufferLimit = Unlimited
2131 , maxWorkerLimit = Unlimited
2132 -- Used only for diagnostics
2133 , yieldRateInfo = rateInfo
2134 , outputDoorBell = outQMv
2135 , readOutputQ = readOutputQPar sv
2136 , postProcess = allThreadsDone sv
2137 , workerThreads = running
2138 , workLoop = undefined
2139 , enqueue = undefined
2140 , isWorkDone = undefined
2141 , isQueueDone = undefined
2142 , needDoorBell = undefined
2143 , svarStyle = ParallelVar
2145 , workerCount = active
2146 , accountThread = modifyThread sv
2147 , workerStopMVar = undefined
2149 , svarInspectMode = getInspectMode st
2151 , aheadWorkQueue = undefined
2152 , outputHeap = undefined
2159 readOutputQPar sv = liftIO $ do
2160 withDiagMVar sv "readOutputQPar: doorbell"
2161 $ takeMVar (outputDoorBell sv)
2162 case yieldRateInfo sv of
2163 Nothing -> return ()
2164 Just yinfo -> void $ collectLatency sv yinfo
2165 fst `fmap` readOutputQRaw sv
2167sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
2168sendFirstWorker sv m = do
2169 -- Note: We must have all the work on the queue before sending the
2170 -- pushworker, otherwise the pushworker may exit before we even get a
2172 liftIO $ enqueue sv m
2173 case yieldRateInfo sv of
2174 Nothing -> pushWorker 0 sv
2176 if svarLatencyTarget yinfo == maxBound
2177 then liftIO $ threadDelay maxBound
2178 else pushWorker 1 sv
2181{-# INLINABLE newAheadVar #-}
2182newAheadVar :: MonadAsync m
2185 -> ( IORef ([t m a], Int)
2186 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2192newAheadVar st m wloop = do
2193 mrun <- captureMonadState
2194 sv <- liftIO $ getAheadSVar st wloop mrun
2195 sendFirstWorker sv m
2197{-# INLINABLE newParallelVar #-}
2198newParallelVar :: MonadAsync m => State t m a -> m (SVar t m a)
2199newParallelVar st = do
2200 mrun <- captureMonadState
2201 liftIO $ getParallelSVar st mrun
2203-- XXX this errors out for Parallel/Ahead SVars
2204-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
2205-- be read back from the SVar using 'fromSVar'.
2206toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
2207toStreamVar sv m = do
2208 liftIO $ enqueue sv m
2209 done <- allThreadsDone sv
2210 -- XXX This is safe only when called from the consumer thread or when no
2211 -- consumer is present. There may be a race if we are not running in the
2213 -- XXX do this only if the work queue is not empty. The work may have been
2214 -- carried out by existing workers.
2216 case yieldRateInfo sv of
2217 Nothing -> pushWorker 0 sv
2218 Just _ -> pushWorker 1 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