2{-# LANGUAGE KindSignatures #-}
3{-# LANGUAGE ConstraintKinds #-}
4{-# LANGUAGE ExistentialQuantification #-}
5{-# LANGUAGE FlexibleContexts #-}
6{-# LANGUAGE FlexibleInstances #-}
7{-# LANGUAGE GeneralizedNewtypeDeriving #-}
8{-# LANGUAGE LambdaCase #-}
9{-# LANGUAGE MagicHash #-}
10{-# LANGUAGE MultiParamTypeClasses #-}
11{-# LANGUAGE RankNTypes #-}
12{-# LANGUAGE ScopedTypeVariables #-}
13{-# LANGUAGE UnboxedTuples #-}
16-- Module : Streamly.Internal.Data.SVar
17-- Copyright : (c) 2017 Harendra Kumar
20-- Maintainer : streamly@composewell.com
21-- Stability : experimental
24module Streamly.Internal.Data.SVar
31 -- State threaded around the stream
50 , cleanupSVarFromWorker
73 , handleChildException
79 , HeapDequeueResult(..)
97 , decrementBufferLimit
98 , incrementBufferLimit
104 , dispatchWorkerPaced
120import Control.Concurrent
121 (ThreadId, myThreadId, threadDelay, throwTo, forkIO, killThread)
122import Control.Concurrent.MVar
123 (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar,
125import Control.Exception
126 (SomeException(..), catch, mask, assert, Exception, catches,
127 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
128 BlockedIndefinitelyOnSTM(..))
129import Control.Monad (when)
130import Control.Monad.Catch (MonadThrow)
131import Control.Monad.IO.Class (MonadIO(..))
132import Control.Monad.Trans.Control
133 (MonadBaseControl, control, StM, liftBaseDiscard)
134import Streamly.Internal.Data.Atomics
135 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
137import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
138import Data.Functor (void)
139import Data.Heap (Heap, Entry(..))
140import Data.Int (Int64)
141#if __GLASGOW_HASKELL__ >= 800
142import Data.Kind (Type)
145 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
146import Data.Maybe (fromJust)
147#if __GLASGOW_HASKELL__ < 808
148import Data.Semigroup ((<>))
151import GHC.Conc (ThreadId(..))
153import GHC.IO (IO(..))
154import System.IO (hPutStrLn, stderr)
155import System.Mem.Weak (addFinalizer)
157import Streamly.Internal.Data.Time.Clock (Clock(..), getTime)
158import Streamly.Internal.Data.Time.Units
159 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
160 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
162import qualified Data.Heap as H
163import qualified Data.Set as S
165newtype Count = Count Int64
177------------------------------------------------------------------------------
178-- Parent child thread communication type
179------------------------------------------------------------------------------
181data ThreadAbort = ThreadAbort deriving Show
183instance Exception ThreadAbort
185-- | Events that a child thread may send to a parent thread.
188 | ChildStop ThreadId (Maybe SomeException)
190#if __GLASGOW_HASKELL__ < 800
193-- | Sorting out-of-turn outputs in a heap for Ahead style streams
194data AheadHeapEntry (t :: (Type -> Type) -> Type -> Type) m a =
197 | AheadEntryStream (t m a)
200------------------------------------------------------------------------------
201-- State threaded around the monad for thread management
202------------------------------------------------------------------------------
204-- | Identify the type of the SVar. Two computations using the same style can
205-- be scheduled on the same SVar.
207 AsyncVar -- depth first concurrent
208 | WAsyncVar -- breadth first concurrent
209 | ParallelVar -- all parallel
210 | AheadVar -- Concurrent look ahead
213-- | An SVar or a Stream Var is a conduit to the output from multiple streams
214-- running concurrently and asynchronously. An SVar can be thought of as an
215-- asynchronous IO handle. We can write any number of streams to an SVar in a
216-- non-blocking manner and then read them back at any time at any pace. The
217-- SVar would run the streams asynchronously and accumulate results. An SVar
218-- may not really execute the stream completely and accumulate all the results.
219-- However, it ensures that the reader can read the results at whatever paces
220-- it wants to read. The SVar monitors and adapts to the consumer's pace.
222-- An SVar is a mini scheduler, it has an associated workLoop that holds the
223-- stream tasks to be picked and run by a pool of worker threads. It has an
224-- associated output queue where the output stream elements are placed by the
225-- worker threads. A outputDoorBell is used by the worker threads to intimate the
226-- consumer thread about availability of new results in the output queue. More
227-- workers are added to the SVar by 'fromStreamVar' on demand if the output
228-- produced is not keeping pace with the consumer. On bounded SVars, workers
229-- block on the output queue to provide throttling of the producer when the
230-- consumer is not pulling fast enough. The number of workers may even get
231-- reduced depending on the consuming pace.
233-- New work is enqueued either at the time of creation of the SVar or as a
234-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
235-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
237-- We measure the individual worker latencies to estimate the number of workers
238-- needed or the amount of time we have to sleep between dispatches to achieve
239-- a particular rate when controlled pace mode it used.
240data WorkerInfo = WorkerInfo
241 { workerYieldMax :: Count -- 0 means unlimited
242 -- total number of yields by the worker till now
243 , workerYieldCount :: IORef Count
244 -- yieldCount at start, timestamp
245 , workerLatencyStart :: IORef (Count, AbsTime)
249-- | Specifies the stream yield rate in yields per second (@Hertz@).
250-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
251-- allow only as many yields as we have accumulated as per 'rateGoal' since the
252-- start of time. If the consumer or the producer is slower or faster, the
253-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
254-- between the two by increasing or decreasing the pull rate from the producer.
255-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
256-- much as 'rateBuffer'.
258-- 'rateLow' puts a bound on how low the instantaneous rate can go when
259-- recovering the rate gap. In other words, it determines the maximum yield
260-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
261-- rate can go when recovering the rate gap. In other words, it determines the
262-- minimum yield latency. We reduce the latency by increasing concurrency,
263-- therefore we can say that it puts an upper bound on concurrency.
265-- If the 'rateGoal' is 0 or negative the stream never yields a value.
266-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
270 { rateLow :: Double -- ^ The lower rate limit
271 , rateGoal :: Double -- ^ The target rate we want to achieve
272 , rateHigh :: Double -- ^ The upper rate limit
273 , rateBuffer :: Int -- ^ Maximum slack from the goal
276data LatencyRange = LatencyRange
277 { minLatency :: NanoSecond64
278 , maxLatency :: NanoSecond64
282data YieldRateInfo = YieldRateInfo
283 { svarLatencyTarget :: NanoSecond64
284 , svarLatencyRange :: LatencyRange
285 , svarRateBuffer :: Int
287 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
288 -- read by the worker threads
289 , svarGainedLostYields :: IORef Count
291 -- Actual latency/througput as seen from the consumer side, we count the
292 -- yields and the time it took to generates those yields. This is used to
293 -- increase or decrease the number of workers needed to achieve the desired
294 -- rate. The idle time of workers is adjusted in this, so that we only
295 -- account for the rate when the consumer actually demands data.
296 -- XXX interval latency is enough, we can move this under diagnostics build
297 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
298 -- read by the worker threads
299 , svarAllTimeLatency :: IORef (Count, AbsTime)
301 -- XXX Worker latency specified by the user to be used before the first
302 -- actual measurement arrives. Not yet implemented
303 , workerBootstrapLatency :: Maybe NanoSecond64
305 -- After how many yields the worker should update the latency information.
306 -- If the latency is high, this count is kept lower and vice-versa. XXX If
307 -- the latency suddenly becomes too high this count may remain too high for
308 -- long time, in such cases the consumer can change it.
309 -- 0 means no latency computation
310 -- XXX this is derivable from workerMeasuredLatency, can be removed.
311 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
312 -- read by the worker threads
313 , workerPollingInterval :: IORef Count
315 -- This is in progress latency stats maintained by the workers which we
316 -- empty into workerCollectedLatency stats at certain intervals - whenever
317 -- we process the stream elements yielded in this period. The first count
318 -- is all yields, the second count is only those yields for which the
319 -- latency was measured to be non-zero (note that if the timer resolution
320 -- is low the measured latency may be zero e.g. on JS platform).
321 -- [LOCKING] Locked access. Modified by the consumer thread as well as
322 -- worker threads. Workers modify it periodically based on
323 -- workerPollingInterval and not on every yield to reduce the locking
325 -- (allYieldCount, yieldCount, timeTaken)
326 , workerPendingLatency :: IORef (Count, Count, NanoSecond64)
328 -- This is the second level stat which is an accmulation from
329 -- workerPendingLatency stats. We keep accumulating latencies in this
330 -- bucket until we have stats for a sufficient period and then we reset it
331 -- to start collecting for the next period and retain the computed average
332 -- latency for the last period in workerMeasuredLatency.
333 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
334 -- read by the worker threads
335 -- (allYieldCount, yieldCount, timeTaken)
336 , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)
338 -- Latency as measured by workers, aggregated for the last period.
339 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
340 -- read by the worker threads
341 , workerMeasuredLatency :: IORef NanoSecond64
344data SVarStats = SVarStats {
345 totalDispatches :: IORef Int
346 , maxWorkers :: IORef Int
347 , maxOutQSize :: IORef Int
348 , maxHeapSize :: IORef Int
349 , maxWorkQSize :: IORef Int
350 , avgWorkerLatency :: IORef (Count, NanoSecond64)
351 , minWorkerLatency :: IORef NanoSecond64
352 , maxWorkerLatency :: IORef NanoSecond64
353 , svarStopTime :: IORef (Maybe AbsTime)
356-- This is essentially a 'Maybe Word' type
357data Limit = Unlimited | Limited Word deriving Show
359instance Eq Limit where
360 Unlimited == Unlimited = True
361 Unlimited == Limited _ = False
362 Limited _ == Unlimited = False
363 Limited x == Limited y = x == y
365instance Ord Limit where
366 Unlimited <= Unlimited = True
367 Unlimited <= Limited _ = False
368 Limited _ <= Unlimited = True
369 Limited x <= Limited y = x <= y
371-- When to stop the composed stream.
373 StopNone -- stops only when all streams are finished
374 | StopAny -- stop when any stream finishes
375 | StopBy -- stop when a specific stream finishes
378-- | Buffering policy for persistent push workers (in ParallelT). In a pull
379-- style SVar (in AsyncT, AheadT etc.), the consumer side dispatches workers on
380-- demand, workers terminate if the buffer is full or if the consumer is not
381-- cosuming fast enough. In a push style SVar, a worker is dispatched only
382-- once, workers are persistent and keep pushing work to the consumer via a
383-- bounded buffer. If the buffer becomes full the worker either blocks, or it
384-- can drop an item from the buffer to make space.
386-- Pull style SVars are useful in lazy stream evaluation whereas push style
387-- SVars are useful in strict left Folds.
389-- XXX Maybe we can separate the implementation in two different types instead
390-- of using a common SVar type.
392data PushBufferPolicy =
393 PushBufferDropNew -- drop the latest element and continue
394 | PushBufferDropOld -- drop the oldest element and continue
395 | PushBufferBlock -- block the thread until space
398-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have
399-- references to the original SVar stored in several functions which will keep
400-- pointing to the original data and the new updates won't reflect there.
401-- Any updateable parts must be kept in mutable references (IORef).
403data SVar t m a = SVar
406 svarStyle :: SVarStyle
407 , svarMrun :: RunInIO m
408 , svarStopStyle :: SVarStopStyle
409 , svarStopBy :: IORef ThreadId
411 -- Shared output queue (events, length)
412 -- XXX For better efficiency we can try a preallocated array type (perhaps
413 -- something like a vector) that allows an O(1) append. That way we will
414 -- avoid constructing and reversing the list. Possibly we can also avoid
415 -- the GC copying overhead. When the size increases we should be able to
416 -- allocate the array in chunks.
418 -- [LOCKING] Frequent locked access. This is updated by workers on each
419 -- yield and once in a while read by the consumer thread. This could have
420 -- big locking overhead if the number of workers is high.
422 -- XXX We can use a per-CPU data structure to reduce the locking overhead.
423 -- However, a per-cpu structure cannot guarantee the exact sequence in
424 -- which the elements were added, though that may not be important.
425 , outputQueue :: IORef ([ChildEvent a], Int)
427 -- [LOCKING] Infrequent MVar. Used when the outputQ transitions from empty
428 -- to non-empty, or a work item is queued by a worker to the work queue and
429 -- needDoorBell is set by the consumer.
430 , outputDoorBell :: MVar () -- signal the consumer about output
431 , readOutputQ :: m [ChildEvent a]
432 , postProcess :: m Bool
434 -- channel to send events from the consumer to the worker. Used to send
435 -- exceptions from a fold driver to the fold computation running as a
436 -- consumer thread in the concurrent fold cases. Currently only one event
437 -- is sent by the fold so we do not really need a queue for it.
438 , outputQueueFromConsumer :: IORef ([ChildEvent a], Int)
439 , outputDoorBellFromConsumer :: MVar ()
441 -- Combined/aggregate parameters
442 -- This is truncated to maxBufferLimit if set to more than that. Otherwise
443 -- potentially each worker may yield one value to the buffer in the worst
444 -- case exceeding the requested buffer size.
445 , maxWorkerLimit :: Limit
446 , maxBufferLimit :: Limit
447 -- These two are valid and used only when maxBufferLimit is Limited.
448 , pushBufferSpace :: IORef Count
449 , pushBufferPolicy :: PushBufferPolicy
450 -- [LOCKING] The consumer puts this MVar after emptying the buffer, workers
451 -- block on it when the buffer becomes full. No overhead unless the buffer
453 , pushBufferMVar :: MVar ()
455 -- [LOCKING] Read only access by consumer when dispatching a worker.
456 -- Decremented by workers when picking work and undo decrement if the
457 -- worker does not yield a value.
458 , remainingWork :: Maybe (IORef Count)
459 , yieldRateInfo :: Maybe YieldRateInfo
461 -- Used only by bounded SVar types
462 , enqueue :: t m a -> IO ()
463 , isWorkDone :: IO Bool
464 , isQueueDone :: IO Bool
465 , needDoorBell :: IORef Bool
466 , workLoop :: Maybe WorkerInfo -> m ()
468 -- Shared, thread tracking
469 -- [LOCKING] Updated unlocked only by consumer thread in case of
470 -- Async/Ahead style SVars. Updated locked by worker threads in case of
472 , workerThreads :: IORef (Set ThreadId)
473 -- [LOCKING] Updated locked by consumer thread when dispatching a worker
474 -- and by the worker threads when the thread stops. This is read unsafely
475 -- at several places where we want to rely on an approximate value.
476 , workerCount :: IORef Int
477 , accountThread :: ThreadId -> m ()
478 , workerStopMVar :: MVar ()
480 , svarStats :: SVarStats
481 -- to track garbage collection of SVar
482 , svarRef :: Maybe (IORef ())
484 -- Only for diagnostics
485 , svarInspectMode :: Bool
486 , svarCreator :: ThreadId
487 , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))
489 -- Shared work queue (stream, seqNo)
490 , aheadWorkQueue :: IORef ([t m a], Int)
493-------------------------------------------------------------------------------
494-- State for concurrency control
495-------------------------------------------------------------------------------
497-- XXX we can put the resettable fields in a oneShotConfig field and others in
498-- a persistentConfig field. That way reset would be fast and scalable
499-- irrespective of the number of fields.
501-- XXX make all these Limited types and use phantom types to distinguish them
502data State t m a = State
503 { -- one shot configuration, automatically reset for each API call
504 streamVar :: Maybe (SVar t m a)
505 , _yieldLimit :: Maybe Count
507 -- persistent configuration, state that remains valid until changed by
508 -- an explicit setting via a combinator.
509 , _threadsHigh :: Limit
510 , _bufferHigh :: Limit
511 -- XXX these two can be collapsed into a single type
512 , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency
513 , _maxStreamRate :: Maybe Rate
514 , _inspectMode :: Bool
517-------------------------------------------------------------------------------
518-- State defaults and reset
519-------------------------------------------------------------------------------
521-- A magical value for the buffer size arrived at by running the smallest
522-- possible task and measuring the optimal value of the buffer for that. This
523-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
525magicMaxBuffer :: Word
528defaultMaxThreads, defaultMaxBuffer :: Limit
529defaultMaxThreads = Limited magicMaxBuffer
530defaultMaxBuffer = Limited magicMaxBuffer
532-- The fields prefixed by an _ are not to be accessed or updated directly but
533-- via smart accessor APIs.
534defState :: State t m a
536 { streamVar = Nothing
537 , _yieldLimit = Nothing
538 , _threadsHigh = defaultMaxThreads
539 , _bufferHigh = defaultMaxBuffer
540 , _maxStreamRate = Nothing
541 , _streamLatency = Nothing
542 , _inspectMode = False
545-- XXX if perf gets affected we can have all the Nothing params in a single
546-- structure so that we reset is fast. We can also use rewrite rules such that
547-- reset occurs only in concurrent streams to reduce the impact on serial
549-- We can optimize this so that we clear it only if it is a Just value, it
550-- results in slightly better perf for zip/zipM but the performance of scan
551-- worsens a lot, it does not fuse.
553-- XXX This has a side effect of clearing the SVar and yieldLimit, therefore it
554-- should not be used to convert from the same type to the same type, unless
555-- you want to clear the SVar. For clearing the SVar you should be using the
556-- appropriate unStream functions instead.
558-- | Adapt the stream state from one type to another.
559adaptState :: State t m a -> State t n b
561 { streamVar = Nothing
562 , _yieldLimit = Nothing
565-------------------------------------------------------------------------------
566-- Smart get/set routines for State
567-------------------------------------------------------------------------------
569-- Use get/set routines instead of directly accessing the State fields
570setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
571setYieldLimit lim st =
578 else Just (fromIntegral n)
581getYieldLimit :: State t m a -> Maybe Count
582getYieldLimit = _yieldLimit
584setMaxThreads :: Int -> State t m a -> State t m a
590 then defaultMaxThreads
591 else Limited (fromIntegral n)
594getMaxThreads :: State t m a -> Limit
595getMaxThreads = _threadsHigh
597setMaxBuffer :: Int -> State t m a -> State t m a
603 then defaultMaxBuffer
604 else Limited (fromIntegral n)
607getMaxBuffer :: State t m a -> Limit
608getMaxBuffer = _bufferHigh
610setStreamRate :: Maybe Rate -> State t m a -> State t m a
611setStreamRate r st = st { _maxStreamRate = r }
613getStreamRate :: State t m a -> Maybe Rate
614getStreamRate = _maxStreamRate
616setStreamLatency :: Int -> State t m a -> State t m a
617setStreamLatency n st =
618 st { _streamLatency =
621 else Just (fromIntegral n)
624getStreamLatency :: State t m a -> Maybe NanoSecond64
625getStreamLatency = _streamLatency
627setInspectMode :: State t m a -> State t m a
628setInspectMode st = st { _inspectMode = True }
630getInspectMode :: State t m a -> Bool
631getInspectMode = _inspectMode
633-------------------------------------------------------------------------------
635-------------------------------------------------------------------------------
637cleanupSVar :: SVar t m a -> IO ()
639 workers <- readIORef (workerThreads sv)
640 Prelude.mapM_ (`throwTo` ThreadAbort)
643cleanupSVarFromWorker :: SVar t m a -> IO ()
644cleanupSVarFromWorker sv = do
645 workers <- readIORef (workerThreads sv)
647 Prelude.mapM_ (`throwTo` ThreadAbort)
648 (Prelude.filter (/= self) $ S.toList workers)
650-------------------------------------------------------------------------------
651-- Worker latency data collection
652-------------------------------------------------------------------------------
654-- Every once in a while workers update the latencies and check the yield rate.
655-- They return if we are above the expected yield rate. If we check too often
656-- it may impact performance, if we check less often we may have a stale
657-- picture. We update every minThreadDelay but we translate that into a yield
658-- count based on latency so that the checking overhead is little.
660-- XXX use a generation count to indicate that the value is updated. If the
661-- value is updated an existing worker must check it again on the next yield.
662-- Otherwise it is possible that we may keep updating it and because of the mod
663-- worker keeps skipping it.
664updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
665updateWorkerPollingInterval yinfo latency = do
666 let periodRef = workerPollingInterval yinfo
667 cnt = max 1 $ minThreadDelay `div` latency
668 period = min cnt (fromIntegral magicMaxBuffer)
670 writeIORef periodRef (fromIntegral period)
672{-# INLINE recordMinMaxLatency #-}
673recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()
674recordMinMaxLatency sv new = do
675 let ss = svarStats sv
676 minLat <- readIORef (minWorkerLatency ss)
677 when (new < minLat || minLat == 0) $
678 writeIORef (minWorkerLatency ss) new
680 maxLat <- readIORef (maxWorkerLatency ss)
681 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
683recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()
684recordAvgLatency sv (count, time) = do
685 let ss = svarStats sv
686 modifyIORef (avgWorkerLatency ss) $
687 \(cnt, t) -> (cnt + count, t + time)
689-- Pour the pending latency stats into a collection bucket
690{-# INLINE collectWorkerPendingLatency #-}
691collectWorkerPendingLatency
692 :: IORef (Count, Count, NanoSecond64)
693 -> IORef (Count, Count, NanoSecond64)
694 -> IO (Count, Maybe (Count, NanoSecond64))
695collectWorkerPendingLatency cur col = do
696 (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
698 (fcnt, cnt, t) <- readIORef col
699 let totalCount = fcnt + fcount
700 latCount = cnt + count
702 writeIORef col (totalCount, latCount, latTime)
704 assert (latCount == 0 || latTime /= 0) (return ())
706 if latCount > 0 && latTime > 0
707 then Just $ (latCount, latTime)
709 return (totalCount, latPair)
711{-# INLINE shouldUseCollectedBatch #-}
712shouldUseCollectedBatch
718shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
719 let r = fromIntegral newLat / fromIntegral prevLat :: Double
720 in (collectedYields > fromIntegral magicMaxBuffer)
721 || (collectedTime > minThreadDelay)
722 || (prevLat > 0 && (r > 2 || r < 0.5))
725-- Returns a triple, (1) yield count since last collection, (2) the base time
726-- when we started counting, (3) average latency in the last measurement
727-- period. The former two are used for accurate measurement of the going rate
728-- whereas the average is used for future estimates e.g. how many workers
729-- should be maintained to maintain the rate.
730-- CAUTION! keep it in sync with getWorkerLatency
731collectLatency :: SVar t m a
734 -> IO (Count, AbsTime, NanoSecond64)
735collectLatency sv yinfo drain = do
736 let cur = workerPendingLatency yinfo
737 col = workerCollectedLatency yinfo
738 longTerm = svarAllTimeLatency yinfo
739 measured = workerMeasuredLatency yinfo
741 (newCount, newLatPair) <- collectWorkerPendingLatency cur col
742 (lcount, ltime) <- readIORef longTerm
743 prevLat <- readIORef measured
745 let newLcount = lcount + newCount
746 retWith lat = return (newLcount, ltime, lat)
749 Nothing -> retWith prevLat
750 Just (count, time) -> do
751 let newLat = time `div` (fromIntegral count)
752 when (svarInspectMode sv) $ recordMinMaxLatency sv newLat
753 -- When we have collected a significant sized batch we compute the
754 -- new latency using that batch and return the new latency,
755 -- otherwise we return the previous latency derived from the
757 if shouldUseCollectedBatch newCount time newLat prevLat || drain
759 -- XXX make this NOINLINE?
760 updateWorkerPollingInterval yinfo (max newLat prevLat)
761 when (svarInspectMode sv) $ recordAvgLatency sv (count, time)
762 writeIORef col (0, 0, 0)
763 writeIORef measured ((prevLat + newLat) `div` 2)
764 modifyIORef longTerm $ \(_, t) -> (newLcount, t)
768-------------------------------------------------------------------------------
769-- Dumping the SVar for debug/diag
770-------------------------------------------------------------------------------
772dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
773dumpSVarStats sv ss style = do
774 case yieldRateInfo sv of
777 _ <- liftIO $ collectLatency sv yinfo True
780 dispatches <- readIORef $ totalDispatches ss
781 maxWrk <- readIORef $ maxWorkers ss
782 maxOq <- readIORef $ maxOutQSize ss
783 maxHp <- readIORef $ maxHeapSize ss
784 minLat <- readIORef $ minWorkerLatency ss
785 maxLat <- readIORef $ maxWorkerLatency ss
786 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
787 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
788 Nothing -> return (0, 0, 0)
790 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
793 t <- readIORef (svarStopTime ss)
794 gl <- readIORef (svarGainedLostYields yinfo)
797 now <- getTime Monotonic
798 let interval = diffAbsTime64 now startTime
799 return (cnt, gl, interval `div` fromIntegral cnt)
801 let interval = diffAbsTime64 stopTime startTime
802 return (cnt, gl, interval `div` fromIntegral cnt)
803 else return (0, 0, 0)
806 [ "total dispatches = " <> show dispatches
807 , "max workers = " <> show maxWrk
808 , "max outQSize = " <> show maxOq
809 <> (if style == AheadVar
810 then "\nheap max size = " <> show maxHp
813 then "\nmin worker latency = " <> showNanoSecond64 minLat
816 then "\nmax worker latency = " <> showNanoSecond64 maxLat
819 then let lat = avgTime `div` fromIntegral avgCnt
820 in "\navg worker latency = " <> showNanoSecond64 lat
823 then "\nSVar latency = " <> showRelTime64 svarLat
826 then "\nSVar yield count = " <> show svarCnt
828 <> (if svarGainLossCnt > 0
829 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
833{-# NOINLINE dumpSVar #-}
834dumpSVar :: SVar t m a -> IO String
836 (oqList, oqLen) <- readIORef $ outputQueue sv
837 db <- tryReadMVar $ outputDoorBell sv
839 if svarStyle sv == AheadVar
841 (oheap, oheapSeq) <- readIORef $ outputHeap sv
842 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
844 [ "heap length = " <> show (H.size oheap)
845 , "heap seqeunce = " <> show oheapSeq
846 , "work queue length = " <> show (length wq)
847 , "work queue sequence = " <> show wqSeq
851 let style = svarStyle sv
853 if style /= ParallelVar
854 then readIORef $ needDoorBell sv
856 rthread <- readIORef $ workerThreads sv
857 workers <- readIORef $ workerCount sv
858 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
862 "Creator tid = " <> show (svarCreator sv),
863 "style = " <> show (svarStyle sv)
864 , "---------CURRENT STATE-----------"
865 , "outputQueue length computed = " <> show (length oqList)
866 , "outputQueue length maintained = " <> show oqLen
867 -- XXX print the types of events in the outputQueue, first 5
868 , "outputDoorBell = " <> show db
872 [ "needDoorBell = " <> show waiting
873 , "running threads = " <> show rthread
874 -- XXX print the status of first 5 threads
875 , "running thread count = " <> show workers
877 <> "---------STATS-----------\n"
880printSVar :: SVar t m a -> String -> IO ()
882 svInfo <- dumpSVar sv
883 hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo
885-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
886-- can keep it on in production to debug problems quickly if and when they
887-- happen, but it may result in unexpected output when threads are left hanging
888-- until they are GCed because the consumer went away.
890{-# NOINLINE mvarExcHandler #-}
891mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
892mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
893 svInfo <- dumpSVar sv
894 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
897{-# NOINLINE stmExcHandler #-}
898stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
899stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
900 svInfo <- dumpSVar sv
901 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
904withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
905withDiagMVar sv label action =
906 if svarInspectMode sv
908 action `catches` [ Handler (mvarExcHandler sv label)
909 , Handler (stmExcHandler sv label)
913------------------------------------------------------------------------------
915------------------------------------------------------------------------------
917-- | A monad that can perform concurrent or parallel IO operations. Streams
918-- that can be composed concurrently require the underlying monad to be
922type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
924-- When we run computations concurrently, we completely isolate the state of
925-- the concurrent computations from the parent computation. The invariant is
926-- that we should never be running two concurrent computations in the same
927-- thread without using the runInIO function. Also, we should never be running
928-- a concurrent computation in the parent thread, otherwise it may affect the
929-- state of the parent which is against the defined semantics of concurrent
931newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }
933captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
934captureMonadState = control $ \run -> run (return $ RunInIO run)
936-- Stolen from the async package. The perf improvement is modest, 2% on a
937-- thread heavy benchmark (parallel composition using noop computations).
938-- A version of forkIO that does not include the outer exception
939-- handler: saves a bit of time when we will be installing our own
941{-# INLINE rawForkIO #-}
942rawForkIO :: IO () -> IO ThreadId
943rawForkIO action = IO $ \ s ->
944 case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)
947doFork :: MonadBaseControl IO m
950 -> (SomeException -> IO ())
952doFork action (RunInIO mrun) exHandler =
954 mask $ \restore -> do
955 tid <- rawForkIO $ catch (restore $ void $ mrun action)
959{-# INLINABLE fork #-}
960fork :: MonadBaseControl IO m => m () -> m ThreadId
961fork = liftBaseDiscard forkIO
963-- | Fork a thread that is automatically killed as soon as the reference to the
964-- returned threadId is garbage collected.
966{-# INLINABLE forkManaged #-}
967forkManaged :: (MonadIO m, MonadBaseControl IO m) => m () -> m ThreadId
968forkManaged action = do
970 liftIO $ addFinalizer tid (killThread tid)
973------------------------------------------------------------------------------
974-- Collecting results from child workers in a streamed fashion
975------------------------------------------------------------------------------
977-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
978-- faster, along with the fields in sv required by send?
979-- XXX make it noinline
981-- XXX we may want to employ an increment and decrement in batches when the
982-- througput is high or when the cost of synchronization is high. For example
983-- if the application is distributed then inc/dec of a shared variable may be
986-- A worker decrements the yield limit before it executes an action. However,
987-- the action may not result in an element being yielded, in that case we have
988-- to increment the yield limit.
990-- Note that we need it to be an Int type so that we have the ability to undo a
991-- decrement that takes it below zero.
992{-# INLINE decrementYieldLimit #-}
993decrementYieldLimit :: SVar t m a -> IO Bool
994decrementYieldLimit sv =
995 case remainingWork sv of
996 Nothing -> return True
998 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
1001{-# INLINE incrementYieldLimit #-}
1002incrementYieldLimit :: SVar t m a -> IO ()
1003incrementYieldLimit sv =
1004 case remainingWork sv of
1005 Nothing -> return ()
1006 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
1008-- XXX exception safety of all atomic/MVar operations
1010-- TBD Each worker can have their own queue and the consumer can empty one
1011-- queue at a time, that way contention can be reduced.
1013-- XXX Only yields should be counted in the buffer limit and not the Stop
1016{-# INLINE decrementBufferLimit #-}
1017decrementBufferLimit :: SVar t m a -> IO ()
1018decrementBufferLimit sv =
1019 case maxBufferLimit sv of
1020 Unlimited -> return ()
1022 let ref = pushBufferSpace sv
1023 old <- atomicModifyIORefCAS ref $ \x ->
1024 (if x >= 1 then x - 1 else x, x)
1026 case pushBufferPolicy sv of
1027 PushBufferBlock -> blockAndRetry
1028 PushBufferDropNew -> do
1029 -- We just drop one item and proceed. It is possible
1030 -- that by the time we drop the item the consumer
1031 -- thread might have run and created space in the
1032 -- buffer, but we do not care about that condition.
1033 -- This is not pedantically correct but it should be
1034 -- fine in practice.
1035 -- XXX we may want to drop only if n == maxBuf
1036 -- otherwise we must have space in the buffer and a
1037 -- decrement should be possible.
1038 block <- atomicModifyIORefCAS (outputQueue sv) $
1041 [] -> (([],n), True)
1042 _ : xs -> ((xs, n - 1), False)
1043 when block blockAndRetry
1044 -- XXX need a dequeue or ring buffer for this
1045 PushBufferDropOld -> undefined
1050 let ref = pushBufferSpace sv
1051 liftIO $ takeMVar (pushBufferMVar sv)
1052 old <- atomicModifyIORefCAS ref $ \x ->
1053 (if x >= 1 then x - 1 else x, x)
1054 -- When multiple threads sleep on takeMVar, the first thread would
1055 -- wakeup due to a putMVar by the consumer, but the rest of the threads
1056 -- would have to put back the MVar after taking it and decrementing the
1057 -- buffer count, otherwise all other threads will remain asleep.
1059 then void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
1060 -- We do not put the MVar back in this case, instead we
1061 -- wait for the consumer to put it.
1064{-# INLINE incrementBufferLimit #-}
1065incrementBufferLimit :: SVar t m a -> IO ()
1066incrementBufferLimit sv =
1067 case maxBufferLimit sv of
1068 Unlimited -> return ()
1070 atomicModifyIORefCAS_ (pushBufferSpace sv) (+ 1)
1072 void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
1074{-# INLINE resetBufferLimit #-}
1075resetBufferLimit :: SVar t m a -> IO ()
1076resetBufferLimit sv =
1077 case maxBufferLimit sv of
1078 Unlimited -> return ()
1079 Limited n -> atomicModifyIORefCAS_ (pushBufferSpace sv)
1080 (const (fromIntegral n))
1082{-# INLINE sendWithDoorBell #-}
1084 IORef ([ChildEvent a], Int) -> MVar () -> ChildEvent a -> IO Int
1085sendWithDoorBell q bell msg = do
1086 -- XXX can the access to outputQueue be made faster somehow?
1087 oldlen <- atomicModifyIORefCAS q $ \(es, n) ->
1088 ((msg : es, n + 1), n)
1089 when (oldlen <= 0) $ do
1090 -- The wake up must happen only after the store has finished otherwise
1091 -- we can have lost wakeup problems.
1093 -- Since multiple workers can try this at the same time, it is possible
1094 -- that we may put a spurious MVar after the consumer has already seen
1095 -- the output. But that's harmless, at worst it may cause the consumer
1096 -- to read the queue again and find it empty.
1097 -- The important point is that the consumer is guaranteed to receive a
1098 -- doorbell if something was added to the queue after it empties it.
1099 void $ tryPutMVar bell ()
1102-- | This function is used by the producer threads to queue output for the
1103-- consumer thread to consume. Returns whether the queue has more space.
1104send :: SVar t m a -> ChildEvent a -> IO Int
1105send sv msg = sendWithDoorBell (outputQueue sv) (outputDoorBell sv) msg
1107-- There is no bound implemented on the buffer, this is assumed to be low
1109sendToProducer :: SVar t m a -> ChildEvent a -> IO Int
1110sendToProducer sv msg = do
1111 -- In case the producer stream is blocked on pushing to the fold buffer
1112 -- then wake it up so that it can check for the stop event or exception
1113 -- being sent to it otherwise we will be deadlocked.
1114 void $ tryPutMVar (pushBufferMVar sv) ()
1115 sendWithDoorBell (outputQueueFromConsumer sv)
1116 (outputDoorBellFromConsumer sv) msg
1118-- {-# NOINLINE sendStopToProducer #-}
1119sendStopToProducer :: MonadIO m => SVar t m a -> m ()
1120sendStopToProducer sv = liftIO $ do
1122 void $ sendToProducer sv (ChildStop tid Nothing)
1124workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
1125workerCollectLatency winfo = do
1126 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
1127 cnt1 <- readIORef (workerYieldCount winfo)
1128 let cnt = cnt1 - cnt0
1132 t1 <- getTime Monotonic
1133 let period = fromRelTime64 $ diffAbsTime64 t1 t0
1134 writeIORef (workerLatencyStart winfo) (cnt1, t1)
1135 return $ Just (cnt, period)
1138-- XXX There are a number of gotchas in measuring latencies.
1139-- 1) We measure latencies only when a worker yields a value
1140-- 2) It is possible that a stream calls the stop continuation, in which case
1141-- the worker would not yield a value and we would not account that worker in
1142-- latencies. Even though this case should ideally be accounted we do not
1143-- account it because we cannot or do not distinguish it from the case
1145-- 3) It is possible that a worker returns without yielding anything because it
1146-- never got a chance to pick up work.
1147-- 4) If the system timer resolution is lower than the latency, the latency
1148-- computation turns out to be zero.
1150-- We can fix this if we measure the latencies by counting the work items
1151-- picked rather than based on the outputs yielded.
1152workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
1153workerUpdateLatency yinfo winfo = do
1154 r <- workerCollectLatency winfo
1156 Just (cnt, period) -> do
1157 -- NOTE: On JS platform the timer resolution could be pretty low. When
1158 -- the timer resolution is low, measurement of latencies could be
1159 -- tricky. All the worker latencies will turn out to be zero if they
1160 -- are lower than the resolution. We only take into account those
1161 -- measurements which are more than the timer resolution.
1163 let ref = workerPendingLatency yinfo
1164 (cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
1165 atomicModifyIORefCAS_ ref $
1166 \(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
1167 Nothing -> return ()
1169updateYieldCount :: WorkerInfo -> IO Count
1170updateYieldCount winfo = do
1171 cnt <- readIORef (workerYieldCount winfo)
1173 writeIORef (workerYieldCount winfo) cnt1
1176isBeyondMaxYield :: Count -> WorkerInfo -> Bool
1177isBeyondMaxYield cnt winfo =
1178 let ymax = workerYieldMax winfo
1179 in ymax /= 0 && cnt >= ymax
1181-- XXX we should do rate control periodically based on the total yields rather
1182-- than based on the worker local yields as other workers may have yielded more
1183-- and we should stop based on the aggregate yields. However, latency update
1184-- period can be based on individual worker yields.
1185{-# NOINLINE checkRatePeriodic #-}
1186checkRatePeriodic :: SVar t m a
1191checkRatePeriodic sv yinfo winfo ycnt = do
1192 i <- readIORef (workerPollingInterval yinfo)
1193 -- XXX use generation count to check if the interval has been updated
1194 if i /= 0 && (ycnt `mod` i) == 0
1196 workerUpdateLatency yinfo winfo
1197 -- XXX not required for parallel streams
1198 isBeyondMaxRate sv yinfo
1201-- CAUTION! this also updates the yield count and therefore should be called
1202-- only when we are actually yielding an element.
1203{-# NOINLINE workerRateControl #-}
1204workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
1205workerRateControl sv yinfo winfo = do
1206 cnt <- updateYieldCount winfo
1207 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
1208 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
1210-- XXX we should do rate control here but not latency update in case of ahead
1211-- streams. latency update must be done when we yield directly to outputQueue
1212-- or when we yield to heap.
1214-- returns whether the worker should continue (True) or stop (False).
1215{-# INLINE sendYield #-}
1216sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
1217sendYield sv mwinfo msg = do
1218 oldlen <- send sv msg
1219 let limit = maxBufferLimit sv
1220 bufferSpaceOk <- case limit of
1221 Unlimited -> return True
1223 active <- readIORef (workerCount sv)
1224 return $ (oldlen + 1) < (fromIntegral lim - active)
1228 case yieldRateInfo sv of
1229 Nothing -> return True
1230 Just yinfo -> workerRateControl sv yinfo winfo
1231 Nothing -> return True
1232 return $ bufferSpaceOk && rateLimitOk
1234{-# INLINE workerStopUpdate #-}
1235workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
1236workerStopUpdate winfo info = do
1237 i <- readIORef (workerPollingInterval info)
1238 when (i /= 0) $ workerUpdateLatency info winfo
1240{-# INLINABLE sendStop #-}
1241sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
1242sendStop sv mwinfo = do
1243 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
1244 case (mwinfo, yieldRateInfo sv) of
1245 (Just winfo, Just info) ->
1246 workerStopUpdate winfo info
1249 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
1251-------------------------------------------------------------------------------
1253-------------------------------------------------------------------------------
1255{-# INLINE ringDoorBell #-}
1256ringDoorBell :: SVar t m a -> IO ()
1259 w <- readIORef $ needDoorBell sv
1261 -- Note: the sequence of operations is important for correctness here.
1262 -- We need to set the flag to false strictly before sending the
1263 -- outputDoorBell, otherwise the outputDoorBell may get processed too
1264 -- early and then we may set the flag to False to later making the
1265 -- consumer lose the flag, even without receiving a outputDoorBell.
1266 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1267 void $ tryPutMVar (outputDoorBell sv) ()
1269-------------------------------------------------------------------------------
1271-------------------------------------------------------------------------------
1273-- Note: For purely right associated expressions this queue should have at most
1274-- one element. It grows to more than one when we have left associcated
1275-- expressions. Large left associated compositions can grow this to a
1277{-# INLINE enqueueLIFO #-}
1278enqueueLIFO :: SVar t m a -> IORef [t m a] -> t m a -> IO ()
1279enqueueLIFO sv q m = do
1280 atomicModifyIORefCAS_ q $ \ms -> m : ms
1283-------------------------------------------------------------------------------
1285-------------------------------------------------------------------------------
1287-- XXX we can use the Ahead style sequence/heap mechanism to make the best
1288-- effort to always try to finish the streams on the left side of an expression
1289-- first as long as possible.
1291{-# INLINE enqueueFIFO #-}
1292enqueueFIFO :: SVar t m a -> LinkedQueue (t m a) -> t m a -> IO ()
1293enqueueFIFO sv q m = do
1297-------------------------------------------------------------------------------
1299-------------------------------------------------------------------------------
1301-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
1302-- but always serve them in the same order as they appear in the stream. To
1303-- implement lookahead streams efficiently we assign a sequence number to each
1304-- task when the task is picked up for execution. When the task finishes, the
1305-- output is tagged with the same sequence number and we rearrange the outputs
1306-- in sequence based on that number.
1308-- To explain the mechanism imagine that the current task at the head of the
1309-- stream has a "token" to yield to the outputQueue. The ownership of the token
1310-- is determined by the current sequence number is maintained in outputHeap.
1311-- Sequence number is assigned when a task is queued. When a thread dequeues a
1312-- task it picks up the sequence number as well and when the output is ready it
1313-- uses the sequence number to queue the output to the outputQueue.
1315-- The thread with current sequence number sends the output directly to the
1316-- outputQueue. Other threads push the output to the outputHeap. When the task
1317-- being queued on the heap is a stream of many elements we evaluate only the
1318-- first element and keep the rest of the unevaluated computation in the heap.
1319-- When such a task gets the "token" for outputQueue it evaluates and directly
1320-- yields all the elements to the outputQueue without checking for the
1323-- Note that no two outputs in the heap can have the same sequence numbers and
1324-- therefore we do not need a stable heap. We have also separated the buffer
1325-- for the current task (outputQueue) and the pending tasks (outputHeap) so
1326-- that the pending tasks cannot interfere with the current task. Note that for
1327-- a single task just the outputQueue is enough and for the case of many
1328-- threads just a heap is good enough. However we balance between these two
1329-- cases, so that both are efficient.
1331-- For bigger streams it may make sense to have separate buffers for each
1332-- stream. However, for singleton streams this may become inefficient. However,
1333-- if we do not have separate buffers, then the streams that come later in
1334-- sequence may hog the buffer, hindering the streams that are ahead. For this
1335-- reason we have a single element buffer limitation for the streams being
1336-- executed in advance.
1338-- This scheme works pretty efficiently with less than 40% extra overhead
1339-- compared to the Async streams where we do not have any kind of sequencing of
1340-- the outputs. It is especially devised so that we are most efficient when we
1341-- have short tasks and need just a single thread. Also when a thread yields
1342-- many items it can hold lockfree access to the outputQueue and do it
1345-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
1346-- that they do not hog the resources and hinder the progress of the threads in
1349-- XXX Left associated ahead expressions are expensive. We start a new SVar for
1350-- each left associative expression. The queue is used only for right
1351-- associated expression, we queue the right expression and execute the left.
1352-- Thererefore the queue never has more than one item in it.
1354-- XXX we can fix this. When we queue more than one item on the queue we can
1355-- mark the previously queued item as not-runnable. The not-runnable item is
1356-- not dequeued until the already running one has finished and at that time we
1357-- would also know the exact sequence number of the already queued item.
1359-- we can even run the already queued items but they will have to be sorted in
1360-- layers in the heap. We can use a list of heaps for that.
1361{-# INLINE enqueueAhead #-}
1362enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1363enqueueAhead sv q m = do
1364 atomicModifyIORefCAS_ q $ \ case
1365 ([], n) -> ([m], n + 1) -- increment sequence
1366 _ -> error "enqueueAhead: queue is not empty"
1369-- enqueue without incrementing the sequence number
1370{-# INLINE reEnqueueAhead #-}
1371reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1372reEnqueueAhead sv q m = do
1373 atomicModifyIORefCAS_ q $ \ case
1374 ([], n) -> ([m], n) -- DO NOT increment sequence
1375 _ -> error "reEnqueueAhead: queue is not empty"
1378-- Normally the thread that has the token should never go away. The token gets
1379-- handed over to another thread, but someone or the other has the token at any
1380-- point of time. But if the task that has the token finds that the outputQueue
1381-- is full, in that case it can go away without even handing over the token to
1382-- another thread. In that case it sets the nextSequence number in the heap its
1383-- own sequence number before going away. To handle this case, any task that
1384-- does not have the token tries to dequeue from the heap first before
1385-- dequeuing from the work queue. If it finds that the task at the top of the
1386-- heap is the one that owns the current sequence number then it grabs the
1387-- token and starts with that.
1389-- XXX instead of queueing just the head element and the remaining computation
1390-- on the heap, evaluate as many as we can and place them on the heap. But we
1391-- need to give higher priority to the lower sequence numbers so that lower
1392-- priority tasks do not fill up the heap making higher priority tasks block
1393-- due to full heap. Maybe we can have a weighted space for them in the heap.
1394-- The weight is inversely proportional to the sequence number.
1396-- XXX review for livelock
1398{-# INLINE queueEmptyAhead #-}
1399queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
1400queueEmptyAhead q = liftIO $ do
1401 (xs, _) <- readIORef q
1404{-# INLINE dequeueAhead #-}
1405dequeueAhead :: MonadIO m
1406 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
1407dequeueAhead q = liftIO $
1408 atomicModifyIORefCAS q $ \case
1409 ([], n) -> (([], n), Nothing)
1410 (x : [], n) -> (([], n), Just (x, n))
1411 _ -> error "more than one item on queue"
1413-------------------------------------------------------------------------------
1415-------------------------------------------------------------------------------
1417withIORef :: IORef a -> (a -> IO b) -> IO b
1418withIORef ref f = readIORef ref >>= f
1420atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
1421atomicModifyIORef_ ref f =
1422 atomicModifyIORef ref $ \x -> (f x, ())
1424data HeapDequeueResult t m a =
1427 | Ready (Entry Int (AheadHeapEntry t m a))
1429{-# INLINE dequeueFromHeap #-}
1431 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1432 -> IO (HeapDequeueResult t m a)
1433dequeueFromHeap hpVar =
1434 atomicModifyIORef hpVar $ \pair@(hp, snum) ->
1436 Nothing -> (pair, Clearing)
1440 Just (ent@(Entry seqNo _ev), hp') ->
1442 then ((hp', Nothing), Ready ent)
1443 else assert (seqNo >= n) (pair, Waiting n)
1444 Nothing -> (pair, Waiting n)
1446{-# INLINE dequeueFromHeapSeq #-}
1448 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1450 -> IO (HeapDequeueResult t m a)
1451dequeueFromHeapSeq hpVar i =
1452 atomicModifyIORef hpVar $ \(hp, snum) ->
1457 Just (ent@(Entry seqNo _ev), hp') ->
1459 then ((hp', Nothing), Ready ent)
1460 else assert (seqNo >= i) ((hp, Just i), Waiting i)
1461 Nothing -> ((hp, Just i), Waiting i)
1462 Just _ -> error "dequeueFromHeapSeq: unreachable"
1464heapIsSane :: Maybe Int -> Int -> Bool
1465heapIsSane snum seqNo =
1468 Just n -> seqNo >= n
1470{-# INLINE requeueOnHeapTop #-}
1472 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1473 -> Entry Int (AheadHeapEntry t m a)
1476requeueOnHeapTop hpVar ent seqNo =
1477 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1478 assert (heapIsSane snum seqNo) (H.insert ent hp, Just seqNo)
1480{-# INLINE updateHeapSeq #-}
1482 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1485updateHeapSeq hpVar seqNo =
1486 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1487 assert (heapIsSane snum seqNo) (hp, Just seqNo)
1489-------------------------------------------------------------------------------
1491-------------------------------------------------------------------------------
1493-- XXX To be implemented. Use a linked queue like WAsync and put back the
1494-- remaining computation at the back of the queue instead of the heap, and
1495-- increment the sequence number.
1497-------------------------------------------------------------------------------
1498-- Dispatching workers and tracking them
1499-------------------------------------------------------------------------------
1501-- Thread tracking is needed for two reasons:
1503-- 1) Killing threads on exceptions. Threads may not be left to go away by
1504-- themselves because they may run for significant times before going away or
1505-- worse they may be stuck in IO and never go away.
1507-- 2) To know when all threads are done and the stream has ended.
1509{-# NOINLINE addThread #-}
1510addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1512 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
1514-- This is cheaper than modifyThread because we do not have to send a
1515-- outputDoorBell This can make a difference when more workers are being
1517{-# INLINE delThread #-}
1518delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1520 liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
1522-- If present then delete else add. This takes care of out of order add and
1523-- delete i.e. a delete arriving before we even added a thread.
1524-- This occurs when the forked thread is done even before the 'addThread' right
1525-- after the fork gets a chance to run.
1526{-# INLINE modifyThread #-}
1527modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1528modifyThread sv tid = do
1529 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1531 then let new = S.delete tid old in (new, new)
1532 else let new = S.insert tid old in (new, old)
1533 when (null changed) $
1536 void $ tryPutMVar (outputDoorBell sv) ()
1538-- | This is safe even if we are adding more threads concurrently because if
1539-- a child thread is adding another thread then anyway 'workerThreads' will
1541{-# INLINE allThreadsDone #-}
1542allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1543allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1545{-# NOINLINE handleChildException #-}
1546handleChildException :: SVar t m a -> SomeException -> IO ()
1547handleChildException sv e = do
1549 void $ send sv (ChildStop tid (Just e))
1551{-# NOINLINE handleFoldException #-}
1552handleFoldException :: SVar t m a -> SomeException -> IO ()
1553handleFoldException sv e = do
1555 void $ sendToProducer sv (ChildStop tid (Just e))
1557{-# NOINLINE recordMaxWorkers #-}
1558recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1559recordMaxWorkers sv = liftIO $ do
1560 active <- readIORef (workerCount sv)
1561 maxWrk <- readIORef (maxWorkers $ svarStats sv)
1562 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1563 modifyIORef (totalDispatches $ svarStats sv) (+1)
1565{-# NOINLINE pushWorker #-}
1566pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1567pushWorker yieldMax sv = do
1568 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1569 when (svarInspectMode sv) $ recordMaxWorkers sv
1570 -- This allocation matters when significant number of workers are being
1571 -- sent. We allocate it only when needed.
1573 case yieldRateInfo sv of
1574 Nothing -> return Nothing
1575 Just _ -> liftIO $ do
1576 cntRef <- newIORef 0
1577 t <- getTime Monotonic
1578 lat <- newIORef (0, t)
1579 return $ Just WorkerInfo
1580 { workerYieldMax = yieldMax
1581 , workerYieldCount = cntRef
1582 , workerLatencyStart = lat
1584 doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
1587-- XXX we can push the workerCount modification in accountThread and use the
1588-- same pushWorker for Parallel case as well.
1590-- | In contrast to pushWorker which always happens only from the consumer
1591-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1592-- producer side. So we need to use a thread safe modification of
1593-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1594-- using a CAS based modification.
1595{-# INLINE pushWorkerPar #-}
1598 => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1599pushWorkerPar sv wloop =
1600 if svarInspectMode sv
1602 else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
1607 {-# NOINLINE forkWithDiag #-}
1609 -- We do not use workerCount in case of ParallelVar but still there is
1610 -- no harm in maintaining it correctly.
1611 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1613 -- This allocation matters when significant number of workers are being
1614 -- sent. We allocate it only when needed. The overhead increases by 4x.
1616 case yieldRateInfo sv of
1617 Nothing -> return Nothing
1618 Just _ -> liftIO $ do
1619 cntRef <- newIORef 0
1620 t <- getTime Monotonic
1621 lat <- newIORef (0, t)
1622 return $ Just WorkerInfo
1623 { workerYieldMax = 0
1624 , workerYieldCount = cntRef
1625 , workerLatencyStart = lat
1628 doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
1632-- True: can dispatch more
1633-- False: cannot dispatch any more
1634dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1635dispatchWorker yieldCount sv = do
1636 let workerLimit = maxWorkerLimit sv
1637 -- XXX in case of Ahead streams we should not send more than one worker
1638 -- when the work queue is done but heap is not done.
1639 done <- liftIO $ isWorkDone sv
1640 -- Note, "done" may not mean that the work is actually finished if there
1641 -- are workers active, because there may be a worker which has not yet
1642 -- queued the leftover work.
1645 qDone <- liftIO $ isQueueDone sv
1646 -- This count may not be accurate as it is decremented by the workers
1647 -- and we have no synchronization with that decrement.
1648 active <- liftIO $ readIORef $ workerCount sv
1651 -- Note that we may deadlock if the previous workers (tasks in the
1652 -- stream) wait/depend on the future workers (tasks in the stream)
1653 -- executing. In that case we should either configure the maxWorker
1654 -- count to higher or use parallel style instead of ahead or async
1656 limit <- case remainingWork sv of
1657 Nothing -> return workerLimit
1659 n <- liftIO $ readIORef ref
1660 case yieldRateInfo sv of
1661 Just _ -> return workerLimit
1665 Unlimited -> Limited (fromIntegral n)
1666 Limited lim -> Limited $ min lim (fromIntegral n)
1668 -- XXX for ahead streams shall we take the heap yields into account
1669 -- for controlling the dispatch? We should not dispatch if the heap
1670 -- has already got the limit covered.
1671 let dispatch = pushWorker yieldCount sv >> return True
1673 Unlimited -> dispatch
1674 -- Note that the use of remainingWork and workerCount is not
1675 -- atomic and the counts may even have changed between reading
1676 -- and using them here, so this is just approximate logic and
1677 -- we cannot rely on it for correctness. We may actually
1678 -- dispatch more workers than required.
1679 Limited lim | lim > fromIntegral active -> dispatch
1682 when (active <= 0) $ pushWorker 0 sv
1686-------------------------------------------------------------------------------
1687-- Dispatch workers with rate control
1688-------------------------------------------------------------------------------
1690-- | This is a magic number and it is overloaded, and used at several places to
1693-- 1. If we have to sleep to slowdown this is the minimum period that we
1694-- accumulate before we sleep. Also, workers do not stop until this much
1695-- sleep time is accumulated.
1696-- 3. Collected latencies are computed and transferred to measured latency
1697-- after a minimum of this period.
1698minThreadDelay :: NanoSecond64
1699minThreadDelay = 1000000
1701-- | Another magic number! When we have to start more workers to cover up a
1702-- number of yields that we are lagging by then we cannot start one worker for
1703-- each yield because that may be a very big number and if the latency of the
1704-- workers is low these number of yields could be very high. We assume that we
1705-- run each extra worker for at least this much time.
1706rateRecoveryTime :: NanoSecond64
1707rateRecoveryTime = 1000000
1709-- We either block, or send one worker with limited yield count or one or more
1710-- workers with unlimited yield count.
1712 = BlockWait NanoSecond64
1713 | PartialWorker Count
1714 | ManyWorkers Int Count
1717-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1727estimateWorkers workerLimit svarYields gainLossYields
1728 svarElapsed wLatency targetLat range =
1729 -- XXX we can have a maxEfficiency combinator as well which runs the
1730 -- producer at the maximal efficiency i.e. the number of workers are chosen
1731 -- such that the latency is minimum or within a range. Or we can call it
1732 -- maxWorkerLatency.
1735 -- How many workers do we need to achieve the required rate?
1737 -- When the workers are IO bound we can increase the throughput by
1738 -- increasing the number of workers as long as the IO device has enough
1739 -- capacity to process all the requests concurrently. If the IO
1740 -- bandwidth is saturated increasing the workers won't help. Also, if
1741 -- the CPU utilization in processing all these requests exceeds the CPU
1742 -- bandwidth, then increasing the number of workers won't help.
1744 -- When the workers are purely CPU bound, increasing the workers beyond
1745 -- the number of CPUs won't help.
1747 -- TODO - measure the CPU and IO requirements of the workers. Have a
1748 -- way to specify the max bandwidth of the underlying IO mechanism and
1749 -- use that to determine the max rate of workers, and also take the CPU
1750 -- bandwidth into account. We can also discover the IO bandwidth if we
1751 -- know that we are not CPU bound, then how much steady state rate are
1752 -- we able to achieve. Design tests for CPU bound and IO bound cases.
1754 -- Calculate how many yields are we ahead or behind to match the exact
1755 -- required rate. Based on that we increase or decrease the effective
1758 -- When the worker latency is lower than required latency we begin with
1759 -- a yield and then wait rather than first waiting and then yielding.
1760 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1761 effectiveYields = svarYields + gainLossYields
1762 deltaYields = fromIntegral targetYields - effectiveYields
1764 -- We recover the deficit by running at a higher/lower rate for a
1765 -- certain amount of time. To keep the effective rate in reasonable
1766 -- limits we use rateRecoveryTime, minLatency and maxLatency.
1767 in if deltaYields > 0
1769 let deltaYieldsFreq :: Double
1771 fromIntegral deltaYields /
1772 fromIntegral rateRecoveryTime
1773 yieldsFreq = 1.0 / fromIntegral targetLat
1774 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1775 requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
1776 adjustedLat = min (max requiredLat (minLatency range))
1778 in assert (adjustedLat > 0) $
1779 if wLatency <= adjustedLat
1780 then PartialWorker deltaYields
1781 else let workers = withLimit $ wLatency `div` adjustedLat
1782 limited = min workers (fromIntegral deltaYields)
1783 in ManyWorkers (fromIntegral limited) deltaYields
1785 let expectedDuration = fromIntegral effectiveYields * targetLat
1786 sleepTime = expectedDuration - svarElapsed
1787 maxSleepTime = maxLatency range - wLatency
1788 s = min sleepTime maxSleepTime
1789 in assert (sleepTime >= 0) $
1790 -- if s is less than 0 it means our maxSleepTime is less
1791 -- than the worker latency.
1792 if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
1797 Limited x -> min n (fromIntegral x)
1799-- | Get the worker latency without resetting workerPendingLatency
1800-- Returns (total yield count, base time, measured latency)
1801-- CAUTION! keep it in sync with collectLatency
1802getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
1803getWorkerLatency yinfo = do
1804 let cur = workerPendingLatency yinfo
1805 col = workerCollectedLatency yinfo
1806 longTerm = svarAllTimeLatency yinfo
1807 measured = workerMeasuredLatency yinfo
1809 (curTotalCount, curCount, curTime) <- readIORef cur
1810 (colTotalCount, colCount, colTime) <- readIORef col
1811 (lcount, ltime) <- readIORef longTerm
1812 prevLat <- readIORef measured
1814 let latCount = colCount + curCount
1815 latTime = colTime + curTime
1816 totalCount = colTotalCount + curTotalCount
1818 if latCount > 0 && latTime > 0
1819 then let lat = latTime `div` fromIntegral latCount
1820 -- XXX Give more weight to new?
1821 in (lat + prevLat) `div` 2
1823 return (lcount + totalCount, ltime, newLat)
1825isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1826isBeyondMaxRate sv yinfo = do
1827 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1828 now <- getTime Monotonic
1829 let duration = fromRelTime64 $ diffAbsTime64 now tstamp
1830 let targetLat = svarLatencyTarget yinfo
1831 gainLoss <- readIORef (svarGainedLostYields yinfo)
1832 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1833 wLatency targetLat (svarLatencyRange yinfo)
1834 cnt <- readIORef $ workerCount sv
1835 return $ case work of
1836 -- XXX set the worker's maxYields or polling interval based on yields
1837 PartialWorker _yields -> cnt > 1
1838 ManyWorkers n _ -> cnt > n
1841-- XXX in case of ahead style stream we need to take the heap size into account
1842-- because we return the workers on the basis of that which causes a condition
1843-- where we keep dispatching and they keep returning. So we must have exactly
1844-- the same logic for not dispatching and for returning.
1847-- True: can dispatch more
1848-- False: full, no more dispatches
1849dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1850dispatchWorkerPaced sv = do
1851 let yinfo = fromJust $ yieldRateInfo sv
1852 (svarYields, svarElapsed, wLatency) <- do
1853 now <- liftIO $ getTime Monotonic
1854 (yieldCount, baseTime, lat) <-
1855 liftIO $ collectLatency sv yinfo False
1856 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
1860 case workerBootstrapLatency yinfo of
1865 return (yieldCount, elapsed, latency)
1868 -- Need to measure the latency with a single worker before we can perform
1872 let workerLimit = maxWorkerLimit sv
1873 let targetLat = svarLatencyTarget yinfo
1874 let range = svarLatencyRange yinfo
1875 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1876 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
1877 wLatency targetLat range
1879 -- XXX we need to take yieldLimit into account here. If we are at the
1880 -- end of the limit as well as the time, we should not be sleeping.
1881 -- If we are not actually planning to dispatch any more workers we need
1882 -- to take that in account.
1885 assert (s >= 0) (return ())
1886 -- XXX note that when we return from here we will block waiting
1887 -- for the result from the existing worker. If that takes too
1888 -- long we won't be able to send another worker until the
1891 -- Sleep only if there are no active workers, otherwise we will
1892 -- defer the output of those. Note we cannot use workerCount
1893 -- here as it is not a reliable way to ensure there are
1894 -- definitely no active workers. When workerCount is 0 we may
1895 -- still have a Stop event waiting in the outputQueue.
1896 done <- allThreadsDone sv
1897 when done $ void $ do
1898 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
1899 liftIO $ threadDelay (fromIntegral us)
1902 PartialWorker yields -> do
1903 assert (yields > 0) (return ())
1904 updateGainedLostYields yinfo yields
1906 done <- allThreadsDone sv
1907 when done $ void $ dispatchWorker yields sv
1909 ManyWorkers netWorkers yields -> do
1910 assert (netWorkers >= 1) (return ())
1911 assert (yields >= 0) (return ())
1912 updateGainedLostYields yinfo yields
1914 let periodRef = workerPollingInterval yinfo
1915 ycnt = max 1 $ yields `div` fromIntegral netWorkers
1916 period = min ycnt (fromIntegral magicMaxBuffer)
1918 old <- liftIO $ readIORef periodRef
1919 when (period < old) $
1920 liftIO $ writeIORef periodRef period
1922 cnt <- liftIO $ readIORef $ workerCount sv
1925 let total = netWorkers - cnt
1926 batch = max 1 $ fromIntegral $
1927 minThreadDelay `div` targetLat
1928 -- XXX stagger the workers over a period?
1929 -- XXX cannot sleep, as that would mean we cannot process
1930 -- the outputs. need to try a different mechanism to
1932 -- when (total > batch) $
1933 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1934 dispatchN (min total batch)
1939 updateGainedLostYields yinfo yields = do
1940 let buf = fromIntegral $ svarRateBuffer yinfo
1941 when (yields /= 0 && abs yields > buf) $ do
1946 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1952 r <- dispatchWorker 0 sv
1954 then dispatchN (n - 1)
1957-------------------------------------------------------------------------------
1958-- Worker dispatch and wait loop
1959-------------------------------------------------------------------------------
1961sendWorkerDelayPaced :: SVar t m a -> IO ()
1962sendWorkerDelayPaced _ = return ()
1964sendWorkerDelay :: SVar t m a -> IO ()
1965sendWorkerDelay _sv =
1966 -- XXX we need a better way to handle this than hardcoded delays. The
1967 -- delays may be different for different systems.
1968 -- If there is a usecase where this is required we can create a combinator
1969 -- to set it as a config in the state.
1972 ncpu <- getNumCapabilities
1975 if (svarStyle sv == AheadVar)
1976 then threadDelay 100
1979 if (svarStyle sv == AheadVar)
1980 then threadDelay 100
1985{-# NOINLINE sendWorkerWait #-}
1988 => (SVar t m a -> IO ())
1989 -> (SVar t m a -> m Bool)
1992sendWorkerWait delay dispatch sv = do
1993 -- Note that we are guaranteed to have at least one outstanding worker when
1994 -- we enter this function. So if we sleep we are guaranteed to be woken up
1995 -- by an outputDoorBell, when the worker exits.
1998 (_, n) <- liftIO $ readIORef (outputQueue sv)
2000 -- The queue may be empty temporarily if the worker has dequeued the
2001 -- work item but has not enqueued the remaining part yet. For the same
2002 -- reason, a worker may come back if it tries to dequeue and finds the
2003 -- queue empty, even though the whole work has not finished yet.
2005 -- If we find that the queue is empty, but it may be empty
2006 -- temporarily, when we checked it. If that's the case we might
2007 -- sleep indefinitely unless the active workers produce some
2008 -- output. We may deadlock specially if the otuput from the active
2009 -- workers depends on the future workers that we may never send.
2010 -- So in case the queue was temporarily empty set a flag to inform
2011 -- the enqueue to send us a doorbell.
2013 -- Note that this is just a best effort mechanism to avoid a
2014 -- deadlock. Deadlocks may still happen if for some weird reason
2015 -- the consuming computation shares an MVar or some other resource
2016 -- with the producing computation and gets blocked on that resource
2017 -- and therefore cannot do any pushworker to add more threads to
2018 -- the producer. In such cases the programmer should use a parallel
2019 -- style so that all the producers are scheduled immediately and
2020 -- unconditionally. We can also use a separate monitor thread to
2021 -- push workers instead of pushing them from the consumer, but then
2022 -- we are no longer using pull based concurrency rate adaptation.
2024 -- XXX update this in the tutorial.
2026 -- Having pending active workers does not mean that we are guaranteed
2027 -- to be woken up if we sleep. In case of Ahead streams, there may be
2028 -- queued items in the heap even though the outputQueue is empty, and
2029 -- we may have active workers which are deadlocked on those items to be
2030 -- processed by the consumer. We should either guarantee that any
2031 -- worker, before returning, clears the heap or we send a worker to
2032 -- clear it. Normally we always send a worker if no output is seen, but
2033 -- if the thread limit is reached or we are using pacing then we may
2034 -- not send a worker. See the concurrentApplication test in the tests,
2035 -- that test case requires at least one yield from the producer to not
2036 -- deadlock, if the last workers output is stuck in the heap then this
2037 -- test fails. This problem can be extended to n threads when the
2038 -- consumer may depend on the evaluation of next n items in the
2041 -- register for the outputDoorBell before we check the queue so that if
2042 -- we sleep because the queue was empty we are guaranteed to get a
2043 -- doorbell on the next enqueue.
2045 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
2046 liftIO storeLoadBarrier
2047 canDoMore <- dispatch sv
2049 -- XXX test for the case when we miss sending a worker when the worker
2050 -- count is more than 1500.
2052 -- XXX Assert here that if the heap is not empty then there is at
2053 -- least one outstanding worker. Otherwise we could be sleeping
2057 then sendWorkerWait delay dispatch sv
2059 liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
2060 $ takeMVar (outputDoorBell sv)
2061 (_, len) <- liftIO $ readIORef (outputQueue sv)
2062 when (len <= 0) $ sendWorkerWait delay dispatch sv
2064-------------------------------------------------------------------------------
2065-- Reading from the workers' output queue/buffer
2066-------------------------------------------------------------------------------
2068{-# INLINE readOutputQBasic #-}
2069readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)
2070readOutputQBasic q = atomicModifyIORefCAS q $ \x -> (([],0), x)
2072{-# INLINE readOutputQRaw #-}
2073readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
2074readOutputQRaw sv = do
2075 (list, len) <- readOutputQBasic (outputQueue sv)
2076 when (svarInspectMode sv) $ do
2077 let ref = maxOutQSize $ svarStats sv
2078 oqLen <- readIORef ref
2079 when (len > oqLen) $ writeIORef ref len
2082readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2083readOutputQBounded sv = do
2084 (list, len) <- liftIO $ readOutputQRaw sv
2085 -- When there is no output seen we dispatch more workers to help
2086 -- out if there is work pending in the work queue.
2090 -- send a worker proactively, if needed, even before we start
2091 -- processing the output. This may degrade single processor
2092 -- perf but improves multi-processor, because of more
2100 cnt <- liftIO $ readIORef $ workerCount sv
2101 when (cnt <= 0) $ do
2102 done <- liftIO $ isWorkDone sv
2103 when (not done) (pushWorker 0 sv)
2105 {-# INLINE blockingRead #-}
2107 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
2108 liftIO (fst `fmap` readOutputQRaw sv)
2110readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2111readOutputQPaced sv = do
2112 (list, len) <- liftIO $ readOutputQRaw sv
2116 -- XXX send a worker proactively, if needed, even before we start
2117 -- processing the output.
2118 void $ dispatchWorkerPaced sv
2123 {-# INLINE blockingRead #-}
2125 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
2126 liftIO (fst `fmap` readOutputQRaw sv)
2128postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
2129postProcessBounded sv = do
2130 workersDone <- allThreadsDone sv
2131 -- There may still be work pending even if there are no workers pending
2132 -- because all the workers may return if the outputQueue becomes full. In
2133 -- that case send off a worker to kickstart the work again.
2135 -- Note that isWorkDone can only be safely checked if all workers are done.
2136 -- When some workers are in progress they may have decremented the yield
2137 -- Limit and later ending up incrementing it again. If we look at the yield
2138 -- limit in that window we may falsely say that it is 0 and therefore we
2142 r <- liftIO $ isWorkDone sv
2143 -- Note that we need to guarantee a worker, therefore we cannot just
2144 -- use dispatchWorker which may or may not send a worker.
2145 when (not r) (pushWorker 0 sv)
2146 -- XXX do we need to dispatch many here?
2147 -- void $ dispatchWorker sv
2151postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
2152postProcessPaced sv = do
2153 workersDone <- allThreadsDone sv
2154 -- XXX If during consumption we figure out we are getting delayed then we
2155 -- should trigger dispatch there as well. We should try to check on the
2156 -- workers after consuming every n item from the buffer?
2159 r <- liftIO $ isWorkDone sv
2161 void $ dispatchWorkerPaced sv
2162 -- Note that we need to guarantee a worker since the work is not
2163 -- finished, therefore we cannot just rely on dispatchWorkerPaced
2164 -- which may or may not send a worker.
2165 noWorker <- allThreadsDone sv
2166 when noWorker $ pushWorker 0 sv
2170-------------------------------------------------------------------------------
2172-------------------------------------------------------------------------------
2174getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
2175getYieldRateInfo st = do
2176 -- convert rate in Hertz to latency in Nanoseconds
2177 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
2178 case getStreamRate st of
2179 Just (Rate low goal high buf) ->
2180 let l = rateToLatency goal
2181 minl = rateToLatency high
2182 maxl = rateToLatency low
2183 in mkYieldRateInfo l (LatencyRange minl maxl) buf
2184 Nothing -> return Nothing
2188 mkYieldRateInfo latency latRange buf = do
2189 measured <- newIORef 0
2190 wcur <- newIORef (0,0,0)
2191 wcol <- newIORef (0,0,0)
2192 now <- getTime Monotonic
2193 wlong <- newIORef (0,now)
2194 period <- newIORef 1
2195 gainLoss <- newIORef (Count 0)
2197 return $ Just YieldRateInfo
2198 { svarLatencyTarget = latency
2199 , svarLatencyRange = latRange
2200 , svarRateBuffer = buf
2201 , svarGainedLostYields = gainLoss
2202 , workerBootstrapLatency = getStreamLatency st
2203 , workerPollingInterval = period
2204 , workerMeasuredLatency = measured
2205 , workerPendingLatency = wcur
2206 , workerCollectedLatency = wcol
2207 , svarAllTimeLatency = wlong
2210newSVarStats :: IO SVarStats
2213 maxWrk <- newIORef 0
2217 avgLat <- newIORef (0, NanoSecond64 0)
2218 maxLat <- newIORef (NanoSecond64 0)
2219 minLat <- newIORef (NanoSecond64 0)
2220 stpTime <- newIORef Nothing
2223 { totalDispatches = disp
2224 , maxWorkers = maxWrk
2225 , maxOutQSize = maxOq
2226 , maxHeapSize = maxHs
2227 , maxWorkQSize = maxWq
2228 , avgWorkerLatency = avgLat
2229 , minWorkerLatency = minLat
2230 , maxWorkerLatency = maxLat
2231 , svarStopTime = stpTime
2234-- XXX remove polymorphism in t, inline f
2235getAheadSVar :: MonadAsync m
2237 -> ( IORef ([t m a], Int)
2238 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2245getAheadSVar st f mrun = do
2246 outQ <- newIORef ([], 0)
2247 -- the second component of the tuple is "Nothing" when heap is being
2248 -- cleared, "Just n" when we are expecting sequence number n to arrive
2249 -- before we can start clearing the heap.
2250 outH <- newIORef (H.empty, Just 0)
2251 outQMv <- newEmptyMVar
2252 active <- newIORef 0
2253 wfw <- newIORef False
2254 running <- newIORef S.empty
2255 -- Sequence number is incremented whenever something is queued, therefore,
2256 -- first sequence number would be 0
2257 q <- newIORef ([], -1)
2258 stopMVar <- newMVar ()
2259 yl <- case getYieldLimit st of
2260 Nothing -> return Nothing
2261 Just x -> Just <$> newIORef x
2262 rateInfo <- getYieldRateInfo st
2264 stats <- newSVarStats
2267 let getSVar sv readOutput postProc = SVar
2268 { outputQueue = outQ
2269 , outputQueueFromConsumer = undefined
2270 , remainingWork = yl
2271 , maxBufferLimit = getMaxBuffer st
2272 , pushBufferSpace = undefined
2273 , pushBufferPolicy = undefined
2274 , pushBufferMVar = undefined
2275 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
2276 , yieldRateInfo = rateInfo
2277 , outputDoorBell = outQMv
2278 , outputDoorBellFromConsumer = undefined
2279 , readOutputQ = readOutput sv
2280 , postProcess = postProc sv
2281 , workerThreads = running
2282 , workLoop = f q outH st{streamVar = Just sv} sv
2283 , enqueue = enqueueAhead sv q
2284 , isWorkDone = isWorkDoneAhead sv q outH
2285 , isQueueDone = isQueueDoneAhead sv q
2286 , needDoorBell = wfw
2287 , svarStyle = AheadVar
2288 , svarStopStyle = StopNone
2289 , svarStopBy = undefined
2291 , workerCount = active
2292 , accountThread = delThread sv
2293 , workerStopMVar = stopMVar
2295 , svarInspectMode = getInspectMode st
2297 , aheadWorkQueue = q
2303 case getStreamRate st of
2304 Nothing -> getSVar sv readOutputQBounded postProcessBounded
2305 Just _ -> getSVar sv readOutputQPaced postProcessPaced
2310 {-# INLINE isQueueDoneAhead #-}
2311 isQueueDoneAhead sv q = do
2312 queueDone <- checkEmpty q
2314 case remainingWork sv of
2318 Nothing -> return False
2319 -- XXX note that yieldsDone can only be authoritative only when there
2320 -- are no workers running. If there are active workers they can
2321 -- later increment the yield count and therefore change the result.
2322 return $ yieldsDone || queueDone
2324 {-# INLINE isWorkDoneAhead #-}
2325 isWorkDoneAhead sv q ref = do
2327 (hp, _) <- readIORef ref
2328 return (H.size hp <= 0)
2329 queueDone <- isQueueDoneAhead sv q
2330 return $ heapDone && queueDone
2333 (xs, _) <- readIORef q
2336getParallelSVar :: MonadIO m
2337 => SVarStopStyle -> State t m a -> RunInIO m -> IO (SVar t m a)
2338getParallelSVar ss st mrun = do
2339 outQ <- newIORef ([], 0)
2340 outQRev <- newIORef ([], 0)
2341 outQMv <- newEmptyMVar
2342 outQMvRev <- newEmptyMVar
2343 active <- newIORef 0
2344 running <- newIORef S.empty
2345 yl <- case getYieldLimit st of
2346 Nothing -> return Nothing
2347 Just x -> Just <$> newIORef x
2348 rateInfo <- getYieldRateInfo st
2350 case getMaxBuffer st of
2351 Unlimited -> undefined
2352 Limited x -> (fromIntegral x)
2353 remBuf <- newIORef bufLim
2354 pbMVar <- newMVar ()
2356 stats <- newSVarStats
2361 StopBy -> liftIO $ newIORef undefined
2362 _ -> return undefined
2365 SVar { outputQueue = outQ
2366 , outputQueueFromConsumer = outQRev
2367 , remainingWork = yl
2368 , maxBufferLimit = getMaxBuffer st
2369 , pushBufferSpace = remBuf
2370 , pushBufferPolicy = PushBufferBlock
2371 , pushBufferMVar = pbMVar
2372 , maxWorkerLimit = Unlimited
2373 -- Used only for diagnostics
2374 , yieldRateInfo = rateInfo
2375 , outputDoorBell = outQMv
2376 , outputDoorBellFromConsumer = outQMvRev
2377 , readOutputQ = readOutputQPar sv
2378 , postProcess = allThreadsDone sv
2379 , workerThreads = running
2380 , workLoop = undefined
2381 , enqueue = undefined
2382 , isWorkDone = undefined
2383 , isQueueDone = undefined
2384 , needDoorBell = undefined
2385 , svarStyle = ParallelVar
2386 , svarStopStyle = ss
2387 , svarStopBy = stopBy
2389 , workerCount = active
2390 , accountThread = modifyThread sv
2391 , workerStopMVar = undefined
2393 , svarInspectMode = getInspectMode st
2395 , aheadWorkQueue = undefined
2396 , outputHeap = undefined
2403 readOutputQPar sv = liftIO $ do
2404 withDiagMVar sv "readOutputQPar: doorbell"
2405 $ takeMVar (outputDoorBell sv)
2406 case yieldRateInfo sv of
2407 Nothing -> return ()
2408 Just yinfo -> void $ collectLatency sv yinfo False
2409 r <- fst `fmap` readOutputQRaw sv
2411 void $ tryTakeMVar (pushBufferMVar sv)
2414 void $ tryPutMVar (pushBufferMVar sv) ()
2417sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
2418sendFirstWorker sv m = do
2419 -- Note: We must have all the work on the queue before sending the
2420 -- pushworker, otherwise the pushworker may exit before we even get a
2422 liftIO $ enqueue sv m
2423 case yieldRateInfo sv of
2424 Nothing -> pushWorker 0 sv
2426 if svarLatencyTarget yinfo == maxBound
2427 then liftIO $ threadDelay maxBound
2428 else pushWorker 1 sv
2431{-# INLINABLE newAheadVar #-}
2432newAheadVar :: MonadAsync m
2435 -> ( IORef ([t m a], Int)
2436 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2442newAheadVar st m wloop = do
2443 mrun <- captureMonadState
2444 sv <- liftIO $ getAheadSVar st wloop mrun
2445 sendFirstWorker sv m
2447{-# INLINABLE newParallelVar #-}
2448newParallelVar :: MonadAsync m
2449 => SVarStopStyle -> State t m a -> m (SVar t m a)
2450newParallelVar ss st = do
2451 mrun <- captureMonadState
2452 liftIO $ getParallelSVar ss st mrun
2454-------------------------------------------------------------------------------
2455-- Write a stream to an SVar
2456-------------------------------------------------------------------------------
2458-- XXX this errors out for Parallel/Ahead SVars
2459-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
2460-- be read back from the SVar using 'fromSVar'.
2461toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
2462toStreamVar sv m = do
2463 liftIO $ enqueue sv m
2464 done <- allThreadsDone sv
2465 -- XXX This is safe only when called from the consumer thread or when no
2466 -- consumer is present. There may be a race if we are not running in the
2468 -- XXX do this only if the work queue is not empty. The work may have been
2469 -- carried out by existing workers.
2471 case yieldRateInfo sv of
2472 Nothing -> pushWorker 0 sv
2473 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