1{-# OPTIONS_HADDOCK hide #-}
3{-# LANGUAGE KindSignatures #-}
4{-# LANGUAGE ConstraintKinds #-}
5{-# LANGUAGE ExistentialQuantification #-}
6{-# LANGUAGE FlexibleContexts #-}
7{-# LANGUAGE FlexibleInstances #-}
8{-# LANGUAGE GeneralizedNewtypeDeriving #-}
9{-# LANGUAGE LambdaCase #-}
10{-# LANGUAGE MagicHash #-}
11{-# LANGUAGE MultiParamTypeClasses #-}
12{-# LANGUAGE RankNTypes #-}
13{-# LANGUAGE ScopedTypeVariables #-}
14{-# LANGUAGE UnboxedTuples #-}
17-- Module : Streamly.Internal.Data.SVar
18-- Copyright : (c) 2017 Harendra Kumar
21-- Maintainer : streamly@composewell.com
22-- Stability : experimental
25module Streamly.Internal.Data.SVar
32 -- State threaded around the stream
50 , cleanupSVarFromWorker
75 , HeapDequeueResult(..)
93 , decrementBufferLimit
94 , incrementBufferLimit
111import Control.Concurrent
112 (ThreadId, myThreadId, threadDelay, throwTo)
113import Control.Concurrent.MVar
114 (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar, tryReadMVar)
115import Control.Exception
116 (SomeException(..), catch, mask, assert, Exception, catches,
117 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
118 BlockedIndefinitelyOnSTM(..))
119import Control.Monad (when)
120import Control.Monad.Catch (MonadThrow)
121import Control.Monad.IO.Class (MonadIO(..))
122import Control.Monad.Trans.Control (MonadBaseControl, control, StM)
123import Streamly.Internal.Data.Atomics
124 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
126import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
127import Data.Functor (void)
128import Data.Heap (Heap, Entry(..))
129import Data.Int (Int64)
130#if __GLASGOW_HASKELL__ >= 800
131import Data.Kind (Type)
134 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
135import Data.Maybe (fromJust)
136#if __GLASGOW_HASKELL__ < 808
137import Data.Semigroup ((<>))
140import GHC.Conc (ThreadId(..))
142import GHC.IO (IO(..))
143import Streamly.Internal.Data.Time.Clock (Clock(..), getTime)
144import Streamly.Internal.Data.Time.Units
145 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
146 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
147import System.IO (hPutStrLn, stderr)
149import qualified Data.Heap as H
150import qualified Data.Set as S
152newtype Count = Count Int64
164------------------------------------------------------------------------------
165-- Parent child thread communication type
166------------------------------------------------------------------------------
168data ThreadAbort = ThreadAbort deriving Show
170instance Exception ThreadAbort
172-- | Events that a child thread may send to a parent thread.
175 | ChildStop ThreadId (Maybe SomeException)
177#if __GLASGOW_HASKELL__ < 800
180-- | Sorting out-of-turn outputs in a heap for Ahead style streams
181data AheadHeapEntry (t :: (Type -> Type) -> Type -> Type) m a =
184 | AheadEntryStream (t m a)
187------------------------------------------------------------------------------
188-- State threaded around the monad for thread management
189------------------------------------------------------------------------------
191-- | Identify the type of the SVar. Two computations using the same style can
192-- be scheduled on the same SVar.
194 AsyncVar -- depth first concurrent
195 | WAsyncVar -- breadth first concurrent
196 | ParallelVar -- all parallel
197 | AheadVar -- Concurrent look ahead
200-- | An SVar or a Stream Var is a conduit to the output from multiple streams
201-- running concurrently and asynchronously. An SVar can be thought of as an
202-- asynchronous IO handle. We can write any number of streams to an SVar in a
203-- non-blocking manner and then read them back at any time at any pace. The
204-- SVar would run the streams asynchronously and accumulate results. An SVar
205-- may not really execute the stream completely and accumulate all the results.
206-- However, it ensures that the reader can read the results at whatever paces
207-- it wants to read. The SVar monitors and adapts to the consumer's pace.
209-- An SVar is a mini scheduler, it has an associated workLoop that holds the
210-- stream tasks to be picked and run by a pool of worker threads. It has an
211-- associated output queue where the output stream elements are placed by the
212-- worker threads. A outputDoorBell is used by the worker threads to intimate the
213-- consumer thread about availability of new results in the output queue. More
214-- workers are added to the SVar by 'fromStreamVar' on demand if the output
215-- produced is not keeping pace with the consumer. On bounded SVars, workers
216-- block on the output queue to provide throttling of the producer when the
217-- consumer is not pulling fast enough. The number of workers may even get
218-- reduced depending on the consuming pace.
220-- New work is enqueued either at the time of creation of the SVar or as a
221-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
222-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
224-- We measure the individual worker latencies to estimate the number of workers
225-- needed or the amount of time we have to sleep between dispatches to achieve
226-- a particular rate when controlled pace mode it used.
227data WorkerInfo = WorkerInfo
228 { workerYieldMax :: Count -- 0 means unlimited
229 -- total number of yields by the worker till now
230 , workerYieldCount :: IORef Count
231 -- yieldCount at start, timestamp
232 , workerLatencyStart :: IORef (Count, AbsTime)
236-- | Specifies the stream yield rate in yields per second (@Hertz@).
237-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
238-- allow only as many yields as we have accumulated as per 'rateGoal' since the
239-- start of time. If the consumer or the producer is slower or faster, the
240-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
241-- between the two by increasing or decreasing the pull rate from the producer.
242-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
243-- much as 'rateBuffer'.
245-- 'rateLow' puts a bound on how low the instantaneous rate can go when
246-- recovering the rate gap. In other words, it determines the maximum yield
247-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
248-- rate can go when recovering the rate gap. In other words, it determines the
249-- minimum yield latency. We reduce the latency by increasing concurrency,
250-- therefore we can say that it puts an upper bound on concurrency.
252-- If the 'rateGoal' is 0 or negative the stream never yields a value.
253-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
257 { rateLow :: Double -- ^ The lower rate limit
258 , rateGoal :: Double -- ^ The target rate we want to achieve
259 , rateHigh :: Double -- ^ The upper rate limit
260 , rateBuffer :: Int -- ^ Maximum slack from the goal
263data LatencyRange = LatencyRange
264 { minLatency :: NanoSecond64
265 , maxLatency :: NanoSecond64
269data YieldRateInfo = YieldRateInfo
270 { svarLatencyTarget :: NanoSecond64
271 , svarLatencyRange :: LatencyRange
272 , svarRateBuffer :: Int
274 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
275 -- read by the worker threads
276 , svarGainedLostYields :: IORef Count
278 -- Actual latency/througput as seen from the consumer side, we count the
279 -- yields and the time it took to generates those yields. This is used to
280 -- increase or decrease the number of workers needed to achieve the desired
281 -- rate. The idle time of workers is adjusted in this, so that we only
282 -- account for the rate when the consumer actually demands data.
283 -- XXX interval latency is enough, we can move this under diagnostics build
284 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
285 -- read by the worker threads
286 , svarAllTimeLatency :: IORef (Count, AbsTime)
288 -- XXX Worker latency specified by the user to be used before the first
289 -- actual measurement arrives. Not yet implemented
290 , workerBootstrapLatency :: Maybe NanoSecond64
292 -- After how many yields the worker should update the latency information.
293 -- If the latency is high, this count is kept lower and vice-versa. XXX If
294 -- the latency suddenly becomes too high this count may remain too high for
295 -- long time, in such cases the consumer can change it.
296 -- 0 means no latency computation
297 -- XXX this is derivable from workerMeasuredLatency, can be removed.
298 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
299 -- read by the worker threads
300 , workerPollingInterval :: IORef Count
302 -- This is in progress latency stats maintained by the workers which we
303 -- empty into workerCollectedLatency stats at certain intervals - whenever
304 -- we process the stream elements yielded in this period. The first count
305 -- is all yields, the second count is only those yields for which the
306 -- latency was measured to be non-zero (note that if the timer resolution
307 -- is low the measured latency may be zero e.g. on JS platform).
308 -- [LOCKING] Locked access. Modified by the consumer thread as well as
309 -- worker threads. Workers modify it periodically based on
310 -- workerPollingInterval and not on every yield to reduce the locking
312 -- (allYieldCount, yieldCount, timeTaken)
313 , workerPendingLatency :: IORef (Count, Count, NanoSecond64)
315 -- This is the second level stat which is an accmulation from
316 -- workerPendingLatency stats. We keep accumulating latencies in this
317 -- bucket until we have stats for a sufficient period and then we reset it
318 -- to start collecting for the next period and retain the computed average
319 -- latency for the last period in workerMeasuredLatency.
320 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
321 -- read by the worker threads
322 -- (allYieldCount, yieldCount, timeTaken)
323 , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)
325 -- Latency as measured by workers, aggregated for the last period.
326 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
327 -- read by the worker threads
328 , workerMeasuredLatency :: IORef NanoSecond64
331data SVarStats = SVarStats {
332 totalDispatches :: IORef Int
333 , maxWorkers :: IORef Int
334 , maxOutQSize :: IORef Int
335 , maxHeapSize :: IORef Int
336 , maxWorkQSize :: IORef Int
337 , avgWorkerLatency :: IORef (Count, NanoSecond64)
338 , minWorkerLatency :: IORef NanoSecond64
339 , maxWorkerLatency :: IORef NanoSecond64
340 , svarStopTime :: IORef (Maybe AbsTime)
343-- This is essentially a 'Maybe Word' type
344data Limit = Unlimited | Limited Word deriving Show
346instance Eq Limit where
347 Unlimited == Unlimited = True
348 Unlimited == Limited _ = False
349 Limited _ == Unlimited = False
350 Limited x == Limited y = x == y
352instance Ord Limit where
353 Unlimited <= Unlimited = True
354 Unlimited <= Limited _ = False
355 Limited _ <= Unlimited = True
356 Limited x <= Limited y = x <= y
358-- When to stop the composed stream.
360 StopNone -- stops only when all streams are finished
361 | StopAny -- stop when any stream finishes
362 | StopBy -- stop when a specific stream finishes
365-- | Buffering policy for persistent push workers (in ParallelT). In a pull
366-- style SVar (in AsyncT, AheadT etc.), the consumer side dispatches workers on
367-- demand, workers terminate if the buffer is full or if the consumer is not
368-- cosuming fast enough. In a push style SVar, a worker is dispatched only
369-- once, workers are persistent and keep pushing work to the consumer via a
370-- bounded buffer. If the buffer becomes full the worker either blocks, or it
371-- can drop an item from the buffer to make space.
373-- Pull style SVars are useful in lazy stream evaluation whereas push style
374-- SVars are useful in strict left Folds.
376-- XXX Maybe we can separate the implementation in two different types instead
377-- of using a common SVar type.
379data PushBufferPolicy =
380 PushBufferDropNew -- drop the latest element and continue
381 | PushBufferDropOld -- drop the oldest element and continue
382 | PushBufferBlock -- block the thread until space
385-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have
386-- references to the original SVar stored in several functions which will keep
387-- pointing to the original data and the new updates won't reflect there.
388-- Any updateable parts must be kept in mutable references (IORef).
390data SVar t m a = SVar
393 svarStyle :: SVarStyle
394 , svarMrun :: RunInIO m
395 , svarStopStyle :: SVarStopStyle
396 , svarStopBy :: IORef ThreadId
398 -- Shared output queue (events, length)
399 -- XXX For better efficiency we can try a preallocated array type (perhaps
400 -- something like a vector) that allows an O(1) append. That way we will
401 -- avoid constructing and reversing the list. Possibly we can also avoid
402 -- the GC copying overhead. When the size increases we should be able to
403 -- allocate the array in chunks.
405 -- [LOCKING] Frequent locked access. This is updated by workers on each
406 -- yield and once in a while read by the consumer thread. This could have
407 -- big locking overhead if the number of workers is high.
409 -- XXX We can use a per-CPU data structure to reduce the locking overhead.
410 -- However, a per-cpu structure cannot guarantee the exact sequence in
411 -- which the elements were added, though that may not be important.
412 , outputQueue :: IORef ([ChildEvent a], Int)
414 -- [LOCKING] Infrequent MVar. Used when the outputQ transitions from empty
415 -- to non-empty, or a work item is queued by a worker to the work queue and
416 -- needDoorBell is set by the consumer.
417 , outputDoorBell :: MVar () -- signal the consumer about output
418 , readOutputQ :: m [ChildEvent a]
419 , postProcess :: m Bool
421 -- Combined/aggregate parameters
422 -- This is truncated to maxBufferLimit if set to more than that. Otherwise
423 -- potentially each worker may yield one value to the buffer in the worst
424 -- case exceeding the requested buffer size.
425 , maxWorkerLimit :: Limit
426 , maxBufferLimit :: Limit
427 -- These two are valid and used only when maxBufferLimit is Limited.
428 , pushBufferSpace :: IORef Count
429 , pushBufferPolicy :: PushBufferPolicy
430 -- [LOCKING] The consumer puts this MVar after emptying the buffer, workers
431 -- block on it when the buffer becomes full. No overhead unless the buffer
433 , pushBufferMVar :: MVar ()
435 -- [LOCKING] Read only access by consumer when dispatching a worker.
436 -- Decremented by workers when picking work and undo decrement if the
437 -- worker does not yield a value.
438 , remainingWork :: Maybe (IORef Count)
439 , yieldRateInfo :: Maybe YieldRateInfo
441 -- Used only by bounded SVar types
442 , enqueue :: t m a -> IO ()
443 , isWorkDone :: IO Bool
444 , isQueueDone :: IO Bool
445 , needDoorBell :: IORef Bool
446 , workLoop :: Maybe WorkerInfo -> m ()
448 -- Shared, thread tracking
449 -- [LOCKING] Updated unlocked only by consumer thread in case of
450 -- Async/Ahead style SVars. Updated locked by worker threads in case of
452 , workerThreads :: IORef (Set ThreadId)
453 -- [LOCKING] Updated locked by consumer thread when dispatching a worker
454 -- and by the worker threads when the thread stops. This is read unsafely
455 -- at several places where we want to rely on an approximate value.
456 , workerCount :: IORef Int
457 , accountThread :: ThreadId -> m ()
458 , workerStopMVar :: MVar ()
460 , svarStats :: SVarStats
461 -- to track garbage collection of SVar
462 , svarRef :: Maybe (IORef ())
464 -- Only for diagnostics
465 , svarInspectMode :: Bool
466 , svarCreator :: ThreadId
467 , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))
469 -- Shared work queue (stream, seqNo)
470 , aheadWorkQueue :: IORef ([t m a], Int)
473-------------------------------------------------------------------------------
474-- State for concurrency control
475-------------------------------------------------------------------------------
477-- XXX we can put the resettable fields in a oneShotConfig field and others in
478-- a persistentConfig field. That way reset would be fast and scalable
479-- irrespective of the number of fields.
481-- XXX make all these Limited types and use phantom types to distinguish them
482data State t m a = State
483 { -- one shot configuration, automatically reset for each API call
484 streamVar :: Maybe (SVar t m a)
485 , _yieldLimit :: Maybe Count
487 -- persistent configuration, state that remains valid until changed by
488 -- an explicit setting via a combinator.
489 , _threadsHigh :: Limit
490 , _bufferHigh :: Limit
491 -- XXX these two can be collapsed into a single type
492 , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency
493 , _maxStreamRate :: Maybe Rate
494 , _inspectMode :: Bool
497-------------------------------------------------------------------------------
498-- State defaults and reset
499-------------------------------------------------------------------------------
501-- A magical value for the buffer size arrived at by running the smallest
502-- possible task and measuring the optimal value of the buffer for that. This
503-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
505magicMaxBuffer :: Word
508defaultMaxThreads, defaultMaxBuffer :: Limit
509defaultMaxThreads = Limited magicMaxBuffer
510defaultMaxBuffer = Limited magicMaxBuffer
512-- The fields prefixed by an _ are not to be accessed or updated directly but
513-- via smart accessor APIs.
514defState :: State t m a
516 { streamVar = Nothing
517 , _yieldLimit = Nothing
518 , _threadsHigh = defaultMaxThreads
519 , _bufferHigh = defaultMaxBuffer
520 , _maxStreamRate = Nothing
521 , _streamLatency = Nothing
522 , _inspectMode = False
525-- XXX if perf gets affected we can have all the Nothing params in a single
526-- structure so that we reset is fast. We can also use rewrite rules such that
527-- reset occurs only in concurrent streams to reduce the impact on serial
529-- We can optimize this so that we clear it only if it is a Just value, it
530-- results in slightly better perf for zip/zipM but the performance of scan
531-- worsens a lot, it does not fuse.
533-- XXX This has a side effect of clearing the SVar and yieldLimit, therefore it
534-- should not be used to convert from the same type to the same type, unless
535-- you want to clear the SVar. For clearing the SVar you should be using the
536-- appropriate unStream functions instead.
538-- | Adapt the stream state from one type to another.
539adaptState :: State t m a -> State t n b
541 { streamVar = Nothing
542 , _yieldLimit = Nothing
545-------------------------------------------------------------------------------
546-- Smart get/set routines for State
547-------------------------------------------------------------------------------
549-- Use get/set routines instead of directly accessing the State fields
550setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
551setYieldLimit lim st =
558 else Just (fromIntegral n)
561getYieldLimit :: State t m a -> Maybe Count
562getYieldLimit = _yieldLimit
564setMaxThreads :: Int -> State t m a -> State t m a
570 then defaultMaxThreads
571 else Limited (fromIntegral n)
574getMaxThreads :: State t m a -> Limit
575getMaxThreads = _threadsHigh
577setMaxBuffer :: Int -> State t m a -> State t m a
583 then defaultMaxBuffer
584 else Limited (fromIntegral n)
587getMaxBuffer :: State t m a -> Limit
588getMaxBuffer = _bufferHigh
590setStreamRate :: Maybe Rate -> State t m a -> State t m a
591setStreamRate r st = st { _maxStreamRate = r }
593getStreamRate :: State t m a -> Maybe Rate
594getStreamRate = _maxStreamRate
596setStreamLatency :: Int -> State t m a -> State t m a
597setStreamLatency n st =
598 st { _streamLatency =
601 else Just (fromIntegral n)
604getStreamLatency :: State t m a -> Maybe NanoSecond64
605getStreamLatency = _streamLatency
607setInspectMode :: State t m a -> State t m a
608setInspectMode st = st { _inspectMode = True }
610getInspectMode :: State t m a -> Bool
611getInspectMode = _inspectMode
613-------------------------------------------------------------------------------
615-------------------------------------------------------------------------------
617cleanupSVar :: SVar t m a -> IO ()
619 workers <- readIORef (workerThreads sv)
620 Prelude.mapM_ (`throwTo` ThreadAbort)
623cleanupSVarFromWorker :: SVar t m a -> IO ()
624cleanupSVarFromWorker sv = do
625 workers <- readIORef (workerThreads sv)
627 Prelude.mapM_ (`throwTo` ThreadAbort)
628 (Prelude.filter (/= self) $ S.toList workers)
630-------------------------------------------------------------------------------
631-- Worker latency data collection
632-------------------------------------------------------------------------------
634-- Every once in a while workers update the latencies and check the yield rate.
635-- They return if we are above the expected yield rate. If we check too often
636-- it may impact performance, if we check less often we may have a stale
637-- picture. We update every minThreadDelay but we translate that into a yield
638-- count based on latency so that the checking overhead is little.
640-- XXX use a generation count to indicate that the value is updated. If the
641-- value is updated an existing worker must check it again on the next yield.
642-- Otherwise it is possible that we may keep updating it and because of the mod
643-- worker keeps skipping it.
644updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
645updateWorkerPollingInterval yinfo latency = do
646 let periodRef = workerPollingInterval yinfo
647 cnt = max 1 $ minThreadDelay `div` latency
648 period = min cnt (fromIntegral magicMaxBuffer)
650 writeIORef periodRef (fromIntegral period)
652{-# INLINE recordMinMaxLatency #-}
653recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()
654recordMinMaxLatency sv new = do
655 let ss = svarStats sv
656 minLat <- readIORef (minWorkerLatency ss)
657 when (new < minLat || minLat == 0) $
658 writeIORef (minWorkerLatency ss) new
660 maxLat <- readIORef (maxWorkerLatency ss)
661 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
663recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()
664recordAvgLatency sv (count, time) = do
665 let ss = svarStats sv
666 modifyIORef (avgWorkerLatency ss) $
667 \(cnt, t) -> (cnt + count, t + time)
669-- Pour the pending latency stats into a collection bucket
670{-# INLINE collectWorkerPendingLatency #-}
671collectWorkerPendingLatency
672 :: IORef (Count, Count, NanoSecond64)
673 -> IORef (Count, Count, NanoSecond64)
674 -> IO (Count, Maybe (Count, NanoSecond64))
675collectWorkerPendingLatency cur col = do
676 (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
678 (fcnt, cnt, t) <- readIORef col
679 let totalCount = fcnt + fcount
680 latCount = cnt + count
682 writeIORef col (totalCount, latCount, latTime)
684 assert (latCount == 0 || latTime /= 0) (return ())
686 if latCount > 0 && latTime > 0
687 then Just $ (latCount, latTime)
689 return (totalCount, latPair)
691{-# INLINE shouldUseCollectedBatch #-}
692shouldUseCollectedBatch
698shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
699 let r = fromIntegral newLat / fromIntegral prevLat :: Double
700 in (collectedYields > fromIntegral magicMaxBuffer)
701 || (collectedTime > minThreadDelay)
702 || (prevLat > 0 && (r > 2 || r < 0.5))
705-- Returns a triple, (1) yield count since last collection, (2) the base time
706-- when we started counting, (3) average latency in the last measurement
707-- period. The former two are used for accurate measurement of the going rate
708-- whereas the average is used for future estimates e.g. how many workers
709-- should be maintained to maintain the rate.
710-- CAUTION! keep it in sync with getWorkerLatency
711collectLatency :: SVar t m a
714 -> IO (Count, AbsTime, NanoSecond64)
715collectLatency sv yinfo drain = do
716 let cur = workerPendingLatency yinfo
717 col = workerCollectedLatency yinfo
718 longTerm = svarAllTimeLatency yinfo
719 measured = workerMeasuredLatency yinfo
721 (newCount, newLatPair) <- collectWorkerPendingLatency cur col
722 (lcount, ltime) <- readIORef longTerm
723 prevLat <- readIORef measured
725 let newLcount = lcount + newCount
726 retWith lat = return (newLcount, ltime, lat)
729 Nothing -> retWith prevLat
730 Just (count, time) -> do
731 let newLat = time `div` (fromIntegral count)
732 when (svarInspectMode sv) $ recordMinMaxLatency sv newLat
733 -- When we have collected a significant sized batch we compute the
734 -- new latency using that batch and return the new latency,
735 -- otherwise we return the previous latency derived from the
737 if shouldUseCollectedBatch newCount time newLat prevLat || drain
739 -- XXX make this NOINLINE?
740 updateWorkerPollingInterval yinfo (max newLat prevLat)
741 when (svarInspectMode sv) $ recordAvgLatency sv (count, time)
742 writeIORef col (0, 0, 0)
743 writeIORef measured ((prevLat + newLat) `div` 2)
744 modifyIORef longTerm $ \(_, t) -> (newLcount, t)
748-------------------------------------------------------------------------------
749-- Dumping the SVar for debug/diag
750-------------------------------------------------------------------------------
752dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
753dumpSVarStats sv ss style = do
754 case yieldRateInfo sv of
757 _ <- liftIO $ collectLatency sv yinfo True
760 dispatches <- readIORef $ totalDispatches ss
761 maxWrk <- readIORef $ maxWorkers ss
762 maxOq <- readIORef $ maxOutQSize ss
763 maxHp <- readIORef $ maxHeapSize ss
764 minLat <- readIORef $ minWorkerLatency ss
765 maxLat <- readIORef $ maxWorkerLatency ss
766 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
767 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
768 Nothing -> return (0, 0, 0)
770 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
773 t <- readIORef (svarStopTime ss)
774 gl <- readIORef (svarGainedLostYields yinfo)
777 now <- getTime Monotonic
778 let interval = diffAbsTime64 now startTime
779 return (cnt, gl, interval `div` fromIntegral cnt)
781 let interval = diffAbsTime64 stopTime startTime
782 return (cnt, gl, interval `div` fromIntegral cnt)
783 else return (0, 0, 0)
786 [ "total dispatches = " <> show dispatches
787 , "max workers = " <> show maxWrk
788 , "max outQSize = " <> show maxOq
789 <> (if style == AheadVar
790 then "\nheap max size = " <> show maxHp
793 then "\nmin worker latency = " <> showNanoSecond64 minLat
796 then "\nmax worker latency = " <> showNanoSecond64 maxLat
799 then let lat = avgTime `div` fromIntegral avgCnt
800 in "\navg worker latency = " <> showNanoSecond64 lat
803 then "\nSVar latency = " <> showRelTime64 svarLat
806 then "\nSVar yield count = " <> show svarCnt
808 <> (if svarGainLossCnt > 0
809 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
813{-# NOINLINE dumpSVar #-}
814dumpSVar :: SVar t m a -> IO String
816 (oqList, oqLen) <- readIORef $ outputQueue sv
817 db <- tryReadMVar $ outputDoorBell sv
819 if svarStyle sv == AheadVar
821 (oheap, oheapSeq) <- readIORef $ outputHeap sv
822 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
824 [ "heap length = " <> show (H.size oheap)
825 , "heap seqeunce = " <> show oheapSeq
826 , "work queue length = " <> show (length wq)
827 , "work queue sequence = " <> show wqSeq
831 let style = svarStyle sv
833 if style /= ParallelVar
834 then readIORef $ needDoorBell sv
836 rthread <- readIORef $ workerThreads sv
837 workers <- readIORef $ workerCount sv
838 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
842 "Creator tid = " <> show (svarCreator sv),
843 "style = " <> show (svarStyle sv)
844 , "---------CURRENT STATE-----------"
845 , "outputQueue length computed = " <> show (length oqList)
846 , "outputQueue length maintained = " <> show oqLen
847 -- XXX print the types of events in the outputQueue, first 5
848 , "outputDoorBell = " <> show db
852 [ "needDoorBell = " <> show waiting
853 , "running threads = " <> show rthread
854 -- XXX print the status of first 5 threads
855 , "running thread count = " <> show workers
857 <> "---------STATS-----------\n"
860-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
861-- can keep it on in production to debug problems quickly if and when they
862-- happen, but it may result in unexpected output when threads are left hanging
863-- until they are GCed because the consumer went away.
865{-# NOINLINE mvarExcHandler #-}
866mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
867mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
868 svInfo <- dumpSVar sv
869 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
872{-# NOINLINE stmExcHandler #-}
873stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
874stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
875 svInfo <- dumpSVar sv
876 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
879withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
880withDiagMVar sv label action =
881 if svarInspectMode sv
883 action `catches` [ Handler (mvarExcHandler sv label)
884 , Handler (stmExcHandler sv label)
888------------------------------------------------------------------------------
890------------------------------------------------------------------------------
892-- | A monad that can perform concurrent or parallel IO operations. Streams
893-- that can be composed concurrently require the underlying monad to be
897type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
899-- When we run computations concurrently, we completely isolate the state of
900-- the concurrent computations from the parent computation. The invariant is
901-- that we should never be running two concurrent computations in the same
902-- thread without using the runInIO function. Also, we should never be running
903-- a concurrent computation in the parent thread, otherwise it may affect the
904-- state of the parent which is against the defined semantics of concurrent
906newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }
908captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
909captureMonadState = control $ \run -> run (return $ RunInIO run)
911-- Stolen from the async package. The perf improvement is modest, 2% on a
912-- thread heavy benchmark (parallel composition using noop computations).
913-- A version of forkIO that does not include the outer exception
914-- handler: saves a bit of time when we will be installing our own
916{-# INLINE rawForkIO #-}
917rawForkIO :: IO () -> IO ThreadId
918rawForkIO action = IO $ \ s ->
919 case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)
922doFork :: MonadBaseControl IO m
925 -> (SomeException -> IO ())
927doFork action (RunInIO mrun) exHandler =
929 mask $ \restore -> do
930 tid <- rawForkIO $ catch (restore $ void $ mrun action)
934------------------------------------------------------------------------------
935-- Collecting results from child workers in a streamed fashion
936------------------------------------------------------------------------------
938-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
939-- faster, along with the fields in sv required by send?
940-- XXX make it noinline
942-- XXX we may want to employ an increment and decrement in batches when the
943-- througput is high or when the cost of synchronization is high. For example
944-- if the application is distributed then inc/dec of a shared variable may be
947-- A worker decrements the yield limit before it executes an action. However,
948-- the action may not result in an element being yielded, in that case we have
949-- to increment the yield limit.
951-- Note that we need it to be an Int type so that we have the ability to undo a
952-- decrement that takes it below zero.
953{-# INLINE decrementYieldLimit #-}
954decrementYieldLimit :: SVar t m a -> IO Bool
955decrementYieldLimit sv =
956 case remainingWork sv of
957 Nothing -> return True
959 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
962{-# INLINE incrementYieldLimit #-}
963incrementYieldLimit :: SVar t m a -> IO ()
964incrementYieldLimit sv =
965 case remainingWork sv of
967 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
969-- XXX exception safety of all atomic/MVar operations
971-- TBD Each worker can have their own queue and the consumer can empty one
972-- queue at a time, that way contention can be reduced.
974-- XXX Only yields should be counted in the buffer limit and not the Stop
977{-# INLINE decrementBufferLimit #-}
978decrementBufferLimit :: SVar t m a -> IO ()
979decrementBufferLimit sv =
980 case maxBufferLimit sv of
981 Unlimited -> return ()
983 let ref = pushBufferSpace sv
984 old <- atomicModifyIORefCAS ref $ \x ->
985 (if x >= 1 then x - 1 else x, x)
987 case pushBufferPolicy sv of
988 PushBufferBlock -> blockAndRetry
989 PushBufferDropNew -> do
990 -- We just drop one item and proceed. It is possible
991 -- that by the time we drop the item the consumer
992 -- thread might have run and created space in the
993 -- buffer, but we do not care about that condition.
994 -- This is not pedantically correct but it should be
996 -- XXX we may want to drop only if n == maxBuf
997 -- otherwise we must have space in the buffer and a
998 -- decrement should be possible.
999 block <- atomicModifyIORefCAS (outputQueue sv) $
1002 [] -> (([],n), True)
1003 _ : xs -> ((xs, n - 1), False)
1004 when block blockAndRetry
1005 -- XXX need a dequeue or ring buffer for this
1006 PushBufferDropOld -> undefined
1011 let ref = pushBufferSpace sv
1012 liftIO $ takeMVar (pushBufferMVar sv)
1013 old <- atomicModifyIORefCAS ref $ \x ->
1014 (if x >= 1 then x - 1 else x, x)
1015 -- When multiple threads sleep on takeMVar, the first thread would
1016 -- wakeup due to a putMVar by the consumer, but the rest of the threads
1017 -- would have to put back the MVar after taking it and decrementing the
1018 -- buffer count, otherwise all other threads will remain asleep.
1020 then void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
1021 -- We do not put the MVar back in this case, instead we
1022 -- wait for the consumer to put it.
1025{-# INLINE incrementBufferLimit #-}
1026incrementBufferLimit :: SVar t m a -> IO ()
1027incrementBufferLimit sv =
1028 case maxBufferLimit sv of
1029 Unlimited -> return ()
1031 atomicModifyIORefCAS_ (pushBufferSpace sv) (+ 1)
1033 void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
1035{-# INLINE resetBufferLimit #-}
1036resetBufferLimit :: SVar t m a -> IO ()
1037resetBufferLimit sv =
1038 case maxBufferLimit sv of
1039 Unlimited -> return ()
1040 Limited n -> atomicModifyIORefCAS_ (pushBufferSpace sv)
1041 (const (fromIntegral n))
1043-- | This function is used by the producer threads to queue output for the
1044-- consumer thread to consume. Returns whether the queue has more space.
1045send :: SVar t m a -> ChildEvent a -> IO Int
1047 -- XXX can the access to outputQueue and maxBufferLimit be made faster
1049 oldlen <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
1050 ((msg : es, n + 1), n)
1051 when (oldlen <= 0) $ do
1052 -- The wake up must happen only after the store has finished otherwise
1053 -- we can have lost wakeup problems.
1055 -- Since multiple workers can try this at the same time, it is possible
1056 -- that we may put a spurious MVar after the consumer has already seen
1057 -- the output. But that's harmless, at worst it may cause the consumer
1058 -- to read the queue again and find it empty.
1059 -- The important point is that the consumer is guaranteed to receive a
1060 -- doorbell if something was added to the queue after it empties it.
1061 void $ tryPutMVar (outputDoorBell sv) ()
1064workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
1065workerCollectLatency winfo = do
1066 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
1067 cnt1 <- readIORef (workerYieldCount winfo)
1068 let cnt = cnt1 - cnt0
1072 t1 <- getTime Monotonic
1073 let period = fromRelTime64 $ diffAbsTime64 t1 t0
1074 writeIORef (workerLatencyStart winfo) (cnt1, t1)
1075 return $ Just (cnt, period)
1078-- XXX There are a number of gotchas in measuring latencies.
1079-- 1) We measure latencies only when a worker yields a value
1080-- 2) It is possible that a stream calls the stop continuation, in which case
1081-- the worker would not yield a value and we would not account that worker in
1082-- latencies. Even though this case should ideally be accounted we do not
1083-- account it because we cannot or do not distinguish it from the case
1085-- 3) It is possible that a worker returns without yielding anything because it
1086-- never got a chance to pick up work.
1087-- 4) If the system timer resolution is lower than the latency, the latency
1088-- computation turns out to be zero.
1090-- We can fix this if we measure the latencies by counting the work items
1091-- picked rather than based on the outputs yielded.
1092workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
1093workerUpdateLatency yinfo winfo = do
1094 r <- workerCollectLatency winfo
1096 Just (cnt, period) -> do
1097 -- NOTE: On JS platform the timer resolution could be pretty low. When
1098 -- the timer resolution is low, measurement of latencies could be
1099 -- tricky. All the worker latencies will turn out to be zero if they
1100 -- are lower than the resolution. We only take into account those
1101 -- measurements which are more than the timer resolution.
1103 let ref = workerPendingLatency yinfo
1104 (cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
1105 atomicModifyIORefCAS_ ref $
1106 \(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
1107 Nothing -> return ()
1109updateYieldCount :: WorkerInfo -> IO Count
1110updateYieldCount winfo = do
1111 cnt <- readIORef (workerYieldCount winfo)
1113 writeIORef (workerYieldCount winfo) cnt1
1116isBeyondMaxYield :: Count -> WorkerInfo -> Bool
1117isBeyondMaxYield cnt winfo =
1118 let ymax = workerYieldMax winfo
1119 in ymax /= 0 && cnt >= ymax
1121-- XXX we should do rate control periodically based on the total yields rather
1122-- than based on the worker local yields as other workers may have yielded more
1123-- and we should stop based on the aggregate yields. However, latency update
1124-- period can be based on individual worker yields.
1125{-# NOINLINE checkRatePeriodic #-}
1126checkRatePeriodic :: SVar t m a
1131checkRatePeriodic sv yinfo winfo ycnt = do
1132 i <- readIORef (workerPollingInterval yinfo)
1133 -- XXX use generation count to check if the interval has been updated
1134 if i /= 0 && (ycnt `mod` i) == 0
1136 workerUpdateLatency yinfo winfo
1137 -- XXX not required for parallel streams
1138 isBeyondMaxRate sv yinfo
1141-- CAUTION! this also updates the yield count and therefore should be called
1142-- only when we are actually yielding an element.
1143{-# NOINLINE workerRateControl #-}
1144workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
1145workerRateControl sv yinfo winfo = do
1146 cnt <- updateYieldCount winfo
1147 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
1148 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
1150-- XXX we should do rate control here but not latency update in case of ahead
1151-- streams. latency update must be done when we yield directly to outputQueue
1152-- or when we yield to heap.
1154-- returns whether the worker should continue (True) or stop (False).
1155{-# INLINE sendYield #-}
1156sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
1157sendYield sv mwinfo msg = do
1158 oldlen <- send sv msg
1159 let limit = maxBufferLimit sv
1160 bufferSpaceOk <- case limit of
1161 Unlimited -> return True
1163 active <- readIORef (workerCount sv)
1164 return $ (oldlen + 1) < (fromIntegral lim - active)
1168 case yieldRateInfo sv of
1169 Nothing -> return True
1170 Just yinfo -> workerRateControl sv yinfo winfo
1171 Nothing -> return True
1172 return $ bufferSpaceOk && rateLimitOk
1174{-# INLINE workerStopUpdate #-}
1175workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
1176workerStopUpdate winfo info = do
1177 i <- readIORef (workerPollingInterval info)
1178 when (i /= 0) $ workerUpdateLatency info winfo
1180{-# INLINABLE sendStop #-}
1181sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
1182sendStop sv mwinfo = do
1183 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
1184 case (mwinfo, yieldRateInfo sv) of
1185 (Just winfo, Just info) ->
1186 workerStopUpdate winfo info
1189 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
1191-------------------------------------------------------------------------------
1193-------------------------------------------------------------------------------
1195{-# INLINE ringDoorBell #-}
1196ringDoorBell :: SVar t m a -> IO ()
1199 w <- readIORef $ needDoorBell sv
1201 -- Note: the sequence of operations is important for correctness here.
1202 -- We need to set the flag to false strictly before sending the
1203 -- outputDoorBell, otherwise the outputDoorBell may get processed too
1204 -- early and then we may set the flag to False to later making the
1205 -- consumer lose the flag, even without receiving a outputDoorBell.
1206 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1207 void $ tryPutMVar (outputDoorBell sv) ()
1209-------------------------------------------------------------------------------
1211-------------------------------------------------------------------------------
1213-- Note: For purely right associated expressions this queue should have at most
1214-- one element. It grows to more than one when we have left associcated
1215-- expressions. Large left associated compositions can grow this to a
1217{-# INLINE enqueueLIFO #-}
1218enqueueLIFO :: SVar t m a -> IORef [t m a] -> t m a -> IO ()
1219enqueueLIFO sv q m = do
1220 atomicModifyIORefCAS_ q $ \ms -> m : ms
1223-------------------------------------------------------------------------------
1225-------------------------------------------------------------------------------
1227-- XXX we can use the Ahead style sequence/heap mechanism to make the best
1228-- effort to always try to finish the streams on the left side of an expression
1229-- first as long as possible.
1231{-# INLINE enqueueFIFO #-}
1232enqueueFIFO :: SVar t m a -> LinkedQueue (t m a) -> t m a -> IO ()
1233enqueueFIFO sv q m = do
1237-------------------------------------------------------------------------------
1239-------------------------------------------------------------------------------
1241-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
1242-- but always serve them in the same order as they appear in the stream. To
1243-- implement lookahead streams efficiently we assign a sequence number to each
1244-- task when the task is picked up for execution. When the task finishes, the
1245-- output is tagged with the same sequence number and we rearrange the outputs
1246-- in sequence based on that number.
1248-- To explain the mechanism imagine that the current task at the head of the
1249-- stream has a "token" to yield to the outputQueue. The ownership of the token
1250-- is determined by the current sequence number is maintained in outputHeap.
1251-- Sequence number is assigned when a task is queued. When a thread dequeues a
1252-- task it picks up the sequence number as well and when the output is ready it
1253-- uses the sequence number to queue the output to the outputQueue.
1255-- The thread with current sequence number sends the output directly to the
1256-- outputQueue. Other threads push the output to the outputHeap. When the task
1257-- being queued on the heap is a stream of many elements we evaluate only the
1258-- first element and keep the rest of the unevaluated computation in the heap.
1259-- When such a task gets the "token" for outputQueue it evaluates and directly
1260-- yields all the elements to the outputQueue without checking for the
1263-- Note that no two outputs in the heap can have the same sequence numbers and
1264-- therefore we do not need a stable heap. We have also separated the buffer
1265-- for the current task (outputQueue) and the pending tasks (outputHeap) so
1266-- that the pending tasks cannot interfere with the current task. Note that for
1267-- a single task just the outputQueue is enough and for the case of many
1268-- threads just a heap is good enough. However we balance between these two
1269-- cases, so that both are efficient.
1271-- For bigger streams it may make sense to have separate buffers for each
1272-- stream. However, for singleton streams this may become inefficient. However,
1273-- if we do not have separate buffers, then the streams that come later in
1274-- sequence may hog the buffer, hindering the streams that are ahead. For this
1275-- reason we have a single element buffer limitation for the streams being
1276-- executed in advance.
1278-- This scheme works pretty efficiently with less than 40% extra overhead
1279-- compared to the Async streams where we do not have any kind of sequencing of
1280-- the outputs. It is especially devised so that we are most efficient when we
1281-- have short tasks and need just a single thread. Also when a thread yields
1282-- many items it can hold lockfree access to the outputQueue and do it
1285-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
1286-- that they do not hog the resources and hinder the progress of the threads in
1289-- XXX Left associated ahead expressions are expensive. We start a new SVar for
1290-- each left associative expression. The queue is used only for right
1291-- associated expression, we queue the right expression and execute the left.
1292-- Thererefore the queue never has more than one item in it.
1294-- XXX we can fix this. When we queue more than one item on the queue we can
1295-- mark the previously queued item as not-runnable. The not-runnable item is
1296-- not dequeued until the already running one has finished and at that time we
1297-- would also know the exact sequence number of the already queued item.
1299-- we can even run the already queued items but they will have to be sorted in
1300-- layers in the heap. We can use a list of heaps for that.
1301{-# INLINE enqueueAhead #-}
1302enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1303enqueueAhead sv q m = do
1304 atomicModifyIORefCAS_ q $ \ case
1305 ([], n) -> ([m], n + 1) -- increment sequence
1306 _ -> error "enqueueAhead: queue is not empty"
1309-- enqueue without incrementing the sequence number
1310{-# INLINE reEnqueueAhead #-}
1311reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1312reEnqueueAhead sv q m = do
1313 atomicModifyIORefCAS_ q $ \ case
1314 ([], n) -> ([m], n) -- DO NOT increment sequence
1315 _ -> error "reEnqueueAhead: queue is not empty"
1318-- Normally the thread that has the token should never go away. The token gets
1319-- handed over to another thread, but someone or the other has the token at any
1320-- point of time. But if the task that has the token finds that the outputQueue
1321-- is full, in that case it can go away without even handing over the token to
1322-- another thread. In that case it sets the nextSequence number in the heap its
1323-- own sequence number before going away. To handle this case, any task that
1324-- does not have the token tries to dequeue from the heap first before
1325-- dequeuing from the work queue. If it finds that the task at the top of the
1326-- heap is the one that owns the current sequence number then it grabs the
1327-- token and starts with that.
1329-- XXX instead of queueing just the head element and the remaining computation
1330-- on the heap, evaluate as many as we can and place them on the heap. But we
1331-- need to give higher priority to the lower sequence numbers so that lower
1332-- priority tasks do not fill up the heap making higher priority tasks block
1333-- due to full heap. Maybe we can have a weighted space for them in the heap.
1334-- The weight is inversely proportional to the sequence number.
1336-- XXX review for livelock
1338{-# INLINE queueEmptyAhead #-}
1339queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
1340queueEmptyAhead q = liftIO $ do
1341 (xs, _) <- readIORef q
1344{-# INLINE dequeueAhead #-}
1345dequeueAhead :: MonadIO m
1346 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
1347dequeueAhead q = liftIO $
1348 atomicModifyIORefCAS q $ \case
1349 ([], n) -> (([], n), Nothing)
1350 (x : [], n) -> (([], n), Just (x, n))
1351 _ -> error "more than one item on queue"
1353-------------------------------------------------------------------------------
1355-------------------------------------------------------------------------------
1357withIORef :: IORef a -> (a -> IO b) -> IO b
1358withIORef ref f = readIORef ref >>= f
1360atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
1361atomicModifyIORef_ ref f =
1362 atomicModifyIORef ref $ \x -> (f x, ())
1364data HeapDequeueResult t m a =
1367 | Ready (Entry Int (AheadHeapEntry t m a))
1369{-# INLINE dequeueFromHeap #-}
1371 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1372 -> IO (HeapDequeueResult t m a)
1373dequeueFromHeap hpVar =
1374 atomicModifyIORef hpVar $ \pair@(hp, snum) ->
1376 Nothing -> (pair, Clearing)
1380 Just (ent@(Entry seqNo _ev), hp') ->
1382 then ((hp', Nothing), Ready ent)
1383 else assert (seqNo >= n) (pair, Waiting n)
1384 Nothing -> (pair, Waiting n)
1386{-# INLINE dequeueFromHeapSeq #-}
1388 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1390 -> IO (HeapDequeueResult t m a)
1391dequeueFromHeapSeq hpVar i =
1392 atomicModifyIORef hpVar $ \(hp, snum) ->
1397 Just (ent@(Entry seqNo _ev), hp') ->
1399 then ((hp', Nothing), Ready ent)
1400 else assert (seqNo >= i) ((hp, Just i), Waiting i)
1401 Nothing -> ((hp, Just i), Waiting i)
1402 Just _ -> error "dequeueFromHeapSeq: unreachable"
1404heapIsSane :: Maybe Int -> Int -> Bool
1405heapIsSane snum seqNo =
1408 Just n -> seqNo >= n
1410{-# INLINE requeueOnHeapTop #-}
1412 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1413 -> Entry Int (AheadHeapEntry t m a)
1416requeueOnHeapTop hpVar ent seqNo =
1417 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1418 assert (heapIsSane snum seqNo) (H.insert ent hp, Just seqNo)
1420{-# INLINE updateHeapSeq #-}
1422 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1425updateHeapSeq hpVar seqNo =
1426 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1427 assert (heapIsSane snum seqNo) (hp, Just seqNo)
1429-------------------------------------------------------------------------------
1431-------------------------------------------------------------------------------
1433-- XXX To be implemented. Use a linked queue like WAsync and put back the
1434-- remaining computation at the back of the queue instead of the heap, and
1435-- increment the sequence number.
1437-------------------------------------------------------------------------------
1438-- Dispatching workers and tracking them
1439-------------------------------------------------------------------------------
1441-- Thread tracking is needed for two reasons:
1443-- 1) Killing threads on exceptions. Threads may not be left to go away by
1444-- themselves because they may run for significant times before going away or
1445-- worse they may be stuck in IO and never go away.
1447-- 2) To know when all threads are done and the stream has ended.
1449{-# NOINLINE addThread #-}
1450addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1452 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
1454-- This is cheaper than modifyThread because we do not have to send a
1455-- outputDoorBell This can make a difference when more workers are being
1457{-# INLINE delThread #-}
1458delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1460 liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
1462-- If present then delete else add. This takes care of out of order add and
1463-- delete i.e. a delete arriving before we even added a thread.
1464-- This occurs when the forked thread is done even before the 'addThread' right
1465-- after the fork gets a chance to run.
1466{-# INLINE modifyThread #-}
1467modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1468modifyThread sv tid = do
1469 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1471 then let new = S.delete tid old in (new, new)
1472 else let new = S.insert tid old in (new, old)
1473 when (null changed) $
1476 void $ tryPutMVar (outputDoorBell sv) ()
1478-- | This is safe even if we are adding more threads concurrently because if
1479-- a child thread is adding another thread then anyway 'workerThreads' will
1481{-# INLINE allThreadsDone #-}
1482allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1483allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1485{-# NOINLINE handleChildException #-}
1486handleChildException :: SVar t m a -> SomeException -> IO ()
1487handleChildException sv e = do
1489 void $ send sv (ChildStop tid (Just e))
1491{-# NOINLINE recordMaxWorkers #-}
1492recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1493recordMaxWorkers sv = liftIO $ do
1494 active <- readIORef (workerCount sv)
1495 maxWrk <- readIORef (maxWorkers $ svarStats sv)
1496 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1497 modifyIORef (totalDispatches $ svarStats sv) (+1)
1499{-# NOINLINE pushWorker #-}
1500pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1501pushWorker yieldMax sv = do
1502 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1503 when (svarInspectMode sv) $ recordMaxWorkers sv
1504 -- This allocation matters when significant number of workers are being
1505 -- sent. We allocate it only when needed.
1507 case yieldRateInfo sv of
1508 Nothing -> return Nothing
1509 Just _ -> liftIO $ do
1510 cntRef <- newIORef 0
1511 t <- getTime Monotonic
1512 lat <- newIORef (0, t)
1513 return $ Just WorkerInfo
1514 { workerYieldMax = yieldMax
1515 , workerYieldCount = cntRef
1516 , workerLatencyStart = lat
1518 doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
1521-- XXX we can push the workerCount modification in accountThread and use the
1522-- same pushWorker for Parallel case as well.
1524-- | In contrast to pushWorker which always happens only from the consumer
1525-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1526-- producer side. So we need to use a thread safe modification of
1527-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1528-- using a CAS based modification.
1529{-# INLINE pushWorkerPar #-}
1532 => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1533pushWorkerPar sv wloop =
1534 if svarInspectMode sv
1536 else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
1541 {-# NOINLINE forkWithDiag #-}
1543 -- We do not use workerCount in case of ParallelVar but still there is
1544 -- no harm in maintaining it correctly.
1545 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1547 -- This allocation matters when significant number of workers are being
1548 -- sent. We allocate it only when needed. The overhead increases by 4x.
1550 case yieldRateInfo sv of
1551 Nothing -> return Nothing
1552 Just _ -> liftIO $ do
1553 cntRef <- newIORef 0
1554 t <- getTime Monotonic
1555 lat <- newIORef (0, t)
1556 return $ Just WorkerInfo
1557 { workerYieldMax = 0
1558 , workerYieldCount = cntRef
1559 , workerLatencyStart = lat
1562 doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
1566-- True: can dispatch more
1567-- False: cannot dispatch any more
1568dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1569dispatchWorker yieldCount sv = do
1570 let workerLimit = maxWorkerLimit sv
1571 -- XXX in case of Ahead streams we should not send more than one worker
1572 -- when the work queue is done but heap is not done.
1573 done <- liftIO $ isWorkDone sv
1574 -- Note, "done" may not mean that the work is actually finished if there
1575 -- are workers active, because there may be a worker which has not yet
1576 -- queued the leftover work.
1579 qDone <- liftIO $ isQueueDone sv
1580 -- This count may not be accurate as it is decremented by the workers
1581 -- and we have no synchronization with that decrement.
1582 active <- liftIO $ readIORef $ workerCount sv
1585 -- Note that we may deadlock if the previous workers (tasks in the
1586 -- stream) wait/depend on the future workers (tasks in the stream)
1587 -- executing. In that case we should either configure the maxWorker
1588 -- count to higher or use parallel style instead of ahead or async
1590 limit <- case remainingWork sv of
1591 Nothing -> return workerLimit
1593 n <- liftIO $ readIORef ref
1594 case yieldRateInfo sv of
1595 Just _ -> return workerLimit
1599 Unlimited -> Limited (fromIntegral n)
1600 Limited lim -> Limited $ min lim (fromIntegral n)
1602 -- XXX for ahead streams shall we take the heap yields into account
1603 -- for controlling the dispatch? We should not dispatch if the heap
1604 -- has already got the limit covered.
1605 let dispatch = pushWorker yieldCount sv >> return True
1607 Unlimited -> dispatch
1608 -- Note that the use of remainingWork and workerCount is not
1609 -- atomic and the counts may even have changed between reading
1610 -- and using them here, so this is just approximate logic and
1611 -- we cannot rely on it for correctness. We may actually
1612 -- dispatch more workers than required.
1613 Limited lim | lim > fromIntegral active -> dispatch
1616 when (active <= 0) $ pushWorker 0 sv
1620-------------------------------------------------------------------------------
1621-- Dispatch workers with rate control
1622-------------------------------------------------------------------------------
1624-- | This is a magic number and it is overloaded, and used at several places to
1627-- 1. If we have to sleep to slowdown this is the minimum period that we
1628-- accumulate before we sleep. Also, workers do not stop until this much
1629-- sleep time is accumulated.
1630-- 3. Collected latencies are computed and transferred to measured latency
1631-- after a minimum of this period.
1632minThreadDelay :: NanoSecond64
1633minThreadDelay = 1000000
1635-- | Another magic number! When we have to start more workers to cover up a
1636-- number of yields that we are lagging by then we cannot start one worker for
1637-- each yield because that may be a very big number and if the latency of the
1638-- workers is low these number of yields could be very high. We assume that we
1639-- run each extra worker for at least this much time.
1640rateRecoveryTime :: NanoSecond64
1641rateRecoveryTime = 1000000
1643-- We either block, or send one worker with limited yield count or one or more
1644-- workers with unlimited yield count.
1646 = BlockWait NanoSecond64
1647 | PartialWorker Count
1648 | ManyWorkers Int Count
1651-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1661estimateWorkers workerLimit svarYields gainLossYields
1662 svarElapsed wLatency targetLat range =
1663 -- XXX we can have a maxEfficiency combinator as well which runs the
1664 -- producer at the maximal efficiency i.e. the number of workers are chosen
1665 -- such that the latency is minimum or within a range. Or we can call it
1666 -- maxWorkerLatency.
1669 -- How many workers do we need to acheive the required rate?
1671 -- When the workers are IO bound we can increase the throughput by
1672 -- increasing the number of workers as long as the IO device has enough
1673 -- capacity to process all the requests concurrently. If the IO
1674 -- bandwidth is saturated increasing the workers won't help. Also, if
1675 -- the CPU utilization in processing all these requests exceeds the CPU
1676 -- bandwidth, then increasing the number of workers won't help.
1678 -- When the workers are purely CPU bound, increasing the workers beyond
1679 -- the number of CPUs won't help.
1681 -- TODO - measure the CPU and IO requirements of the workers. Have a
1682 -- way to specify the max bandwidth of the underlying IO mechanism and
1683 -- use that to determine the max rate of workers, and also take the CPU
1684 -- bandwidth into account. We can also discover the IO bandwidth if we
1685 -- know that we are not CPU bound, then how much steady state rate are
1686 -- we able to acheive. Design tests for CPU bound and IO bound cases.
1688 -- Calculate how many yields are we ahead or behind to match the exact
1689 -- required rate. Based on that we increase or decrease the effective
1692 -- When the worker latency is lower than required latency we begin with
1693 -- a yield and then wait rather than first waiting and then yielding.
1694 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1695 effectiveYields = svarYields + gainLossYields
1696 deltaYields = fromIntegral targetYields - effectiveYields
1698 -- We recover the deficit by running at a higher/lower rate for a
1699 -- certain amount of time. To keep the effective rate in reasonable
1700 -- limits we use rateRecoveryTime, minLatency and maxLatency.
1701 in if deltaYields > 0
1703 let deltaYieldsFreq :: Double
1705 fromIntegral deltaYields /
1706 fromIntegral rateRecoveryTime
1707 yieldsFreq = 1.0 / fromIntegral targetLat
1708 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1709 requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
1710 adjustedLat = min (max requiredLat (minLatency range))
1712 in assert (adjustedLat > 0) $
1713 if wLatency <= adjustedLat
1714 then PartialWorker deltaYields
1715 else let workers = withLimit $ wLatency `div` adjustedLat
1716 limited = min workers (fromIntegral deltaYields)
1717 in ManyWorkers (fromIntegral limited) deltaYields
1719 let expectedDuration = fromIntegral effectiveYields * targetLat
1720 sleepTime = expectedDuration - svarElapsed
1721 maxSleepTime = maxLatency range - wLatency
1722 s = min sleepTime maxSleepTime
1723 in assert (sleepTime >= 0) $
1724 -- if s is less than 0 it means our maxSleepTime is less
1725 -- than the worker latency.
1726 if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
1731 Limited x -> min n (fromIntegral x)
1733-- | Get the worker latency without resetting workerPendingLatency
1734-- Returns (total yield count, base time, measured latency)
1735-- CAUTION! keep it in sync with collectLatency
1736getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
1737getWorkerLatency yinfo = do
1738 let cur = workerPendingLatency yinfo
1739 col = workerCollectedLatency yinfo
1740 longTerm = svarAllTimeLatency yinfo
1741 measured = workerMeasuredLatency yinfo
1743 (curTotalCount, curCount, curTime) <- readIORef cur
1744 (colTotalCount, colCount, colTime) <- readIORef col
1745 (lcount, ltime) <- readIORef longTerm
1746 prevLat <- readIORef measured
1748 let latCount = colCount + curCount
1749 latTime = colTime + curTime
1750 totalCount = colTotalCount + curTotalCount
1752 if latCount > 0 && latTime > 0
1753 then let lat = latTime `div` fromIntegral latCount
1754 -- XXX Give more weight to new?
1755 in (lat + prevLat) `div` 2
1757 return (lcount + totalCount, ltime, newLat)
1759isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1760isBeyondMaxRate sv yinfo = do
1761 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1762 now <- getTime Monotonic
1763 let duration = fromRelTime64 $ diffAbsTime64 now tstamp
1764 let targetLat = svarLatencyTarget yinfo
1765 gainLoss <- readIORef (svarGainedLostYields yinfo)
1766 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1767 wLatency targetLat (svarLatencyRange yinfo)
1768 cnt <- readIORef $ workerCount sv
1769 return $ case work of
1770 -- XXX set the worker's maxYields or polling interval based on yields
1771 PartialWorker _yields -> cnt > 1
1772 ManyWorkers n _ -> cnt > n
1775-- XXX in case of ahead style stream we need to take the heap size into account
1776-- because we return the workers on the basis of that which causes a condition
1777-- where we keep dispatching and they keep returning. So we must have exactly
1778-- the same logic for not dispatching and for returning.
1781-- True: can dispatch more
1782-- False: full, no more dispatches
1783dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1784dispatchWorkerPaced sv = do
1785 let yinfo = fromJust $ yieldRateInfo sv
1786 (svarYields, svarElapsed, wLatency) <- do
1787 now <- liftIO $ getTime Monotonic
1788 (yieldCount, baseTime, lat) <-
1789 liftIO $ collectLatency sv yinfo False
1790 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
1794 case workerBootstrapLatency yinfo of
1799 return (yieldCount, elapsed, latency)
1802 -- Need to measure the latency with a single worker before we can perform
1806 let workerLimit = maxWorkerLimit sv
1807 let targetLat = svarLatencyTarget yinfo
1808 let range = svarLatencyRange yinfo
1809 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1810 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
1811 wLatency targetLat range
1813 -- XXX we need to take yieldLimit into account here. If we are at the
1814 -- end of the limit as well as the time, we should not be sleeping.
1815 -- If we are not actually planning to dispatch any more workers we need
1816 -- to take that in account.
1819 assert (s >= 0) (return ())
1820 -- XXX note that when we return from here we will block waiting
1821 -- for the result from the existing worker. If that takes too
1822 -- long we won't be able to send another worker until the
1825 -- Sleep only if there are no active workers, otherwise we will
1826 -- defer the output of those. Note we cannot use workerCount
1827 -- here as it is not a reliable way to ensure there are
1828 -- definitely no active workers. When workerCount is 0 we may
1829 -- still have a Stop event waiting in the outputQueue.
1830 done <- allThreadsDone sv
1831 when done $ void $ do
1832 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
1833 liftIO $ threadDelay (fromIntegral us)
1836 PartialWorker yields -> do
1837 assert (yields > 0) (return ())
1838 updateGainedLostYields yinfo yields
1840 done <- allThreadsDone sv
1841 when done $ void $ dispatchWorker yields sv
1843 ManyWorkers netWorkers yields -> do
1844 assert (netWorkers >= 1) (return ())
1845 assert (yields >= 0) (return ())
1846 updateGainedLostYields yinfo yields
1848 let periodRef = workerPollingInterval yinfo
1849 ycnt = max 1 $ yields `div` fromIntegral netWorkers
1850 period = min ycnt (fromIntegral magicMaxBuffer)
1852 old <- liftIO $ readIORef periodRef
1853 when (period < old) $
1854 liftIO $ writeIORef periodRef period
1856 cnt <- liftIO $ readIORef $ workerCount sv
1859 let total = netWorkers - cnt
1860 batch = max 1 $ fromIntegral $
1861 minThreadDelay `div` targetLat
1862 -- XXX stagger the workers over a period?
1863 -- XXX cannot sleep, as that would mean we cannot process
1864 -- the outputs. need to try a different mechanism to
1866 -- when (total > batch) $
1867 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1868 dispatchN (min total batch)
1873 updateGainedLostYields yinfo yields = do
1874 let buf = fromIntegral $ svarRateBuffer yinfo
1875 when (yields /= 0 && abs yields > buf) $ do
1880 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1886 r <- dispatchWorker 0 sv
1888 then dispatchN (n - 1)
1891-------------------------------------------------------------------------------
1892-- Worker dispatch and wait loop
1893-------------------------------------------------------------------------------
1895sendWorkerDelayPaced :: SVar t m a -> IO ()
1896sendWorkerDelayPaced _ = return ()
1898sendWorkerDelay :: SVar t m a -> IO ()
1899sendWorkerDelay _sv =
1900 -- XXX we need a better way to handle this than hardcoded delays. The
1901 -- delays may be different for different systems.
1902 -- If there is a usecase where this is required we can create a combinator
1903 -- to set it as a config in the state.
1906 ncpu <- getNumCapabilities
1909 if (svarStyle sv == AheadVar)
1910 then threadDelay 100
1913 if (svarStyle sv == AheadVar)
1914 then threadDelay 100
1919{-# NOINLINE sendWorkerWait #-}
1922 => (SVar t m a -> IO ())
1923 -> (SVar t m a -> m Bool)
1926sendWorkerWait delay dispatch sv = do
1927 -- Note that we are guaranteed to have at least one outstanding worker when
1928 -- we enter this function. So if we sleep we are guaranteed to be woken up
1929 -- by an outputDoorBell, when the worker exits.
1932 (_, n) <- liftIO $ readIORef (outputQueue sv)
1934 -- The queue may be empty temporarily if the worker has dequeued the
1935 -- work item but has not enqueued the remaining part yet. For the same
1936 -- reason, a worker may come back if it tries to dequeue and finds the
1937 -- queue empty, even though the whole work has not finished yet.
1939 -- If we find that the queue is empty, but it may be empty
1940 -- temporarily, when we checked it. If that's the case we might
1941 -- sleep indefinitely unless the active workers produce some
1942 -- output. We may deadlock specially if the otuput from the active
1943 -- workers depends on the future workers that we may never send.
1944 -- So in case the queue was temporarily empty set a flag to inform
1945 -- the enqueue to send us a doorbell.
1947 -- Note that this is just a best effort mechanism to avoid a
1948 -- deadlock. Deadlocks may still happen if for some weird reason
1949 -- the consuming computation shares an MVar or some other resource
1950 -- with the producing computation and gets blocked on that resource
1951 -- and therefore cannot do any pushworker to add more threads to
1952 -- the producer. In such cases the programmer should use a parallel
1953 -- style so that all the producers are scheduled immediately and
1954 -- unconditionally. We can also use a separate monitor thread to
1955 -- push workers instead of pushing them from the consumer, but then
1956 -- we are no longer using pull based concurrency rate adaptation.
1958 -- XXX update this in the tutorial.
1960 -- Having pending active workers does not mean that we are guaranteed
1961 -- to be woken up if we sleep. In case of Ahead streams, there may be
1962 -- queued items in the heap even though the outputQueue is empty, and
1963 -- we may have active workers which are deadlocked on those items to be
1964 -- processed by the consumer. We should either guarantee that any
1965 -- worker, before returning, clears the heap or we send a worker to
1966 -- clear it. Normally we always send a worker if no output is seen, but
1967 -- if the thread limit is reached or we are using pacing then we may
1968 -- not send a worker. See the concurrentApplication test in the tests,
1969 -- that test case requires at least one yield from the producer to not
1970 -- deadlock, if the last workers output is stuck in the heap then this
1971 -- test fails. This problem can be extended to n threads when the
1972 -- consumer may depend on the evaluation of next n items in the
1975 -- register for the outputDoorBell before we check the queue so that if
1976 -- we sleep because the queue was empty we are guaranteed to get a
1977 -- doorbell on the next enqueue.
1979 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
1980 liftIO storeLoadBarrier
1981 canDoMore <- dispatch sv
1983 -- XXX test for the case when we miss sending a worker when the worker
1984 -- count is more than 1500.
1986 -- XXX Assert here that if the heap is not empty then there is at
1987 -- least one outstanding worker. Otherwise we could be sleeping
1991 then sendWorkerWait delay dispatch sv
1993 liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
1994 $ takeMVar (outputDoorBell sv)
1995 (_, len) <- liftIO $ readIORef (outputQueue sv)
1996 when (len <= 0) $ sendWorkerWait delay dispatch sv
1998-------------------------------------------------------------------------------
1999-- Reading from the workers' output queue/buffer
2000-------------------------------------------------------------------------------
2002{-# INLINE readOutputQRaw #-}
2003readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
2004readOutputQRaw sv = do
2005 (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
2006 when (svarInspectMode sv) $ do
2007 let ref = maxOutQSize $ svarStats sv
2008 oqLen <- readIORef ref
2009 when (len > oqLen) $ writeIORef ref len
2012readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2013readOutputQBounded sv = do
2014 (list, len) <- liftIO $ readOutputQRaw sv
2015 -- When there is no output seen we dispatch more workers to help
2016 -- out if there is work pending in the work queue.
2020 -- send a worker proactively, if needed, even before we start
2021 -- processing the output. This may degrade single processor
2022 -- perf but improves multi-processor, because of more
2030 cnt <- liftIO $ readIORef $ workerCount sv
2031 when (cnt <= 0) $ do
2032 done <- liftIO $ isWorkDone sv
2033 when (not done) (pushWorker 0 sv)
2035 {-# INLINE blockingRead #-}
2037 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
2038 liftIO (fst `fmap` readOutputQRaw sv)
2040readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2041readOutputQPaced sv = do
2042 (list, len) <- liftIO $ readOutputQRaw sv
2046 -- XXX send a worker proactively, if needed, even before we start
2047 -- processing the output.
2048 void $ dispatchWorkerPaced sv
2053 {-# INLINE blockingRead #-}
2055 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
2056 liftIO (fst `fmap` readOutputQRaw sv)
2058postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
2059postProcessBounded sv = do
2060 workersDone <- allThreadsDone sv
2061 -- There may still be work pending even if there are no workers pending
2062 -- because all the workers may return if the outputQueue becomes full. In
2063 -- that case send off a worker to kickstart the work again.
2065 -- Note that isWorkDone can only be safely checked if all workers are done.
2066 -- When some workers are in progress they may have decremented the yield
2067 -- Limit and later ending up incrementing it again. If we look at the yield
2068 -- limit in that window we may falsely say that it is 0 and therefore we
2072 r <- liftIO $ isWorkDone sv
2073 -- Note that we need to guarantee a worker, therefore we cannot just
2074 -- use dispatchWorker which may or may not send a worker.
2075 when (not r) (pushWorker 0 sv)
2076 -- XXX do we need to dispatch many here?
2077 -- void $ dispatchWorker sv
2081postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
2082postProcessPaced sv = do
2083 workersDone <- allThreadsDone sv
2084 -- XXX If during consumption we figure out we are getting delayed then we
2085 -- should trigger dispatch there as well. We should try to check on the
2086 -- workers after consuming every n item from the buffer?
2089 r <- liftIO $ isWorkDone sv
2091 void $ dispatchWorkerPaced sv
2092 -- Note that we need to guarantee a worker since the work is not
2093 -- finished, therefore we cannot just rely on dispatchWorkerPaced
2094 -- which may or may not send a worker.
2095 noWorker <- allThreadsDone sv
2096 when noWorker $ pushWorker 0 sv
2100-------------------------------------------------------------------------------
2102-------------------------------------------------------------------------------
2104getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
2105getYieldRateInfo st = do
2106 -- convert rate in Hertz to latency in Nanoseconds
2107 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
2108 case getStreamRate st of
2109 Just (Rate low goal high buf) ->
2110 let l = rateToLatency goal
2111 minl = rateToLatency high
2112 maxl = rateToLatency low
2113 in mkYieldRateInfo l (LatencyRange minl maxl) buf
2114 Nothing -> return Nothing
2118 mkYieldRateInfo latency latRange buf = do
2119 measured <- newIORef 0
2120 wcur <- newIORef (0,0,0)
2121 wcol <- newIORef (0,0,0)
2122 now <- getTime Monotonic
2123 wlong <- newIORef (0,now)
2124 period <- newIORef 1
2125 gainLoss <- newIORef (Count 0)
2127 return $ Just YieldRateInfo
2128 { svarLatencyTarget = latency
2129 , svarLatencyRange = latRange
2130 , svarRateBuffer = buf
2131 , svarGainedLostYields = gainLoss
2132 , workerBootstrapLatency = getStreamLatency st
2133 , workerPollingInterval = period
2134 , workerMeasuredLatency = measured
2135 , workerPendingLatency = wcur
2136 , workerCollectedLatency = wcol
2137 , svarAllTimeLatency = wlong
2140newSVarStats :: IO SVarStats
2143 maxWrk <- newIORef 0
2147 avgLat <- newIORef (0, NanoSecond64 0)
2148 maxLat <- newIORef (NanoSecond64 0)
2149 minLat <- newIORef (NanoSecond64 0)
2150 stpTime <- newIORef Nothing
2153 { totalDispatches = disp
2154 , maxWorkers = maxWrk
2155 , maxOutQSize = maxOq
2156 , maxHeapSize = maxHs
2157 , maxWorkQSize = maxWq
2158 , avgWorkerLatency = avgLat
2159 , minWorkerLatency = minLat
2160 , maxWorkerLatency = maxLat
2161 , svarStopTime = stpTime
2164-- XXX remove polymorphism in t, inline f
2165getAheadSVar :: MonadAsync m
2167 -> ( IORef ([t m a], Int)
2168 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2175getAheadSVar st f mrun = do
2176 outQ <- newIORef ([], 0)
2177 -- the second component of the tuple is "Nothing" when heap is being
2178 -- cleared, "Just n" when we are expecting sequence number n to arrive
2179 -- before we can start clearing the heap.
2180 outH <- newIORef (H.empty, Just 0)
2181 outQMv <- newEmptyMVar
2182 active <- newIORef 0
2183 wfw <- newIORef False
2184 running <- newIORef S.empty
2185 -- Sequence number is incremented whenever something is queued, therefore,
2186 -- first sequence number would be 0
2187 q <- newIORef ([], -1)
2188 stopMVar <- newMVar ()
2189 yl <- case getYieldLimit st of
2190 Nothing -> return Nothing
2191 Just x -> Just <$> newIORef x
2192 rateInfo <- getYieldRateInfo st
2194 stats <- newSVarStats
2197 let getSVar sv readOutput postProc = SVar
2198 { outputQueue = outQ
2199 , remainingWork = yl
2200 , maxBufferLimit = getMaxBuffer st
2201 , pushBufferSpace = undefined
2202 , pushBufferPolicy = undefined
2203 , pushBufferMVar = undefined
2204 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
2205 , yieldRateInfo = rateInfo
2206 , outputDoorBell = outQMv
2207 , readOutputQ = readOutput sv
2208 , postProcess = postProc sv
2209 , workerThreads = running
2210 , workLoop = f q outH st{streamVar = Just sv} sv
2211 , enqueue = enqueueAhead sv q
2212 , isWorkDone = isWorkDoneAhead sv q outH
2213 , isQueueDone = isQueueDoneAhead sv q
2214 , needDoorBell = wfw
2215 , svarStyle = AheadVar
2216 , svarStopStyle = StopNone
2217 , svarStopBy = undefined
2219 , workerCount = active
2220 , accountThread = delThread sv
2221 , workerStopMVar = stopMVar
2223 , svarInspectMode = getInspectMode st
2225 , aheadWorkQueue = q
2231 case getStreamRate st of
2232 Nothing -> getSVar sv readOutputQBounded postProcessBounded
2233 Just _ -> getSVar sv readOutputQPaced postProcessPaced
2238 {-# INLINE isQueueDoneAhead #-}
2239 isQueueDoneAhead sv q = do
2240 queueDone <- checkEmpty q
2242 case remainingWork sv of
2246 Nothing -> return False
2247 -- XXX note that yieldsDone can only be authoritative only when there
2248 -- are no workers running. If there are active workers they can
2249 -- later increment the yield count and therefore change the result.
2250 return $ yieldsDone || queueDone
2252 {-# INLINE isWorkDoneAhead #-}
2253 isWorkDoneAhead sv q ref = do
2255 (hp, _) <- readIORef ref
2256 return (H.size hp <= 0)
2257 queueDone <- isQueueDoneAhead sv q
2258 return $ heapDone && queueDone
2261 (xs, _) <- readIORef q
2264getParallelSVar :: MonadIO m
2265 => SVarStopStyle -> State t m a -> RunInIO m -> IO (SVar t m a)
2266getParallelSVar ss st mrun = do
2267 outQ <- newIORef ([], 0)
2268 outQMv <- newEmptyMVar
2269 active <- newIORef 0
2270 running <- newIORef S.empty
2271 yl <- case getYieldLimit st of
2272 Nothing -> return Nothing
2273 Just x -> Just <$> newIORef x
2274 rateInfo <- getYieldRateInfo st
2276 case getMaxBuffer st of
2277 Unlimited -> undefined
2278 Limited x -> (fromIntegral x)
2279 remBuf <- newIORef bufLim
2280 pbMVar <- newMVar ()
2282 stats <- newSVarStats
2287 StopBy -> liftIO $ newIORef undefined
2288 _ -> return undefined
2291 SVar { outputQueue = outQ
2292 , remainingWork = yl
2293 , maxBufferLimit = getMaxBuffer st
2294 , pushBufferSpace = remBuf
2295 , pushBufferPolicy = PushBufferBlock
2296 , pushBufferMVar = pbMVar
2297 , maxWorkerLimit = Unlimited
2298 -- Used only for diagnostics
2299 , yieldRateInfo = rateInfo
2300 , outputDoorBell = outQMv
2301 , readOutputQ = readOutputQPar sv
2302 , postProcess = allThreadsDone sv
2303 , workerThreads = running
2304 , workLoop = undefined
2305 , enqueue = undefined
2306 , isWorkDone = undefined
2307 , isQueueDone = undefined
2308 , needDoorBell = undefined
2309 , svarStyle = ParallelVar
2310 , svarStopStyle = ss
2311 , svarStopBy = stopBy
2313 , workerCount = active
2314 , accountThread = modifyThread sv
2315 , workerStopMVar = undefined
2317 , svarInspectMode = getInspectMode st
2319 , aheadWorkQueue = undefined
2320 , outputHeap = undefined
2327 readOutputQPar sv = liftIO $ do
2328 withDiagMVar sv "readOutputQPar: doorbell"
2329 $ takeMVar (outputDoorBell sv)
2330 case yieldRateInfo sv of
2331 Nothing -> return ()
2332 Just yinfo -> void $ collectLatency sv yinfo False
2333 r <- fst `fmap` readOutputQRaw sv
2335 void $ tryTakeMVar (pushBufferMVar sv)
2338 void $ tryPutMVar (pushBufferMVar sv) ()
2341sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
2342sendFirstWorker sv m = do
2343 -- Note: We must have all the work on the queue before sending the
2344 -- pushworker, otherwise the pushworker may exit before we even get a
2346 liftIO $ enqueue sv m
2347 case yieldRateInfo sv of
2348 Nothing -> pushWorker 0 sv
2350 if svarLatencyTarget yinfo == maxBound
2351 then liftIO $ threadDelay maxBound
2352 else pushWorker 1 sv
2355{-# INLINABLE newAheadVar #-}
2356newAheadVar :: MonadAsync m
2359 -> ( IORef ([t m a], Int)
2360 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2366newAheadVar st m wloop = do
2367 mrun <- captureMonadState
2368 sv <- liftIO $ getAheadSVar st wloop mrun
2369 sendFirstWorker sv m
2371{-# INLINABLE newParallelVar #-}
2372newParallelVar :: MonadAsync m
2373 => SVarStopStyle -> State t m a -> m (SVar t m a)
2374newParallelVar ss st = do
2375 mrun <- captureMonadState
2376 liftIO $ getParallelSVar ss st mrun
2378-------------------------------------------------------------------------------
2379-- Write a stream to an SVar
2380-------------------------------------------------------------------------------
2382-- XXX this errors out for Parallel/Ahead SVars
2383-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
2384-- be read back from the SVar using 'fromSVar'.
2385toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
2386toStreamVar sv m = do
2387 liftIO $ enqueue sv m
2388 done <- allThreadsDone sv
2389 -- XXX This is safe only when called from the consumer thread or when no
2390 -- consumer is present. There may be a race if we are not running in the
2392 -- XXX do this only if the work queue is not empty. The work may have been
2393 -- carried out by existing workers.
2395 case yieldRateInfo sv of
2396 Nothing -> pushWorker 0 sv
2397 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