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
73 , HeapDequeueResult(..)
90 , decrementYieldLimitPost
106import Control.Concurrent
107 (ThreadId, myThreadId, threadDelay, throwTo)
108import Control.Concurrent.MVar
109 (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar, tryReadMVar)
110import Control.Exception
111 (SomeException(..), catch, mask, assert, Exception, catches,
112 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
113 BlockedIndefinitelyOnSTM(..))
114import Control.Monad (when)
115import Control.Monad.Catch (MonadThrow)
116import Control.Monad.IO.Class (MonadIO(..))
117import Control.Monad.Trans.Control (MonadBaseControl, control, StM)
118import Streamly.Atomics
119 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
121import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
122import Data.Functor (void)
123import Data.Heap (Heap, Entry(..))
124import Data.Int (Int64)
126 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
127import Data.List ((\\))
128import Data.Maybe (fromJust)
129import Data.Semigroup ((<>))
131import GHC.Conc (ThreadId(..))
133import GHC.IO (IO(..))
134import Streamly.Time.Clock (Clock(..), getTime)
135import Streamly.Time.Units
136 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
137 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
138import System.IO (hPutStrLn, stderr)
140import qualified Data.Heap as H
141import qualified Data.Set as S
143newtype Count = Count Int64
155------------------------------------------------------------------------------
156-- Parent child thread communication type
157------------------------------------------------------------------------------
159data ThreadAbort = ThreadAbort deriving Show
161instance Exception ThreadAbort
163-- | Events that a child thread may send to a parent thread.
166 | ChildStop ThreadId (Maybe SomeException)
168-- | Sorting out-of-turn outputs in a heap for Ahead style streams
169data AheadHeapEntry (t :: (* -> *) -> * -> *) m a =
172 | AheadEntryStream (t m a)
174------------------------------------------------------------------------------
175-- State threaded around the monad for thread management
176------------------------------------------------------------------------------
178-- | Identify the type of the SVar. Two computations using the same style can
179-- be scheduled on the same SVar.
181 AsyncVar -- depth first concurrent
182 | WAsyncVar -- breadth first concurrent
183 | ParallelVar -- all parallel
184 | AheadVar -- Concurrent look ahead
187-- | An SVar or a Stream Var is a conduit to the output from multiple streams
188-- running concurrently and asynchronously. An SVar can be thought of as an
189-- asynchronous IO handle. We can write any number of streams to an SVar in a
190-- non-blocking manner and then read them back at any time at any pace. The
191-- SVar would run the streams asynchronously and accumulate results. An SVar
192-- may not really execute the stream completely and accumulate all the results.
193-- However, it ensures that the reader can read the results at whatever paces
194-- it wants to read. The SVar monitors and adapts to the consumer's pace.
196-- An SVar is a mini scheduler, it has an associated workLoop that holds the
197-- stream tasks to be picked and run by a pool of worker threads. It has an
198-- associated output queue where the output stream elements are placed by the
199-- worker threads. A outputDoorBell is used by the worker threads to intimate the
200-- consumer thread about availability of new results in the output queue. More
201-- workers are added to the SVar by 'fromStreamVar' on demand if the output
202-- produced is not keeping pace with the consumer. On bounded SVars, workers
203-- block on the output queue to provide throttling of the producer when the
204-- consumer is not pulling fast enough. The number of workers may even get
205-- reduced depending on the consuming pace.
207-- New work is enqueued either at the time of creation of the SVar or as a
208-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
209-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
211-- We measure the individual worker latencies to estimate the number of workers
212-- needed or the amount of time we have to sleep between dispatches to achieve
213-- a particular rate when controlled pace mode it used.
214data WorkerInfo = WorkerInfo
215 { workerYieldMax :: Count -- 0 means unlimited
216 -- total number of yields by the worker till now
217 , workerYieldCount :: IORef Count
218 -- yieldCount at start, timestamp
219 , workerLatencyStart :: IORef (Count, AbsTime)
223-- | Specifies the stream yield rate in yields per second (@Hertz@).
224-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
225-- allow only as many yields as we have accumulated as per 'rateGoal' since the
226-- start of time. If the consumer or the producer is slower or faster, the
227-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
228-- between the two by increasing or decreasing the pull rate from the producer.
229-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
230-- much as 'rateBuffer'.
232-- 'rateLow' puts a bound on how low the instantaneous rate can go when
233-- recovering the rate gap. In other words, it determines the maximum yield
234-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
235-- rate can go when recovering the rate gap. In other words, it determines the
236-- minimum yield latency. We reduce the latency by increasing concurrency,
237-- therefore we can say that it puts an upper bound on concurrency.
239-- If the 'rateGoal' is 0 or negative the stream never yields a value.
240-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
244 { rateLow :: Double -- ^ The lower rate limit
245 , rateGoal :: Double -- ^ The target rate we want to achieve
246 , rateHigh :: Double -- ^ The upper rate limit
247 , rateBuffer :: Int -- ^ Maximum slack from the goal
250data LatencyRange = LatencyRange
251 { minLatency :: NanoSecond64
252 , maxLatency :: NanoSecond64
256data YieldRateInfo = YieldRateInfo
257 { svarLatencyTarget :: NanoSecond64
258 , svarLatencyRange :: LatencyRange
259 , svarRateBuffer :: Int
260 , svarGainedLostYields :: IORef Count
262 -- Actual latency/througput as seen from the consumer side, we count the
263 -- yields and the time it took to generates those yields. This is used to
264 -- increase or decrease the number of workers needed to achieve the desired
265 -- rate. The idle time of workers is adjusted in this, so that we only
266 -- account for the rate when the consumer actually demands data.
267 -- XXX interval latency is enough, we can move this under diagnostics build
268 , svarAllTimeLatency :: IORef (Count, AbsTime)
270 -- XXX Worker latency specified by the user to be used before the first
271 -- actual measurement arrives. Not yet implemented
272 , workerBootstrapLatency :: Maybe NanoSecond64
274 -- After how many yields the worker should update the latency information.
275 -- If the latency is high, this count is kept lower and vice-versa. XXX If
276 -- the latency suddenly becomes too high this count may remain too high for
277 -- long time, in such cases the consumer can change it.
278 -- 0 means no latency computation
279 -- XXX this is derivable from workerMeasuredLatency, can be removed.
280 , workerPollingInterval :: IORef Count
282 -- This is in progress latency stats maintained by the workers which we
283 -- empty into workerCollectedLatency stats at certain intervals - whenever
284 -- we process the stream elements yielded in this period. The first count
285 -- is all yields, the second count is only those yields for which the
286 -- latency was measured to be non-zero (note that if the timer resolution
287 -- is low the measured latency may be zero e.g. on JS platform).
288 -- (allYieldCount, yieldCount, timeTaken)
289 , workerPendingLatency :: IORef (Count, Count, NanoSecond64)
291 -- This is the second level stat which is an accmulation from
292 -- workerPendingLatency stats. We keep accumulating latencies in this
293 -- bucket until we have stats for a sufficient period and then we reset it
294 -- to start collecting for the next period and retain the computed average
295 -- latency for the last period in workerMeasuredLatency.
296 -- (allYieldCount, yieldCount, timeTaken)
297 , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)
299 -- Latency as measured by workers, aggregated for the last period.
300 , workerMeasuredLatency :: IORef NanoSecond64
303data SVarStats = SVarStats {
304 totalDispatches :: IORef Int
305 , maxWorkers :: IORef Int
306 , maxOutQSize :: IORef Int
307 , maxHeapSize :: IORef Int
308 , maxWorkQSize :: IORef Int
309 , avgWorkerLatency :: IORef (Count, NanoSecond64)
310 , minWorkerLatency :: IORef NanoSecond64
311 , maxWorkerLatency :: IORef NanoSecond64
312 , svarStopTime :: IORef (Maybe AbsTime)
315data Limit = Unlimited | Limited Word deriving Show
317data SVar t m a = SVar
320 svarStyle :: SVarStyle
321 , svarMrun :: RunInIO m
323 -- Shared output queue (events, length)
324 -- XXX For better efficiency we can try a preallocated array type (perhaps
325 -- something like a vector) that allows an O(1) append. That way we will
326 -- avoid constructing and reversing the list. Possibly we can also avoid
327 -- the GC copying overhead. When the size increases we should be able to
328 -- allocate the array in chunks.
329 , outputQueue :: IORef ([ChildEvent a], Int)
330 , outputDoorBell :: MVar () -- signal the consumer about output
331 , readOutputQ :: m [ChildEvent a]
332 , postProcess :: m Bool
334 -- Combined/aggregate parameters
335 , maxWorkerLimit :: Limit
336 , maxBufferLimit :: Limit
337 , remainingWork :: Maybe (IORef Count)
338 , yieldRateInfo :: Maybe YieldRateInfo
340 -- Used only by bounded SVar types
341 , enqueue :: t m a -> IO ()
342 , isWorkDone :: IO Bool
343 , isQueueDone :: IO Bool
344 , needDoorBell :: IORef Bool
345 , workLoop :: Maybe WorkerInfo -> m ()
347 -- Shared, thread tracking
348 , workerThreads :: IORef (Set ThreadId)
349 , workerCount :: IORef Int
350 , accountThread :: ThreadId -> m ()
351 , workerStopMVar :: MVar ()
353 , svarStats :: SVarStats
354 -- to track garbage collection of SVar
355 , svarRef :: Maybe (IORef ())
357 -- Only for diagnostics
358 , svarInspectMode :: Bool
359 , svarCreator :: ThreadId
360 , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))
362 -- Shared work queue (stream, seqNo)
363 , aheadWorkQueue :: IORef ([t m a], Int)
366-------------------------------------------------------------------------------
367-- State for concurrency control
368-------------------------------------------------------------------------------
370-- XXX we can put the resettable fields in a oneShotConfig field and others in
371-- a persistentConfig field. That way reset would be fast and scalable
372-- irrespective of the number of fields.
374-- XXX make all these Limited types and use phantom types to distinguish them
375data State t m a = State
376 { -- one shot configuration, automatically reset for each API call
377 streamVar :: Maybe (SVar t m a)
378 , _yieldLimit :: Maybe Count
380 -- persistent configuration, state that remains valid until changed by
381 -- an explicit setting via a combinator.
382 , _threadsHigh :: Limit
383 , _bufferHigh :: Limit
384 -- XXX these two can be collapsed into a single type
385 , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency
386 , _maxStreamRate :: Maybe Rate
387 , _inspectMode :: Bool
390-------------------------------------------------------------------------------
391-- State defaults and reset
392-------------------------------------------------------------------------------
394-- A magical value for the buffer size arrived at by running the smallest
395-- possible task and measuring the optimal value of the buffer for that. This
396-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
398magicMaxBuffer :: Word
401defaultMaxThreads, defaultMaxBuffer :: Limit
402defaultMaxThreads = Limited magicMaxBuffer
403defaultMaxBuffer = Limited magicMaxBuffer
405-- The fields prefixed by an _ are not to be accessed or updated directly but
406-- via smart accessor APIs.
407defState :: State t m a
409 { streamVar = Nothing
410 , _yieldLimit = Nothing
411 , _threadsHigh = defaultMaxThreads
412 , _bufferHigh = defaultMaxBuffer
413 , _maxStreamRate = Nothing
414 , _streamLatency = Nothing
415 , _inspectMode = False
418-- XXX if perf gets affected we can have all the Nothing params in a single
419-- structure so that we reset is fast. We can also use rewrite rules such that
420-- reset occurs only in concurrent streams to reduce the impact on serial
422-- We can optimize this so that we clear it only if it is a Just value, it
423-- results in slightly better perf for zip/zipM but the performance of scan
424-- worsens a lot, it does not fuse.
426-- XXX This has a side effect of clearing the SVar and yieldLimit, therefore it
427-- should not be used to convert from the same type to the same type, unless
428-- you want to clear the SVar. For clearing the SVar you should be using the
429-- appropriate unStream functions instead.
431-- | Adapt the stream state from one type to another.
432adaptState :: State t m a -> State t m b
434 { streamVar = Nothing
435 , _yieldLimit = Nothing
438-------------------------------------------------------------------------------
439-- Smart get/set routines for State
440-------------------------------------------------------------------------------
442-- Use get/set routines instead of directly accessing the State fields
443setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
444setYieldLimit lim st =
451 else Just (fromIntegral n)
454getYieldLimit :: State t m a -> Maybe Count
455getYieldLimit = _yieldLimit
457setMaxThreads :: Int -> State t m a -> State t m a
463 then defaultMaxThreads
464 else Limited (fromIntegral n)
467getMaxThreads :: State t m a -> Limit
468getMaxThreads = _threadsHigh
470setMaxBuffer :: Int -> State t m a -> State t m a
476 then defaultMaxBuffer
477 else Limited (fromIntegral n)
480getMaxBuffer :: State t m a -> Limit
481getMaxBuffer = _bufferHigh
483setStreamRate :: Maybe Rate -> State t m a -> State t m a
484setStreamRate r st = st { _maxStreamRate = r }
486getStreamRate :: State t m a -> Maybe Rate
487getStreamRate = _maxStreamRate
489setStreamLatency :: Int -> State t m a -> State t m a
490setStreamLatency n st =
491 st { _streamLatency =
494 else Just (fromIntegral n)
497getStreamLatency :: State t m a -> Maybe NanoSecond64
498getStreamLatency = _streamLatency
500setInspectMode :: State t m a -> State t m a
501setInspectMode st = st { _inspectMode = True }
503getInspectMode :: State t m a -> Bool
504getInspectMode = _inspectMode
506-------------------------------------------------------------------------------
508-------------------------------------------------------------------------------
510cleanupSVar :: SVar t m a -> IO ()
512 workers <- readIORef (workerThreads sv)
513 Prelude.mapM_ (`throwTo` ThreadAbort)
516cleanupSVarFromWorker :: SVar t m a -> IO ()
517cleanupSVarFromWorker sv = do
518 workers <- readIORef (workerThreads sv)
520 mapM_ (`throwTo` ThreadAbort)
521 (S.toList workers \\ [self])
523-------------------------------------------------------------------------------
524-- Worker latency data collection
525-------------------------------------------------------------------------------
527-- Every once in a while workers update the latencies and check the yield rate.
528-- They return if we are above the expected yield rate. If we check too often
529-- it may impact performance, if we check less often we may have a stale
530-- picture. We update every minThreadDelay but we translate that into a yield
531-- count based on latency so that the checking overhead is little.
533-- XXX use a generation count to indicate that the value is updated. If the
534-- value is updated an existing worker must check it again on the next yield.
535-- Otherwise it is possible that we may keep updating it and because of the mod
536-- worker keeps skipping it.
537updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
538updateWorkerPollingInterval yinfo latency = do
539 let periodRef = workerPollingInterval yinfo
540 cnt = max 1 $ minThreadDelay `div` latency
541 period = min cnt (fromIntegral magicMaxBuffer)
543 writeIORef periodRef (fromIntegral period)
545{-# INLINE recordMinMaxLatency #-}
546recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()
547recordMinMaxLatency sv new = do
548 let ss = svarStats sv
549 minLat <- readIORef (minWorkerLatency ss)
550 when (new < minLat || minLat == 0) $
551 writeIORef (minWorkerLatency ss) new
553 maxLat <- readIORef (maxWorkerLatency ss)
554 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
556recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()
557recordAvgLatency sv (count, time) = do
558 let ss = svarStats sv
559 modifyIORef (avgWorkerLatency ss) $
560 \(cnt, t) -> (cnt + count, t + time)
562-- Pour the pending latency stats into a collection bucket
563{-# INLINE collectWorkerPendingLatency #-}
564collectWorkerPendingLatency
565 :: IORef (Count, Count, NanoSecond64)
566 -> IORef (Count, Count, NanoSecond64)
567 -> IO (Count, Maybe (Count, NanoSecond64))
568collectWorkerPendingLatency cur col = do
569 (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
571 (fcnt, cnt, t) <- readIORef col
572 let totalCount = fcnt + fcount
573 latCount = cnt + count
575 writeIORef col (totalCount, latCount, latTime)
577 assert (latCount == 0 || latTime /= 0) (return ())
579 if latCount > 0 && latTime > 0
580 then Just $ (latCount, latTime)
582 return (totalCount, latPair)
584{-# INLINE shouldUseCollectedBatch #-}
585shouldUseCollectedBatch
591shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
592 let r = fromIntegral newLat / fromIntegral prevLat :: Double
593 in (collectedYields > fromIntegral magicMaxBuffer)
594 || (collectedTime > minThreadDelay)
595 || (prevLat > 0 && (r > 2 || r < 0.5))
598-- Returns a triple, (1) yield count since last collection, (2) the base time
599-- when we started counting, (3) average latency in the last measurement
600-- period. The former two are used for accurate measurement of the going rate
601-- whereas the average is used for future estimates e.g. how many workers
602-- should be maintained to maintain the rate.
603-- CAUTION! keep it in sync with getWorkerLatency
604collectLatency :: SVar t m a
607 -> IO (Count, AbsTime, NanoSecond64)
608collectLatency sv yinfo drain = do
609 let cur = workerPendingLatency yinfo
610 col = workerCollectedLatency yinfo
611 longTerm = svarAllTimeLatency yinfo
612 measured = workerMeasuredLatency yinfo
614 (newCount, newLatPair) <- collectWorkerPendingLatency cur col
615 (lcount, ltime) <- readIORef longTerm
616 prevLat <- readIORef measured
618 let newLcount = lcount + newCount
619 retWith lat = return (newLcount, ltime, lat)
622 Nothing -> retWith prevLat
623 Just (count, time) -> do
624 let newLat = time `div` (fromIntegral count)
625 when (svarInspectMode sv) $ recordMinMaxLatency sv newLat
626 -- When we have collected a significant sized batch we compute the
627 -- new latency using that batch and return the new latency,
628 -- otherwise we return the previous latency derived from the
630 if shouldUseCollectedBatch newCount time newLat prevLat || drain
632 -- XXX make this NOINLINE?
633 updateWorkerPollingInterval yinfo (max newLat prevLat)
634 when (svarInspectMode sv) $ recordAvgLatency sv (count, time)
635 writeIORef col (0, 0, 0)
636 writeIORef measured ((prevLat + newLat) `div` 2)
637 modifyIORef longTerm $ \(_, t) -> (newLcount, t)
641-------------------------------------------------------------------------------
642-- Dumping the SVar for debug/diag
643-------------------------------------------------------------------------------
645dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
646dumpSVarStats sv ss style = do
647 case yieldRateInfo sv of
650 _ <- liftIO $ collectLatency sv yinfo True
653 dispatches <- readIORef $ totalDispatches ss
654 maxWrk <- readIORef $ maxWorkers ss
655 maxOq <- readIORef $ maxOutQSize ss
656 maxHp <- readIORef $ maxHeapSize ss
657 minLat <- readIORef $ minWorkerLatency ss
658 maxLat <- readIORef $ maxWorkerLatency ss
659 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
660 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
661 Nothing -> return (0, 0, 0)
663 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
666 t <- readIORef (svarStopTime ss)
667 gl <- readIORef (svarGainedLostYields yinfo)
670 now <- getTime Monotonic
671 let interval = diffAbsTime64 now startTime
672 return (cnt, gl, interval `div` fromIntegral cnt)
674 let interval = diffAbsTime64 stopTime startTime
675 return (cnt, gl, interval `div` fromIntegral cnt)
676 else return (0, 0, 0)
679 [ "total dispatches = " <> show dispatches
680 , "max workers = " <> show maxWrk
681 , "max outQSize = " <> show maxOq
682 <> (if style == AheadVar
683 then "\nheap max size = " <> show maxHp
686 then "\nmin worker latency = " <> showNanoSecond64 minLat
689 then "\nmax worker latency = " <> showNanoSecond64 maxLat
692 then let lat = avgTime `div` fromIntegral avgCnt
693 in "\navg worker latency = " <> showNanoSecond64 lat
696 then "\nSVar latency = " <> showRelTime64 svarLat
699 then "\nSVar yield count = " <> show svarCnt
701 <> (if svarGainLossCnt > 0
702 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
706{-# NOINLINE dumpSVar #-}
707dumpSVar :: SVar t m a -> IO String
709 (oqList, oqLen) <- readIORef $ outputQueue sv
710 db <- tryReadMVar $ outputDoorBell sv
712 if svarStyle sv == AheadVar
714 (oheap, oheapSeq) <- readIORef $ outputHeap sv
715 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
717 [ "heap length = " <> show (H.size oheap)
718 , "heap seqeunce = " <> show oheapSeq
719 , "work queue length = " <> show (length wq)
720 , "work queue sequence = " <> show wqSeq
724 let style = svarStyle sv
726 if style /= ParallelVar
727 then readIORef $ needDoorBell sv
729 rthread <- readIORef $ workerThreads sv
730 workers <- readIORef $ workerCount sv
731 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
735 "Creator tid = " <> show (svarCreator sv),
736 "style = " <> show (svarStyle sv)
737 , "---------CURRENT STATE-----------"
738 , "outputQueue length computed = " <> show (length oqList)
739 , "outputQueue length maintained = " <> show oqLen
740 -- XXX print the types of events in the outputQueue, first 5
741 , "outputDoorBell = " <> show db
745 [ "needDoorBell = " <> show waiting
746 , "running threads = " <> show rthread
747 -- XXX print the status of first 5 threads
748 , "running thread count = " <> show workers
750 <> "---------STATS-----------\n"
753-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
754-- can keep it on in production to debug problems quickly if and when they
755-- happen, but it may result in unexpected output when threads are left hanging
756-- until they are GCed because the consumer went away.
758{-# NOINLINE mvarExcHandler #-}
759mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
760mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
761 svInfo <- dumpSVar sv
762 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
765{-# NOINLINE stmExcHandler #-}
766stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
767stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
768 svInfo <- dumpSVar sv
769 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
772withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
773withDiagMVar sv label action =
774 if svarInspectMode sv
776 action `catches` [ Handler (mvarExcHandler sv label)
777 , Handler (stmExcHandler sv label)
781------------------------------------------------------------------------------
783------------------------------------------------------------------------------
785-- | A monad that can perform concurrent or parallel IO operations. Streams
786-- that can be composed concurrently require the underlying monad to be
790type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
792-- When we run computations concurrently, we completely isolate the state of
793-- the concurrent computations from the parent computation. The invariant is
794-- that we should never be running two concurrent computations in the same
795-- thread without using the runInIO function. Also, we should never be running
796-- a concurrent computation in the parent thread, otherwise it may affect the
797-- state of the parent which is against the defined semantics of concurrent
799newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }
801captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
802captureMonadState = control $ \run -> run (return $ RunInIO run)
804-- Stolen from the async package. The perf improvement is modest, 2% on a
805-- thread heavy benchmark (parallel composition using noop computations).
806-- A version of forkIO that does not include the outer exception
807-- handler: saves a bit of time when we will be installing our own
809{-# INLINE rawForkIO #-}
810rawForkIO :: IO () -> IO ThreadId
811rawForkIO action = IO $ \ s ->
812 case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)
815doFork :: MonadBaseControl IO m
818 -> (SomeException -> IO ())
820doFork action (RunInIO mrun) exHandler =
822 mask $ \restore -> do
823 tid <- rawForkIO $ catch (restore $ void $ mrun action)
827------------------------------------------------------------------------------
828-- Collecting results from child workers in a streamed fashion
829------------------------------------------------------------------------------
831-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
832-- faster, along with the fields in sv required by send?
833-- XXX make it noinline
835-- XXX we may want to employ an increment and decrement in batches when the
836-- througput is high or when the cost of synchronization is high. For example
837-- if the application is distributed then inc/dec of a shared variable may be
840-- Note that we need it to be an Int type so that we have the ability to undo a
841-- decrement that takes below zero.
842{-# INLINE decrementYieldLimit #-}
843decrementYieldLimit :: SVar t m a -> IO Bool
844decrementYieldLimit sv =
845 case remainingWork sv of
846 Nothing -> return True
848 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
851-- decrementYieldLimit returns False when the old limit is 0. This one returns
852-- False when the old limit is 1.
853{-# INLINE decrementYieldLimitPost #-}
854decrementYieldLimitPost :: SVar t m a -> IO Bool
855decrementYieldLimitPost sv =
856 case remainingWork sv of
857 Nothing -> return True
859 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
862{-# INLINE incrementYieldLimit #-}
863incrementYieldLimit :: SVar t m a -> IO ()
864incrementYieldLimit sv =
865 case remainingWork sv of
867 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
869-- XXX exception safety of all atomic/MVar operations
871-- TBD Each worker can have their own queue and the consumer can empty one
872-- queue at a time, that way contention can be reduced.
874-- XXX Only yields should be counted in the buffer limit and not the Stop
877-- | This function is used by the producer threads to queue output for the
878-- consumer thread to consume. Returns whether the queue has more space.
879send :: SVar t m a -> ChildEvent a -> IO Bool
881 -- XXX can the access to outputQueue and maxBufferLimit be made faster
883 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
884 ((msg : es, n + 1), n)
886 -- The wake up must happen only after the store has finished otherwise
887 -- we can have lost wakeup problems.
889 -- Since multiple workers can try this at the same time, it is possible
890 -- that we may put a spurious MVar after the consumer has already seen
891 -- the output. But that's harmless, at worst it may cause the consumer
892 -- to read the queue again and find it empty.
893 -- The important point is that the consumer is guaranteed to receive a
894 -- doorbell if something was added to the queue after it empties it.
895 void $ tryPutMVar (outputDoorBell sv) ()
897 -- XXX we should reserve the buffer when we pick up the work from the
898 -- queue, instead of checking it here when it is too late.
899 let limit = maxBufferLimit sv
901 Unlimited -> return True
903 active <- readIORef (workerCount sv)
904 return $ len < (fromIntegral lim - active)
906workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
907workerCollectLatency winfo = do
908 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
909 cnt1 <- readIORef (workerYieldCount winfo)
910 let cnt = cnt1 - cnt0
914 t1 <- getTime Monotonic
915 let period = fromRelTime64 $ diffAbsTime64 t1 t0
916 writeIORef (workerLatencyStart winfo) (cnt1, t1)
917 return $ Just (cnt, period)
920-- XXX There are a number of gotchas in measuring latencies.
921-- 1) We measure latencies only when a worker yields a value
922-- 2) It is possible that a stream calls the stop continuation, in which case
923-- the worker would not yield a value and we would not account that worker in
924-- latencies. Even though this case should ideally be accounted we do not
925-- account it because we cannot or do not distinguish it from the case
927-- 3) It is possible that a worker returns without yielding anything because it
928-- never got a chance to pick up work.
929-- 4) If the system timer resolution is lower than the latency, the latency
930-- computation turns out to be zero.
932-- We can fix this if we measure the latencies by counting the work items
933-- picked rather than based on the outputs yielded.
934workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
935workerUpdateLatency yinfo winfo = do
936 r <- workerCollectLatency winfo
938 Just (cnt, period) -> do
939 -- NOTE: On JS platform the timer resolution could be pretty low. When
940 -- the timer resolution is low, measurement of latencies could be
941 -- tricky. All the worker latencies will turn out to be zero if they
942 -- are lower than the resolution. We only take into account those
943 -- measurements which are more than the timer resolution.
945 let ref = workerPendingLatency yinfo
946 (cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
947 atomicModifyIORefCAS_ ref $
948 \(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
951updateYieldCount :: WorkerInfo -> IO Count
952updateYieldCount winfo = do
953 cnt <- readIORef (workerYieldCount winfo)
955 writeIORef (workerYieldCount winfo) cnt1
958isBeyondMaxYield :: Count -> WorkerInfo -> Bool
959isBeyondMaxYield cnt winfo =
960 let ymax = workerYieldMax winfo
961 in ymax /= 0 && cnt >= ymax
963-- XXX we should do rate control periodically based on the total yields rather
964-- than based on the worker local yields as other workers may have yielded more
965-- and we should stop based on the aggregate yields. However, latency update
966-- period can be based on individual worker yields.
967{-# NOINLINE checkRatePeriodic #-}
968checkRatePeriodic :: SVar t m a
973checkRatePeriodic sv yinfo winfo ycnt = do
974 i <- readIORef (workerPollingInterval yinfo)
975 -- XXX use generation count to check if the interval has been updated
976 if i /= 0 && (ycnt `mod` i) == 0
978 workerUpdateLatency yinfo winfo
979 -- XXX not required for parallel streams
980 isBeyondMaxRate sv yinfo
983-- CAUTION! this also updates the yield count and therefore should be called
984-- only when we are actually yielding an element.
985{-# NOINLINE workerRateControl #-}
986workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
987workerRateControl sv yinfo winfo = do
988 cnt <- updateYieldCount winfo
989 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
990 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
992-- XXX we should do rate control here but not latency update in case of ahead
993-- streams. latency update must be done when we yield directly to outputQueue
994-- or when we yield to heap.
995{-# INLINE sendYield #-}
996sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
997sendYield sv mwinfo msg = do
1002 case yieldRateInfo sv of
1003 Nothing -> return True
1004 Just yinfo -> workerRateControl sv yinfo winfo
1005 Nothing -> return True
1006 return $ r && rateLimitOk
1008{-# INLINE workerStopUpdate #-}
1009workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
1010workerStopUpdate winfo info = do
1011 i <- readIORef (workerPollingInterval info)
1012 when (i /= 0) $ workerUpdateLatency info winfo
1014{-# INLINABLE sendStop #-}
1015sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
1016sendStop sv mwinfo = do
1017 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
1018 case (mwinfo, yieldRateInfo sv) of
1019 (Just winfo, Just info) ->
1020 workerStopUpdate winfo info
1023 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
1025-------------------------------------------------------------------------------
1027-------------------------------------------------------------------------------
1029{-# INLINE ringDoorBell #-}
1030ringDoorBell :: SVar t m a -> IO ()
1033 w <- readIORef $ needDoorBell sv
1035 -- Note: the sequence of operations is important for correctness here.
1036 -- We need to set the flag to false strictly before sending the
1037 -- outputDoorBell, otherwise the outputDoorBell may get processed too
1038 -- early and then we may set the flag to False to later making the
1039 -- consumer lose the flag, even without receiving a outputDoorBell.
1040 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1041 void $ tryPutMVar (outputDoorBell sv) ()
1043-------------------------------------------------------------------------------
1045-------------------------------------------------------------------------------
1047-- Note: For purely right associated expressions this queue should have at most
1048-- one element. It grows to more than one when we have left associcated
1049-- expressions. Large left associated compositions can grow this to a
1051{-# INLINE enqueueLIFO #-}
1052enqueueLIFO :: SVar t m a -> IORef [t m a] -> t m a -> IO ()
1053enqueueLIFO sv q m = do
1054 atomicModifyIORefCAS_ q $ \ms -> m : ms
1057-------------------------------------------------------------------------------
1059-------------------------------------------------------------------------------
1061-- XXX we can use the Ahead style sequence/heap mechanism to make the best
1062-- effort to always try to finish the streams on the left side of an expression
1063-- first as long as possible.
1065{-# INLINE enqueueFIFO #-}
1066enqueueFIFO :: SVar t m a -> LinkedQueue (t m a) -> t m a -> IO ()
1067enqueueFIFO sv q m = do
1071-------------------------------------------------------------------------------
1073-------------------------------------------------------------------------------
1075-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
1076-- but always serve them in the same order as they appear in the stream. To
1077-- implement lookahead streams efficiently we assign a sequence number to each
1078-- task when the task is picked up for execution. When the task finishes, the
1079-- output is tagged with the same sequence number and we rearrange the outputs
1080-- in sequence based on that number.
1082-- To explain the mechanism imagine that the current task at the head of the
1083-- stream has a "token" to yield to the outputQueue. The ownership of the token
1084-- is determined by the current sequence number is maintained in outputHeap.
1085-- Sequence number is assigned when a task is queued. When a thread dequeues a
1086-- task it picks up the sequence number as well and when the output is ready it
1087-- uses the sequence number to queue the output to the outputQueue.
1089-- The thread with current sequence number sends the output directly to the
1090-- outputQueue. Other threads push the output to the outputHeap. When the task
1091-- being queued on the heap is a stream of many elements we evaluate only the
1092-- first element and keep the rest of the unevaluated computation in the heap.
1093-- When such a task gets the "token" for outputQueue it evaluates and directly
1094-- yields all the elements to the outputQueue without checking for the
1097-- Note that no two outputs in the heap can have the same sequence numbers and
1098-- therefore we do not need a stable heap. We have also separated the buffer
1099-- for the current task (outputQueue) and the pending tasks (outputHeap) so
1100-- that the pending tasks cannot interfere with the current task. Note that for
1101-- a single task just the outputQueue is enough and for the case of many
1102-- threads just a heap is good enough. However we balance between these two
1103-- cases, so that both are efficient.
1105-- For bigger streams it may make sense to have separate buffers for each
1106-- stream. However, for singleton streams this may become inefficient. However,
1107-- if we do not have separate buffers, then the streams that come later in
1108-- sequence may hog the buffer, hindering the streams that are ahead. For this
1109-- reason we have a single element buffer limitation for the streams being
1110-- executed in advance.
1112-- This scheme works pretty efficiently with less than 40% extra overhead
1113-- compared to the Async streams where we do not have any kind of sequencing of
1114-- the outputs. It is especially devised so that we are most efficient when we
1115-- have short tasks and need just a single thread. Also when a thread yields
1116-- many items it can hold lockfree access to the outputQueue and do it
1119-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
1120-- that they do not hog the resources and hinder the progress of the threads in
1123-- XXX Left associated ahead expressions are expensive. We start a new SVar for
1124-- each left associative expression. The queue is used only for right
1125-- associated expression, we queue the right expression and execute the left.
1126-- Thererefore the queue never has more than one item in it.
1128-- XXX we can fix this. When we queue more than one item on the queue we can
1129-- mark the previously queued item as not-runnable. The not-runnable item is
1130-- not dequeued until the already running one has finished and at that time we
1131-- would also know the exact sequence number of the already queued item.
1133-- we can even run the already queued items but they will have to be sorted in
1134-- layers in the heap. We can use a list of heaps for that.
1135{-# INLINE enqueueAhead #-}
1136enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1137enqueueAhead sv q m = do
1138 atomicModifyIORefCAS_ q $ \ case
1139 ([], n) -> ([m], n + 1) -- increment sequence
1140 _ -> error "not empty"
1143-- enqueue without incrementing the sequence number
1144{-# INLINE reEnqueueAhead #-}
1145reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1146reEnqueueAhead sv q m = do
1147 atomicModifyIORefCAS_ q $ \ case
1148 ([], n) -> ([m], n) -- DO NOT increment sequence
1149 _ -> error "not empty"
1152-- Normally the thread that has the token should never go away. The token gets
1153-- handed over to another thread, but someone or the other has the token at any
1154-- point of time. But if the task that has the token finds that the outputQueue
1155-- is full, in that case it can go away without even handing over the token to
1156-- another thread. In that case it sets the nextSequence number in the heap its
1157-- own sequence number before going away. To handle this case, any task that
1158-- does not have the token tries to dequeue from the heap first before
1159-- dequeuing from the work queue. If it finds that the task at the top of the
1160-- heap is the one that owns the current sequence number then it grabs the
1161-- token and starts with that.
1163-- XXX instead of queueing just the head element and the remaining computation
1164-- on the heap, evaluate as many as we can and place them on the heap. But we
1165-- need to give higher priority to the lower sequence numbers so that lower
1166-- priority tasks do not fill up the heap making higher priority tasks block
1167-- due to full heap. Maybe we can have a weighted space for them in the heap.
1168-- The weight is inversely proportional to the sequence number.
1170-- XXX review for livelock
1172{-# INLINE queueEmptyAhead #-}
1173queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
1174queueEmptyAhead q = liftIO $ do
1175 (xs, _) <- readIORef q
1178{-# INLINE dequeueAhead #-}
1179dequeueAhead :: MonadIO m
1180 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
1181dequeueAhead q = liftIO $
1182 atomicModifyIORefCAS q $ \case
1183 ([], n) -> (([], n), Nothing)
1184 (x : [], n) -> (([], n), Just (x, n))
1185 _ -> error "more than one item on queue"
1187-------------------------------------------------------------------------------
1189-------------------------------------------------------------------------------
1191withIORef :: IORef a -> (a -> IO b) -> IO b
1192withIORef ref f = readIORef ref >>= f
1194atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
1195atomicModifyIORef_ ref f =
1196 atomicModifyIORef ref $ \x -> (f x, ())
1198data HeapDequeueResult t m a =
1201 | Ready (Entry Int (AheadHeapEntry t m a))
1203{-# INLINE dequeueFromHeap #-}
1205 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1206 -> IO (HeapDequeueResult t m a)
1207dequeueFromHeap hpVar =
1208 atomicModifyIORef hpVar $ \pair@(hp, snum) ->
1210 Nothing -> (pair, Clearing)
1214 Just (ent@(Entry seqNo _ev), hp') ->
1216 then ((hp', Nothing), Ready ent)
1217 else assert (seqNo >= n) (pair, Waiting n)
1218 Nothing -> (pair, Waiting n)
1220{-# INLINE dequeueFromHeapSeq #-}
1222 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1224 -> IO (HeapDequeueResult t m a)
1225dequeueFromHeapSeq hpVar i =
1226 atomicModifyIORef hpVar $ \(hp, snum) ->
1231 Just (ent@(Entry seqNo _ev), hp') ->
1233 then ((hp', Nothing), Ready ent)
1234 else assert (seqNo >= i) ((hp, Just i), Waiting i)
1235 Nothing -> ((hp, Just i), Waiting i)
1236 Just _ -> error "dequeueFromHeapSeq: unreachable"
1238heapIsSane :: Maybe Int -> Int -> Bool
1239heapIsSane snum seqNo =
1242 Just n -> seqNo >= n
1244{-# INLINE requeueOnHeapTop #-}
1246 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1247 -> Entry Int (AheadHeapEntry t m a)
1250requeueOnHeapTop hpVar ent seqNo =
1251 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1252 assert (heapIsSane snum seqNo) (H.insert ent hp, Just seqNo)
1254{-# INLINE updateHeapSeq #-}
1256 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1259updateHeapSeq hpVar seqNo =
1260 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1261 assert (heapIsSane snum seqNo) (hp, Just seqNo)
1263-------------------------------------------------------------------------------
1265-------------------------------------------------------------------------------
1267-- XXX To be implemented. Use a linked queue like WAsync and put back the
1268-- remaining computation at the back of the queue instead of the heap, and
1269-- increment the sequence number.
1271-------------------------------------------------------------------------------
1272-- Dispatching workers and tracking them
1273-------------------------------------------------------------------------------
1275-- Thread tracking is needed for two reasons:
1277-- 1) Killing threads on exceptions. Threads may not be left to go away by
1278-- themselves because they may run for significant times before going away or
1279-- worse they may be stuck in IO and never go away.
1281-- 2) To know when all threads are done and the stream has ended.
1283{-# NOINLINE addThread #-}
1284addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1286 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
1288-- This is cheaper than modifyThread because we do not have to send a
1289-- outputDoorBell This can make a difference when more workers are being
1291{-# INLINE delThread #-}
1292delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1294 liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
1296-- If present then delete else add. This takes care of out of order add and
1297-- delete i.e. a delete arriving before we even added a thread.
1298-- This occurs when the forked thread is done even before the 'addThread' right
1299-- after the fork gets a chance to run.
1300{-# INLINE modifyThread #-}
1301modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1302modifyThread sv tid = do
1303 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1305 then let new = S.delete tid old in (new, new)
1306 else let new = S.insert tid old in (new, old)
1307 when (null changed) $
1310 void $ tryPutMVar (outputDoorBell sv) ()
1312-- | This is safe even if we are adding more threads concurrently because if
1313-- a child thread is adding another thread then anyway 'workerThreads' will
1315{-# INLINE allThreadsDone #-}
1316allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1317allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1319{-# NOINLINE handleChildException #-}
1320handleChildException :: SVar t m a -> SomeException -> IO ()
1321handleChildException sv e = do
1323 void $ send sv (ChildStop tid (Just e))
1325{-# NOINLINE recordMaxWorkers #-}
1326recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1327recordMaxWorkers sv = liftIO $ do
1328 active <- readIORef (workerCount sv)
1329 maxWrk <- readIORef (maxWorkers $ svarStats sv)
1330 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1331 modifyIORef (totalDispatches $ svarStats sv) (+1)
1333{-# NOINLINE pushWorker #-}
1334pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1335pushWorker yieldMax sv = do
1336 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1337 when (svarInspectMode sv) $ recordMaxWorkers sv
1338 -- This allocation matters when significant number of workers are being
1339 -- sent. We allocate it only when needed.
1341 case yieldRateInfo sv of
1342 Nothing -> return Nothing
1343 Just _ -> liftIO $ do
1344 cntRef <- newIORef 0
1345 t <- getTime Monotonic
1346 lat <- newIORef (0, t)
1347 return $ Just WorkerInfo
1348 { workerYieldMax = yieldMax
1349 , workerYieldCount = cntRef
1350 , workerLatencyStart = lat
1352 doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
1355-- XXX we can push the workerCount modification in accountThread and use the
1356-- same pushWorker for Parallel case as well.
1358-- | In contrast to pushWorker which always happens only from the consumer
1359-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1360-- producer side. So we need to use a thread safe modification of
1361-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1362-- using a CAS based modification.
1363{-# INLINE pushWorkerPar #-}
1366 => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1367pushWorkerPar sv wloop =
1368 if svarInspectMode sv
1370 else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
1375 {-# NOINLINE forkWithDiag #-}
1377 -- We do not use workerCount in case of ParallelVar but still there is
1378 -- no harm in maintaining it correctly.
1379 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1381 -- This allocation matters when significant number of workers are being
1382 -- sent. We allocate it only when needed. The overhead increases by 4x.
1384 case yieldRateInfo sv of
1385 Nothing -> return Nothing
1386 Just _ -> liftIO $ do
1387 cntRef <- newIORef 0
1388 t <- getTime Monotonic
1389 lat <- newIORef (0, t)
1390 return $ Just WorkerInfo
1391 { workerYieldMax = 0
1392 , workerYieldCount = cntRef
1393 , workerLatencyStart = lat
1396 doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
1400-- True: can dispatch more
1401-- False: cannot dispatch any more
1402dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1403dispatchWorker yieldCount sv = do
1404 let workerLimit = maxWorkerLimit sv
1405 -- XXX in case of Ahead streams we should not send more than one worker
1406 -- when the work queue is done but heap is not done.
1407 done <- liftIO $ isWorkDone sv
1408 -- Note, "done" may not mean that the work is actually finished if there
1409 -- are workers active, because there may be a worker which has not yet
1410 -- queued the leftover work.
1413 qDone <- liftIO $ isQueueDone sv
1414 -- Note that the worker count is only decremented during event
1415 -- processing in fromStreamVar and therefore it is safe to read and
1416 -- use it without a lock.
1417 active <- liftIO $ readIORef $ workerCount sv
1420 -- Note that we may deadlock if the previous workers (tasks in the
1421 -- stream) wait/depend on the future workers (tasks in the stream)
1422 -- executing. In that case we should either configure the maxWorker
1423 -- count to higher or use parallel style instead of ahead or async
1425 limit <- case remainingWork sv of
1426 Nothing -> return workerLimit
1428 n <- liftIO $ readIORef ref
1429 case yieldRateInfo sv of
1430 Just _ -> return workerLimit
1434 Unlimited -> Limited (fromIntegral n)
1435 Limited lim -> Limited $ min lim (fromIntegral n)
1437 -- XXX for ahead streams shall we take the heap yields into account
1438 -- for controlling the dispatch? We should not dispatch if the heap
1439 -- has already got the limit covered.
1440 let dispatch = pushWorker yieldCount sv >> return True
1442 Unlimited -> dispatch
1443 -- Note that the use of remainingWork and workerCount is not
1444 -- atomic and the counts may even have changed between reading
1445 -- and using them here, so this is just approximate logic and
1446 -- we cannot rely on it for correctness. We may actually
1447 -- dispatch more workers than required.
1448 Limited lim | lim > fromIntegral active -> dispatch
1451 when (active <= 0) $ pushWorker 0 sv
1455-------------------------------------------------------------------------------
1456-- Dispatch workers with rate control
1457-------------------------------------------------------------------------------
1459-- | This is a magic number and it is overloaded, and used at several places to
1462-- 1. If we have to sleep to slowdown this is the minimum period that we
1463-- accumulate before we sleep. Also, workers do not stop until this much
1464-- sleep time is accumulated.
1465-- 3. Collected latencies are computed and transferred to measured latency
1466-- after a minimum of this period.
1467minThreadDelay :: NanoSecond64
1468minThreadDelay = 1000000
1470-- | Another magic number! When we have to start more workers to cover up a
1471-- number of yields that we are lagging by then we cannot start one worker for
1472-- each yield because that may be a very big number and if the latency of the
1473-- workers is low these number of yields could be very high. We assume that we
1474-- run each extra worker for at least this much time.
1475rateRecoveryTime :: NanoSecond64
1476rateRecoveryTime = 1000000
1478-- We either block, or send one worker with limited yield count or one or more
1479-- workers with unlimited yield count.
1481 = BlockWait NanoSecond64
1482 | PartialWorker Count
1483 | ManyWorkers Int Count
1486-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1496estimateWorkers workerLimit svarYields gainLossYields
1497 svarElapsed wLatency targetLat range =
1498 -- XXX we can have a maxEfficiency combinator as well which runs the
1499 -- producer at the maximal efficiency i.e. the number of workers are chosen
1500 -- such that the latency is minimum or within a range. Or we can call it
1501 -- maxWorkerLatency.
1504 -- How many workers do we need to acheive the required rate?
1506 -- When the workers are IO bound we can increase the throughput by
1507 -- increasing the number of workers as long as the IO device has enough
1508 -- capacity to process all the requests concurrently. If the IO
1509 -- bandwidth is saturated increasing the workers won't help. Also, if
1510 -- the CPU utilization in processing all these requests exceeds the CPU
1511 -- bandwidth, then increasing the number of workers won't help.
1513 -- When the workers are purely CPU bound, increasing the workers beyond
1514 -- the number of CPUs won't help.
1516 -- TODO - measure the CPU and IO requirements of the workers. Have a
1517 -- way to specify the max bandwidth of the underlying IO mechanism and
1518 -- use that to determine the max rate of workers, and also take the CPU
1519 -- bandwidth into account. We can also discover the IO bandwidth if we
1520 -- know that we are not CPU bound, then how much steady state rate are
1521 -- we able to acheive. Design tests for CPU bound and IO bound cases.
1523 -- Calculate how many yields are we ahead or behind to match the exact
1524 -- required rate. Based on that we increase or decrease the effective
1527 -- When the worker latency is lower than required latency we begin with
1528 -- a yield and then wait rather than first waiting and then yielding.
1529 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1530 effectiveYields = svarYields + gainLossYields
1531 deltaYields = fromIntegral targetYields - effectiveYields
1533 -- We recover the deficit by running at a higher/lower rate for a
1534 -- certain amount of time. To keep the effective rate in reasonable
1535 -- limits we use rateRecoveryTime, minLatency and maxLatency.
1536 in if deltaYields > 0
1538 let deltaYieldsFreq :: Double
1540 fromIntegral deltaYields /
1541 fromIntegral rateRecoveryTime
1542 yieldsFreq = 1.0 / fromIntegral targetLat
1543 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1544 requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
1545 adjustedLat = min (max requiredLat (minLatency range))
1547 in assert (adjustedLat > 0) $
1548 if wLatency <= adjustedLat
1549 then PartialWorker deltaYields
1550 else let workers = withLimit $ wLatency `div` adjustedLat
1551 limited = min workers (fromIntegral deltaYields)
1552 in ManyWorkers (fromIntegral limited) deltaYields
1554 let expectedDuration = fromIntegral effectiveYields * targetLat
1555 sleepTime = expectedDuration - svarElapsed
1556 maxSleepTime = maxLatency range - wLatency
1557 s = min sleepTime maxSleepTime
1558 in assert (sleepTime >= 0) $
1559 -- if s is less than 0 it means our maxSleepTime is less
1560 -- than the worker latency.
1561 if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
1566 Limited x -> min n (fromIntegral x)
1568-- | Get the worker latency without resetting workerPendingLatency
1569-- Returns (total yield count, base time, measured latency)
1570-- CAUTION! keep it in sync with collectLatency
1571getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
1572getWorkerLatency yinfo = do
1573 let cur = workerPendingLatency yinfo
1574 col = workerCollectedLatency yinfo
1575 longTerm = svarAllTimeLatency yinfo
1576 measured = workerMeasuredLatency yinfo
1578 (curTotalCount, curCount, curTime) <- readIORef cur
1579 (colTotalCount, colCount, colTime) <- readIORef col
1580 (lcount, ltime) <- readIORef longTerm
1581 prevLat <- readIORef measured
1583 let latCount = colCount + curCount
1584 latTime = colTime + curTime
1585 totalCount = colTotalCount + curTotalCount
1587 if latCount > 0 && latTime > 0
1588 then let lat = latTime `div` fromIntegral latCount
1589 -- XXX Give more weight to new?
1590 in (lat + prevLat) `div` 2
1592 return (lcount + totalCount, ltime, newLat)
1594isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1595isBeyondMaxRate sv yinfo = do
1596 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1597 now <- getTime Monotonic
1598 let duration = fromRelTime64 $ diffAbsTime64 now tstamp
1599 let targetLat = svarLatencyTarget yinfo
1600 gainLoss <- readIORef (svarGainedLostYields yinfo)
1601 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1602 wLatency targetLat (svarLatencyRange yinfo)
1603 cnt <- readIORef $ workerCount sv
1604 return $ case work of
1605 -- XXX set the worker's maxYields or polling interval based on yields
1606 PartialWorker _yields -> cnt > 1
1607 ManyWorkers n _ -> cnt > n
1610-- XXX in case of ahead style stream we need to take the heap size into account
1611-- because we return the workers on the basis of that which causes a condition
1612-- where we keep dispatching and they keep returning. So we must have exactly
1613-- the same logic for not dispatching and for returning.
1616-- True: can dispatch more
1617-- False: full, no more dispatches
1618dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1619dispatchWorkerPaced sv = do
1620 let yinfo = fromJust $ yieldRateInfo sv
1621 (svarYields, svarElapsed, wLatency) <- do
1622 now <- liftIO $ getTime Monotonic
1623 (yieldCount, baseTime, lat) <-
1624 liftIO $ collectLatency sv yinfo False
1625 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
1629 case workerBootstrapLatency yinfo of
1634 return (yieldCount, elapsed, latency)
1637 -- Need to measure the latency with a single worker before we can perform
1641 let workerLimit = maxWorkerLimit sv
1642 let targetLat = svarLatencyTarget yinfo
1643 let range = svarLatencyRange yinfo
1644 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1645 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
1646 wLatency targetLat range
1648 -- XXX we need to take yieldLimit into account here. If we are at the
1649 -- end of the limit as well as the time, we should not be sleeping.
1650 -- If we are not actually planning to dispatch any more workers we need
1651 -- to take that in account.
1654 assert (s >= 0) (return ())
1655 -- XXX note that when we return from here we will block waiting
1656 -- for the result from the existing worker. If that takes too
1657 -- long we won't be able to send another worker until the
1660 -- Sleep only if there are no active workers, otherwise we will
1661 -- defer the output of those. Note we cannot use workerCount
1662 -- here as it is not a reliable way to ensure there are
1663 -- definitely no active workers. When workerCount is 0 we may
1664 -- still have a Stop event waiting in the outputQueue.
1665 done <- allThreadsDone sv
1666 when done $ void $ do
1667 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
1668 liftIO $ threadDelay (fromIntegral us)
1671 PartialWorker yields -> do
1672 assert (yields > 0) (return ())
1673 updateGainedLostYields yinfo yields
1675 done <- allThreadsDone sv
1676 when done $ void $ dispatchWorker yields sv
1678 ManyWorkers netWorkers yields -> do
1679 assert (netWorkers >= 1) (return ())
1680 assert (yields >= 0) (return ())
1681 updateGainedLostYields yinfo yields
1683 let periodRef = workerPollingInterval yinfo
1684 ycnt = max 1 $ yields `div` fromIntegral netWorkers
1685 period = min ycnt (fromIntegral magicMaxBuffer)
1687 old <- liftIO $ readIORef periodRef
1688 when (period < old) $
1689 liftIO $ writeIORef periodRef period
1691 cnt <- liftIO $ readIORef $ workerCount sv
1694 let total = netWorkers - cnt
1695 batch = max 1 $ fromIntegral $
1696 minThreadDelay `div` targetLat
1697 -- XXX stagger the workers over a period?
1698 -- XXX cannot sleep, as that would mean we cannot process
1699 -- the outputs. need to try a different mechanism to
1701 -- when (total > batch) $
1702 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1703 dispatchN (min total batch)
1708 updateGainedLostYields yinfo yields = do
1709 let buf = fromIntegral $ svarRateBuffer yinfo
1710 when (yields /= 0 && abs yields > buf) $ do
1715 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1721 r <- dispatchWorker 0 sv
1723 then dispatchN (n - 1)
1726-------------------------------------------------------------------------------
1727-- Worker dispatch and wait loop
1728-------------------------------------------------------------------------------
1730sendWorkerDelayPaced :: SVar t m a -> IO ()
1731sendWorkerDelayPaced _ = return ()
1733sendWorkerDelay :: SVar t m a -> IO ()
1734sendWorkerDelay _sv =
1735 -- XXX we need a better way to handle this than hardcoded delays. The
1736 -- delays may be different for different systems.
1737 -- If there is a usecase where this is required we can create a combinator
1738 -- to set it as a config in the state.
1741 ncpu <- getNumCapabilities
1744 if (svarStyle sv == AheadVar)
1745 then threadDelay 100
1748 if (svarStyle sv == AheadVar)
1749 then threadDelay 100
1754{-# NOINLINE sendWorkerWait #-}
1757 => (SVar t m a -> IO ())
1758 -> (SVar t m a -> m Bool)
1761sendWorkerWait delay dispatch sv = do
1762 -- Note that we are guaranteed to have at least one outstanding worker when
1763 -- we enter this function. So if we sleep we are guaranteed to be woken up
1764 -- by an outputDoorBell, when the worker exits.
1767 (_, n) <- liftIO $ readIORef (outputQueue sv)
1769 -- The queue may be empty temporarily if the worker has dequeued the
1770 -- work item but has not enqueued the remaining part yet. For the same
1771 -- reason, a worker may come back if it tries to dequeue and finds the
1772 -- queue empty, even though the whole work has not finished yet.
1774 -- If we find that the queue is empty, but it may be empty
1775 -- temporarily, when we checked it. If that's the case we might
1776 -- sleep indefinitely unless the active workers produce some
1777 -- output. We may deadlock specially if the otuput from the active
1778 -- workers depends on the future workers that we may never send.
1779 -- So in case the queue was temporarily empty set a flag to inform
1780 -- the enqueue to send us a doorbell.
1782 -- Note that this is just a best effort mechanism to avoid a
1783 -- deadlock. Deadlocks may still happen if for some weird reason
1784 -- the consuming computation shares an MVar or some other resource
1785 -- with the producing computation and gets blocked on that resource
1786 -- and therefore cannot do any pushworker to add more threads to
1787 -- the producer. In such cases the programmer should use a parallel
1788 -- style so that all the producers are scheduled immediately and
1789 -- unconditionally. We can also use a separate monitor thread to
1790 -- push workers instead of pushing them from the consumer, but then
1791 -- we are no longer using pull based concurrency rate adaptation.
1793 -- XXX update this in the tutorial.
1795 -- Having pending active workers does not mean that we are guaranteed
1796 -- to be woken up if we sleep. In case of Ahead streams, there may be
1797 -- queued items in the heap even though the outputQueue is empty, and
1798 -- we may have active workers which are deadlocked on those items to be
1799 -- processed by the consumer. We should either guarantee that any
1800 -- worker, before returning, clears the heap or we send a worker to
1801 -- clear it. Normally we always send a worker if no output is seen, but
1802 -- if the thread limit is reached or we are using pacing then we may
1803 -- not send a worker. See the concurrentApplication test in the tests,
1804 -- that test case requires at least one yield from the producer to not
1805 -- deadlock, if the last workers output is stuck in the heap then this
1806 -- test fails. This problem can be extended to n threads when the
1807 -- consumer may depend on the evaluation of next n items in the
1810 -- register for the outputDoorBell before we check the queue so that if
1811 -- we sleep because the queue was empty we are guaranteed to get a
1812 -- doorbell on the next enqueue.
1814 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
1815 liftIO storeLoadBarrier
1816 canDoMore <- dispatch sv
1818 -- XXX test for the case when we miss sending a worker when the worker
1819 -- count is more than 1500.
1821 -- XXX Assert here that if the heap is not empty then there is at
1822 -- least one outstanding worker. Otherwise we could be sleeping
1826 then sendWorkerWait delay dispatch sv
1828 liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
1829 $ takeMVar (outputDoorBell sv)
1830 (_, len) <- liftIO $ readIORef (outputQueue sv)
1831 when (len <= 0) $ sendWorkerWait delay dispatch sv
1833-------------------------------------------------------------------------------
1834-- Reading from the workers' output queue/buffer
1835-------------------------------------------------------------------------------
1837{-# INLINE readOutputQRaw #-}
1838readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
1839readOutputQRaw sv = do
1840 (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
1841 when (svarInspectMode sv) $ do
1842 let ref = maxOutQSize $ svarStats sv
1843 oqLen <- readIORef ref
1844 when (len > oqLen) $ writeIORef ref len
1847readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1848readOutputQBounded sv = do
1849 (list, len) <- liftIO $ readOutputQRaw sv
1850 -- When there is no output seen we dispatch more workers to help
1851 -- out if there is work pending in the work queue.
1855 -- send a worker proactively, if needed, even before we start
1856 -- processing the output. This may degrade single processor
1857 -- perf but improves multi-processor, because of more
1865 cnt <- liftIO $ readIORef $ workerCount sv
1866 when (cnt <= 0) $ do
1867 done <- liftIO $ isWorkDone sv
1868 when (not done) (pushWorker 0 sv)
1870 {-# INLINE blockingRead #-}
1872 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
1873 liftIO (fst `fmap` readOutputQRaw sv)
1875readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1876readOutputQPaced sv = do
1877 (list, len) <- liftIO $ readOutputQRaw sv
1881 -- XXX send a worker proactively, if needed, even before we start
1882 -- processing the output.
1883 void $ dispatchWorkerPaced sv
1888 {-# INLINE blockingRead #-}
1890 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
1891 liftIO (fst `fmap` readOutputQRaw sv)
1893postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
1894postProcessBounded sv = do
1895 workersDone <- allThreadsDone sv
1896 -- There may still be work pending even if there are no workers pending
1897 -- because all the workers may return if the outputQueue becomes full. In
1898 -- that case send off a worker to kickstart the work again.
1900 -- Note that isWorkDone can only be safely checked if all workers are done.
1901 -- When some workers are in progress they may have decremented the yield
1902 -- Limit and later ending up incrementing it again. If we look at the yield
1903 -- limit in that window we may falsely say that it is 0 and therefore we
1907 r <- liftIO $ isWorkDone sv
1908 -- Note that we need to guarantee a worker, therefore we cannot just
1909 -- use dispatchWorker which may or may not send a worker.
1910 when (not r) (pushWorker 0 sv)
1911 -- XXX do we need to dispatch many here?
1912 -- void $ dispatchWorker sv
1916postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
1917postProcessPaced sv = do
1918 workersDone <- allThreadsDone sv
1919 -- XXX If during consumption we figure out we are getting delayed then we
1920 -- should trigger dispatch there as well. We should try to check on the
1921 -- workers after consuming every n item from the buffer?
1924 r <- liftIO $ isWorkDone sv
1926 void $ dispatchWorkerPaced sv
1927 -- Note that we need to guarantee a worker since the work is not
1928 -- finished, therefore we cannot just rely on dispatchWorkerPaced
1929 -- which may or may not send a worker.
1930 noWorker <- allThreadsDone sv
1931 when noWorker $ pushWorker 0 sv
1935-------------------------------------------------------------------------------
1937-------------------------------------------------------------------------------
1939getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
1940getYieldRateInfo st = do
1941 -- convert rate in Hertz to latency in Nanoseconds
1942 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
1943 case getStreamRate st of
1944 Just (Rate low goal high buf) ->
1945 let l = rateToLatency goal
1946 minl = rateToLatency high
1947 maxl = rateToLatency low
1948 in mkYieldRateInfo l (LatencyRange minl maxl) buf
1949 Nothing -> return Nothing
1953 mkYieldRateInfo latency latRange buf = do
1954 measured <- newIORef 0
1955 wcur <- newIORef (0,0,0)
1956 wcol <- newIORef (0,0,0)
1957 now <- getTime Monotonic
1958 wlong <- newIORef (0,now)
1959 period <- newIORef 1
1960 gainLoss <- newIORef (Count 0)
1962 return $ Just YieldRateInfo
1963 { svarLatencyTarget = latency
1964 , svarLatencyRange = latRange
1965 , svarRateBuffer = buf
1966 , svarGainedLostYields = gainLoss
1967 , workerBootstrapLatency = getStreamLatency st
1968 , workerPollingInterval = period
1969 , workerMeasuredLatency = measured
1970 , workerPendingLatency = wcur
1971 , workerCollectedLatency = wcol
1972 , svarAllTimeLatency = wlong
1975newSVarStats :: IO SVarStats
1978 maxWrk <- newIORef 0
1982 avgLat <- newIORef (0, NanoSecond64 0)
1983 maxLat <- newIORef (NanoSecond64 0)
1984 minLat <- newIORef (NanoSecond64 0)
1985 stpTime <- newIORef Nothing
1988 { totalDispatches = disp
1989 , maxWorkers = maxWrk
1990 , maxOutQSize = maxOq
1991 , maxHeapSize = maxHs
1992 , maxWorkQSize = maxWq
1993 , avgWorkerLatency = avgLat
1994 , minWorkerLatency = minLat
1995 , maxWorkerLatency = maxLat
1996 , svarStopTime = stpTime
1999-- XXX remove polymorphism in t, inline f
2000getAheadSVar :: MonadAsync m
2002 -> ( IORef ([t m a], Int)
2003 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2010getAheadSVar st f mrun = do
2011 outQ <- newIORef ([], 0)
2012 -- the second component of the tuple is "Nothing" when heap is being
2013 -- cleared, "Just n" when we are expecting sequence number n to arrive
2014 -- before we can start clearing the heap.
2015 outH <- newIORef (H.empty, Just 0)
2016 outQMv <- newEmptyMVar
2017 active <- newIORef 0
2018 wfw <- newIORef False
2019 running <- newIORef S.empty
2020 -- Sequence number is incremented whenever something is queued, therefore,
2021 -- first sequence number would be 0
2022 q <- newIORef ([], -1)
2023 stopMVar <- newMVar ()
2024 yl <- case getYieldLimit st of
2025 Nothing -> return Nothing
2026 Just x -> Just <$> newIORef x
2027 rateInfo <- getYieldRateInfo st
2029 stats <- newSVarStats
2032 let getSVar sv readOutput postProc = SVar
2033 { outputQueue = outQ
2034 , remainingWork = yl
2035 , maxBufferLimit = getMaxBuffer st
2036 , maxWorkerLimit = getMaxThreads st
2037 , yieldRateInfo = rateInfo
2038 , outputDoorBell = outQMv
2039 , readOutputQ = readOutput sv
2040 , postProcess = postProc sv
2041 , workerThreads = running
2042 , workLoop = f q outH st{streamVar = Just sv} sv
2043 , enqueue = enqueueAhead sv q
2044 , isWorkDone = isWorkDoneAhead sv q outH
2045 , isQueueDone = isQueueDoneAhead sv q
2046 , needDoorBell = wfw
2047 , svarStyle = AheadVar
2049 , workerCount = active
2050 , accountThread = delThread sv
2051 , workerStopMVar = stopMVar
2053 , svarInspectMode = getInspectMode st
2055 , aheadWorkQueue = q
2061 case getStreamRate st of
2062 Nothing -> getSVar sv readOutputQBounded postProcessBounded
2063 Just _ -> getSVar sv readOutputQPaced postProcessPaced
2068 {-# INLINE isQueueDoneAhead #-}
2069 isQueueDoneAhead sv q = do
2070 queueDone <- checkEmpty q
2072 case remainingWork sv of
2076 Nothing -> return False
2077 -- XXX note that yieldsDone can only be authoritative only when there
2078 -- are no workers running. If there are active workers they can
2079 -- later increment the yield count and therefore change the result.
2080 return $ yieldsDone || queueDone
2082 {-# INLINE isWorkDoneAhead #-}
2083 isWorkDoneAhead sv q ref = do
2085 (hp, _) <- readIORef ref
2086 return (H.size hp <= 0)
2087 queueDone <- isQueueDoneAhead sv q
2088 return $ heapDone && queueDone
2091 (xs, _) <- readIORef q
2094getParallelSVar :: MonadIO m => State t m a -> RunInIO m -> IO (SVar t m a)
2095getParallelSVar st mrun = do
2096 outQ <- newIORef ([], 0)
2097 outQMv <- newEmptyMVar
2098 active <- newIORef 0
2099 running <- newIORef S.empty
2100 yl <- case getYieldLimit st of
2101 Nothing -> return Nothing
2102 Just x -> Just <$> newIORef x
2103 rateInfo <- getYieldRateInfo st
2105 stats <- newSVarStats
2109 SVar { outputQueue = outQ
2110 , remainingWork = yl
2111 , maxBufferLimit = Unlimited
2112 , maxWorkerLimit = Unlimited
2113 -- Used only for diagnostics
2114 , yieldRateInfo = rateInfo
2115 , outputDoorBell = outQMv
2116 , readOutputQ = readOutputQPar sv
2117 , postProcess = allThreadsDone sv
2118 , workerThreads = running
2119 , workLoop = undefined
2120 , enqueue = undefined
2121 , isWorkDone = undefined
2122 , isQueueDone = undefined
2123 , needDoorBell = undefined
2124 , svarStyle = ParallelVar
2126 , workerCount = active
2127 , accountThread = modifyThread sv
2128 , workerStopMVar = undefined
2130 , svarInspectMode = getInspectMode st
2132 , aheadWorkQueue = undefined
2133 , outputHeap = undefined
2140 readOutputQPar sv = liftIO $ do
2141 withDiagMVar sv "readOutputQPar: doorbell"
2142 $ takeMVar (outputDoorBell sv)
2143 case yieldRateInfo sv of
2144 Nothing -> return ()
2145 Just yinfo -> void $ collectLatency sv yinfo False
2146 fst `fmap` readOutputQRaw sv
2148sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
2149sendFirstWorker sv m = do
2150 -- Note: We must have all the work on the queue before sending the
2151 -- pushworker, otherwise the pushworker may exit before we even get a
2153 liftIO $ enqueue sv m
2154 case yieldRateInfo sv of
2155 Nothing -> pushWorker 0 sv
2157 if svarLatencyTarget yinfo == maxBound
2158 then liftIO $ threadDelay maxBound
2159 else pushWorker 1 sv
2162{-# INLINABLE newAheadVar #-}
2163newAheadVar :: MonadAsync m
2166 -> ( IORef ([t m a], Int)
2167 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2173newAheadVar st m wloop = do
2174 mrun <- captureMonadState
2175 sv <- liftIO $ getAheadSVar st wloop mrun
2176 sendFirstWorker sv m
2178{-# INLINABLE newParallelVar #-}
2179newParallelVar :: MonadAsync m => State t m a -> m (SVar t m a)
2180newParallelVar st = do
2181 mrun <- captureMonadState
2182 liftIO $ getParallelSVar st mrun
2184-------------------------------------------------------------------------------
2185-- Write a stream to an SVar
2186-------------------------------------------------------------------------------
2188-- XXX this errors out for Parallel/Ahead SVars
2189-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
2190-- be read back from the SVar using 'fromSVar'.
2191toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
2192toStreamVar sv m = do
2193 liftIO $ enqueue sv m
2194 done <- allThreadsDone sv
2195 -- XXX This is safe only when called from the consumer thread or when no
2196 -- consumer is present. There may be a race if we are not running in the
2198 -- XXX do this only if the work queue is not empty. The work may have been
2199 -- carried out by existing workers.
2201 case yieldRateInfo sv of
2202 Nothing -> pushWorker 0 sv
2203 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