1{-# LANGUAGE UnboxedTuples #-}
4-- Module : Streamly.Internal.Data.SVar
5-- Copyright : (c) 2017 Composewell Technologies
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
12module Streamly.Internal.Data.SVar
19 -- State threaded around the stream
38 , cleanupSVarFromWorker
61 , handleChildException
67 , HeapDequeueResult(..)
85 , decrementBufferLimit
86 , incrementBufferLimit
108import Control.Concurrent (ThreadId, myThreadId, threadDelay, throwTo)
109import Control.Concurrent.MVar
110 (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar,
112import Control.Exception
113 (SomeException(..), assert, Exception, catches,
114 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
115 BlockedIndefinitelyOnSTM(..))
116import Control.Monad (when)
117import Control.Monad.IO.Class (MonadIO(..))
118import Control.Monad.Trans.Control (MonadBaseControl, control)
119import Streamly.Internal.Data.Atomics
120 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
122import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
123import Data.Functor (void)
124import Data.Heap (Heap, Entry(..))
125import Data.Int (Int64)
127 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
128import Data.Kind (Type)
129import Data.Maybe (fromJust, fromMaybe)
130#if __GLASGOW_HASKELL__ < 808
131import Data.Semigroup ((<>))
134import System.IO (hPutStrLn, stderr)
136import Streamly.Internal.Control.Concurrent
137 (MonadAsync, RunInIO(..), doFork, fork, forkManaged)
138import Streamly.Internal.Data.Time.Clock.Type (Clock(..), getTime)
139import Streamly.Internal.Data.Time.Units
140 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
141 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
143import qualified Data.Heap as H
144import qualified Data.Set as S
146newtype Count = Count Int64
158------------------------------------------------------------------------------
159-- Parent child thread communication type
160------------------------------------------------------------------------------
162data ThreadAbort = ThreadAbort deriving Show
164instance Exception ThreadAbort
166-- | Events that a child thread may send to a parent thread.
169 | ChildStop ThreadId (Maybe SomeException)
171-- | Sorting out-of-turn outputs in a heap for Ahead style streams
172data AheadHeapEntry (t :: (Type -> Type) -> Type -> Type) m a =
175 | AheadEntryStream (RunInIO m, t m a)
178------------------------------------------------------------------------------
179-- State threaded around the monad for thread management
180------------------------------------------------------------------------------
182-- | Identify the type of the SVar. Two computations using the same style can
183-- be scheduled on the same SVar.
185 AsyncVar -- depth first concurrent
186 | WAsyncVar -- breadth first concurrent
187 | ParallelVar -- all parallel
188 | AheadVar -- Concurrent look ahead
191-- | An SVar or a Stream Var is a conduit to the output from multiple streams
192-- running concurrently and asynchronously. An SVar can be thought of as an
193-- asynchronous IO handle. We can write any number of streams to an SVar in a
194-- non-blocking manner and then read them back at any time at any pace. The
195-- SVar would run the streams asynchronously and accumulate results. An SVar
196-- may not really execute the stream completely and accumulate all the results.
197-- However, it ensures that the reader can read the results at whatever paces
198-- it wants to read. The SVar monitors and adapts to the consumer's pace.
200-- An SVar is a mini scheduler, it has an associated workLoop that holds the
201-- stream tasks to be picked and run by a pool of worker threads. It has an
202-- associated output queue where the output stream elements are placed by the
203-- worker threads. A outputDoorBell is used by the worker threads to intimate the
204-- consumer thread about availability of new results in the output queue. More
205-- workers are added to the SVar by 'fromStreamVar' on demand if the output
206-- produced is not keeping pace with the consumer. On bounded SVars, workers
207-- block on the output queue to provide throttling of the producer when the
208-- consumer is not pulling fast enough. The number of workers may even get
209-- reduced depending on the consuming pace.
211-- New work is enqueued either at the time of creation of the SVar or as a
212-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
213-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
215-- We measure the individual worker latencies to estimate the number of workers
216-- needed or the amount of time we have to sleep between dispatches to achieve
217-- a particular rate when controlled pace mode it used.
218data WorkerInfo = WorkerInfo
219 { workerYieldMax :: Count -- 0 means unlimited
220 -- total number of yields by the worker till now
221 , workerYieldCount :: IORef Count
222 -- yieldCount at start, timestamp
223 , workerLatencyStart :: IORef (Count, AbsTime)
226-- | Specifies the stream yield rate in yields per second (@Hertz@).
227-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
228-- allow only as many yields as we have accumulated as per 'rateGoal' since the
229-- start of time. If the consumer or the producer is slower or faster, the
230-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
231-- between the two by increasing or decreasing the pull rate from the producer.
232-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
233-- much as 'rateBuffer'.
235-- 'rateLow' puts a bound on how low the instantaneous rate can go when
236-- recovering the rate gap. In other words, it determines the maximum yield
237-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
238-- rate can go when recovering the rate gap. In other words, it determines the
239-- minimum yield latency. We reduce the latency by increasing concurrency,
240-- therefore we can say that it puts an upper bound on concurrency.
242-- If the 'rateGoal' is 0 or negative the stream never yields a value.
243-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
245-- /Since: 0.5.0 ("Streamly")/
249 { rateLow :: Double -- ^ The lower rate limit
250 , rateGoal :: Double -- ^ The target rate we want to achieve
251 , rateHigh :: Double -- ^ The upper rate limit
252 , rateBuffer :: Int -- ^ Maximum slack from the goal
255data LatencyRange = LatencyRange
256 { minLatency :: NanoSecond64
257 , maxLatency :: NanoSecond64
261data YieldRateInfo = YieldRateInfo
262 { svarLatencyTarget :: NanoSecond64
263 , svarLatencyRange :: LatencyRange
264 , svarRateBuffer :: Int
266 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
267 -- read by the worker threads
268 , svarGainedLostYields :: IORef Count
270 -- Actual latency/througput as seen from the consumer side, we count the
271 -- yields and the time it took to generates those yields. This is used to
272 -- increase or decrease the number of workers needed to achieve the desired
273 -- rate. The idle time of workers is adjusted in this, so that we only
274 -- account for the rate when the consumer actually demands data.
275 -- XXX interval latency is enough, we can move this under diagnostics build
276 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
277 -- read by the worker threads
278 , svarAllTimeLatency :: IORef (Count, AbsTime)
280 -- XXX Worker latency specified by the user to be used before the first
281 -- actual measurement arrives. Not yet implemented
282 , workerBootstrapLatency :: Maybe NanoSecond64
284 -- After how many yields the worker should update the latency information.
285 -- If the latency is high, this count is kept lower and vice-versa. XXX If
286 -- the latency suddenly becomes too high this count may remain too high for
287 -- long time, in such cases the consumer can change it.
288 -- 0 means no latency computation
289 -- XXX this is derivable from workerMeasuredLatency, can be removed.
290 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
291 -- read by the worker threads
292 , workerPollingInterval :: IORef Count
294 -- This is in progress latency stats maintained by the workers which we
295 -- empty into workerCollectedLatency stats at certain intervals - whenever
296 -- we process the stream elements yielded in this period. The first count
297 -- is all yields, the second count is only those yields for which the
298 -- latency was measured to be non-zero (note that if the timer resolution
299 -- is low the measured latency may be zero e.g. on JS platform).
300 -- [LOCKING] Locked access. Modified by the consumer thread as well as
301 -- worker threads. Workers modify it periodically based on
302 -- workerPollingInterval and not on every yield to reduce the locking
304 -- (allYieldCount, yieldCount, timeTaken)
305 , workerPendingLatency :: IORef (Count, Count, NanoSecond64)
307 -- This is the second level stat which is an accmulation from
308 -- workerPendingLatency stats. We keep accumulating latencies in this
309 -- bucket until we have stats for a sufficient period and then we reset it
310 -- to start collecting for the next period and retain the computed average
311 -- latency for the last period in workerMeasuredLatency.
312 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
313 -- read by the worker threads
314 -- (allYieldCount, yieldCount, timeTaken)
315 , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)
317 -- Latency as measured by workers, aggregated for the last period.
318 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
319 -- read by the worker threads
320 , workerMeasuredLatency :: IORef NanoSecond64
323data SVarStats = SVarStats {
324 totalDispatches :: IORef Int
325 , maxWorkers :: IORef Int
326 , maxOutQSize :: IORef Int
327 , maxHeapSize :: IORef Int
328 , maxWorkQSize :: IORef Int
329 , avgWorkerLatency :: IORef (Count, NanoSecond64)
330 , minWorkerLatency :: IORef NanoSecond64
331 , maxWorkerLatency :: IORef NanoSecond64
332 , svarStopTime :: IORef (Maybe AbsTime)
335-- This is essentially a 'Maybe Word' type
336data Limit = Unlimited | Limited Word deriving Show
338instance Eq Limit where
339 Unlimited == Unlimited = True
340 Unlimited == Limited _ = False
341 Limited _ == Unlimited = False
342 Limited x == Limited y = x == y
344instance Ord Limit where
345 Unlimited <= Unlimited = True
346 Unlimited <= Limited _ = False
347 Limited _ <= Unlimited = True
348 Limited x <= Limited y = x <= y
350-- When to stop the composed stream.
352 StopNone -- stops only when all streams are finished
353 | StopAny -- stop when any stream finishes
354 | StopBy -- stop when a specific stream finishes
357-- | Buffering policy for persistent push workers (in ParallelT). In a pull
358-- style SVar (in AsyncT, AheadT etc.), the consumer side dispatches workers on
359-- demand, workers terminate if the buffer is full or if the consumer is not
360-- cosuming fast enough. In a push style SVar, a worker is dispatched only
361-- once, workers are persistent and keep pushing work to the consumer via a
362-- bounded buffer. If the buffer becomes full the worker either blocks, or it
363-- can drop an item from the buffer to make space.
365-- Pull style SVars are useful in lazy stream evaluation whereas push style
366-- SVars are useful in strict left Folds.
368-- XXX Maybe we can separate the implementation in two different types instead
369-- of using a common SVar type.
371data PushBufferPolicy =
372 PushBufferDropNew -- drop the latest element and continue
373 | PushBufferDropOld -- drop the oldest element and continue
374 | PushBufferBlock -- block the thread until space
377-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have
378-- references to the original SVar stored in several functions which will keep
379-- pointing to the original data and the new updates won't reflect there.
380-- Any updateable parts must be kept in mutable references (IORef).
382data SVar t m a = SVar
385 svarStyle :: SVarStyle
386 , svarMrun :: RunInIO m
387 , svarStopStyle :: SVarStopStyle
388 , svarStopBy :: IORef ThreadId
390 -- Shared output queue (events, length)
391 -- XXX For better efficiency we can try a preallocated array type (perhaps
392 -- something like a vector) that allows an O(1) append. That way we will
393 -- avoid constructing and reversing the list. Possibly we can also avoid
394 -- the GC copying overhead. When the size increases we should be able to
395 -- allocate the array in chunks.
397 -- [LOCKING] Frequent locked access. This is updated by workers on each
398 -- yield and once in a while read by the consumer thread. This could have
399 -- big locking overhead if the number of workers is high.
401 -- XXX We can use a per-CPU data structure to reduce the locking overhead.
402 -- However, a per-cpu structure cannot guarantee the exact sequence in
403 -- which the elements were added, though that may not be important.
404 , outputQueue :: IORef ([ChildEvent a], Int)
406 -- [LOCKING] Infrequent MVar. Used when the outputQ transitions from empty
407 -- to non-empty, or a work item is queued by a worker to the work queue and
408 -- needDoorBell is set by the consumer.
409 , outputDoorBell :: MVar () -- signal the consumer about output
410 , readOutputQ :: m [ChildEvent a]
411 , postProcess :: m Bool
413 -- channel to send events from the consumer to the worker. Used to send
414 -- exceptions from a fold driver to the fold computation running as a
415 -- consumer thread in the concurrent fold cases. Currently only one event
416 -- is sent by the fold so we do not really need a queue for it.
417 , outputQueueFromConsumer :: IORef ([ChildEvent a], Int)
418 , outputDoorBellFromConsumer :: MVar ()
420 -- Combined/aggregate parameters
421 -- This is truncated to maxBufferLimit if set to more than that. Otherwise
422 -- potentially each worker may yield one value to the buffer in the worst
423 -- case exceeding the requested buffer size.
424 , maxWorkerLimit :: Limit
425 , maxBufferLimit :: Limit
426 -- These two are valid and used only when maxBufferLimit is Limited.
427 , pushBufferSpace :: IORef Count
428 , pushBufferPolicy :: PushBufferPolicy
429 -- [LOCKING] The consumer puts this MVar after emptying the buffer, workers
430 -- block on it when the buffer becomes full. No overhead unless the buffer
432 , pushBufferMVar :: MVar ()
434 -- [LOCKING] Read only access by consumer when dispatching a worker.
435 -- Decremented by workers when picking work and undo decrement if the
436 -- worker does not yield a value.
437 , remainingWork :: Maybe (IORef Count)
438 , yieldRateInfo :: Maybe YieldRateInfo
440 -- Used only by bounded SVar types
441 , enqueue :: (RunInIO m, t m a) -> IO ()
442 , isWorkDone :: IO Bool
443 , isQueueDone :: IO Bool
444 , needDoorBell :: IORef Bool
445 , workLoop :: Maybe WorkerInfo -> m ()
447 -- Shared, thread tracking
448 -- [LOCKING] Updated unlocked only by consumer thread in case of
449 -- Async/Ahead style SVars. Updated locked by worker threads in case of
451 , workerThreads :: IORef (Set ThreadId)
452 -- [LOCKING] Updated locked by consumer thread when dispatching a worker
453 -- and by the worker threads when the thread stops. This is read unsafely
454 -- at several places where we want to rely on an approximate value.
455 , workerCount :: IORef Int
456 , accountThread :: ThreadId -> m ()
457 , workerStopMVar :: MVar ()
459 , svarStats :: SVarStats
460 -- to track garbage collection of SVar
461 , svarRef :: Maybe (IORef ())
463 -- Only for diagnostics
464 , svarInspectMode :: Bool
465 , svarCreator :: ThreadId
466 , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))
468 -- Shared work queue (stream, seqNo)
469 , aheadWorkQueue :: IORef ([t m a], Int)
472-------------------------------------------------------------------------------
473-- State for concurrency control
474-------------------------------------------------------------------------------
476-- XXX we can put the resettable fields in a oneShotConfig field and others in
477-- a persistentConfig field. That way reset would be fast and scalable
478-- irrespective of the number of fields.
480-- XXX make all these Limited types and use phantom types to distinguish them
481data State t m a = State
482 { -- one shot configuration, automatically reset for each API call
483 streamVar :: Maybe (SVar t m a)
484 , _yieldLimit :: Maybe Count
486 -- persistent configuration, state that remains valid until changed by
487 -- an explicit setting via a combinator.
488 , _threadsHigh :: Limit
489 , _bufferHigh :: Limit
490 -- XXX these two can be collapsed into a single type
491 , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency
492 , _maxStreamRate :: Maybe Rate
493 , _inspectMode :: Bool
496-------------------------------------------------------------------------------
497-- State defaults and reset
498-------------------------------------------------------------------------------
500-- A magical value for the buffer size arrived at by running the smallest
501-- possible task and measuring the optimal value of the buffer for that. This
502-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
504magicMaxBuffer :: Word
507defaultMaxThreads, defaultMaxBuffer :: Limit
508defaultMaxThreads = Limited magicMaxBuffer
509defaultMaxBuffer = Limited magicMaxBuffer
511-- The fields prefixed by an _ are not to be accessed or updated directly but
512-- via smart accessor APIs.
513defState :: State t m a
515 { streamVar = Nothing
516 , _yieldLimit = Nothing
517 , _threadsHigh = defaultMaxThreads
518 , _bufferHigh = defaultMaxBuffer
519 , _maxStreamRate = Nothing
520 , _streamLatency = Nothing
521 , _inspectMode = False
524-- XXX if perf gets affected we can have all the Nothing params in a single
525-- structure so that we reset is fast. We can also use rewrite rules such that
526-- reset occurs only in concurrent streams to reduce the impact on serial
528-- We can optimize this so that we clear it only if it is a Just value, it
529-- results in slightly better perf for zip/zipM but the performance of scan
530-- worsens a lot, it does not fuse.
532-- XXX This has a side effect of clearing the SVar and yieldLimit, therefore it
533-- should not be used to convert from the same type to the same type, unless
534-- you want to clear the SVar. For clearing the SVar you should be using the
535-- appropriate unStream functions instead.
537-- | Adapt the stream state from one type to another.
538adaptState :: State t m a -> State t n b
540 { streamVar = Nothing
541 , _yieldLimit = Nothing
544-------------------------------------------------------------------------------
545-- Smart get/set routines for State
546-------------------------------------------------------------------------------
548-- Use get/set routines instead of directly accessing the State fields
549setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
550setYieldLimit lim st =
557 else Just (fromIntegral n)
560getYieldLimit :: State t m a -> Maybe Count
561getYieldLimit = _yieldLimit
563setMaxThreads :: Int -> State t m a -> State t m a
569 then defaultMaxThreads
570 else Limited (fromIntegral n)
573getMaxThreads :: State t m a -> Limit
574getMaxThreads = _threadsHigh
576setMaxBuffer :: Int -> State t m a -> State t m a
582 then defaultMaxBuffer
583 else Limited (fromIntegral n)
586getMaxBuffer :: State t m a -> Limit
587getMaxBuffer = _bufferHigh
589setStreamRate :: Maybe Rate -> State t m a -> State t m a
590setStreamRate r st = st { _maxStreamRate = r }
592getStreamRate :: State t m a -> Maybe Rate
593getStreamRate = _maxStreamRate
595setStreamLatency :: Int -> State t m a -> State t m a
596setStreamLatency n st =
597 st { _streamLatency =
600 else Just (fromIntegral n)
603getStreamLatency :: State t m a -> Maybe NanoSecond64
604getStreamLatency = _streamLatency
606setInspectMode :: State t m a -> State t m a
607setInspectMode st = st { _inspectMode = True }
609getInspectMode :: State t m a -> Bool
610getInspectMode = _inspectMode
612-------------------------------------------------------------------------------
614-------------------------------------------------------------------------------
616cleanupSVar :: SVar t m a -> IO ()
618 workers <- readIORef (workerThreads sv)
619 Prelude.mapM_ (`throwTo` ThreadAbort)
622cleanupSVarFromWorker :: SVar t m a -> IO ()
623cleanupSVarFromWorker sv = do
624 workers <- readIORef (workerThreads sv)
626 Prelude.mapM_ (`throwTo` ThreadAbort)
627 (Prelude.filter (/= self) $ S.toList workers)
629-------------------------------------------------------------------------------
630-- Worker latency data collection
631-------------------------------------------------------------------------------
633-- Every once in a while workers update the latencies and check the yield rate.
634-- They return if we are above the expected yield rate. If we check too often
635-- it may impact performance, if we check less often we may have a stale
636-- picture. We update every minThreadDelay but we translate that into a yield
637-- count based on latency so that the checking overhead is little.
639-- XXX use a generation count to indicate that the value is updated. If the
640-- value is updated an existing worker must check it again on the next yield.
641-- Otherwise it is possible that we may keep updating it and because of the mod
642-- worker keeps skipping it.
643updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
644updateWorkerPollingInterval yinfo latency = do
645 let periodRef = workerPollingInterval yinfo
646 cnt = max 1 $ minThreadDelay `div` latency
647 period = min cnt (fromIntegral magicMaxBuffer)
649 writeIORef periodRef (fromIntegral period)
651{-# INLINE recordMinMaxLatency #-}
652recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()
653recordMinMaxLatency sv new = do
654 let ss = svarStats sv
655 minLat <- readIORef (minWorkerLatency ss)
656 when (new < minLat || minLat == 0) $
657 writeIORef (minWorkerLatency ss) new
659 maxLat <- readIORef (maxWorkerLatency ss)
660 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
662recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()
663recordAvgLatency sv (count, time) = do
664 let ss = svarStats sv
665 modifyIORef (avgWorkerLatency ss) $
666 \(cnt, t) -> (cnt + count, t + time)
668-- Pour the pending latency stats into a collection bucket
669{-# INLINE collectWorkerPendingLatency #-}
670collectWorkerPendingLatency
671 :: IORef (Count, Count, NanoSecond64)
672 -> IORef (Count, Count, NanoSecond64)
673 -> IO (Count, Maybe (Count, NanoSecond64))
674collectWorkerPendingLatency cur col = do
675 (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
677 (fcnt, cnt, t) <- readIORef col
678 let totalCount = fcnt + fcount
679 latCount = cnt + count
681 writeIORef col (totalCount, latCount, latTime)
683 assert (latCount == 0 || latTime /= 0) (return ())
685 if latCount > 0 && latTime > 0
686 then Just (latCount, latTime)
688 return (totalCount, latPair)
690{-# INLINE shouldUseCollectedBatch #-}
691shouldUseCollectedBatch
697shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
698 let r = fromIntegral newLat / fromIntegral prevLat :: Double
699 in (collectedYields > fromIntegral magicMaxBuffer)
700 || (collectedTime > minThreadDelay)
701 || (prevLat > 0 && (r > 2 || r < 0.5))
704-- Returns a triple, (1) yield count since last collection, (2) the base time
705-- when we started counting, (3) average latency in the last measurement
706-- period. The former two are used for accurate measurement of the going rate
707-- whereas the average is used for future estimates e.g. how many workers
708-- should be maintained to maintain the rate.
709-- CAUTION! keep it in sync with getWorkerLatency
710collectLatency :: SVar t m a
713 -> IO (Count, AbsTime, NanoSecond64)
714collectLatency sv yinfo drain = do
715 let cur = workerPendingLatency yinfo
716 col = workerCollectedLatency yinfo
717 longTerm = svarAllTimeLatency yinfo
718 measured = workerMeasuredLatency yinfo
720 (newCount, newLatPair) <- collectWorkerPendingLatency cur col
721 (lcount, ltime) <- readIORef longTerm
722 prevLat <- readIORef measured
724 let newLcount = lcount + newCount
725 retWith lat = return (newLcount, ltime, lat)
728 Nothing -> retWith prevLat
729 Just (count, time) -> do
730 let newLat = time `div` fromIntegral count
731 when (svarInspectMode sv) $ recordMinMaxLatency sv newLat
732 -- When we have collected a significant sized batch we compute the
733 -- new latency using that batch and return the new latency,
734 -- otherwise we return the previous latency derived from the
736 if shouldUseCollectedBatch newCount time newLat prevLat || drain
738 -- XXX make this NOINLINE?
739 updateWorkerPollingInterval yinfo (max newLat prevLat)
740 when (svarInspectMode sv) $ recordAvgLatency sv (count, time)
741 writeIORef col (0, 0, 0)
742 writeIORef measured ((prevLat + newLat) `div` 2)
743 modifyIORef longTerm $ \(_, t) -> (newLcount, t)
747-------------------------------------------------------------------------------
748-- Dumping the SVar for debug/diag
749-------------------------------------------------------------------------------
751dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
752dumpSVarStats sv ss style = do
753 case yieldRateInfo sv of
756 _ <- liftIO $ collectLatency sv yinfo True
759 dispatches <- readIORef $ totalDispatches ss
760 maxWrk <- readIORef $ maxWorkers ss
761 maxOq <- readIORef $ maxOutQSize ss
762 maxHp <- readIORef $ maxHeapSize ss
763 minLat <- readIORef $ minWorkerLatency ss
764 maxLat <- readIORef $ maxWorkerLatency ss
765 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
766 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
767 Nothing -> return (0, 0, 0)
769 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
772 t <- readIORef (svarStopTime ss)
773 gl <- readIORef (svarGainedLostYields yinfo)
776 now <- getTime Monotonic
777 let interval = diffAbsTime64 now startTime
778 return (cnt, gl, interval `div` fromIntegral cnt)
780 let interval = diffAbsTime64 stopTime startTime
781 return (cnt, gl, interval `div` fromIntegral cnt)
782 else return (0, 0, 0)
785 [ "total dispatches = " <> show dispatches
786 , "max workers = " <> show maxWrk
787 , "max outQSize = " <> show maxOq
788 <> (if style == AheadVar
789 then "\nheap max size = " <> show maxHp
792 then "\nmin worker latency = " <> showNanoSecond64 minLat
795 then "\nmax worker latency = " <> showNanoSecond64 maxLat
798 then let lat = avgTime `div` fromIntegral avgCnt
799 in "\navg worker latency = " <> showNanoSecond64 lat
802 then "\nSVar latency = " <> showRelTime64 svarLat
805 then "\nSVar yield count = " <> show svarCnt
807 <> (if svarGainLossCnt > 0
808 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
812{-# NOINLINE dumpSVar #-}
813dumpSVar :: SVar t m a -> IO String
815 (oqList, oqLen) <- readIORef $ outputQueue sv
816 db <- tryReadMVar $ outputDoorBell sv
818 if svarStyle sv == AheadVar
820 (oheap, oheapSeq) <- readIORef $ outputHeap sv
821 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
823 [ "heap length = " <> show (H.size oheap)
824 , "heap seqeunce = " <> show oheapSeq
825 , "work queue length = " <> show (length wq)
826 , "work queue sequence = " <> show wqSeq
830 let style = svarStyle sv
832 if style /= ParallelVar
833 then readIORef $ needDoorBell sv
835 rthread <- readIORef $ workerThreads sv
836 workers <- readIORef $ workerCount sv
837 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
841 "Creator tid = " <> show (svarCreator sv),
842 "style = " <> show (svarStyle sv)
843 , "---------CURRENT STATE-----------"
844 , "outputQueue length computed = " <> show (length oqList)
845 , "outputQueue length maintained = " <> show oqLen
846 -- XXX print the types of events in the outputQueue, first 5
847 , "outputDoorBell = " <> show db
851 [ "needDoorBell = " <> show waiting
852 , "running threads = " <> show rthread
853 -- XXX print the status of first 5 threads
854 , "running thread count = " <> show workers
856 <> "---------STATS-----------\n"
859printSVar :: SVar t m a -> String -> IO ()
861 svInfo <- dumpSVar sv
862 hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo
864-- MVar diagnostics has some overhead - around 5% on AsyncT null benchmark, we
865-- can keep it on in production to debug problems quickly if and when they
866-- happen, but it may result in unexpected output when threads are left hanging
867-- until they are GCed because the consumer went away.
869{-# NOINLINE mvarExcHandler #-}
870mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
871mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
872 svInfo <- dumpSVar sv
873 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
876{-# NOINLINE stmExcHandler #-}
877stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
878stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
879 svInfo <- dumpSVar sv
880 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
883withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
884withDiagMVar sv label action =
885 if svarInspectMode sv
887 action `catches` [ Handler (mvarExcHandler sv label)
888 , Handler (stmExcHandler sv label)
892------------------------------------------------------------------------------
894------------------------------------------------------------------------------
896-- | When we run computations concurrently, we completely isolate the state of
897-- the concurrent computations from the parent computation. The invariant is
898-- that we should never be running two concurrent computations in the same
899-- thread without using the runInIO function. Also, we should never be running
900-- a concurrent computation in the parent thread, otherwise it may affect the
901-- state of the parent which is against the defined semantics of concurrent
903captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
904captureMonadState = control $ \run -> run (return $ RunInIO run)
906------------------------------------------------------------------------------
907-- Collecting results from child workers in a streamed fashion
908------------------------------------------------------------------------------
910-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
911-- faster, along with the fields in sv required by send?
912-- XXX make it noinline
914-- XXX we may want to employ an increment and decrement in batches when the
915-- througput is high or when the cost of synchronization is high. For example
916-- if the application is distributed then inc/dec of a shared variable may be
919-- A worker decrements the yield limit before it executes an action. However,
920-- the action may not result in an element being yielded, in that case we have
921-- to increment the yield limit.
923-- Note that we need it to be an Int type so that we have the ability to undo a
924-- decrement that takes it below zero.
925{-# INLINE decrementYieldLimit #-}
926decrementYieldLimit :: SVar t m a -> IO Bool
927decrementYieldLimit sv =
928 case remainingWork sv of
929 Nothing -> return True
931 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
934{-# INLINE incrementYieldLimit #-}
935incrementYieldLimit :: SVar t m a -> IO ()
936incrementYieldLimit sv =
937 case remainingWork sv of
939 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
941-- XXX exception safety of all atomic/MVar operations
943-- TBD Each worker can have their own queue and the consumer can empty one
944-- queue at a time, that way contention can be reduced.
946-- XXX Only yields should be counted in the buffer limit and not the Stop
949{-# INLINE decrementBufferLimit #-}
950decrementBufferLimit :: SVar t m a -> IO ()
951decrementBufferLimit sv =
952 case maxBufferLimit sv of
953 Unlimited -> return ()
955 let ref = pushBufferSpace sv
956 old <- atomicModifyIORefCAS ref $ \x ->
957 (if x >= 1 then x - 1 else x, x)
959 case pushBufferPolicy sv of
960 PushBufferBlock -> blockAndRetry
961 PushBufferDropNew -> do
962 -- We just drop one item and proceed. It is possible
963 -- that by the time we drop the item the consumer
964 -- thread might have run and created space in the
965 -- buffer, but we do not care about that condition.
966 -- This is not pedantically correct but it should be
968 -- XXX we may want to drop only if n == maxBuf
969 -- otherwise we must have space in the buffer and a
970 -- decrement should be possible.
971 block <- atomicModifyIORefCAS (outputQueue sv) $
975 _ : xs -> ((xs, n - 1), False)
976 when block blockAndRetry
977 -- XXX need a dequeue or ring buffer for this
978 PushBufferDropOld -> undefined
983 let ref = pushBufferSpace sv
984 liftIO $ takeMVar (pushBufferMVar sv)
985 old <- atomicModifyIORefCAS ref $ \x ->
986 (if x >= 1 then x - 1 else x, x)
987 -- When multiple threads sleep on takeMVar, the first thread would
988 -- wakeup due to a putMVar by the consumer, but the rest of the threads
989 -- would have to put back the MVar after taking it and decrementing the
990 -- buffer count, otherwise all other threads will remain asleep.
992 then void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
993 -- We do not put the MVar back in this case, instead we
994 -- wait for the consumer to put it.
997{-# INLINE incrementBufferLimit #-}
998incrementBufferLimit :: SVar t m a -> IO ()
999incrementBufferLimit sv =
1000 case maxBufferLimit sv of
1001 Unlimited -> return ()
1003 atomicModifyIORefCAS_ (pushBufferSpace sv) (+ 1)
1005 void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
1007{-# INLINE resetBufferLimit #-}
1008resetBufferLimit :: SVar t m a -> IO ()
1009resetBufferLimit sv =
1010 case maxBufferLimit sv of
1011 Unlimited -> return ()
1012 Limited n -> atomicModifyIORefCAS_ (pushBufferSpace sv)
1013 (const (fromIntegral n))
1015{-# INLINE sendWithDoorBell #-}
1017 IORef ([ChildEvent a], Int) -> MVar () -> ChildEvent a -> IO Int
1018sendWithDoorBell q bell msg = do
1019 -- XXX can the access to outputQueue be made faster somehow?
1020 oldlen <- atomicModifyIORefCAS q $ \(es, n) ->
1021 ((msg : es, n + 1), n)
1022 when (oldlen <= 0) $ do
1023 -- The wake up must happen only after the store has finished otherwise
1024 -- we can have lost wakeup problems.
1026 -- Since multiple workers can try this at the same time, it is possible
1027 -- that we may put a spurious MVar after the consumer has already seen
1028 -- the output. But that's harmless, at worst it may cause the consumer
1029 -- to read the queue again and find it empty.
1030 -- The important point is that the consumer is guaranteed to receive a
1031 -- doorbell if something was added to the queue after it empties it.
1032 void $ tryPutMVar bell ()
1035-- | This function is used by the producer threads to queue output for the
1036-- consumer thread to consume. Returns whether the queue has more space.
1037send :: SVar t m a -> ChildEvent a -> IO Int
1038send sv = sendWithDoorBell (outputQueue sv) (outputDoorBell sv)
1040-- There is no bound implemented on the buffer, this is assumed to be low
1042sendToProducer :: SVar t m a -> ChildEvent a -> IO Int
1043sendToProducer sv msg = do
1044 -- In case the producer stream is blocked on pushing to the fold buffer
1045 -- then wake it up so that it can check for the stop event or exception
1046 -- being sent to it otherwise we will be deadlocked.
1047 void $ tryPutMVar (pushBufferMVar sv) ()
1048 sendWithDoorBell (outputQueueFromConsumer sv)
1049 (outputDoorBellFromConsumer sv) msg
1051-- {-# NOINLINE sendStopToProducer #-}
1052sendStopToProducer :: MonadIO m => SVar t m a -> m ()
1053sendStopToProducer sv = liftIO $ do
1055 void $ sendToProducer sv (ChildStop tid Nothing)
1057workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
1058workerCollectLatency winfo = do
1059 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
1060 cnt1 <- readIORef (workerYieldCount winfo)
1061 let cnt = cnt1 - cnt0
1065 t1 <- getTime Monotonic
1066 let period = fromRelTime64 $ diffAbsTime64 t1 t0
1067 writeIORef (workerLatencyStart winfo) (cnt1, t1)
1068 return $ Just (cnt, period)
1071-- XXX There are a number of gotchas in measuring latencies.
1072-- 1) We measure latencies only when a worker yields a value
1073-- 2) It is possible that a stream calls the stop continuation, in which case
1074-- the worker would not yield a value and we would not account that worker in
1075-- latencies. Even though this case should ideally be accounted we do not
1076-- account it because we cannot or do not distinguish it from the case
1078-- 3) It is possible that a worker returns without yielding anything because it
1079-- never got a chance to pick up work.
1080-- 4) If the system timer resolution is lower than the latency, the latency
1081-- computation turns out to be zero.
1083-- We can fix this if we measure the latencies by counting the work items
1084-- picked rather than based on the outputs yielded.
1085workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
1086workerUpdateLatency yinfo winfo = do
1087 r <- workerCollectLatency winfo
1089 Just (cnt, period) -> do
1090 -- NOTE: On JS platform the timer resolution could be pretty low. When
1091 -- the timer resolution is low, measurement of latencies could be
1092 -- tricky. All the worker latencies will turn out to be zero if they
1093 -- are lower than the resolution. We only take into account those
1094 -- measurements which are more than the timer resolution.
1096 let ref = workerPendingLatency yinfo
1097 (cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
1098 atomicModifyIORefCAS_ ref $
1099 \(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
1100 Nothing -> return ()
1102updateYieldCount :: WorkerInfo -> IO Count
1103updateYieldCount winfo = do
1104 cnt <- readIORef (workerYieldCount winfo)
1106 writeIORef (workerYieldCount winfo) cnt1
1109isBeyondMaxYield :: Count -> WorkerInfo -> Bool
1110isBeyondMaxYield cnt winfo =
1111 let ymax = workerYieldMax winfo
1112 in ymax /= 0 && cnt >= ymax
1114-- XXX we should do rate control periodically based on the total yields rather
1115-- than based on the worker local yields as other workers may have yielded more
1116-- and we should stop based on the aggregate yields. However, latency update
1117-- period can be based on individual worker yields.
1118{-# NOINLINE checkRatePeriodic #-}
1119checkRatePeriodic :: SVar t m a
1124checkRatePeriodic sv yinfo winfo ycnt = do
1125 i <- readIORef (workerPollingInterval yinfo)
1126 -- XXX use generation count to check if the interval has been updated
1127 if i /= 0 && (ycnt `mod` i) == 0
1129 workerUpdateLatency yinfo winfo
1130 -- XXX not required for parallel streams
1131 isBeyondMaxRate sv yinfo
1134-- CAUTION! this also updates the yield count and therefore should be called
1135-- only when we are actually yielding an element.
1136{-# NOINLINE workerRateControl #-}
1137workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
1138workerRateControl sv yinfo winfo = do
1139 cnt <- updateYieldCount winfo
1140 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
1141 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
1143-- XXX we should do rate control here but not latency update in case of ahead
1144-- streams. latency update must be done when we yield directly to outputQueue
1145-- or when we yield to heap.
1147-- returns whether the worker should continue (True) or stop (False).
1148{-# INLINE sendYield #-}
1149sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
1150sendYield sv mwinfo msg = do
1151 oldlen <- send sv msg
1152 let limit = maxBufferLimit sv
1153 bufferSpaceOk <- case limit of
1154 Unlimited -> return True
1156 active <- readIORef (workerCount sv)
1157 return $ (oldlen + 1) < (fromIntegral lim - active)
1161 case yieldRateInfo sv of
1162 Nothing -> return True
1163 Just yinfo -> workerRateControl sv yinfo winfo
1164 Nothing -> return True
1165 return $ bufferSpaceOk && rateLimitOk
1167{-# INLINE workerStopUpdate #-}
1168workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
1169workerStopUpdate winfo info = do
1170 i <- readIORef (workerPollingInterval info)
1171 when (i /= 0) $ workerUpdateLatency info winfo
1173{-# INLINABLE sendStop #-}
1174sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
1175sendStop sv mwinfo = do
1176 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
1177 case (mwinfo, yieldRateInfo sv) of
1178 (Just winfo, Just info) ->
1179 workerStopUpdate winfo info
1182 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
1184-------------------------------------------------------------------------------
1186-------------------------------------------------------------------------------
1188{-# INLINE ringDoorBell #-}
1189ringDoorBell :: SVar t m a -> IO ()
1192 w <- readIORef $ needDoorBell sv
1194 -- Note: the sequence of operations is important for correctness here.
1195 -- We need to set the flag to false strictly before sending the
1196 -- outputDoorBell, otherwise the outputDoorBell may get processed too
1197 -- early and then we may set the flag to False to later making the
1198 -- consumer lose the flag, even without receiving a outputDoorBell.
1199 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1200 void $ tryPutMVar (outputDoorBell sv) ()
1202-------------------------------------------------------------------------------
1204-------------------------------------------------------------------------------
1206-- Note: For purely right associated expressions this queue should have at most
1207-- one element. It grows to more than one when we have left associcated
1208-- expressions. Large left associated compositions can grow this to a
1210{-# INLINE enqueueLIFO #-}
1212 SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
1213enqueueLIFO sv q m = do
1214 atomicModifyIORefCAS_ q $ \ms -> m : ms
1217-------------------------------------------------------------------------------
1219-------------------------------------------------------------------------------
1221-- XXX we can use the Ahead style sequence/heap mechanism to make the best
1222-- effort to always try to finish the streams on the left side of an expression
1223-- first as long as possible.
1225{-# INLINE enqueueFIFO #-}
1228 -> LinkedQueue (RunInIO m, t m a)
1229 -> (RunInIO m, t m a)
1231enqueueFIFO sv q m = do
1235-------------------------------------------------------------------------------
1237-------------------------------------------------------------------------------
1239-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
1240-- but always serve them in the same order as they appear in the stream. To
1241-- implement lookahead streams efficiently we assign a sequence number to each
1242-- task when the task is picked up for execution. When the task finishes, the
1243-- output is tagged with the same sequence number and we rearrange the outputs
1244-- in sequence based on that number.
1246-- To explain the mechanism imagine that the current task at the head of the
1247-- stream has a "token" to yield to the outputQueue. The ownership of the token
1248-- is determined by the current sequence number is maintained in outputHeap.
1249-- Sequence number is assigned when a task is queued. When a thread dequeues a
1250-- task it picks up the sequence number as well and when the output is ready it
1251-- uses the sequence number to queue the output to the outputQueue.
1253-- The thread with current sequence number sends the output directly to the
1254-- outputQueue. Other threads push the output to the outputHeap. When the task
1255-- being queued on the heap is a stream of many elements we evaluate only the
1256-- first element and keep the rest of the unevaluated computation in the heap.
1257-- When such a task gets the "token" for outputQueue it evaluates and directly
1258-- yields all the elements to the outputQueue without checking for the
1261-- Note that no two outputs in the heap can have the same sequence numbers and
1262-- therefore we do not need a stable heap. We have also separated the buffer
1263-- for the current task (outputQueue) and the pending tasks (outputHeap) so
1264-- that the pending tasks cannot interfere with the current task. Note that for
1265-- a single task just the outputQueue is enough and for the case of many
1266-- threads just a heap is good enough. However we balance between these two
1267-- cases, so that both are efficient.
1269-- For bigger streams it may make sense to have separate buffers for each
1270-- stream. However, for singleton streams this may become inefficient. However,
1271-- if we do not have separate buffers, then the streams that come later in
1272-- sequence may hog the buffer, hindering the streams that are ahead. For this
1273-- reason we have a single element buffer limitation for the streams being
1274-- executed in advance.
1276-- This scheme works pretty efficiently with less than 40% extra overhead
1277-- compared to the Async streams where we do not have any kind of sequencing of
1278-- the outputs. It is especially devised so that we are most efficient when we
1279-- have short tasks and need just a single thread. Also when a thread yields
1280-- many items it can hold lockfree access to the outputQueue and do it
1283-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
1284-- that they do not hog the resources and hinder the progress of the threads in
1287-- XXX Left associated ahead expressions are expensive. We start a new SVar for
1288-- each left associative expression. The queue is used only for right
1289-- associated expression, we queue the right expression and execute the left.
1290-- Thererefore the queue never has more than one item in it.
1292-- XXX we can fix this. When we queue more than one item on the queue we can
1293-- mark the previously queued item as not-runnable. The not-runnable item is
1294-- not dequeued until the already running one has finished and at that time we
1295-- would also know the exact sequence number of the already queued item.
1297-- we can even run the already queued items but they will have to be sorted in
1298-- layers in the heap. We can use a list of heaps for that.
1299{-# INLINE enqueueAhead #-}
1300enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> (RunInIO m, t m a) -> IO ()
1301enqueueAhead sv q m = do
1302 atomicModifyIORefCAS_ q $ \ case
1303 ([], n) -> ([snd m], n + 1) -- increment sequence
1304 _ -> error "enqueueAhead: queue is not empty"
1307-- enqueue without incrementing the sequence number
1308{-# INLINE reEnqueueAhead #-}
1309reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1310reEnqueueAhead sv q m = do
1311 atomicModifyIORefCAS_ q $ \ case
1312 ([], n) -> ([m], n) -- DO NOT increment sequence
1313 _ -> error "reEnqueueAhead: queue is not empty"
1316-- Normally the thread that has the token should never go away. The token gets
1317-- handed over to another thread, but someone or the other has the token at any
1318-- point of time. But if the task that has the token finds that the outputQueue
1319-- is full, in that case it can go away without even handing over the token to
1320-- another thread. In that case it sets the nextSequence number in the heap its
1321-- own sequence number before going away. To handle this case, any task that
1322-- does not have the token tries to dequeue from the heap first before
1323-- dequeuing from the work queue. If it finds that the task at the top of the
1324-- heap is the one that owns the current sequence number then it grabs the
1325-- token and starts with that.
1327-- XXX instead of queueing just the head element and the remaining computation
1328-- on the heap, evaluate as many as we can and place them on the heap. But we
1329-- need to give higher priority to the lower sequence numbers so that lower
1330-- priority tasks do not fill up the heap making higher priority tasks block
1331-- due to full heap. Maybe we can have a weighted space for them in the heap.
1332-- The weight is inversely proportional to the sequence number.
1334-- XXX review for livelock
1336{-# INLINE queueEmptyAhead #-}
1337queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
1338queueEmptyAhead q = liftIO $ do
1339 (xs, _) <- readIORef q
1342{-# INLINE dequeueAhead #-}
1343dequeueAhead :: MonadIO m
1344 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
1345dequeueAhead q = liftIO $
1346 atomicModifyIORefCAS q $ \case
1347 ([], n) -> (([], n), Nothing)
1348 (x : [], n) -> (([], n), Just (x, n))
1349 _ -> error "more than one item on queue"
1351-------------------------------------------------------------------------------
1353-------------------------------------------------------------------------------
1355withIORef :: IORef a -> (a -> IO b) -> IO b
1356withIORef ref f = readIORef ref >>= f
1358atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
1359atomicModifyIORef_ ref f =
1360 atomicModifyIORef ref $ \x -> (f x, ())
1362data HeapDequeueResult t m a =
1365 | Ready (Entry Int (AheadHeapEntry t m a))
1367{-# INLINE dequeueFromHeap #-}
1369 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1370 -> IO (HeapDequeueResult t m a)
1371dequeueFromHeap hpVar =
1372 atomicModifyIORef hpVar $ \pair@(hp, snum) ->
1374 Nothing -> (pair, Clearing)
1378 Just (ent@(Entry seqNo _ev), hp') ->
1380 then ((hp', Nothing), Ready ent)
1381 else assert (seqNo >= n) (pair, Waiting n)
1382 Nothing -> (pair, Waiting n)
1384{-# INLINE dequeueFromHeapSeq #-}
1386 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1388 -> IO (HeapDequeueResult t m a)
1389dequeueFromHeapSeq hpVar i =
1390 atomicModifyIORef hpVar $ \(hp, snum) ->
1395 Just (ent@(Entry seqNo _ev), hp') ->
1397 then ((hp', Nothing), Ready ent)
1398 else assert (seqNo >= i) ((hp, Just i), Waiting i)
1399 Nothing -> ((hp, Just i), Waiting i)
1400 Just _ -> error "dequeueFromHeapSeq: unreachable"
1402heapIsSane :: Maybe Int -> Int -> Bool
1403heapIsSane snum seqNo =
1406 Just n -> seqNo >= n
1408{-# INLINE requeueOnHeapTop #-}
1410 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1411 -> Entry Int (AheadHeapEntry t m a)
1414requeueOnHeapTop hpVar ent seqNo =
1415 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1416 assert (heapIsSane snum seqNo) (H.insert ent hp, Just seqNo)
1418{-# INLINE updateHeapSeq #-}
1420 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1423updateHeapSeq hpVar seqNo =
1424 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1425 assert (heapIsSane snum seqNo) (hp, Just seqNo)
1427-------------------------------------------------------------------------------
1429-------------------------------------------------------------------------------
1431-- XXX To be implemented. Use a linked queue like WAsync and put back the
1432-- remaining computation at the back of the queue instead of the heap, and
1433-- increment the sequence number.
1435-------------------------------------------------------------------------------
1436-- Dispatching workers and tracking them
1437-------------------------------------------------------------------------------
1439-- Thread tracking is needed for two reasons:
1441-- 1) Killing threads on exceptions. Threads may not be left to go away by
1442-- themselves because they may run for significant times before going away or
1443-- worse they may be stuck in IO and never go away.
1445-- 2) To know when all threads are done and the stream has ended.
1447{-# NOINLINE addThread #-}
1448addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1450 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
1452-- This is cheaper than modifyThread because we do not have to send a
1453-- outputDoorBell This can make a difference when more workers are being
1455{-# INLINE delThread #-}
1456delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1458 liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
1460-- If present then delete else add. This takes care of out of order add and
1461-- delete i.e. a delete arriving before we even added a thread.
1462-- This occurs when the forked thread is done even before the 'addThread' right
1463-- after the fork gets a chance to run.
1464{-# INLINE modifyThread #-}
1465modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1466modifyThread sv tid = do
1467 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1469 then let new = S.delete tid old in (new, new)
1470 else let new = S.insert tid old in (new, old)
1471 when (null changed) $
1474 void $ tryPutMVar (outputDoorBell sv) ()
1476-- | This is safe even if we are adding more threads concurrently because if
1477-- a child thread is adding another thread then anyway 'workerThreads' will
1479{-# INLINE allThreadsDone #-}
1480allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1481allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1483{-# NOINLINE handleChildException #-}
1484handleChildException :: SVar t m a -> SomeException -> IO ()
1485handleChildException sv e = do
1487 void $ send sv (ChildStop tid (Just e))
1489{-# NOINLINE handleFoldException #-}
1490handleFoldException :: SVar t m a -> SomeException -> IO ()
1491handleFoldException sv e = do
1493 void $ sendToProducer sv (ChildStop tid (Just e))
1495{-# NOINLINE recordMaxWorkers #-}
1496recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1497recordMaxWorkers sv = liftIO $ do
1498 active <- readIORef (workerCount sv)
1499 maxWrk <- readIORef (maxWorkers $ svarStats sv)
1500 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1501 modifyIORef (totalDispatches $ svarStats sv) (+1)
1503{-# NOINLINE pushWorker #-}
1504pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1505pushWorker yieldMax sv = do
1506 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1507 when (svarInspectMode sv) $ recordMaxWorkers sv
1508 -- This allocation matters when significant number of workers are being
1509 -- sent. We allocate it only when needed.
1511 case yieldRateInfo sv of
1512 Nothing -> return Nothing
1513 Just _ -> liftIO $ do
1514 cntRef <- newIORef 0
1515 t <- getTime Monotonic
1516 lat <- newIORef (0, t)
1517 return $ Just WorkerInfo
1518 { workerYieldMax = yieldMax
1519 , workerYieldCount = cntRef
1520 , workerLatencyStart = lat
1522 doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
1525-- XXX we can push the workerCount modification in accountThread and use the
1526-- same pushWorker for Parallel case as well.
1528-- | In contrast to pushWorker which always happens only from the consumer
1529-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1530-- producer side. So we need to use a thread safe modification of
1531-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1532-- using a CAS based modification.
1533{-# INLINE pushWorkerPar #-}
1536 => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1537pushWorkerPar sv wloop =
1538 if svarInspectMode sv
1540 else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
1545 {-# NOINLINE forkWithDiag #-}
1547 -- We do not use workerCount in case of ParallelVar but still there is
1548 -- no harm in maintaining it correctly.
1549 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1551 -- This allocation matters when significant number of workers are being
1552 -- sent. We allocate it only when needed. The overhead increases by 4x.
1554 case yieldRateInfo sv of
1555 Nothing -> return Nothing
1556 Just _ -> liftIO $ do
1557 cntRef <- newIORef 0
1558 t <- getTime Monotonic
1559 lat <- newIORef (0, t)
1560 return $ Just WorkerInfo
1561 { workerYieldMax = 0
1562 , workerYieldCount = cntRef
1563 , workerLatencyStart = lat
1566 doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
1570-- True: can dispatch more
1571-- False: cannot dispatch any more
1572dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1573dispatchWorker yieldCount sv = do
1574 let workerLimit = maxWorkerLimit sv
1575 -- XXX in case of Ahead streams we should not send more than one worker
1576 -- when the work queue is done but heap is not done.
1577 done <- liftIO $ isWorkDone sv
1578 -- Note, "done" may not mean that the work is actually finished if there
1579 -- are workers active, because there may be a worker which has not yet
1580 -- queued the leftover work.
1583 qDone <- liftIO $ isQueueDone sv
1584 -- This count may not be accurate as it is decremented by the workers
1585 -- and we have no synchronization with that decrement.
1586 active <- liftIO $ readIORef $ workerCount sv
1589 -- Note that we may deadlock if the previous workers (tasks in the
1590 -- stream) wait/depend on the future workers (tasks in the stream)
1591 -- executing. In that case we should either configure the maxWorker
1592 -- count to higher or use parallel style instead of ahead or async
1594 limit <- case remainingWork sv of
1595 Nothing -> return workerLimit
1597 n <- liftIO $ readIORef ref
1598 case yieldRateInfo sv of
1599 Just _ -> return workerLimit
1603 Unlimited -> Limited (fromIntegral n)
1604 Limited lim -> Limited $ min lim (fromIntegral n)
1606 -- XXX for ahead streams shall we take the heap yields into account
1607 -- for controlling the dispatch? We should not dispatch if the heap
1608 -- has already got the limit covered.
1609 let dispatch = pushWorker yieldCount sv >> return True
1611 Unlimited -> dispatch
1612 -- Note that the use of remainingWork and workerCount is not
1613 -- atomic and the counts may even have changed between reading
1614 -- and using them here, so this is just approximate logic and
1615 -- we cannot rely on it for correctness. We may actually
1616 -- dispatch more workers than required.
1617 Limited lim | lim > fromIntegral active -> dispatch
1620 when (active <= 0) $ pushWorker 0 sv
1624-------------------------------------------------------------------------------
1625-- Dispatch workers with rate control
1626-------------------------------------------------------------------------------
1628-- | This is a magic number and it is overloaded, and used at several places to
1631-- 1. If we have to sleep to slowdown this is the minimum period that we
1632-- accumulate before we sleep. Also, workers do not stop until this much
1633-- sleep time is accumulated.
1634-- 3. Collected latencies are computed and transferred to measured latency
1635-- after a minimum of this period.
1636minThreadDelay :: NanoSecond64
1637minThreadDelay = 1000000
1639-- | Another magic number! When we have to start more workers to cover up a
1640-- number of yields that we are lagging by then we cannot start one worker for
1641-- each yield because that may be a very big number and if the latency of the
1642-- workers is low these number of yields could be very high. We assume that we
1643-- run each extra worker for at least this much time.
1644rateRecoveryTime :: NanoSecond64
1645rateRecoveryTime = 1000000
1647-- We either block, or send one worker with limited yield count or one or more
1648-- workers with unlimited yield count.
1650 = BlockWait NanoSecond64
1651 | PartialWorker Count
1652 | ManyWorkers Int Count
1655-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1665estimateWorkers workerLimit svarYields gainLossYields
1666 svarElapsed wLatency targetLat range =
1667 -- XXX we can have a maxEfficiency combinator as well which runs the
1668 -- producer at the maximal efficiency i.e. the number of workers are chosen
1669 -- such that the latency is minimum or within a range. Or we can call it
1670 -- maxWorkerLatency.
1673 -- How many workers do we need to achieve the required rate?
1675 -- When the workers are IO bound we can increase the throughput by
1676 -- increasing the number of workers as long as the IO device has enough
1677 -- capacity to process all the requests concurrently. If the IO
1678 -- bandwidth is saturated increasing the workers won't help. Also, if
1679 -- the CPU utilization in processing all these requests exceeds the CPU
1680 -- bandwidth, then increasing the number of workers won't help.
1682 -- When the workers are purely CPU bound, increasing the workers beyond
1683 -- the number of CPUs won't help.
1685 -- TODO - measure the CPU and IO requirements of the workers. Have a
1686 -- way to specify the max bandwidth of the underlying IO mechanism and
1687 -- use that to determine the max rate of workers, and also take the CPU
1688 -- bandwidth into account. We can also discover the IO bandwidth if we
1689 -- know that we are not CPU bound, then how much steady state rate are
1690 -- we able to achieve. Design tests for CPU bound and IO bound cases.
1692 -- Calculate how many yields are we ahead or behind to match the exact
1693 -- required rate. Based on that we increase or decrease the effective
1696 -- When the worker latency is lower than required latency we begin with
1697 -- a yield and then wait rather than first waiting and then yielding.
1698 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1699 effectiveYields = svarYields + gainLossYields
1700 deltaYields = fromIntegral targetYields - effectiveYields
1702 -- We recover the deficit by running at a higher/lower rate for a
1703 -- certain amount of time. To keep the effective rate in reasonable
1704 -- limits we use rateRecoveryTime, minLatency and maxLatency.
1705 in if deltaYields > 0
1707 let deltaYieldsFreq :: Double
1709 fromIntegral deltaYields /
1710 fromIntegral rateRecoveryTime
1711 yieldsFreq = 1.0 / fromIntegral targetLat
1712 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1713 requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
1714 adjustedLat = min (max requiredLat (minLatency range))
1716 in assert (adjustedLat > 0) $
1717 if wLatency <= adjustedLat
1718 then PartialWorker deltaYields
1719 else let workers = withLimit $ wLatency `div` adjustedLat
1720 limited = min workers (fromIntegral deltaYields)
1721 in ManyWorkers (fromIntegral limited) deltaYields
1723 let expectedDuration = fromIntegral effectiveYields * targetLat
1724 sleepTime = expectedDuration - svarElapsed
1725 maxSleepTime = maxLatency range - wLatency
1726 s = min sleepTime maxSleepTime
1727 in assert (sleepTime >= 0) $
1728 -- if s is less than 0 it means our maxSleepTime is less
1729 -- than the worker latency.
1730 if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
1735 Limited x -> min n (fromIntegral x)
1737-- | Get the worker latency without resetting workerPendingLatency
1738-- Returns (total yield count, base time, measured latency)
1739-- CAUTION! keep it in sync with collectLatency
1740getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
1741getWorkerLatency yinfo = do
1742 let cur = workerPendingLatency yinfo
1743 col = workerCollectedLatency yinfo
1744 longTerm = svarAllTimeLatency yinfo
1745 measured = workerMeasuredLatency yinfo
1747 (curTotalCount, curCount, curTime) <- readIORef cur
1748 (colTotalCount, colCount, colTime) <- readIORef col
1749 (lcount, ltime) <- readIORef longTerm
1750 prevLat <- readIORef measured
1752 let latCount = colCount + curCount
1753 latTime = colTime + curTime
1754 totalCount = colTotalCount + curTotalCount
1756 if latCount > 0 && latTime > 0
1757 then let lat = latTime `div` fromIntegral latCount
1758 -- XXX Give more weight to new?
1759 in (lat + prevLat) `div` 2
1761 return (lcount + totalCount, ltime, newLat)
1763isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1764isBeyondMaxRate sv yinfo = do
1765 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1766 now <- getTime Monotonic
1767 let duration = fromRelTime64 $ diffAbsTime64 now tstamp
1768 let targetLat = svarLatencyTarget yinfo
1769 gainLoss <- readIORef (svarGainedLostYields yinfo)
1770 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1771 wLatency targetLat (svarLatencyRange yinfo)
1772 cnt <- readIORef $ workerCount sv
1773 return $ case work of
1774 -- XXX set the worker's maxYields or polling interval based on yields
1775 PartialWorker _yields -> cnt > 1
1776 ManyWorkers n _ -> cnt > n
1779-- XXX in case of ahead style stream we need to take the heap size into account
1780-- because we return the workers on the basis of that which causes a condition
1781-- where we keep dispatching and they keep returning. So we must have exactly
1782-- the same logic for not dispatching and for returning.
1785-- True: can dispatch more
1786-- False: full, no more dispatches
1787dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1788dispatchWorkerPaced sv = do
1789 let yinfo = fromJust $ yieldRateInfo sv
1790 (svarYields, svarElapsed, wLatency) <- do
1791 now <- liftIO $ getTime Monotonic
1792 (yieldCount, baseTime, lat) <-
1793 liftIO $ collectLatency sv yinfo False
1794 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
1797 then fromMaybe lat (workerBootstrapLatency yinfo)
1800 return (yieldCount, elapsed, latency)
1803 -- Need to measure the latency with a single worker before we can perform
1807 let workerLimit = maxWorkerLimit sv
1808 let targetLat = svarLatencyTarget yinfo
1809 let range = svarLatencyRange yinfo
1810 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1811 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
1812 wLatency targetLat range
1814 -- XXX we need to take yieldLimit into account here. If we are at the
1815 -- end of the limit as well as the time, we should not be sleeping.
1816 -- If we are not actually planning to dispatch any more workers we need
1817 -- to take that in account.
1820 assert (s >= 0) (return ())
1821 -- XXX note that when we return from here we will block waiting
1822 -- for the result from the existing worker. If that takes too
1823 -- long we won't be able to send another worker until the
1826 -- Sleep only if there are no active workers, otherwise we will
1827 -- defer the output of those. Note we cannot use workerCount
1828 -- here as it is not a reliable way to ensure there are
1829 -- definitely no active workers. When workerCount is 0 we may
1830 -- still have a Stop event waiting in the outputQueue.
1831 done <- allThreadsDone sv
1832 when done $ void $ do
1833 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
1834 liftIO $ threadDelay (fromIntegral us)
1837 PartialWorker yields -> do
1838 assert (yields > 0) (return ())
1839 updateGainedLostYields yinfo yields
1841 done <- allThreadsDone sv
1842 when done $ void $ dispatchWorker yields sv
1844 ManyWorkers netWorkers yields -> do
1845 assert (netWorkers >= 1) (return ())
1846 assert (yields >= 0) (return ())
1847 updateGainedLostYields yinfo yields
1849 let periodRef = workerPollingInterval yinfo
1850 ycnt = max 1 $ yields `div` fromIntegral netWorkers
1851 period = min ycnt (fromIntegral magicMaxBuffer)
1853 old <- liftIO $ readIORef periodRef
1854 when (period < old) $
1855 liftIO $ writeIORef periodRef period
1857 cnt <- liftIO $ readIORef $ workerCount sv
1860 let total = netWorkers - cnt
1861 batch = max 1 $ fromIntegral $
1862 minThreadDelay `div` targetLat
1863 -- XXX stagger the workers over a period?
1864 -- XXX cannot sleep, as that would mean we cannot process
1865 -- the outputs. need to try a different mechanism to
1867 -- when (total > batch) $
1868 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1869 dispatchN (min total batch)
1874 updateGainedLostYields yinfo yields = do
1875 let buf = fromIntegral $ svarRateBuffer yinfo
1876 when (yields /= 0 && abs yields > buf) $ do
1881 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1887 r <- dispatchWorker 0 sv
1889 then dispatchN (n - 1)
1892-------------------------------------------------------------------------------
1893-- Worker dispatch and wait loop
1894-------------------------------------------------------------------------------
1896sendWorkerDelayPaced :: SVar t m a -> IO ()
1897sendWorkerDelayPaced _ = return ()
1899sendWorkerDelay :: SVar t m a -> IO ()
1900sendWorkerDelay _sv =
1901 -- XXX we need a better way to handle this than hardcoded delays. The
1902 -- delays may be different for different systems.
1903 -- If there is a usecase where this is required we can create a combinator
1904 -- to set it as a config in the state.
1907 ncpu <- getNumCapabilities
1910 if (svarStyle sv == AheadVar)
1911 then threadDelay 100
1914 if (svarStyle sv == AheadVar)
1915 then threadDelay 100
1920{-# NOINLINE sendWorkerWait #-}
1923 => (SVar t m a -> IO ())
1924 -> (SVar t m a -> m Bool)
1927sendWorkerWait delay dispatch sv = do
1928 -- Note that we are guaranteed to have at least one outstanding worker when
1929 -- we enter this function. So if we sleep we are guaranteed to be woken up
1930 -- by an outputDoorBell, when the worker exits.
1933 (_, n) <- liftIO $ readIORef (outputQueue sv)
1935 -- The queue may be empty temporarily if the worker has dequeued the
1936 -- work item but has not enqueued the remaining part yet. For the same
1937 -- reason, a worker may come back if it tries to dequeue and finds the
1938 -- queue empty, even though the whole work has not finished yet.
1940 -- If we find that the queue is empty, but it may be empty
1941 -- temporarily, when we checked it. If that's the case we might
1942 -- sleep indefinitely unless the active workers produce some
1943 -- output. We may deadlock specially if the otuput from the active
1944 -- workers depends on the future workers that we may never send.
1945 -- So in case the queue was temporarily empty set a flag to inform
1946 -- the enqueue to send us a doorbell.
1948 -- Note that this is just a best effort mechanism to avoid a
1949 -- deadlock. Deadlocks may still happen if for some weird reason
1950 -- the consuming computation shares an MVar or some other resource
1951 -- with the producing computation and gets blocked on that resource
1952 -- and therefore cannot do any pushworker to add more threads to
1953 -- the producer. In such cases the programmer should use a parallel
1954 -- style so that all the producers are scheduled immediately and
1955 -- unconditionally. We can also use a separate monitor thread to
1956 -- push workers instead of pushing them from the consumer, but then
1957 -- we are no longer using pull based concurrency rate adaptation.
1959 -- XXX update this in the tutorial.
1961 -- Having pending active workers does not mean that we are guaranteed
1962 -- to be woken up if we sleep. In case of Ahead streams, there may be
1963 -- queued items in the heap even though the outputQueue is empty, and
1964 -- we may have active workers which are deadlocked on those items to be
1965 -- processed by the consumer. We should either guarantee that any
1966 -- worker, before returning, clears the heap or we send a worker to
1967 -- clear it. Normally we always send a worker if no output is seen, but
1968 -- if the thread limit is reached or we are using pacing then we may
1969 -- not send a worker. See the concurrentApplication test in the tests,
1970 -- that test case requires at least one yield from the producer to not
1971 -- deadlock, if the last workers output is stuck in the heap then this
1972 -- test fails. This problem can be extended to n threads when the
1973 -- consumer may depend on the evaluation of next n items in the
1976 -- register for the outputDoorBell before we check the queue so that if
1977 -- we sleep because the queue was empty we are guaranteed to get a
1978 -- doorbell on the next enqueue.
1980 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
1981 liftIO storeLoadBarrier
1982 canDoMore <- dispatch sv
1984 -- XXX test for the case when we miss sending a worker when the worker
1985 -- count is more than 1500.
1987 -- XXX Assert here that if the heap is not empty then there is at
1988 -- least one outstanding worker. Otherwise we could be sleeping
1992 then sendWorkerWait delay dispatch sv
1994 liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
1995 $ takeMVar (outputDoorBell sv)
1996 (_, len) <- liftIO $ readIORef (outputQueue sv)
1997 when (len <= 0) $ sendWorkerWait delay dispatch sv
1999-------------------------------------------------------------------------------
2000-- Reading from the workers' output queue/buffer
2001-------------------------------------------------------------------------------
2003{-# INLINE readOutputQBasic #-}
2004readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)
2005readOutputQBasic q = atomicModifyIORefCAS q $ \x -> (([],0), x)
2007{-# INLINE readOutputQRaw #-}
2008readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
2009readOutputQRaw sv = do
2010 (list, len) <- readOutputQBasic (outputQueue sv)
2011 when (svarInspectMode sv) $ do
2012 let ref = maxOutQSize $ svarStats sv
2013 oqLen <- readIORef ref
2014 when (len > oqLen) $ writeIORef ref len
2017readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2018readOutputQBounded sv = do
2019 (list, len) <- liftIO $ readOutputQRaw sv
2020 -- When there is no output seen we dispatch more workers to help
2021 -- out if there is work pending in the work queue.
2025 -- send a worker proactively, if needed, even before we start
2026 -- processing the output. This may degrade single processor
2027 -- perf but improves multi-processor, because of more
2035 cnt <- liftIO $ readIORef $ workerCount sv
2036 when (cnt <= 0) $ do
2037 done <- liftIO $ isWorkDone sv
2038 when (not done) (pushWorker 0 sv)
2040 {-# INLINE blockingRead #-}
2042 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
2043 liftIO (fst `fmap` readOutputQRaw sv)
2045readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2046readOutputQPaced sv = do
2047 (list, len) <- liftIO $ readOutputQRaw sv
2051 -- XXX send a worker proactively, if needed, even before we start
2052 -- processing the output.
2053 void $ dispatchWorkerPaced sv
2058 {-# INLINE blockingRead #-}
2060 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
2061 liftIO (fst `fmap` readOutputQRaw sv)
2063postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
2064postProcessBounded sv = do
2065 workersDone <- allThreadsDone sv
2066 -- There may still be work pending even if there are no workers pending
2067 -- because all the workers may return if the outputQueue becomes full. In
2068 -- that case send off a worker to kickstart the work again.
2070 -- Note that isWorkDone can only be safely checked if all workers are done.
2071 -- When some workers are in progress they may have decremented the yield
2072 -- Limit and later ending up incrementing it again. If we look at the yield
2073 -- limit in that window we may falsely say that it is 0 and therefore we
2077 r <- liftIO $ isWorkDone sv
2078 -- Note that we need to guarantee a worker, therefore we cannot just
2079 -- use dispatchWorker which may or may not send a worker.
2080 when (not r) (pushWorker 0 sv)
2081 -- XXX do we need to dispatch many here?
2082 -- void $ dispatchWorker sv
2086postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
2087postProcessPaced sv = do
2088 workersDone <- allThreadsDone sv
2089 -- XXX If during consumption we figure out we are getting delayed then we
2090 -- should trigger dispatch there as well. We should try to check on the
2091 -- workers after consuming every n item from the buffer?
2094 r <- liftIO $ isWorkDone sv
2096 void $ dispatchWorkerPaced sv
2097 -- Note that we need to guarantee a worker since the work is not
2098 -- finished, therefore we cannot just rely on dispatchWorkerPaced
2099 -- which may or may not send a worker.
2100 noWorker <- allThreadsDone sv
2101 when noWorker $ pushWorker 0 sv
2105-------------------------------------------------------------------------------
2107-------------------------------------------------------------------------------
2109getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
2110getYieldRateInfo st = do
2111 -- convert rate in Hertz to latency in Nanoseconds
2112 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
2113 case getStreamRate st of
2114 Just (Rate low goal high buf) ->
2115 let l = rateToLatency goal
2116 minl = rateToLatency high
2117 maxl = rateToLatency low
2118 in mkYieldRateInfo l (LatencyRange minl maxl) buf
2119 Nothing -> return Nothing
2123 mkYieldRateInfo latency latRange buf = do
2124 measured <- newIORef 0
2125 wcur <- newIORef (0,0,0)
2126 wcol <- newIORef (0,0,0)
2127 now <- getTime Monotonic
2128 wlong <- newIORef (0,now)
2129 period <- newIORef 1
2130 gainLoss <- newIORef (Count 0)
2132 return $ Just YieldRateInfo
2133 { svarLatencyTarget = latency
2134 , svarLatencyRange = latRange
2135 , svarRateBuffer = buf
2136 , svarGainedLostYields = gainLoss
2137 , workerBootstrapLatency = getStreamLatency st
2138 , workerPollingInterval = period
2139 , workerMeasuredLatency = measured
2140 , workerPendingLatency = wcur
2141 , workerCollectedLatency = wcol
2142 , svarAllTimeLatency = wlong
2145newSVarStats :: IO SVarStats
2148 maxWrk <- newIORef 0
2152 avgLat <- newIORef (0, NanoSecond64 0)
2153 maxLat <- newIORef (NanoSecond64 0)
2154 minLat <- newIORef (NanoSecond64 0)
2155 stpTime <- newIORef Nothing
2158 { totalDispatches = disp
2159 , maxWorkers = maxWrk
2160 , maxOutQSize = maxOq
2161 , maxHeapSize = maxHs
2162 , maxWorkQSize = maxWq
2163 , avgWorkerLatency = avgLat
2164 , minWorkerLatency = minLat
2165 , maxWorkerLatency = maxLat
2166 , svarStopTime = stpTime
2169-- XXX remove polymorphism in t, inline f
2170getAheadSVar :: MonadAsync m
2172 -> ( IORef ([t m a], Int)
2173 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2180getAheadSVar st f mrun = do
2181 outQ <- newIORef ([], 0)
2182 -- the second component of the tuple is "Nothing" when heap is being
2183 -- cleared, "Just n" when we are expecting sequence number n to arrive
2184 -- before we can start clearing the heap.
2185 outH <- newIORef (H.empty, Just 0)
2186 outQMv <- newEmptyMVar
2187 active <- newIORef 0
2188 wfw <- newIORef False
2189 running <- newIORef S.empty
2190 -- Sequence number is incremented whenever something is queued, therefore,
2191 -- first sequence number would be 0
2192 q <- newIORef ([], -1)
2193 stopMVar <- newMVar ()
2194 yl <- case getYieldLimit st of
2195 Nothing -> return Nothing
2196 Just x -> Just <$> newIORef x
2197 rateInfo <- getYieldRateInfo st
2199 stats <- newSVarStats
2202 let getSVar sv readOutput postProc = SVar
2203 { outputQueue = outQ
2204 , outputQueueFromConsumer = undefined
2205 , remainingWork = yl
2206 , maxBufferLimit = getMaxBuffer st
2207 , pushBufferSpace = undefined
2208 , pushBufferPolicy = undefined
2209 , pushBufferMVar = undefined
2210 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
2211 , yieldRateInfo = rateInfo
2212 , outputDoorBell = outQMv
2213 , outputDoorBellFromConsumer = undefined
2214 , readOutputQ = readOutput sv
2215 , postProcess = postProc sv
2216 , workerThreads = running
2217 , workLoop = f q outH st{streamVar = Just sv} sv
2218 , enqueue = enqueueAhead sv q
2219 , isWorkDone = isWorkDoneAhead sv q outH
2220 , isQueueDone = isQueueDoneAhead sv q
2221 , needDoorBell = wfw
2222 , svarStyle = AheadVar
2223 , svarStopStyle = StopNone
2224 , svarStopBy = undefined
2226 , workerCount = active
2227 , accountThread = delThread sv
2228 , workerStopMVar = stopMVar
2230 , svarInspectMode = getInspectMode st
2232 , aheadWorkQueue = q
2238 case getStreamRate st of
2239 Nothing -> getSVar sv readOutputQBounded postProcessBounded
2240 Just _ -> getSVar sv readOutputQPaced postProcessPaced
2245 {-# INLINE isQueueDoneAhead #-}
2246 isQueueDoneAhead sv q = do
2247 queueDone <- checkEmpty q
2249 case remainingWork sv of
2253 Nothing -> return False
2254 -- XXX note that yieldsDone can only be authoritative only when there
2255 -- are no workers running. If there are active workers they can
2256 -- later increment the yield count and therefore change the result.
2257 return $ yieldsDone || queueDone
2259 {-# INLINE isWorkDoneAhead #-}
2260 isWorkDoneAhead sv q ref = do
2262 (hp, _) <- readIORef ref
2263 return (H.size hp <= 0)
2264 queueDone <- isQueueDoneAhead sv q
2265 return $ heapDone && queueDone
2268 (xs, _) <- readIORef q
2271getParallelSVar :: MonadIO m
2272 => SVarStopStyle -> State t m a -> RunInIO m -> IO (SVar t m a)
2273getParallelSVar ss st mrun = do
2274 outQ <- newIORef ([], 0)
2275 outQRev <- newIORef ([], 0)
2276 outQMv <- newEmptyMVar
2277 outQMvRev <- newEmptyMVar
2278 active <- newIORef 0
2279 running <- newIORef S.empty
2280 yl <- case getYieldLimit st of
2281 Nothing -> return Nothing
2282 Just x -> Just <$> newIORef x
2283 rateInfo <- getYieldRateInfo st
2285 case getMaxBuffer st of
2286 Unlimited -> undefined
2287 Limited x -> fromIntegral x
2288 remBuf <- newIORef bufLim
2289 pbMVar <- newMVar ()
2291 stats <- newSVarStats
2296 StopBy -> liftIO $ newIORef undefined
2297 _ -> return undefined
2300 SVar { outputQueue = outQ
2301 , outputQueueFromConsumer = outQRev
2302 , remainingWork = yl
2303 , maxBufferLimit = getMaxBuffer st
2304 , pushBufferSpace = remBuf
2305 , pushBufferPolicy = PushBufferBlock
2306 , pushBufferMVar = pbMVar
2307 , maxWorkerLimit = Unlimited
2308 -- Used only for diagnostics
2309 , yieldRateInfo = rateInfo
2310 , outputDoorBell = outQMv
2311 , outputDoorBellFromConsumer = outQMvRev
2312 , readOutputQ = readOutputQPar sv
2313 , postProcess = allThreadsDone sv
2314 , workerThreads = running
2315 , workLoop = undefined
2316 , enqueue = undefined
2317 , isWorkDone = undefined
2318 , isQueueDone = undefined
2319 , needDoorBell = undefined
2320 , svarStyle = ParallelVar
2321 , svarStopStyle = ss
2322 , svarStopBy = stopBy
2324 , workerCount = active
2325 , accountThread = modifyThread sv
2326 , workerStopMVar = undefined
2328 , svarInspectMode = getInspectMode st
2330 , aheadWorkQueue = undefined
2331 , outputHeap = undefined
2338 readOutputQPar sv = liftIO $ do
2339 withDiagMVar sv "readOutputQPar: doorbell"
2340 $ takeMVar (outputDoorBell sv)
2341 case yieldRateInfo sv of
2342 Nothing -> return ()
2343 Just yinfo -> void $ collectLatency sv yinfo False
2344 r <- fst `fmap` readOutputQRaw sv
2346 void $ tryTakeMVar (pushBufferMVar sv)
2349 void $ tryPutMVar (pushBufferMVar sv) ()
2352sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
2353sendFirstWorker sv m = do
2354 -- Note: We must have all the work on the queue before sending the
2355 -- pushworker, otherwise the pushworker may exit before we even get a
2357 runIn <- captureMonadState
2358 liftIO $ enqueue sv (runIn, m)
2359 case yieldRateInfo sv of
2360 Nothing -> pushWorker 0 sv
2362 if svarLatencyTarget yinfo == maxBound
2363 then liftIO $ threadDelay maxBound
2364 else pushWorker 1 sv
2367{-# INLINABLE newAheadVar #-}
2368newAheadVar :: MonadAsync m
2371 -> ( IORef ([t m a], Int)
2372 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2378newAheadVar st m wloop = do
2379 mrun <- captureMonadState
2380 sv <- liftIO $ getAheadSVar st wloop mrun
2381 sendFirstWorker sv m
2383{-# INLINABLE newParallelVar #-}
2384newParallelVar :: MonadAsync m
2385 => SVarStopStyle -> State t m a -> m (SVar t m a)
2386newParallelVar ss st = do
2387 mrun <- captureMonadState
2388 liftIO $ getParallelSVar ss st mrun
2390-------------------------------------------------------------------------------
2391-- Write a stream to an SVar
2392-------------------------------------------------------------------------------
2394-- XXX this errors out for Parallel/Ahead SVars
2395-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
2396-- be read back from the SVar using 'fromSVar'.
2397toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
2398toStreamVar sv m = do
2399 runIn <- captureMonadState
2400 liftIO $ enqueue sv (runIn, m)
2401 done <- allThreadsDone sv
2402 -- XXX This is safe only when called from the consumer thread or when no
2403 -- consumer is present. There may be a race if we are not running in the
2405 -- XXX do this only if the work queue is not empty. The work may have been
2406 -- carried out by existing workers.
2408 case yieldRateInfo sv of
2409 Nothing -> pushWorker 0 sv
2410 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