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
109 (ThreadId, myThreadId, threadDelay, throwTo, forkIO, killThread)
110import Control.Concurrent.MVar
111 (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar,
113import Control.Exception
114 (SomeException(..), catch, mask, assert, Exception, catches,
115 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
116 BlockedIndefinitelyOnSTM(..))
117import Control.Monad (when)
118import Control.Monad.Catch (MonadThrow)
119import Control.Monad.IO.Class (MonadIO(..))
120import Control.Monad.Trans.Control
121 (MonadBaseControl, control, StM, liftBaseDiscard)
122import Streamly.Internal.Data.Atomics
123 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
125import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
126import Data.Functor (void)
127import Data.Heap (Heap, Entry(..))
128import Data.Int (Int64)
129import Data.Kind (Type)
131 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
132import Data.Maybe (fromJust, fromMaybe)
133#if __GLASGOW_HASKELL__ < 808
134import Data.Semigroup ((<>))
137import GHC.Conc (ThreadId(..))
139import GHC.IO (IO(..))
140import System.IO (hPutStrLn, stderr)
141import System.Mem.Weak (addFinalizer)
143import Streamly.Internal.Data.Time.Clock (Clock(..), getTime)
144import Streamly.Internal.Data.Time.Units
145 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
146 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
148import qualified Data.Heap as H
149import qualified Data.Set as S
151newtype Count = Count Int64
163------------------------------------------------------------------------------
164-- Parent child thread communication type
165------------------------------------------------------------------------------
167data ThreadAbort = ThreadAbort deriving Show
169instance Exception ThreadAbort
171-- | Events that a child thread may send to a parent thread.
174 | ChildStop ThreadId (Maybe SomeException)
176-- | Sorting out-of-turn outputs in a heap for Ahead style streams
177data AheadHeapEntry (t :: (Type -> Type) -> Type -> Type) m a =
180 | AheadEntryStream (RunInIO m, t m a)
183------------------------------------------------------------------------------
184-- State threaded around the monad for thread management
185------------------------------------------------------------------------------
187-- | Identify the type of the SVar. Two computations using the same style can
188-- be scheduled on the same SVar.
190 AsyncVar -- depth first concurrent
191 | WAsyncVar -- breadth first concurrent
192 | ParallelVar -- all parallel
193 | AheadVar -- Concurrent look ahead
196-- | An SVar or a Stream Var is a conduit to the output from multiple streams
197-- running concurrently and asynchronously. An SVar can be thought of as an
198-- asynchronous IO handle. We can write any number of streams to an SVar in a
199-- non-blocking manner and then read them back at any time at any pace. The
200-- SVar would run the streams asynchronously and accumulate results. An SVar
201-- may not really execute the stream completely and accumulate all the results.
202-- However, it ensures that the reader can read the results at whatever paces
203-- it wants to read. The SVar monitors and adapts to the consumer's pace.
205-- An SVar is a mini scheduler, it has an associated workLoop that holds the
206-- stream tasks to be picked and run by a pool of worker threads. It has an
207-- associated output queue where the output stream elements are placed by the
208-- worker threads. A outputDoorBell is used by the worker threads to intimate the
209-- consumer thread about availability of new results in the output queue. More
210-- workers are added to the SVar by 'fromStreamVar' on demand if the output
211-- produced is not keeping pace with the consumer. On bounded SVars, workers
212-- block on the output queue to provide throttling of the producer when the
213-- consumer is not pulling fast enough. The number of workers may even get
214-- reduced depending on the consuming pace.
216-- New work is enqueued either at the time of creation of the SVar or as a
217-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
218-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
220-- We measure the individual worker latencies to estimate the number of workers
221-- needed or the amount of time we have to sleep between dispatches to achieve
222-- a particular rate when controlled pace mode it used.
223data WorkerInfo = WorkerInfo
224 { workerYieldMax :: Count -- 0 means unlimited
225 -- total number of yields by the worker till now
226 , workerYieldCount :: IORef Count
227 -- yieldCount at start, timestamp
228 , workerLatencyStart :: IORef (Count, AbsTime)
231-- | Specifies the stream yield rate in yields per second (@Hertz@).
232-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
233-- allow only as many yields as we have accumulated as per 'rateGoal' since the
234-- start of time. If the consumer or the producer is slower or faster, the
235-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
236-- between the two by increasing or decreasing the pull rate from the producer.
237-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
238-- much as 'rateBuffer'.
240-- 'rateLow' puts a bound on how low the instantaneous rate can go when
241-- recovering the rate gap. In other words, it determines the maximum yield
242-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
243-- rate can go when recovering the rate gap. In other words, it determines the
244-- minimum yield latency. We reduce the latency by increasing concurrency,
245-- therefore we can say that it puts an upper bound on concurrency.
247-- If the 'rateGoal' is 0 or negative the stream never yields a value.
248-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
250-- /Since: 0.5.0 ("Streamly")/
254 { rateLow :: Double -- ^ The lower rate limit
255 , rateGoal :: Double -- ^ The target rate we want to achieve
256 , rateHigh :: Double -- ^ The upper rate limit
257 , rateBuffer :: Int -- ^ Maximum slack from the goal
260data LatencyRange = LatencyRange
261 { minLatency :: NanoSecond64
262 , maxLatency :: NanoSecond64
266data YieldRateInfo = YieldRateInfo
267 { svarLatencyTarget :: NanoSecond64
268 , svarLatencyRange :: LatencyRange
269 , svarRateBuffer :: Int
271 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
272 -- read by the worker threads
273 , svarGainedLostYields :: IORef Count
275 -- Actual latency/througput as seen from the consumer side, we count the
276 -- yields and the time it took to generates those yields. This is used to
277 -- increase or decrease the number of workers needed to achieve the desired
278 -- rate. The idle time of workers is adjusted in this, so that we only
279 -- account for the rate when the consumer actually demands data.
280 -- XXX interval latency is enough, we can move this under diagnostics build
281 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
282 -- read by the worker threads
283 , svarAllTimeLatency :: IORef (Count, AbsTime)
285 -- XXX Worker latency specified by the user to be used before the first
286 -- actual measurement arrives. Not yet implemented
287 , workerBootstrapLatency :: Maybe NanoSecond64
289 -- After how many yields the worker should update the latency information.
290 -- If the latency is high, this count is kept lower and vice-versa. XXX If
291 -- the latency suddenly becomes too high this count may remain too high for
292 -- long time, in such cases the consumer can change it.
293 -- 0 means no latency computation
294 -- XXX this is derivable from workerMeasuredLatency, can be removed.
295 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
296 -- read by the worker threads
297 , workerPollingInterval :: IORef Count
299 -- This is in progress latency stats maintained by the workers which we
300 -- empty into workerCollectedLatency stats at certain intervals - whenever
301 -- we process the stream elements yielded in this period. The first count
302 -- is all yields, the second count is only those yields for which the
303 -- latency was measured to be non-zero (note that if the timer resolution
304 -- is low the measured latency may be zero e.g. on JS platform).
305 -- [LOCKING] Locked access. Modified by the consumer thread as well as
306 -- worker threads. Workers modify it periodically based on
307 -- workerPollingInterval and not on every yield to reduce the locking
309 -- (allYieldCount, yieldCount, timeTaken)
310 , workerPendingLatency :: IORef (Count, Count, NanoSecond64)
312 -- This is the second level stat which is an accmulation from
313 -- workerPendingLatency stats. We keep accumulating latencies in this
314 -- bucket until we have stats for a sufficient period and then we reset it
315 -- to start collecting for the next period and retain the computed average
316 -- latency for the last period in workerMeasuredLatency.
317 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
318 -- read by the worker threads
319 -- (allYieldCount, yieldCount, timeTaken)
320 , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)
322 -- Latency as measured by workers, aggregated for the last period.
323 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
324 -- read by the worker threads
325 , workerMeasuredLatency :: IORef NanoSecond64
328data SVarStats = SVarStats {
329 totalDispatches :: IORef Int
330 , maxWorkers :: IORef Int
331 , maxOutQSize :: IORef Int
332 , maxHeapSize :: IORef Int
333 , maxWorkQSize :: IORef Int
334 , avgWorkerLatency :: IORef (Count, NanoSecond64)
335 , minWorkerLatency :: IORef NanoSecond64
336 , maxWorkerLatency :: IORef NanoSecond64
337 , svarStopTime :: IORef (Maybe AbsTime)
340-- This is essentially a 'Maybe Word' type
341data Limit = Unlimited | Limited Word deriving Show
343instance Eq Limit where
344 Unlimited == Unlimited = True
345 Unlimited == Limited _ = False
346 Limited _ == Unlimited = False
347 Limited x == Limited y = x == y
349instance Ord Limit where
350 Unlimited <= Unlimited = True
351 Unlimited <= Limited _ = False
352 Limited _ <= Unlimited = True
353 Limited x <= Limited y = x <= y
355-- When to stop the composed stream.
357 StopNone -- stops only when all streams are finished
358 | StopAny -- stop when any stream finishes
359 | StopBy -- stop when a specific stream finishes
362-- | Buffering policy for persistent push workers (in ParallelT). In a pull
363-- style SVar (in AsyncT, AheadT etc.), the consumer side dispatches workers on
364-- demand, workers terminate if the buffer is full or if the consumer is not
365-- cosuming fast enough. In a push style SVar, a worker is dispatched only
366-- once, workers are persistent and keep pushing work to the consumer via a
367-- bounded buffer. If the buffer becomes full the worker either blocks, or it
368-- can drop an item from the buffer to make space.
370-- Pull style SVars are useful in lazy stream evaluation whereas push style
371-- SVars are useful in strict left Folds.
373-- XXX Maybe we can separate the implementation in two different types instead
374-- of using a common SVar type.
376data PushBufferPolicy =
377 PushBufferDropNew -- drop the latest element and continue
378 | PushBufferDropOld -- drop the oldest element and continue
379 | PushBufferBlock -- block the thread until space
382-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have
383-- references to the original SVar stored in several functions which will keep
384-- pointing to the original data and the new updates won't reflect there.
385-- Any updateable parts must be kept in mutable references (IORef).
387data SVar t m a = SVar
390 svarStyle :: SVarStyle
391 , svarMrun :: RunInIO m
392 , svarStopStyle :: SVarStopStyle
393 , svarStopBy :: IORef ThreadId
395 -- Shared output queue (events, length)
396 -- XXX For better efficiency we can try a preallocated array type (perhaps
397 -- something like a vector) that allows an O(1) append. That way we will
398 -- avoid constructing and reversing the list. Possibly we can also avoid
399 -- the GC copying overhead. When the size increases we should be able to
400 -- allocate the array in chunks.
402 -- [LOCKING] Frequent locked access. This is updated by workers on each
403 -- yield and once in a while read by the consumer thread. This could have
404 -- big locking overhead if the number of workers is high.
406 -- XXX We can use a per-CPU data structure to reduce the locking overhead.
407 -- However, a per-cpu structure cannot guarantee the exact sequence in
408 -- which the elements were added, though that may not be important.
409 , outputQueue :: IORef ([ChildEvent a], Int)
411 -- [LOCKING] Infrequent MVar. Used when the outputQ transitions from empty
412 -- to non-empty, or a work item is queued by a worker to the work queue and
413 -- needDoorBell is set by the consumer.
414 , outputDoorBell :: MVar () -- signal the consumer about output
415 , readOutputQ :: m [ChildEvent a]
416 , postProcess :: m Bool
418 -- channel to send events from the consumer to the worker. Used to send
419 -- exceptions from a fold driver to the fold computation running as a
420 -- consumer thread in the concurrent fold cases. Currently only one event
421 -- is sent by the fold so we do not really need a queue for it.
422 , outputQueueFromConsumer :: IORef ([ChildEvent a], Int)
423 , outputDoorBellFromConsumer :: MVar ()
425 -- Combined/aggregate parameters
426 -- This is truncated to maxBufferLimit if set to more than that. Otherwise
427 -- potentially each worker may yield one value to the buffer in the worst
428 -- case exceeding the requested buffer size.
429 , maxWorkerLimit :: Limit
430 , maxBufferLimit :: Limit
431 -- These two are valid and used only when maxBufferLimit is Limited.
432 , pushBufferSpace :: IORef Count
433 , pushBufferPolicy :: PushBufferPolicy
434 -- [LOCKING] The consumer puts this MVar after emptying the buffer, workers
435 -- block on it when the buffer becomes full. No overhead unless the buffer
437 , pushBufferMVar :: MVar ()
439 -- [LOCKING] Read only access by consumer when dispatching a worker.
440 -- Decremented by workers when picking work and undo decrement if the
441 -- worker does not yield a value.
442 , remainingWork :: Maybe (IORef Count)
443 , yieldRateInfo :: Maybe YieldRateInfo
445 -- Used only by bounded SVar types
446 , enqueue :: (RunInIO m, t m a) -> IO ()
447 , isWorkDone :: IO Bool
448 , isQueueDone :: IO Bool
449 , needDoorBell :: IORef Bool
450 , workLoop :: Maybe WorkerInfo -> m ()
452 -- Shared, thread tracking
453 -- [LOCKING] Updated unlocked only by consumer thread in case of
454 -- Async/Ahead style SVars. Updated locked by worker threads in case of
456 , workerThreads :: IORef (Set ThreadId)
457 -- [LOCKING] Updated locked by consumer thread when dispatching a worker
458 -- and by the worker threads when the thread stops. This is read unsafely
459 -- at several places where we want to rely on an approximate value.
460 , workerCount :: IORef Int
461 , accountThread :: ThreadId -> m ()
462 , workerStopMVar :: MVar ()
464 , svarStats :: SVarStats
465 -- to track garbage collection of SVar
466 , svarRef :: Maybe (IORef ())
468 -- Only for diagnostics
469 , svarInspectMode :: Bool
470 , svarCreator :: ThreadId
471 , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))
473 -- Shared work queue (stream, seqNo)
474 , aheadWorkQueue :: IORef ([t m a], Int)
477-------------------------------------------------------------------------------
478-- State for concurrency control
479-------------------------------------------------------------------------------
481-- XXX we can put the resettable fields in a oneShotConfig field and others in
482-- a persistentConfig field. That way reset would be fast and scalable
483-- irrespective of the number of fields.
485-- XXX make all these Limited types and use phantom types to distinguish them
486data State t m a = State
487 { -- one shot configuration, automatically reset for each API call
488 streamVar :: Maybe (SVar t m a)
489 , _yieldLimit :: Maybe Count
491 -- persistent configuration, state that remains valid until changed by
492 -- an explicit setting via a combinator.
493 , _threadsHigh :: Limit
494 , _bufferHigh :: Limit
495 -- XXX these two can be collapsed into a single type
496 , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency
497 , _maxStreamRate :: Maybe Rate
498 , _inspectMode :: Bool
501-------------------------------------------------------------------------------
502-- State defaults and reset
503-------------------------------------------------------------------------------
505-- A magical value for the buffer size arrived at by running the smallest
506-- possible task and measuring the optimal value of the buffer for that. This
507-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
509magicMaxBuffer :: Word
512defaultMaxThreads, defaultMaxBuffer :: Limit
513defaultMaxThreads = Limited magicMaxBuffer
514defaultMaxBuffer = Limited magicMaxBuffer
516-- The fields prefixed by an _ are not to be accessed or updated directly but
517-- via smart accessor APIs.
518defState :: State t m a
520 { streamVar = Nothing
521 , _yieldLimit = Nothing
522 , _threadsHigh = defaultMaxThreads
523 , _bufferHigh = defaultMaxBuffer
524 , _maxStreamRate = Nothing
525 , _streamLatency = Nothing
526 , _inspectMode = False
529-- XXX if perf gets affected we can have all the Nothing params in a single
530-- structure so that we reset is fast. We can also use rewrite rules such that
531-- reset occurs only in concurrent streams to reduce the impact on serial
533-- We can optimize this so that we clear it only if it is a Just value, it
534-- results in slightly better perf for zip/zipM but the performance of scan
535-- worsens a lot, it does not fuse.
537-- XXX This has a side effect of clearing the SVar and yieldLimit, therefore it
538-- should not be used to convert from the same type to the same type, unless
539-- you want to clear the SVar. For clearing the SVar you should be using the
540-- appropriate unStream functions instead.
542-- | Adapt the stream state from one type to another.
543adaptState :: State t m a -> State t n b
545 { streamVar = Nothing
546 , _yieldLimit = Nothing
549-------------------------------------------------------------------------------
550-- Smart get/set routines for State
551-------------------------------------------------------------------------------
553-- Use get/set routines instead of directly accessing the State fields
554setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
555setYieldLimit lim st =
562 else Just (fromIntegral n)
565getYieldLimit :: State t m a -> Maybe Count
566getYieldLimit = _yieldLimit
568setMaxThreads :: Int -> State t m a -> State t m a
574 then defaultMaxThreads
575 else Limited (fromIntegral n)
578getMaxThreads :: State t m a -> Limit
579getMaxThreads = _threadsHigh
581setMaxBuffer :: Int -> State t m a -> State t m a
587 then defaultMaxBuffer
588 else Limited (fromIntegral n)
591getMaxBuffer :: State t m a -> Limit
592getMaxBuffer = _bufferHigh
594setStreamRate :: Maybe Rate -> State t m a -> State t m a
595setStreamRate r st = st { _maxStreamRate = r }
597getStreamRate :: State t m a -> Maybe Rate
598getStreamRate = _maxStreamRate
600setStreamLatency :: Int -> State t m a -> State t m a
601setStreamLatency n st =
602 st { _streamLatency =
605 else Just (fromIntegral n)
608getStreamLatency :: State t m a -> Maybe NanoSecond64
609getStreamLatency = _streamLatency
611setInspectMode :: State t m a -> State t m a
612setInspectMode st = st { _inspectMode = True }
614getInspectMode :: State t m a -> Bool
615getInspectMode = _inspectMode
617-------------------------------------------------------------------------------
619-------------------------------------------------------------------------------
621cleanupSVar :: SVar t m a -> IO ()
623 workers <- readIORef (workerThreads sv)
624 Prelude.mapM_ (`throwTo` ThreadAbort)
627cleanupSVarFromWorker :: SVar t m a -> IO ()
628cleanupSVarFromWorker sv = do
629 workers <- readIORef (workerThreads sv)
631 Prelude.mapM_ (`throwTo` ThreadAbort)
632 (Prelude.filter (/= self) $ S.toList workers)
634-------------------------------------------------------------------------------
635-- Worker latency data collection
636-------------------------------------------------------------------------------
638-- Every once in a while workers update the latencies and check the yield rate.
639-- They return if we are above the expected yield rate. If we check too often
640-- it may impact performance, if we check less often we may have a stale
641-- picture. We update every minThreadDelay but we translate that into a yield
642-- count based on latency so that the checking overhead is little.
644-- XXX use a generation count to indicate that the value is updated. If the
645-- value is updated an existing worker must check it again on the next yield.
646-- Otherwise it is possible that we may keep updating it and because of the mod
647-- worker keeps skipping it.
648updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
649updateWorkerPollingInterval yinfo latency = do
650 let periodRef = workerPollingInterval yinfo
651 cnt = max 1 $ minThreadDelay `div` latency
652 period = min cnt (fromIntegral magicMaxBuffer)
654 writeIORef periodRef (fromIntegral period)
656{-# INLINE recordMinMaxLatency #-}
657recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()
658recordMinMaxLatency sv new = do
659 let ss = svarStats sv
660 minLat <- readIORef (minWorkerLatency ss)
661 when (new < minLat || minLat == 0) $
662 writeIORef (minWorkerLatency ss) new
664 maxLat <- readIORef (maxWorkerLatency ss)
665 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
667recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()
668recordAvgLatency sv (count, time) = do
669 let ss = svarStats sv
670 modifyIORef (avgWorkerLatency ss) $
671 \(cnt, t) -> (cnt + count, t + time)
673-- Pour the pending latency stats into a collection bucket
674{-# INLINE collectWorkerPendingLatency #-}
675collectWorkerPendingLatency
676 :: IORef (Count, Count, NanoSecond64)
677 -> IORef (Count, Count, NanoSecond64)
678 -> IO (Count, Maybe (Count, NanoSecond64))
679collectWorkerPendingLatency cur col = do
680 (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
682 (fcnt, cnt, t) <- readIORef col
683 let totalCount = fcnt + fcount
684 latCount = cnt + count
686 writeIORef col (totalCount, latCount, latTime)
688 assert (latCount == 0 || latTime /= 0) (return ())
690 if latCount > 0 && latTime > 0
691 then Just (latCount, latTime)
693 return (totalCount, latPair)
695{-# INLINE shouldUseCollectedBatch #-}
696shouldUseCollectedBatch
702shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
703 let r = fromIntegral newLat / fromIntegral prevLat :: Double
704 in (collectedYields > fromIntegral magicMaxBuffer)
705 || (collectedTime > minThreadDelay)
706 || (prevLat > 0 && (r > 2 || r < 0.5))
709-- Returns a triple, (1) yield count since last collection, (2) the base time
710-- when we started counting, (3) average latency in the last measurement
711-- period. The former two are used for accurate measurement of the going rate
712-- whereas the average is used for future estimates e.g. how many workers
713-- should be maintained to maintain the rate.
714-- CAUTION! keep it in sync with getWorkerLatency
715collectLatency :: SVar t m a
718 -> IO (Count, AbsTime, NanoSecond64)
719collectLatency sv yinfo drain = do
720 let cur = workerPendingLatency yinfo
721 col = workerCollectedLatency yinfo
722 longTerm = svarAllTimeLatency yinfo
723 measured = workerMeasuredLatency yinfo
725 (newCount, newLatPair) <- collectWorkerPendingLatency cur col
726 (lcount, ltime) <- readIORef longTerm
727 prevLat <- readIORef measured
729 let newLcount = lcount + newCount
730 retWith lat = return (newLcount, ltime, lat)
733 Nothing -> retWith prevLat
734 Just (count, time) -> do
735 let newLat = time `div` fromIntegral count
736 when (svarInspectMode sv) $ recordMinMaxLatency sv newLat
737 -- When we have collected a significant sized batch we compute the
738 -- new latency using that batch and return the new latency,
739 -- otherwise we return the previous latency derived from the
741 if shouldUseCollectedBatch newCount time newLat prevLat || drain
743 -- XXX make this NOINLINE?
744 updateWorkerPollingInterval yinfo (max newLat prevLat)
745 when (svarInspectMode sv) $ recordAvgLatency sv (count, time)
746 writeIORef col (0, 0, 0)
747 writeIORef measured ((prevLat + newLat) `div` 2)
748 modifyIORef longTerm $ \(_, t) -> (newLcount, t)
752-------------------------------------------------------------------------------
753-- Dumping the SVar for debug/diag
754-------------------------------------------------------------------------------
756dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
757dumpSVarStats sv ss style = do
758 case yieldRateInfo sv of
761 _ <- liftIO $ collectLatency sv yinfo True
764 dispatches <- readIORef $ totalDispatches ss
765 maxWrk <- readIORef $ maxWorkers ss
766 maxOq <- readIORef $ maxOutQSize ss
767 maxHp <- readIORef $ maxHeapSize ss
768 minLat <- readIORef $ minWorkerLatency ss
769 maxLat <- readIORef $ maxWorkerLatency ss
770 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
771 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
772 Nothing -> return (0, 0, 0)
774 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
777 t <- readIORef (svarStopTime ss)
778 gl <- readIORef (svarGainedLostYields yinfo)
781 now <- getTime Monotonic
782 let interval = diffAbsTime64 now startTime
783 return (cnt, gl, interval `div` fromIntegral cnt)
785 let interval = diffAbsTime64 stopTime startTime
786 return (cnt, gl, interval `div` fromIntegral cnt)
787 else return (0, 0, 0)
790 [ "total dispatches = " <> show dispatches
791 , "max workers = " <> show maxWrk
792 , "max outQSize = " <> show maxOq
793 <> (if style == AheadVar
794 then "\nheap max size = " <> show maxHp
797 then "\nmin worker latency = " <> showNanoSecond64 minLat
800 then "\nmax worker latency = " <> showNanoSecond64 maxLat
803 then let lat = avgTime `div` fromIntegral avgCnt
804 in "\navg worker latency = " <> showNanoSecond64 lat
807 then "\nSVar latency = " <> showRelTime64 svarLat
810 then "\nSVar yield count = " <> show svarCnt
812 <> (if svarGainLossCnt > 0
813 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
817{-# NOINLINE dumpSVar #-}
818dumpSVar :: SVar t m a -> IO String
820 (oqList, oqLen) <- readIORef $ outputQueue sv
821 db <- tryReadMVar $ outputDoorBell sv
823 if svarStyle sv == AheadVar
825 (oheap, oheapSeq) <- readIORef $ outputHeap sv
826 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
828 [ "heap length = " <> show (H.size oheap)
829 , "heap seqeunce = " <> show oheapSeq
830 , "work queue length = " <> show (length wq)
831 , "work queue sequence = " <> show wqSeq
835 let style = svarStyle sv
837 if style /= ParallelVar
838 then readIORef $ needDoorBell sv
840 rthread <- readIORef $ workerThreads sv
841 workers <- readIORef $ workerCount sv
842 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
846 "Creator tid = " <> show (svarCreator sv),
847 "style = " <> show (svarStyle sv)
848 , "---------CURRENT STATE-----------"
849 , "outputQueue length computed = " <> show (length oqList)
850 , "outputQueue length maintained = " <> show oqLen
851 -- XXX print the types of events in the outputQueue, first 5
852 , "outputDoorBell = " <> show db
856 [ "needDoorBell = " <> show waiting
857 , "running threads = " <> show rthread
858 -- XXX print the status of first 5 threads
859 , "running thread count = " <> show workers
861 <> "---------STATS-----------\n"
864printSVar :: SVar t m a -> String -> IO ()
866 svInfo <- dumpSVar sv
867 hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo
869-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
870-- can keep it on in production to debug problems quickly if and when they
871-- happen, but it may result in unexpected output when threads are left hanging
872-- until they are GCed because the consumer went away.
874{-# NOINLINE mvarExcHandler #-}
875mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
876mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
877 svInfo <- dumpSVar sv
878 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
881{-# NOINLINE stmExcHandler #-}
882stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
883stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
884 svInfo <- dumpSVar sv
885 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
888withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
889withDiagMVar sv label action =
890 if svarInspectMode sv
892 action `catches` [ Handler (mvarExcHandler sv label)
893 , Handler (stmExcHandler sv label)
897------------------------------------------------------------------------------
899------------------------------------------------------------------------------
901-- /Since: 0.8.0 ("Streamly.Prelude")/
903-- | A monad that can perform concurrent or parallel IO operations. Streams
904-- that can be composed concurrently require the underlying monad to be
907-- /Since: 0.1.0 ("Streamly")/
910type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
912-- When we run computations concurrently, we completely isolate the state of
913-- the concurrent computations from the parent computation. The invariant is
914-- that we should never be running two concurrent computations in the same
915-- thread without using the runInIO function. Also, we should never be running
916-- a concurrent computation in the parent thread, otherwise it may affect the
917-- state of the parent which is against the defined semantics of concurrent
919newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }
921captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
922captureMonadState = control $ \run -> run (return $ RunInIO run)
924-- Stolen from the async package. The perf improvement is modest, 2% on a
925-- thread heavy benchmark (parallel composition using noop computations).
926-- A version of forkIO that does not include the outer exception
927-- handler: saves a bit of time when we will be installing our own
929{-# INLINE rawForkIO #-}
930rawForkIO :: IO () -> IO ThreadId
931rawForkIO action = IO $ \ s ->
932 case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)
935doFork :: MonadBaseControl IO m
938 -> (SomeException -> IO ())
940doFork action (RunInIO mrun) exHandler =
942 mask $ \restore -> do
943 tid <- rawForkIO $ catch (restore $ void $ mrun action)
947{-# INLINABLE fork #-}
948fork :: MonadBaseControl IO m => m () -> m ThreadId
949fork = liftBaseDiscard forkIO
951-- | Fork a thread that is automatically killed as soon as the reference to the
952-- returned threadId is garbage collected.
954{-# INLINABLE forkManaged #-}
955forkManaged :: (MonadIO m, MonadBaseControl IO m) => m () -> m ThreadId
956forkManaged action = do
958 liftIO $ addFinalizer tid (killThread tid)
961------------------------------------------------------------------------------
962-- Collecting results from child workers in a streamed fashion
963------------------------------------------------------------------------------
965-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
966-- faster, along with the fields in sv required by send?
967-- XXX make it noinline
969-- XXX we may want to employ an increment and decrement in batches when the
970-- througput is high or when the cost of synchronization is high. For example
971-- if the application is distributed then inc/dec of a shared variable may be
974-- A worker decrements the yield limit before it executes an action. However,
975-- the action may not result in an element being yielded, in that case we have
976-- to increment the yield limit.
978-- Note that we need it to be an Int type so that we have the ability to undo a
979-- decrement that takes it below zero.
980{-# INLINE decrementYieldLimit #-}
981decrementYieldLimit :: SVar t m a -> IO Bool
982decrementYieldLimit sv =
983 case remainingWork sv of
984 Nothing -> return True
986 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
989{-# INLINE incrementYieldLimit #-}
990incrementYieldLimit :: SVar t m a -> IO ()
991incrementYieldLimit sv =
992 case remainingWork sv of
994 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
996-- XXX exception safety of all atomic/MVar operations
998-- TBD Each worker can have their own queue and the consumer can empty one
999-- queue at a time, that way contention can be reduced.
1001-- XXX Only yields should be counted in the buffer limit and not the Stop
1004{-# INLINE decrementBufferLimit #-}
1005decrementBufferLimit :: SVar t m a -> IO ()
1006decrementBufferLimit sv =
1007 case maxBufferLimit sv of
1008 Unlimited -> return ()
1010 let ref = pushBufferSpace sv
1011 old <- atomicModifyIORefCAS ref $ \x ->
1012 (if x >= 1 then x - 1 else x, x)
1014 case pushBufferPolicy sv of
1015 PushBufferBlock -> blockAndRetry
1016 PushBufferDropNew -> do
1017 -- We just drop one item and proceed. It is possible
1018 -- that by the time we drop the item the consumer
1019 -- thread might have run and created space in the
1020 -- buffer, but we do not care about that condition.
1021 -- This is not pedantically correct but it should be
1022 -- fine in practice.
1023 -- XXX we may want to drop only if n == maxBuf
1024 -- otherwise we must have space in the buffer and a
1025 -- decrement should be possible.
1026 block <- atomicModifyIORefCAS (outputQueue sv) $
1029 [] -> (([],n), True)
1030 _ : xs -> ((xs, n - 1), False)
1031 when block blockAndRetry
1032 -- XXX need a dequeue or ring buffer for this
1033 PushBufferDropOld -> undefined
1038 let ref = pushBufferSpace sv
1039 liftIO $ takeMVar (pushBufferMVar sv)
1040 old <- atomicModifyIORefCAS ref $ \x ->
1041 (if x >= 1 then x - 1 else x, x)
1042 -- When multiple threads sleep on takeMVar, the first thread would
1043 -- wakeup due to a putMVar by the consumer, but the rest of the threads
1044 -- would have to put back the MVar after taking it and decrementing the
1045 -- buffer count, otherwise all other threads will remain asleep.
1047 then void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
1048 -- We do not put the MVar back in this case, instead we
1049 -- wait for the consumer to put it.
1052{-# INLINE incrementBufferLimit #-}
1053incrementBufferLimit :: SVar t m a -> IO ()
1054incrementBufferLimit sv =
1055 case maxBufferLimit sv of
1056 Unlimited -> return ()
1058 atomicModifyIORefCAS_ (pushBufferSpace sv) (+ 1)
1060 void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
1062{-# INLINE resetBufferLimit #-}
1063resetBufferLimit :: SVar t m a -> IO ()
1064resetBufferLimit sv =
1065 case maxBufferLimit sv of
1066 Unlimited -> return ()
1067 Limited n -> atomicModifyIORefCAS_ (pushBufferSpace sv)
1068 (const (fromIntegral n))
1070{-# INLINE sendWithDoorBell #-}
1072 IORef ([ChildEvent a], Int) -> MVar () -> ChildEvent a -> IO Int
1073sendWithDoorBell q bell msg = do
1074 -- XXX can the access to outputQueue be made faster somehow?
1075 oldlen <- atomicModifyIORefCAS q $ \(es, n) ->
1076 ((msg : es, n + 1), n)
1077 when (oldlen <= 0) $ do
1078 -- The wake up must happen only after the store has finished otherwise
1079 -- we can have lost wakeup problems.
1081 -- Since multiple workers can try this at the same time, it is possible
1082 -- that we may put a spurious MVar after the consumer has already seen
1083 -- the output. But that's harmless, at worst it may cause the consumer
1084 -- to read the queue again and find it empty.
1085 -- The important point is that the consumer is guaranteed to receive a
1086 -- doorbell if something was added to the queue after it empties it.
1087 void $ tryPutMVar bell ()
1090-- | This function is used by the producer threads to queue output for the
1091-- consumer thread to consume. Returns whether the queue has more space.
1092send :: SVar t m a -> ChildEvent a -> IO Int
1093send sv = sendWithDoorBell (outputQueue sv) (outputDoorBell sv)
1095-- There is no bound implemented on the buffer, this is assumed to be low
1097sendToProducer :: SVar t m a -> ChildEvent a -> IO Int
1098sendToProducer sv msg = do
1099 -- In case the producer stream is blocked on pushing to the fold buffer
1100 -- then wake it up so that it can check for the stop event or exception
1101 -- being sent to it otherwise we will be deadlocked.
1102 void $ tryPutMVar (pushBufferMVar sv) ()
1103 sendWithDoorBell (outputQueueFromConsumer sv)
1104 (outputDoorBellFromConsumer sv) msg
1106-- {-# NOINLINE sendStopToProducer #-}
1107sendStopToProducer :: MonadIO m => SVar t m a -> m ()
1108sendStopToProducer sv = liftIO $ do
1110 void $ sendToProducer sv (ChildStop tid Nothing)
1112workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
1113workerCollectLatency winfo = do
1114 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
1115 cnt1 <- readIORef (workerYieldCount winfo)
1116 let cnt = cnt1 - cnt0
1120 t1 <- getTime Monotonic
1121 let period = fromRelTime64 $ diffAbsTime64 t1 t0
1122 writeIORef (workerLatencyStart winfo) (cnt1, t1)
1123 return $ Just (cnt, period)
1126-- XXX There are a number of gotchas in measuring latencies.
1127-- 1) We measure latencies only when a worker yields a value
1128-- 2) It is possible that a stream calls the stop continuation, in which case
1129-- the worker would not yield a value and we would not account that worker in
1130-- latencies. Even though this case should ideally be accounted we do not
1131-- account it because we cannot or do not distinguish it from the case
1133-- 3) It is possible that a worker returns without yielding anything because it
1134-- never got a chance to pick up work.
1135-- 4) If the system timer resolution is lower than the latency, the latency
1136-- computation turns out to be zero.
1138-- We can fix this if we measure the latencies by counting the work items
1139-- picked rather than based on the outputs yielded.
1140workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
1141workerUpdateLatency yinfo winfo = do
1142 r <- workerCollectLatency winfo
1144 Just (cnt, period) -> do
1145 -- NOTE: On JS platform the timer resolution could be pretty low. When
1146 -- the timer resolution is low, measurement of latencies could be
1147 -- tricky. All the worker latencies will turn out to be zero if they
1148 -- are lower than the resolution. We only take into account those
1149 -- measurements which are more than the timer resolution.
1151 let ref = workerPendingLatency yinfo
1152 (cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
1153 atomicModifyIORefCAS_ ref $
1154 \(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
1155 Nothing -> return ()
1157updateYieldCount :: WorkerInfo -> IO Count
1158updateYieldCount winfo = do
1159 cnt <- readIORef (workerYieldCount winfo)
1161 writeIORef (workerYieldCount winfo) cnt1
1164isBeyondMaxYield :: Count -> WorkerInfo -> Bool
1165isBeyondMaxYield cnt winfo =
1166 let ymax = workerYieldMax winfo
1167 in ymax /= 0 && cnt >= ymax
1169-- XXX we should do rate control periodically based on the total yields rather
1170-- than based on the worker local yields as other workers may have yielded more
1171-- and we should stop based on the aggregate yields. However, latency update
1172-- period can be based on individual worker yields.
1173{-# NOINLINE checkRatePeriodic #-}
1174checkRatePeriodic :: SVar t m a
1179checkRatePeriodic sv yinfo winfo ycnt = do
1180 i <- readIORef (workerPollingInterval yinfo)
1181 -- XXX use generation count to check if the interval has been updated
1182 if i /= 0 && (ycnt `mod` i) == 0
1184 workerUpdateLatency yinfo winfo
1185 -- XXX not required for parallel streams
1186 isBeyondMaxRate sv yinfo
1189-- CAUTION! this also updates the yield count and therefore should be called
1190-- only when we are actually yielding an element.
1191{-# NOINLINE workerRateControl #-}
1192workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
1193workerRateControl sv yinfo winfo = do
1194 cnt <- updateYieldCount winfo
1195 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
1196 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
1198-- XXX we should do rate control here but not latency update in case of ahead
1199-- streams. latency update must be done when we yield directly to outputQueue
1200-- or when we yield to heap.
1202-- returns whether the worker should continue (True) or stop (False).
1203{-# INLINE sendYield #-}
1204sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
1205sendYield sv mwinfo msg = do
1206 oldlen <- send sv msg
1207 let limit = maxBufferLimit sv
1208 bufferSpaceOk <- case limit of
1209 Unlimited -> return True
1211 active <- readIORef (workerCount sv)
1212 return $ (oldlen + 1) < (fromIntegral lim - active)
1216 case yieldRateInfo sv of
1217 Nothing -> return True
1218 Just yinfo -> workerRateControl sv yinfo winfo
1219 Nothing -> return True
1220 return $ bufferSpaceOk && rateLimitOk
1222{-# INLINE workerStopUpdate #-}
1223workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
1224workerStopUpdate winfo info = do
1225 i <- readIORef (workerPollingInterval info)
1226 when (i /= 0) $ workerUpdateLatency info winfo
1228{-# INLINABLE sendStop #-}
1229sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
1230sendStop sv mwinfo = do
1231 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
1232 case (mwinfo, yieldRateInfo sv) of
1233 (Just winfo, Just info) ->
1234 workerStopUpdate winfo info
1237 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
1239-------------------------------------------------------------------------------
1241-------------------------------------------------------------------------------
1243{-# INLINE ringDoorBell #-}
1244ringDoorBell :: SVar t m a -> IO ()
1247 w <- readIORef $ needDoorBell sv
1249 -- Note: the sequence of operations is important for correctness here.
1250 -- We need to set the flag to false strictly before sending the
1251 -- outputDoorBell, otherwise the outputDoorBell may get processed too
1252 -- early and then we may set the flag to False to later making the
1253 -- consumer lose the flag, even without receiving a outputDoorBell.
1254 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1255 void $ tryPutMVar (outputDoorBell sv) ()
1257-------------------------------------------------------------------------------
1259-------------------------------------------------------------------------------
1261-- Note: For purely right associated expressions this queue should have at most
1262-- one element. It grows to more than one when we have left associcated
1263-- expressions. Large left associated compositions can grow this to a
1265{-# INLINE enqueueLIFO #-}
1267 SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
1268enqueueLIFO sv q m = do
1269 atomicModifyIORefCAS_ q $ \ms -> m : ms
1272-------------------------------------------------------------------------------
1274-------------------------------------------------------------------------------
1276-- XXX we can use the Ahead style sequence/heap mechanism to make the best
1277-- effort to always try to finish the streams on the left side of an expression
1278-- first as long as possible.
1280{-# INLINE enqueueFIFO #-}
1283 -> LinkedQueue (RunInIO m, t m a)
1284 -> (RunInIO m, t m a)
1286enqueueFIFO sv q m = do
1290-------------------------------------------------------------------------------
1292-------------------------------------------------------------------------------
1294-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
1295-- but always serve them in the same order as they appear in the stream. To
1296-- implement lookahead streams efficiently we assign a sequence number to each
1297-- task when the task is picked up for execution. When the task finishes, the
1298-- output is tagged with the same sequence number and we rearrange the outputs
1299-- in sequence based on that number.
1301-- To explain the mechanism imagine that the current task at the head of the
1302-- stream has a "token" to yield to the outputQueue. The ownership of the token
1303-- is determined by the current sequence number is maintained in outputHeap.
1304-- Sequence number is assigned when a task is queued. When a thread dequeues a
1305-- task it picks up the sequence number as well and when the output is ready it
1306-- uses the sequence number to queue the output to the outputQueue.
1308-- The thread with current sequence number sends the output directly to the
1309-- outputQueue. Other threads push the output to the outputHeap. When the task
1310-- being queued on the heap is a stream of many elements we evaluate only the
1311-- first element and keep the rest of the unevaluated computation in the heap.
1312-- When such a task gets the "token" for outputQueue it evaluates and directly
1313-- yields all the elements to the outputQueue without checking for the
1316-- Note that no two outputs in the heap can have the same sequence numbers and
1317-- therefore we do not need a stable heap. We have also separated the buffer
1318-- for the current task (outputQueue) and the pending tasks (outputHeap) so
1319-- that the pending tasks cannot interfere with the current task. Note that for
1320-- a single task just the outputQueue is enough and for the case of many
1321-- threads just a heap is good enough. However we balance between these two
1322-- cases, so that both are efficient.
1324-- For bigger streams it may make sense to have separate buffers for each
1325-- stream. However, for singleton streams this may become inefficient. However,
1326-- if we do not have separate buffers, then the streams that come later in
1327-- sequence may hog the buffer, hindering the streams that are ahead. For this
1328-- reason we have a single element buffer limitation for the streams being
1329-- executed in advance.
1331-- This scheme works pretty efficiently with less than 40% extra overhead
1332-- compared to the Async streams where we do not have any kind of sequencing of
1333-- the outputs. It is especially devised so that we are most efficient when we
1334-- have short tasks and need just a single thread. Also when a thread yields
1335-- many items it can hold lockfree access to the outputQueue and do it
1338-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
1339-- that they do not hog the resources and hinder the progress of the threads in
1342-- XXX Left associated ahead expressions are expensive. We start a new SVar for
1343-- each left associative expression. The queue is used only for right
1344-- associated expression, we queue the right expression and execute the left.
1345-- Thererefore the queue never has more than one item in it.
1347-- XXX we can fix this. When we queue more than one item on the queue we can
1348-- mark the previously queued item as not-runnable. The not-runnable item is
1349-- not dequeued until the already running one has finished and at that time we
1350-- would also know the exact sequence number of the already queued item.
1352-- we can even run the already queued items but they will have to be sorted in
1353-- layers in the heap. We can use a list of heaps for that.
1354{-# INLINE enqueueAhead #-}
1355enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> (RunInIO m, t m a) -> IO ()
1356enqueueAhead sv q m = do
1357 atomicModifyIORefCAS_ q $ \ case
1358 ([], n) -> ([snd m], n + 1) -- increment sequence
1359 _ -> error "enqueueAhead: queue is not empty"
1362-- enqueue without incrementing the sequence number
1363{-# INLINE reEnqueueAhead #-}
1364reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1365reEnqueueAhead sv q m = do
1366 atomicModifyIORefCAS_ q $ \ case
1367 ([], n) -> ([m], n) -- DO NOT increment sequence
1368 _ -> error "reEnqueueAhead: queue is not empty"
1371-- Normally the thread that has the token should never go away. The token gets
1372-- handed over to another thread, but someone or the other has the token at any
1373-- point of time. But if the task that has the token finds that the outputQueue
1374-- is full, in that case it can go away without even handing over the token to
1375-- another thread. In that case it sets the nextSequence number in the heap its
1376-- own sequence number before going away. To handle this case, any task that
1377-- does not have the token tries to dequeue from the heap first before
1378-- dequeuing from the work queue. If it finds that the task at the top of the
1379-- heap is the one that owns the current sequence number then it grabs the
1380-- token and starts with that.
1382-- XXX instead of queueing just the head element and the remaining computation
1383-- on the heap, evaluate as many as we can and place them on the heap. But we
1384-- need to give higher priority to the lower sequence numbers so that lower
1385-- priority tasks do not fill up the heap making higher priority tasks block
1386-- due to full heap. Maybe we can have a weighted space for them in the heap.
1387-- The weight is inversely proportional to the sequence number.
1389-- XXX review for livelock
1391{-# INLINE queueEmptyAhead #-}
1392queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
1393queueEmptyAhead q = liftIO $ do
1394 (xs, _) <- readIORef q
1397{-# INLINE dequeueAhead #-}
1398dequeueAhead :: MonadIO m
1399 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
1400dequeueAhead q = liftIO $
1401 atomicModifyIORefCAS q $ \case
1402 ([], n) -> (([], n), Nothing)
1403 (x : [], n) -> (([], n), Just (x, n))
1404 _ -> error "more than one item on queue"
1406-------------------------------------------------------------------------------
1408-------------------------------------------------------------------------------
1410withIORef :: IORef a -> (a -> IO b) -> IO b
1411withIORef ref f = readIORef ref >>= f
1413atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
1414atomicModifyIORef_ ref f =
1415 atomicModifyIORef ref $ \x -> (f x, ())
1417data HeapDequeueResult t m a =
1420 | Ready (Entry Int (AheadHeapEntry t m a))
1422{-# INLINE dequeueFromHeap #-}
1424 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1425 -> IO (HeapDequeueResult t m a)
1426dequeueFromHeap hpVar =
1427 atomicModifyIORef hpVar $ \pair@(hp, snum) ->
1429 Nothing -> (pair, Clearing)
1433 Just (ent@(Entry seqNo _ev), hp') ->
1435 then ((hp', Nothing), Ready ent)
1436 else assert (seqNo >= n) (pair, Waiting n)
1437 Nothing -> (pair, Waiting n)
1439{-# INLINE dequeueFromHeapSeq #-}
1441 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1443 -> IO (HeapDequeueResult t m a)
1444dequeueFromHeapSeq hpVar i =
1445 atomicModifyIORef hpVar $ \(hp, snum) ->
1450 Just (ent@(Entry seqNo _ev), hp') ->
1452 then ((hp', Nothing), Ready ent)
1453 else assert (seqNo >= i) ((hp, Just i), Waiting i)
1454 Nothing -> ((hp, Just i), Waiting i)
1455 Just _ -> error "dequeueFromHeapSeq: unreachable"
1457heapIsSane :: Maybe Int -> Int -> Bool
1458heapIsSane snum seqNo =
1461 Just n -> seqNo >= n
1463{-# INLINE requeueOnHeapTop #-}
1465 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1466 -> Entry Int (AheadHeapEntry t m a)
1469requeueOnHeapTop hpVar ent seqNo =
1470 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1471 assert (heapIsSane snum seqNo) (H.insert ent hp, Just seqNo)
1473{-# INLINE updateHeapSeq #-}
1475 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1478updateHeapSeq hpVar seqNo =
1479 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1480 assert (heapIsSane snum seqNo) (hp, Just seqNo)
1482-------------------------------------------------------------------------------
1484-------------------------------------------------------------------------------
1486-- XXX To be implemented. Use a linked queue like WAsync and put back the
1487-- remaining computation at the back of the queue instead of the heap, and
1488-- increment the sequence number.
1490-------------------------------------------------------------------------------
1491-- Dispatching workers and tracking them
1492-------------------------------------------------------------------------------
1494-- Thread tracking is needed for two reasons:
1496-- 1) Killing threads on exceptions. Threads may not be left to go away by
1497-- themselves because they may run for significant times before going away or
1498-- worse they may be stuck in IO and never go away.
1500-- 2) To know when all threads are done and the stream has ended.
1502{-# NOINLINE addThread #-}
1503addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1505 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
1507-- This is cheaper than modifyThread because we do not have to send a
1508-- outputDoorBell This can make a difference when more workers are being
1510{-# INLINE delThread #-}
1511delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1513 liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
1515-- If present then delete else add. This takes care of out of order add and
1516-- delete i.e. a delete arriving before we even added a thread.
1517-- This occurs when the forked thread is done even before the 'addThread' right
1518-- after the fork gets a chance to run.
1519{-# INLINE modifyThread #-}
1520modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1521modifyThread sv tid = do
1522 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1524 then let new = S.delete tid old in (new, new)
1525 else let new = S.insert tid old in (new, old)
1526 when (null changed) $
1529 void $ tryPutMVar (outputDoorBell sv) ()
1531-- | This is safe even if we are adding more threads concurrently because if
1532-- a child thread is adding another thread then anyway 'workerThreads' will
1534{-# INLINE allThreadsDone #-}
1535allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1536allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1538{-# NOINLINE handleChildException #-}
1539handleChildException :: SVar t m a -> SomeException -> IO ()
1540handleChildException sv e = do
1542 void $ send sv (ChildStop tid (Just e))
1544{-# NOINLINE handleFoldException #-}
1545handleFoldException :: SVar t m a -> SomeException -> IO ()
1546handleFoldException sv e = do
1548 void $ sendToProducer sv (ChildStop tid (Just e))
1550{-# NOINLINE recordMaxWorkers #-}
1551recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1552recordMaxWorkers sv = liftIO $ do
1553 active <- readIORef (workerCount sv)
1554 maxWrk <- readIORef (maxWorkers $ svarStats sv)
1555 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1556 modifyIORef (totalDispatches $ svarStats sv) (+1)
1558{-# NOINLINE pushWorker #-}
1559pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1560pushWorker yieldMax sv = do
1561 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1562 when (svarInspectMode sv) $ recordMaxWorkers sv
1563 -- This allocation matters when significant number of workers are being
1564 -- sent. We allocate it only when needed.
1566 case yieldRateInfo sv of
1567 Nothing -> return Nothing
1568 Just _ -> liftIO $ do
1569 cntRef <- newIORef 0
1570 t <- getTime Monotonic
1571 lat <- newIORef (0, t)
1572 return $ Just WorkerInfo
1573 { workerYieldMax = yieldMax
1574 , workerYieldCount = cntRef
1575 , workerLatencyStart = lat
1577 doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
1580-- XXX we can push the workerCount modification in accountThread and use the
1581-- same pushWorker for Parallel case as well.
1583-- | In contrast to pushWorker which always happens only from the consumer
1584-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1585-- producer side. So we need to use a thread safe modification of
1586-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1587-- using a CAS based modification.
1588{-# INLINE pushWorkerPar #-}
1591 => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1592pushWorkerPar sv wloop =
1593 if svarInspectMode sv
1595 else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
1600 {-# NOINLINE forkWithDiag #-}
1602 -- We do not use workerCount in case of ParallelVar but still there is
1603 -- no harm in maintaining it correctly.
1604 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1606 -- This allocation matters when significant number of workers are being
1607 -- sent. We allocate it only when needed. The overhead increases by 4x.
1609 case yieldRateInfo sv of
1610 Nothing -> return Nothing
1611 Just _ -> liftIO $ do
1612 cntRef <- newIORef 0
1613 t <- getTime Monotonic
1614 lat <- newIORef (0, t)
1615 return $ Just WorkerInfo
1616 { workerYieldMax = 0
1617 , workerYieldCount = cntRef
1618 , workerLatencyStart = lat
1621 doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
1625-- True: can dispatch more
1626-- False: cannot dispatch any more
1627dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1628dispatchWorker yieldCount sv = do
1629 let workerLimit = maxWorkerLimit sv
1630 -- XXX in case of Ahead streams we should not send more than one worker
1631 -- when the work queue is done but heap is not done.
1632 done <- liftIO $ isWorkDone sv
1633 -- Note, "done" may not mean that the work is actually finished if there
1634 -- are workers active, because there may be a worker which has not yet
1635 -- queued the leftover work.
1638 qDone <- liftIO $ isQueueDone sv
1639 -- This count may not be accurate as it is decremented by the workers
1640 -- and we have no synchronization with that decrement.
1641 active <- liftIO $ readIORef $ workerCount sv
1644 -- Note that we may deadlock if the previous workers (tasks in the
1645 -- stream) wait/depend on the future workers (tasks in the stream)
1646 -- executing. In that case we should either configure the maxWorker
1647 -- count to higher or use parallel style instead of ahead or async
1649 limit <- case remainingWork sv of
1650 Nothing -> return workerLimit
1652 n <- liftIO $ readIORef ref
1653 case yieldRateInfo sv of
1654 Just _ -> return workerLimit
1658 Unlimited -> Limited (fromIntegral n)
1659 Limited lim -> Limited $ min lim (fromIntegral n)
1661 -- XXX for ahead streams shall we take the heap yields into account
1662 -- for controlling the dispatch? We should not dispatch if the heap
1663 -- has already got the limit covered.
1664 let dispatch = pushWorker yieldCount sv >> return True
1666 Unlimited -> dispatch
1667 -- Note that the use of remainingWork and workerCount is not
1668 -- atomic and the counts may even have changed between reading
1669 -- and using them here, so this is just approximate logic and
1670 -- we cannot rely on it for correctness. We may actually
1671 -- dispatch more workers than required.
1672 Limited lim | lim > fromIntegral active -> dispatch
1675 when (active <= 0) $ pushWorker 0 sv
1679-------------------------------------------------------------------------------
1680-- Dispatch workers with rate control
1681-------------------------------------------------------------------------------
1683-- | This is a magic number and it is overloaded, and used at several places to
1686-- 1. If we have to sleep to slowdown this is the minimum period that we
1687-- accumulate before we sleep. Also, workers do not stop until this much
1688-- sleep time is accumulated.
1689-- 3. Collected latencies are computed and transferred to measured latency
1690-- after a minimum of this period.
1691minThreadDelay :: NanoSecond64
1692minThreadDelay = 1000000
1694-- | Another magic number! When we have to start more workers to cover up a
1695-- number of yields that we are lagging by then we cannot start one worker for
1696-- each yield because that may be a very big number and if the latency of the
1697-- workers is low these number of yields could be very high. We assume that we
1698-- run each extra worker for at least this much time.
1699rateRecoveryTime :: NanoSecond64
1700rateRecoveryTime = 1000000
1702-- We either block, or send one worker with limited yield count or one or more
1703-- workers with unlimited yield count.
1705 = BlockWait NanoSecond64
1706 | PartialWorker Count
1707 | ManyWorkers Int Count
1710-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1720estimateWorkers workerLimit svarYields gainLossYields
1721 svarElapsed wLatency targetLat range =
1722 -- XXX we can have a maxEfficiency combinator as well which runs the
1723 -- producer at the maximal efficiency i.e. the number of workers are chosen
1724 -- such that the latency is minimum or within a range. Or we can call it
1725 -- maxWorkerLatency.
1728 -- How many workers do we need to achieve the required rate?
1730 -- When the workers are IO bound we can increase the throughput by
1731 -- increasing the number of workers as long as the IO device has enough
1732 -- capacity to process all the requests concurrently. If the IO
1733 -- bandwidth is saturated increasing the workers won't help. Also, if
1734 -- the CPU utilization in processing all these requests exceeds the CPU
1735 -- bandwidth, then increasing the number of workers won't help.
1737 -- When the workers are purely CPU bound, increasing the workers beyond
1738 -- the number of CPUs won't help.
1740 -- TODO - measure the CPU and IO requirements of the workers. Have a
1741 -- way to specify the max bandwidth of the underlying IO mechanism and
1742 -- use that to determine the max rate of workers, and also take the CPU
1743 -- bandwidth into account. We can also discover the IO bandwidth if we
1744 -- know that we are not CPU bound, then how much steady state rate are
1745 -- we able to achieve. Design tests for CPU bound and IO bound cases.
1747 -- Calculate how many yields are we ahead or behind to match the exact
1748 -- required rate. Based on that we increase or decrease the effective
1751 -- When the worker latency is lower than required latency we begin with
1752 -- a yield and then wait rather than first waiting and then yielding.
1753 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1754 effectiveYields = svarYields + gainLossYields
1755 deltaYields = fromIntegral targetYields - effectiveYields
1757 -- We recover the deficit by running at a higher/lower rate for a
1758 -- certain amount of time. To keep the effective rate in reasonable
1759 -- limits we use rateRecoveryTime, minLatency and maxLatency.
1760 in if deltaYields > 0
1762 let deltaYieldsFreq :: Double
1764 fromIntegral deltaYields /
1765 fromIntegral rateRecoveryTime
1766 yieldsFreq = 1.0 / fromIntegral targetLat
1767 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1768 requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
1769 adjustedLat = min (max requiredLat (minLatency range))
1771 in assert (adjustedLat > 0) $
1772 if wLatency <= adjustedLat
1773 then PartialWorker deltaYields
1774 else let workers = withLimit $ wLatency `div` adjustedLat
1775 limited = min workers (fromIntegral deltaYields)
1776 in ManyWorkers (fromIntegral limited) deltaYields
1778 let expectedDuration = fromIntegral effectiveYields * targetLat
1779 sleepTime = expectedDuration - svarElapsed
1780 maxSleepTime = maxLatency range - wLatency
1781 s = min sleepTime maxSleepTime
1782 in assert (sleepTime >= 0) $
1783 -- if s is less than 0 it means our maxSleepTime is less
1784 -- than the worker latency.
1785 if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
1790 Limited x -> min n (fromIntegral x)
1792-- | Get the worker latency without resetting workerPendingLatency
1793-- Returns (total yield count, base time, measured latency)
1794-- CAUTION! keep it in sync with collectLatency
1795getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
1796getWorkerLatency yinfo = do
1797 let cur = workerPendingLatency yinfo
1798 col = workerCollectedLatency yinfo
1799 longTerm = svarAllTimeLatency yinfo
1800 measured = workerMeasuredLatency yinfo
1802 (curTotalCount, curCount, curTime) <- readIORef cur
1803 (colTotalCount, colCount, colTime) <- readIORef col
1804 (lcount, ltime) <- readIORef longTerm
1805 prevLat <- readIORef measured
1807 let latCount = colCount + curCount
1808 latTime = colTime + curTime
1809 totalCount = colTotalCount + curTotalCount
1811 if latCount > 0 && latTime > 0
1812 then let lat = latTime `div` fromIntegral latCount
1813 -- XXX Give more weight to new?
1814 in (lat + prevLat) `div` 2
1816 return (lcount + totalCount, ltime, newLat)
1818isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1819isBeyondMaxRate sv yinfo = do
1820 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1821 now <- getTime Monotonic
1822 let duration = fromRelTime64 $ diffAbsTime64 now tstamp
1823 let targetLat = svarLatencyTarget yinfo
1824 gainLoss <- readIORef (svarGainedLostYields yinfo)
1825 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1826 wLatency targetLat (svarLatencyRange yinfo)
1827 cnt <- readIORef $ workerCount sv
1828 return $ case work of
1829 -- XXX set the worker's maxYields or polling interval based on yields
1830 PartialWorker _yields -> cnt > 1
1831 ManyWorkers n _ -> cnt > n
1834-- XXX in case of ahead style stream we need to take the heap size into account
1835-- because we return the workers on the basis of that which causes a condition
1836-- where we keep dispatching and they keep returning. So we must have exactly
1837-- the same logic for not dispatching and for returning.
1840-- True: can dispatch more
1841-- False: full, no more dispatches
1842dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1843dispatchWorkerPaced sv = do
1844 let yinfo = fromJust $ yieldRateInfo sv
1845 (svarYields, svarElapsed, wLatency) <- do
1846 now <- liftIO $ getTime Monotonic
1847 (yieldCount, baseTime, lat) <-
1848 liftIO $ collectLatency sv yinfo False
1849 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
1852 then fromMaybe lat (workerBootstrapLatency yinfo)
1855 return (yieldCount, elapsed, latency)
1858 -- Need to measure the latency with a single worker before we can perform
1862 let workerLimit = maxWorkerLimit sv
1863 let targetLat = svarLatencyTarget yinfo
1864 let range = svarLatencyRange yinfo
1865 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1866 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
1867 wLatency targetLat range
1869 -- XXX we need to take yieldLimit into account here. If we are at the
1870 -- end of the limit as well as the time, we should not be sleeping.
1871 -- If we are not actually planning to dispatch any more workers we need
1872 -- to take that in account.
1875 assert (s >= 0) (return ())
1876 -- XXX note that when we return from here we will block waiting
1877 -- for the result from the existing worker. If that takes too
1878 -- long we won't be able to send another worker until the
1881 -- Sleep only if there are no active workers, otherwise we will
1882 -- defer the output of those. Note we cannot use workerCount
1883 -- here as it is not a reliable way to ensure there are
1884 -- definitely no active workers. When workerCount is 0 we may
1885 -- still have a Stop event waiting in the outputQueue.
1886 done <- allThreadsDone sv
1887 when done $ void $ do
1888 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
1889 liftIO $ threadDelay (fromIntegral us)
1892 PartialWorker yields -> do
1893 assert (yields > 0) (return ())
1894 updateGainedLostYields yinfo yields
1896 done <- allThreadsDone sv
1897 when done $ void $ dispatchWorker yields sv
1899 ManyWorkers netWorkers yields -> do
1900 assert (netWorkers >= 1) (return ())
1901 assert (yields >= 0) (return ())
1902 updateGainedLostYields yinfo yields
1904 let periodRef = workerPollingInterval yinfo
1905 ycnt = max 1 $ yields `div` fromIntegral netWorkers
1906 period = min ycnt (fromIntegral magicMaxBuffer)
1908 old <- liftIO $ readIORef periodRef
1909 when (period < old) $
1910 liftIO $ writeIORef periodRef period
1912 cnt <- liftIO $ readIORef $ workerCount sv
1915 let total = netWorkers - cnt
1916 batch = max 1 $ fromIntegral $
1917 minThreadDelay `div` targetLat
1918 -- XXX stagger the workers over a period?
1919 -- XXX cannot sleep, as that would mean we cannot process
1920 -- the outputs. need to try a different mechanism to
1922 -- when (total > batch) $
1923 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1924 dispatchN (min total batch)
1929 updateGainedLostYields yinfo yields = do
1930 let buf = fromIntegral $ svarRateBuffer yinfo
1931 when (yields /= 0 && abs yields > buf) $ do
1936 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1942 r <- dispatchWorker 0 sv
1944 then dispatchN (n - 1)
1947-------------------------------------------------------------------------------
1948-- Worker dispatch and wait loop
1949-------------------------------------------------------------------------------
1951sendWorkerDelayPaced :: SVar t m a -> IO ()
1952sendWorkerDelayPaced _ = return ()
1954sendWorkerDelay :: SVar t m a -> IO ()
1955sendWorkerDelay _sv =
1956 -- XXX we need a better way to handle this than hardcoded delays. The
1957 -- delays may be different for different systems.
1958 -- If there is a usecase where this is required we can create a combinator
1959 -- to set it as a config in the state.
1962 ncpu <- getNumCapabilities
1965 if (svarStyle sv == AheadVar)
1966 then threadDelay 100
1969 if (svarStyle sv == AheadVar)
1970 then threadDelay 100
1975{-# NOINLINE sendWorkerWait #-}
1978 => (SVar t m a -> IO ())
1979 -> (SVar t m a -> m Bool)
1982sendWorkerWait delay dispatch sv = do
1983 -- Note that we are guaranteed to have at least one outstanding worker when
1984 -- we enter this function. So if we sleep we are guaranteed to be woken up
1985 -- by an outputDoorBell, when the worker exits.
1988 (_, n) <- liftIO $ readIORef (outputQueue sv)
1990 -- The queue may be empty temporarily if the worker has dequeued the
1991 -- work item but has not enqueued the remaining part yet. For the same
1992 -- reason, a worker may come back if it tries to dequeue and finds the
1993 -- queue empty, even though the whole work has not finished yet.
1995 -- If we find that the queue is empty, but it may be empty
1996 -- temporarily, when we checked it. If that's the case we might
1997 -- sleep indefinitely unless the active workers produce some
1998 -- output. We may deadlock specially if the otuput from the active
1999 -- workers depends on the future workers that we may never send.
2000 -- So in case the queue was temporarily empty set a flag to inform
2001 -- the enqueue to send us a doorbell.
2003 -- Note that this is just a best effort mechanism to avoid a
2004 -- deadlock. Deadlocks may still happen if for some weird reason
2005 -- the consuming computation shares an MVar or some other resource
2006 -- with the producing computation and gets blocked on that resource
2007 -- and therefore cannot do any pushworker to add more threads to
2008 -- the producer. In such cases the programmer should use a parallel
2009 -- style so that all the producers are scheduled immediately and
2010 -- unconditionally. We can also use a separate monitor thread to
2011 -- push workers instead of pushing them from the consumer, but then
2012 -- we are no longer using pull based concurrency rate adaptation.
2014 -- XXX update this in the tutorial.
2016 -- Having pending active workers does not mean that we are guaranteed
2017 -- to be woken up if we sleep. In case of Ahead streams, there may be
2018 -- queued items in the heap even though the outputQueue is empty, and
2019 -- we may have active workers which are deadlocked on those items to be
2020 -- processed by the consumer. We should either guarantee that any
2021 -- worker, before returning, clears the heap or we send a worker to
2022 -- clear it. Normally we always send a worker if no output is seen, but
2023 -- if the thread limit is reached or we are using pacing then we may
2024 -- not send a worker. See the concurrentApplication test in the tests,
2025 -- that test case requires at least one yield from the producer to not
2026 -- deadlock, if the last workers output is stuck in the heap then this
2027 -- test fails. This problem can be extended to n threads when the
2028 -- consumer may depend on the evaluation of next n items in the
2031 -- register for the outputDoorBell before we check the queue so that if
2032 -- we sleep because the queue was empty we are guaranteed to get a
2033 -- doorbell on the next enqueue.
2035 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
2036 liftIO storeLoadBarrier
2037 canDoMore <- dispatch sv
2039 -- XXX test for the case when we miss sending a worker when the worker
2040 -- count is more than 1500.
2042 -- XXX Assert here that if the heap is not empty then there is at
2043 -- least one outstanding worker. Otherwise we could be sleeping
2047 then sendWorkerWait delay dispatch sv
2049 liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
2050 $ takeMVar (outputDoorBell sv)
2051 (_, len) <- liftIO $ readIORef (outputQueue sv)
2052 when (len <= 0) $ sendWorkerWait delay dispatch sv
2054-------------------------------------------------------------------------------
2055-- Reading from the workers' output queue/buffer
2056-------------------------------------------------------------------------------
2058{-# INLINE readOutputQBasic #-}
2059readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)
2060readOutputQBasic q = atomicModifyIORefCAS q $ \x -> (([],0), x)
2062{-# INLINE readOutputQRaw #-}
2063readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
2064readOutputQRaw sv = do
2065 (list, len) <- readOutputQBasic (outputQueue sv)
2066 when (svarInspectMode sv) $ do
2067 let ref = maxOutQSize $ svarStats sv
2068 oqLen <- readIORef ref
2069 when (len > oqLen) $ writeIORef ref len
2072readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2073readOutputQBounded sv = do
2074 (list, len) <- liftIO $ readOutputQRaw sv
2075 -- When there is no output seen we dispatch more workers to help
2076 -- out if there is work pending in the work queue.
2080 -- send a worker proactively, if needed, even before we start
2081 -- processing the output. This may degrade single processor
2082 -- perf but improves multi-processor, because of more
2090 cnt <- liftIO $ readIORef $ workerCount sv
2091 when (cnt <= 0) $ do
2092 done <- liftIO $ isWorkDone sv
2093 when (not done) (pushWorker 0 sv)
2095 {-# INLINE blockingRead #-}
2097 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
2098 liftIO (fst `fmap` readOutputQRaw sv)
2100readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2101readOutputQPaced sv = do
2102 (list, len) <- liftIO $ readOutputQRaw sv
2106 -- XXX send a worker proactively, if needed, even before we start
2107 -- processing the output.
2108 void $ dispatchWorkerPaced sv
2113 {-# INLINE blockingRead #-}
2115 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
2116 liftIO (fst `fmap` readOutputQRaw sv)
2118postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
2119postProcessBounded sv = do
2120 workersDone <- allThreadsDone sv
2121 -- There may still be work pending even if there are no workers pending
2122 -- because all the workers may return if the outputQueue becomes full. In
2123 -- that case send off a worker to kickstart the work again.
2125 -- Note that isWorkDone can only be safely checked if all workers are done.
2126 -- When some workers are in progress they may have decremented the yield
2127 -- Limit and later ending up incrementing it again. If we look at the yield
2128 -- limit in that window we may falsely say that it is 0 and therefore we
2132 r <- liftIO $ isWorkDone sv
2133 -- Note that we need to guarantee a worker, therefore we cannot just
2134 -- use dispatchWorker which may or may not send a worker.
2135 when (not r) (pushWorker 0 sv)
2136 -- XXX do we need to dispatch many here?
2137 -- void $ dispatchWorker sv
2141postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
2142postProcessPaced sv = do
2143 workersDone <- allThreadsDone sv
2144 -- XXX If during consumption we figure out we are getting delayed then we
2145 -- should trigger dispatch there as well. We should try to check on the
2146 -- workers after consuming every n item from the buffer?
2149 r <- liftIO $ isWorkDone sv
2151 void $ dispatchWorkerPaced sv
2152 -- Note that we need to guarantee a worker since the work is not
2153 -- finished, therefore we cannot just rely on dispatchWorkerPaced
2154 -- which may or may not send a worker.
2155 noWorker <- allThreadsDone sv
2156 when noWorker $ pushWorker 0 sv
2160-------------------------------------------------------------------------------
2162-------------------------------------------------------------------------------
2164getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
2165getYieldRateInfo st = do
2166 -- convert rate in Hertz to latency in Nanoseconds
2167 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
2168 case getStreamRate st of
2169 Just (Rate low goal high buf) ->
2170 let l = rateToLatency goal
2171 minl = rateToLatency high
2172 maxl = rateToLatency low
2173 in mkYieldRateInfo l (LatencyRange minl maxl) buf
2174 Nothing -> return Nothing
2178 mkYieldRateInfo latency latRange buf = do
2179 measured <- newIORef 0
2180 wcur <- newIORef (0,0,0)
2181 wcol <- newIORef (0,0,0)
2182 now <- getTime Monotonic
2183 wlong <- newIORef (0,now)
2184 period <- newIORef 1
2185 gainLoss <- newIORef (Count 0)
2187 return $ Just YieldRateInfo
2188 { svarLatencyTarget = latency
2189 , svarLatencyRange = latRange
2190 , svarRateBuffer = buf
2191 , svarGainedLostYields = gainLoss
2192 , workerBootstrapLatency = getStreamLatency st
2193 , workerPollingInterval = period
2194 , workerMeasuredLatency = measured
2195 , workerPendingLatency = wcur
2196 , workerCollectedLatency = wcol
2197 , svarAllTimeLatency = wlong
2200newSVarStats :: IO SVarStats
2203 maxWrk <- newIORef 0
2207 avgLat <- newIORef (0, NanoSecond64 0)
2208 maxLat <- newIORef (NanoSecond64 0)
2209 minLat <- newIORef (NanoSecond64 0)
2210 stpTime <- newIORef Nothing
2213 { totalDispatches = disp
2214 , maxWorkers = maxWrk
2215 , maxOutQSize = maxOq
2216 , maxHeapSize = maxHs
2217 , maxWorkQSize = maxWq
2218 , avgWorkerLatency = avgLat
2219 , minWorkerLatency = minLat
2220 , maxWorkerLatency = maxLat
2221 , svarStopTime = stpTime
2224-- XXX remove polymorphism in t, inline f
2225getAheadSVar :: MonadAsync m
2227 -> ( IORef ([t m a], Int)
2228 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2235getAheadSVar st f mrun = do
2236 outQ <- newIORef ([], 0)
2237 -- the second component of the tuple is "Nothing" when heap is being
2238 -- cleared, "Just n" when we are expecting sequence number n to arrive
2239 -- before we can start clearing the heap.
2240 outH <- newIORef (H.empty, Just 0)
2241 outQMv <- newEmptyMVar
2242 active <- newIORef 0
2243 wfw <- newIORef False
2244 running <- newIORef S.empty
2245 -- Sequence number is incremented whenever something is queued, therefore,
2246 -- first sequence number would be 0
2247 q <- newIORef ([], -1)
2248 stopMVar <- newMVar ()
2249 yl <- case getYieldLimit st of
2250 Nothing -> return Nothing
2251 Just x -> Just <$> newIORef x
2252 rateInfo <- getYieldRateInfo st
2254 stats <- newSVarStats
2257 let getSVar sv readOutput postProc = SVar
2258 { outputQueue = outQ
2259 , outputQueueFromConsumer = undefined
2260 , remainingWork = yl
2261 , maxBufferLimit = getMaxBuffer st
2262 , pushBufferSpace = undefined
2263 , pushBufferPolicy = undefined
2264 , pushBufferMVar = undefined
2265 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
2266 , yieldRateInfo = rateInfo
2267 , outputDoorBell = outQMv
2268 , outputDoorBellFromConsumer = undefined
2269 , readOutputQ = readOutput sv
2270 , postProcess = postProc sv
2271 , workerThreads = running
2272 , workLoop = f q outH st{streamVar = Just sv} sv
2273 , enqueue = enqueueAhead sv q
2274 , isWorkDone = isWorkDoneAhead sv q outH
2275 , isQueueDone = isQueueDoneAhead sv q
2276 , needDoorBell = wfw
2277 , svarStyle = AheadVar
2278 , svarStopStyle = StopNone
2279 , svarStopBy = undefined
2281 , workerCount = active
2282 , accountThread = delThread sv
2283 , workerStopMVar = stopMVar
2285 , svarInspectMode = getInspectMode st
2287 , aheadWorkQueue = q
2293 case getStreamRate st of
2294 Nothing -> getSVar sv readOutputQBounded postProcessBounded
2295 Just _ -> getSVar sv readOutputQPaced postProcessPaced
2300 {-# INLINE isQueueDoneAhead #-}
2301 isQueueDoneAhead sv q = do
2302 queueDone <- checkEmpty q
2304 case remainingWork sv of
2308 Nothing -> return False
2309 -- XXX note that yieldsDone can only be authoritative only when there
2310 -- are no workers running. If there are active workers they can
2311 -- later increment the yield count and therefore change the result.
2312 return $ yieldsDone || queueDone
2314 {-# INLINE isWorkDoneAhead #-}
2315 isWorkDoneAhead sv q ref = do
2317 (hp, _) <- readIORef ref
2318 return (H.size hp <= 0)
2319 queueDone <- isQueueDoneAhead sv q
2320 return $ heapDone && queueDone
2323 (xs, _) <- readIORef q
2326getParallelSVar :: MonadIO m
2327 => SVarStopStyle -> State t m a -> RunInIO m -> IO (SVar t m a)
2328getParallelSVar ss st mrun = do
2329 outQ <- newIORef ([], 0)
2330 outQRev <- newIORef ([], 0)
2331 outQMv <- newEmptyMVar
2332 outQMvRev <- newEmptyMVar
2333 active <- newIORef 0
2334 running <- newIORef S.empty
2335 yl <- case getYieldLimit st of
2336 Nothing -> return Nothing
2337 Just x -> Just <$> newIORef x
2338 rateInfo <- getYieldRateInfo st
2340 case getMaxBuffer st of
2341 Unlimited -> undefined
2342 Limited x -> fromIntegral x
2343 remBuf <- newIORef bufLim
2344 pbMVar <- newMVar ()
2346 stats <- newSVarStats
2351 StopBy -> liftIO $ newIORef undefined
2352 _ -> return undefined
2355 SVar { outputQueue = outQ
2356 , outputQueueFromConsumer = outQRev
2357 , remainingWork = yl
2358 , maxBufferLimit = getMaxBuffer st
2359 , pushBufferSpace = remBuf
2360 , pushBufferPolicy = PushBufferBlock
2361 , pushBufferMVar = pbMVar
2362 , maxWorkerLimit = Unlimited
2363 -- Used only for diagnostics
2364 , yieldRateInfo = rateInfo
2365 , outputDoorBell = outQMv
2366 , outputDoorBellFromConsumer = outQMvRev
2367 , readOutputQ = readOutputQPar sv
2368 , postProcess = allThreadsDone sv
2369 , workerThreads = running
2370 , workLoop = undefined
2371 , enqueue = undefined
2372 , isWorkDone = undefined
2373 , isQueueDone = undefined
2374 , needDoorBell = undefined
2375 , svarStyle = ParallelVar
2376 , svarStopStyle = ss
2377 , svarStopBy = stopBy
2379 , workerCount = active
2380 , accountThread = modifyThread sv
2381 , workerStopMVar = undefined
2383 , svarInspectMode = getInspectMode st
2385 , aheadWorkQueue = undefined
2386 , outputHeap = undefined
2393 readOutputQPar sv = liftIO $ do
2394 withDiagMVar sv "readOutputQPar: doorbell"
2395 $ takeMVar (outputDoorBell sv)
2396 case yieldRateInfo sv of
2397 Nothing -> return ()
2398 Just yinfo -> void $ collectLatency sv yinfo False
2399 r <- fst `fmap` readOutputQRaw sv
2401 void $ tryTakeMVar (pushBufferMVar sv)
2404 void $ tryPutMVar (pushBufferMVar sv) ()
2407sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
2408sendFirstWorker sv m = do
2409 -- Note: We must have all the work on the queue before sending the
2410 -- pushworker, otherwise the pushworker may exit before we even get a
2412 runIn <- captureMonadState
2413 liftIO $ enqueue sv (runIn, m)
2414 case yieldRateInfo sv of
2415 Nothing -> pushWorker 0 sv
2417 if svarLatencyTarget yinfo == maxBound
2418 then liftIO $ threadDelay maxBound
2419 else pushWorker 1 sv
2422{-# INLINABLE newAheadVar #-}
2423newAheadVar :: MonadAsync m
2426 -> ( IORef ([t m a], Int)
2427 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2433newAheadVar st m wloop = do
2434 mrun <- captureMonadState
2435 sv <- liftIO $ getAheadSVar st wloop mrun
2436 sendFirstWorker sv m
2438{-# INLINABLE newParallelVar #-}
2439newParallelVar :: MonadAsync m
2440 => SVarStopStyle -> State t m a -> m (SVar t m a)
2441newParallelVar ss st = do
2442 mrun <- captureMonadState
2443 liftIO $ getParallelSVar ss st mrun
2445-------------------------------------------------------------------------------
2446-- Write a stream to an SVar
2447-------------------------------------------------------------------------------
2449-- XXX this errors out for Parallel/Ahead SVars
2450-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
2451-- be read back from the SVar using 'fromSVar'.
2452toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
2453toStreamVar sv m = do
2454 runIn <- captureMonadState
2455 liftIO $ enqueue sv (runIn, m)
2456 done <- allThreadsDone sv
2457 -- XXX This is safe only when called from the consumer thread or when no
2458 -- consumer is present. There may be a race if we are not running in the
2460 -- XXX do this only if the work queue is not empty. The work may have been
2461 -- carried out by existing workers.
2463 case yieldRateInfo sv of
2464 Nothing -> pushWorker 0 sv
2465 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