2{-# LANGUAGE KindSignatures #-}
3{-# LANGUAGE ConstraintKinds #-}
4{-# LANGUAGE ExistentialQuantification #-}
5{-# LANGUAGE FlexibleContexts #-}
6{-# LANGUAGE FlexibleInstances #-}
7{-# LANGUAGE GeneralizedNewtypeDeriving #-}
8{-# LANGUAGE LambdaCase #-}
9{-# LANGUAGE MagicHash #-}
10{-# LANGUAGE MultiParamTypeClasses #-}
11{-# LANGUAGE RankNTypes #-}
12{-# LANGUAGE ScopedTypeVariables #-}
13{-# LANGUAGE UnboxedTuples #-}
16-- Module : Streamly.SVar
17-- Copyright : (c) 2017 Harendra Kumar
20-- Maintainer : harendra.kumar@gmail.com
21-- Stability : experimental
31 -- State threaded around the stream
49 , cleanupSVarFromWorker
74 , HeapDequeueResult(..)
92 , decrementBufferLimit
93 , incrementBufferLimit
110import Control.Concurrent
111 (ThreadId, myThreadId, threadDelay, throwTo)
112import Control.Concurrent.MVar
113 (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar, tryReadMVar)
114import Control.Exception
115 (SomeException(..), catch, mask, assert, Exception, catches,
116 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
117 BlockedIndefinitelyOnSTM(..))
118import Control.Monad (when)
119import Control.Monad.Catch (MonadThrow)
120import Control.Monad.IO.Class (MonadIO(..))
121import Control.Monad.Trans.Control (MonadBaseControl, control, StM)
122import Streamly.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)
130 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
131import Data.List ((\\))
132import Data.Maybe (fromJust)
133import Data.Semigroup ((<>))
135import GHC.Conc (ThreadId(..))
137import GHC.IO (IO(..))
138import Streamly.Time.Clock (Clock(..), getTime)
139import Streamly.Time.Units
140 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
141 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
142import System.IO (hPutStrLn, stderr)
144import qualified Data.Heap as H
145import qualified Data.Set as S
147newtype Count = Count Int64
159------------------------------------------------------------------------------
160-- Parent child thread communication type
161------------------------------------------------------------------------------
163data ThreadAbort = ThreadAbort deriving Show
165instance Exception ThreadAbort
167-- | Events that a child thread may send to a parent thread.
170 | ChildStop ThreadId (Maybe SomeException)
172-- | Sorting out-of-turn outputs in a heap for Ahead style streams
173data AheadHeapEntry (t :: (* -> *) -> * -> *) m a =
176 | AheadEntryStream (t m a)
178------------------------------------------------------------------------------
179-- State threaded around the monad for thread management
180------------------------------------------------------------------------------
182-- | Identify the type of the SVar. Two computations using the same style can
183-- be scheduled on the same SVar.
185 AsyncVar -- depth first concurrent
186 | WAsyncVar -- breadth first concurrent
187 | ParallelVar -- all parallel
188 | AheadVar -- Concurrent look ahead
191-- | An SVar or a Stream Var is a conduit to the output from multiple streams
192-- running concurrently and asynchronously. An SVar can be thought of as an
193-- asynchronous IO handle. We can write any number of streams to an SVar in a
194-- non-blocking manner and then read them back at any time at any pace. The
195-- SVar would run the streams asynchronously and accumulate results. An SVar
196-- may not really execute the stream completely and accumulate all the results.
197-- However, it ensures that the reader can read the results at whatever paces
198-- it wants to read. The SVar monitors and adapts to the consumer's pace.
200-- An SVar is a mini scheduler, it has an associated workLoop that holds the
201-- stream tasks to be picked and run by a pool of worker threads. It has an
202-- associated output queue where the output stream elements are placed by the
203-- worker threads. A outputDoorBell is used by the worker threads to intimate the
204-- consumer thread about availability of new results in the output queue. More
205-- workers are added to the SVar by 'fromStreamVar' on demand if the output
206-- produced is not keeping pace with the consumer. On bounded SVars, workers
207-- block on the output queue to provide throttling of the producer when the
208-- consumer is not pulling fast enough. The number of workers may even get
209-- reduced depending on the consuming pace.
211-- New work is enqueued either at the time of creation of the SVar or as a
212-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
213-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
215-- We measure the individual worker latencies to estimate the number of workers
216-- needed or the amount of time we have to sleep between dispatches to achieve
217-- a particular rate when controlled pace mode it used.
218data WorkerInfo = WorkerInfo
219 { workerYieldMax :: Count -- 0 means unlimited
220 -- total number of yields by the worker till now
221 , workerYieldCount :: IORef Count
222 -- yieldCount at start, timestamp
223 , workerLatencyStart :: IORef (Count, AbsTime)
227-- | Specifies the stream yield rate in yields per second (@Hertz@).
228-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
229-- allow only as many yields as we have accumulated as per 'rateGoal' since the
230-- start of time. If the consumer or the producer is slower or faster, the
231-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
232-- between the two by increasing or decreasing the pull rate from the producer.
233-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
234-- much as 'rateBuffer'.
236-- 'rateLow' puts a bound on how low the instantaneous rate can go when
237-- recovering the rate gap. In other words, it determines the maximum yield
238-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
239-- rate can go when recovering the rate gap. In other words, it determines the
240-- minimum yield latency. We reduce the latency by increasing concurrency,
241-- therefore we can say that it puts an upper bound on concurrency.
243-- If the 'rateGoal' is 0 or negative the stream never yields a value.
244-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
248 { rateLow :: Double -- ^ The lower rate limit
249 , rateGoal :: Double -- ^ The target rate we want to achieve
250 , rateHigh :: Double -- ^ The upper rate limit
251 , rateBuffer :: Int -- ^ Maximum slack from the goal
254data LatencyRange = LatencyRange
255 { minLatency :: NanoSecond64
256 , maxLatency :: NanoSecond64
260data YieldRateInfo = YieldRateInfo
261 { svarLatencyTarget :: NanoSecond64
262 , svarLatencyRange :: LatencyRange
263 , svarRateBuffer :: Int
265 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
266 -- read by the worker threads
267 , svarGainedLostYields :: IORef Count
269 -- Actual latency/througput as seen from the consumer side, we count the
270 -- yields and the time it took to generates those yields. This is used to
271 -- increase or decrease the number of workers needed to achieve the desired
272 -- rate. The idle time of workers is adjusted in this, so that we only
273 -- account for the rate when the consumer actually demands data.
274 -- XXX interval latency is enough, we can move this under diagnostics build
275 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
276 -- read by the worker threads
277 , svarAllTimeLatency :: IORef (Count, AbsTime)
279 -- XXX Worker latency specified by the user to be used before the first
280 -- actual measurement arrives. Not yet implemented
281 , workerBootstrapLatency :: Maybe NanoSecond64
283 -- After how many yields the worker should update the latency information.
284 -- If the latency is high, this count is kept lower and vice-versa. XXX If
285 -- the latency suddenly becomes too high this count may remain too high for
286 -- long time, in such cases the consumer can change it.
287 -- 0 means no latency computation
288 -- XXX this is derivable from workerMeasuredLatency, can be removed.
289 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
290 -- read by the worker threads
291 , workerPollingInterval :: IORef Count
293 -- This is in progress latency stats maintained by the workers which we
294 -- empty into workerCollectedLatency stats at certain intervals - whenever
295 -- we process the stream elements yielded in this period. The first count
296 -- is all yields, the second count is only those yields for which the
297 -- latency was measured to be non-zero (note that if the timer resolution
298 -- is low the measured latency may be zero e.g. on JS platform).
299 -- [LOCKING] Locked access. Modified by the consumer thread as well as
300 -- worker threads. Workers modify it periodically based on
301 -- workerPollingInterval and not on every yield to reduce the locking
303 -- (allYieldCount, yieldCount, timeTaken)
304 , workerPendingLatency :: IORef (Count, Count, NanoSecond64)
306 -- This is the second level stat which is an accmulation from
307 -- workerPendingLatency stats. We keep accumulating latencies in this
308 -- bucket until we have stats for a sufficient period and then we reset it
309 -- to start collecting for the next period and retain the computed average
310 -- latency for the last period in workerMeasuredLatency.
311 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
312 -- read by the worker threads
313 -- (allYieldCount, yieldCount, timeTaken)
314 , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)
316 -- Latency as measured by workers, aggregated for the last period.
317 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
318 -- read by the worker threads
319 , workerMeasuredLatency :: IORef NanoSecond64
322data SVarStats = SVarStats {
323 totalDispatches :: IORef Int
324 , maxWorkers :: IORef Int
325 , maxOutQSize :: IORef Int
326 , maxHeapSize :: IORef Int
327 , maxWorkQSize :: IORef Int
328 , avgWorkerLatency :: IORef (Count, NanoSecond64)
329 , minWorkerLatency :: IORef NanoSecond64
330 , maxWorkerLatency :: IORef NanoSecond64
331 , svarStopTime :: IORef (Maybe AbsTime)
334-- This is essentially a 'Maybe Word' type
335data Limit = Unlimited | Limited Word deriving Show
337instance Eq Limit where
338 Unlimited == Unlimited = True
339 Unlimited == Limited _ = False
340 Limited _ == Unlimited = False
341 Limited x == Limited y = x == y
343instance Ord Limit where
344 Unlimited <= Unlimited = True
345 Unlimited <= Limited _ = False
346 Limited _ <= Unlimited = True
347 Limited x <= Limited y = x <= y
349-- When to stop the composed stream.
351 StopNone -- stops only when all streams are finished
352 | StopAny -- stop when any stream finishes
353 | StopBy -- stop when a specific stream finishes
356-- | Buffering policy for persistent push workers (in ParallelT). In a pull
357-- style SVar (in AsyncT, AheadT etc.), the consumer side dispatches workers on
358-- demand, workers terminate if the buffer is full or if the consumer is not
359-- cosuming fast enough. In a push style SVar, a worker is dispatched only
360-- once, workers are persistent and keep pushing work to the consumer via a
361-- bounded buffer. If the buffer becomes full the worker either blocks, or it
362-- can drop an item from the buffer to make space.
364-- Pull style SVars are useful in lazy stream evaluation whereas push style
365-- SVars are useful in strict left Folds.
367-- XXX Maybe we can separate the implementation in two different types instead
368-- of using a common SVar type.
370data PushBufferPolicy =
371 PushBufferDropNew -- drop the latest element and continue
372 | PushBufferDropOld -- drop the oldest element and continue
373 | PushBufferBlock -- block the thread until space
376-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have
377-- references to the original SVar stored in several functions which will keep
378-- pointing to the original data and the new updates won't reflect there.
379-- Any updateable parts must be kept in mutable references (IORef).
381data SVar t m a = SVar
384 svarStyle :: SVarStyle
385 , svarMrun :: RunInIO m
386 , svarStopStyle :: SVarStopStyle
387 , svarStopBy :: IORef ThreadId
389 -- Shared output queue (events, length)
390 -- XXX For better efficiency we can try a preallocated array type (perhaps
391 -- something like a vector) that allows an O(1) append. That way we will
392 -- avoid constructing and reversing the list. Possibly we can also avoid
393 -- the GC copying overhead. When the size increases we should be able to
394 -- allocate the array in chunks.
396 -- [LOCKING] Frequent locked access. This is updated by workers on each
397 -- yield and once in a while read by the consumer thread. This could have
398 -- big locking overhead if the number of workers is high.
400 -- XXX We can use a per-CPU data structure to reduce the locking overhead.
401 -- However, a per-cpu structure cannot guarantee the exact sequence in
402 -- which the elements were added, though that may not be important.
403 , outputQueue :: IORef ([ChildEvent a], Int)
405 -- [LOCKING] Infrequent MVar. Used when the outputQ transitions from empty
406 -- to non-empty, or a work item is queued by a worker to the work queue and
407 -- needDoorBell is set by the consumer.
408 , outputDoorBell :: MVar () -- signal the consumer about output
409 , readOutputQ :: m [ChildEvent a]
410 , postProcess :: m Bool
412 -- Combined/aggregate parameters
413 -- This is truncated to maxBufferLimit if set to more than that. Otherwise
414 -- potentially each worker may yield one value to the buffer in the worst
415 -- case exceeding the requested buffer size.
416 , maxWorkerLimit :: Limit
417 , maxBufferLimit :: Limit
418 -- These two are valid and used only when maxBufferLimit is Limited.
419 , pushBufferSpace :: IORef Count
420 , pushBufferPolicy :: PushBufferPolicy
421 -- [LOCKING] The consumer puts this MVar after emptying the buffer, workers
422 -- block on it when the buffer becomes full. No overhead unless the buffer
424 , pushBufferMVar :: MVar ()
426 -- [LOCKING] Read only access by consumer when dispatching a worker.
427 -- Decremented by workers when picking work and undo decrement if the
428 -- worker does not yield a value.
429 , remainingWork :: Maybe (IORef Count)
430 , yieldRateInfo :: Maybe YieldRateInfo
432 -- Used only by bounded SVar types
433 , enqueue :: t m a -> IO ()
434 , isWorkDone :: IO Bool
435 , isQueueDone :: IO Bool
436 , needDoorBell :: IORef Bool
437 , workLoop :: Maybe WorkerInfo -> m ()
439 -- Shared, thread tracking
440 -- [LOCKING] Updated unlocked only by consumer thread in case of
441 -- Async/Ahead style SVars. Updated locked by worker threads in case of
443 , workerThreads :: IORef (Set ThreadId)
444 -- [LOCKING] Updated locked by consumer thread when dispatching a worker
445 -- and by the worker threads when the thread stops. This is read unsafely
446 -- at several places where we want to rely on an approximate value.
447 , workerCount :: IORef Int
448 , accountThread :: ThreadId -> m ()
449 , workerStopMVar :: MVar ()
451 , svarStats :: SVarStats
452 -- to track garbage collection of SVar
453 , svarRef :: Maybe (IORef ())
455 -- Only for diagnostics
456 , svarInspectMode :: Bool
457 , svarCreator :: ThreadId
458 , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))
460 -- Shared work queue (stream, seqNo)
461 , aheadWorkQueue :: IORef ([t m a], Int)
464-------------------------------------------------------------------------------
465-- State for concurrency control
466-------------------------------------------------------------------------------
468-- XXX we can put the resettable fields in a oneShotConfig field and others in
469-- a persistentConfig field. That way reset would be fast and scalable
470-- irrespective of the number of fields.
472-- XXX make all these Limited types and use phantom types to distinguish them
473data State t m a = State
474 { -- one shot configuration, automatically reset for each API call
475 streamVar :: Maybe (SVar t m a)
476 , _yieldLimit :: Maybe Count
478 -- persistent configuration, state that remains valid until changed by
479 -- an explicit setting via a combinator.
480 , _threadsHigh :: Limit
481 , _bufferHigh :: Limit
482 -- XXX these two can be collapsed into a single type
483 , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency
484 , _maxStreamRate :: Maybe Rate
485 , _inspectMode :: Bool
488-------------------------------------------------------------------------------
489-- State defaults and reset
490-------------------------------------------------------------------------------
492-- A magical value for the buffer size arrived at by running the smallest
493-- possible task and measuring the optimal value of the buffer for that. This
494-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
496magicMaxBuffer :: Word
499defaultMaxThreads, defaultMaxBuffer :: Limit
500defaultMaxThreads = Limited magicMaxBuffer
501defaultMaxBuffer = Limited magicMaxBuffer
503-- The fields prefixed by an _ are not to be accessed or updated directly but
504-- via smart accessor APIs.
505defState :: State t m a
507 { streamVar = Nothing
508 , _yieldLimit = Nothing
509 , _threadsHigh = defaultMaxThreads
510 , _bufferHigh = defaultMaxBuffer
511 , _maxStreamRate = Nothing
512 , _streamLatency = Nothing
513 , _inspectMode = False
516-- XXX if perf gets affected we can have all the Nothing params in a single
517-- structure so that we reset is fast. We can also use rewrite rules such that
518-- reset occurs only in concurrent streams to reduce the impact on serial
520-- We can optimize this so that we clear it only if it is a Just value, it
521-- results in slightly better perf for zip/zipM but the performance of scan
522-- worsens a lot, it does not fuse.
524-- XXX This has a side effect of clearing the SVar and yieldLimit, therefore it
525-- should not be used to convert from the same type to the same type, unless
526-- you want to clear the SVar. For clearing the SVar you should be using the
527-- appropriate unStream functions instead.
529-- | Adapt the stream state from one type to another.
530adaptState :: State t m a -> State t m b
532 { streamVar = Nothing
533 , _yieldLimit = Nothing
536-------------------------------------------------------------------------------
537-- Smart get/set routines for State
538-------------------------------------------------------------------------------
540-- Use get/set routines instead of directly accessing the State fields
541setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
542setYieldLimit lim st =
549 else Just (fromIntegral n)
552getYieldLimit :: State t m a -> Maybe Count
553getYieldLimit = _yieldLimit
555setMaxThreads :: Int -> State t m a -> State t m a
561 then defaultMaxThreads
562 else Limited (fromIntegral n)
565getMaxThreads :: State t m a -> Limit
566getMaxThreads = _threadsHigh
568setMaxBuffer :: Int -> State t m a -> State t m a
574 then defaultMaxBuffer
575 else Limited (fromIntegral n)
578getMaxBuffer :: State t m a -> Limit
579getMaxBuffer = _bufferHigh
581setStreamRate :: Maybe Rate -> State t m a -> State t m a
582setStreamRate r st = st { _maxStreamRate = r }
584getStreamRate :: State t m a -> Maybe Rate
585getStreamRate = _maxStreamRate
587setStreamLatency :: Int -> State t m a -> State t m a
588setStreamLatency n st =
589 st { _streamLatency =
592 else Just (fromIntegral n)
595getStreamLatency :: State t m a -> Maybe NanoSecond64
596getStreamLatency = _streamLatency
598setInspectMode :: State t m a -> State t m a
599setInspectMode st = st { _inspectMode = True }
601getInspectMode :: State t m a -> Bool
602getInspectMode = _inspectMode
604-------------------------------------------------------------------------------
606-------------------------------------------------------------------------------
608cleanupSVar :: SVar t m a -> IO ()
610 workers <- readIORef (workerThreads sv)
611 Prelude.mapM_ (`throwTo` ThreadAbort)
614cleanupSVarFromWorker :: SVar t m a -> IO ()
615cleanupSVarFromWorker sv = do
616 workers <- readIORef (workerThreads sv)
618 mapM_ (`throwTo` ThreadAbort)
619 (S.toList workers \\ [self])
621-------------------------------------------------------------------------------
622-- Worker latency data collection
623-------------------------------------------------------------------------------
625-- Every once in a while workers update the latencies and check the yield rate.
626-- They return if we are above the expected yield rate. If we check too often
627-- it may impact performance, if we check less often we may have a stale
628-- picture. We update every minThreadDelay but we translate that into a yield
629-- count based on latency so that the checking overhead is little.
631-- XXX use a generation count to indicate that the value is updated. If the
632-- value is updated an existing worker must check it again on the next yield.
633-- Otherwise it is possible that we may keep updating it and because of the mod
634-- worker keeps skipping it.
635updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
636updateWorkerPollingInterval yinfo latency = do
637 let periodRef = workerPollingInterval yinfo
638 cnt = max 1 $ minThreadDelay `div` latency
639 period = min cnt (fromIntegral magicMaxBuffer)
641 writeIORef periodRef (fromIntegral period)
643{-# INLINE recordMinMaxLatency #-}
644recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()
645recordMinMaxLatency sv new = do
646 let ss = svarStats sv
647 minLat <- readIORef (minWorkerLatency ss)
648 when (new < minLat || minLat == 0) $
649 writeIORef (minWorkerLatency ss) new
651 maxLat <- readIORef (maxWorkerLatency ss)
652 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
654recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()
655recordAvgLatency sv (count, time) = do
656 let ss = svarStats sv
657 modifyIORef (avgWorkerLatency ss) $
658 \(cnt, t) -> (cnt + count, t + time)
660-- Pour the pending latency stats into a collection bucket
661{-# INLINE collectWorkerPendingLatency #-}
662collectWorkerPendingLatency
663 :: IORef (Count, Count, NanoSecond64)
664 -> IORef (Count, Count, NanoSecond64)
665 -> IO (Count, Maybe (Count, NanoSecond64))
666collectWorkerPendingLatency cur col = do
667 (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
669 (fcnt, cnt, t) <- readIORef col
670 let totalCount = fcnt + fcount
671 latCount = cnt + count
673 writeIORef col (totalCount, latCount, latTime)
675 assert (latCount == 0 || latTime /= 0) (return ())
677 if latCount > 0 && latTime > 0
678 then Just $ (latCount, latTime)
680 return (totalCount, latPair)
682{-# INLINE shouldUseCollectedBatch #-}
683shouldUseCollectedBatch
689shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
690 let r = fromIntegral newLat / fromIntegral prevLat :: Double
691 in (collectedYields > fromIntegral magicMaxBuffer)
692 || (collectedTime > minThreadDelay)
693 || (prevLat > 0 && (r > 2 || r < 0.5))
696-- Returns a triple, (1) yield count since last collection, (2) the base time
697-- when we started counting, (3) average latency in the last measurement
698-- period. The former two are used for accurate measurement of the going rate
699-- whereas the average is used for future estimates e.g. how many workers
700-- should be maintained to maintain the rate.
701-- CAUTION! keep it in sync with getWorkerLatency
702collectLatency :: SVar t m a
705 -> IO (Count, AbsTime, NanoSecond64)
706collectLatency sv yinfo drain = do
707 let cur = workerPendingLatency yinfo
708 col = workerCollectedLatency yinfo
709 longTerm = svarAllTimeLatency yinfo
710 measured = workerMeasuredLatency yinfo
712 (newCount, newLatPair) <- collectWorkerPendingLatency cur col
713 (lcount, ltime) <- readIORef longTerm
714 prevLat <- readIORef measured
716 let newLcount = lcount + newCount
717 retWith lat = return (newLcount, ltime, lat)
720 Nothing -> retWith prevLat
721 Just (count, time) -> do
722 let newLat = time `div` (fromIntegral count)
723 when (svarInspectMode sv) $ recordMinMaxLatency sv newLat
724 -- When we have collected a significant sized batch we compute the
725 -- new latency using that batch and return the new latency,
726 -- otherwise we return the previous latency derived from the
728 if shouldUseCollectedBatch newCount time newLat prevLat || drain
730 -- XXX make this NOINLINE?
731 updateWorkerPollingInterval yinfo (max newLat prevLat)
732 when (svarInspectMode sv) $ recordAvgLatency sv (count, time)
733 writeIORef col (0, 0, 0)
734 writeIORef measured ((prevLat + newLat) `div` 2)
735 modifyIORef longTerm $ \(_, t) -> (newLcount, t)
739-------------------------------------------------------------------------------
740-- Dumping the SVar for debug/diag
741-------------------------------------------------------------------------------
743dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
744dumpSVarStats sv ss style = do
745 case yieldRateInfo sv of
748 _ <- liftIO $ collectLatency sv yinfo True
751 dispatches <- readIORef $ totalDispatches ss
752 maxWrk <- readIORef $ maxWorkers ss
753 maxOq <- readIORef $ maxOutQSize ss
754 maxHp <- readIORef $ maxHeapSize ss
755 minLat <- readIORef $ minWorkerLatency ss
756 maxLat <- readIORef $ maxWorkerLatency ss
757 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
758 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
759 Nothing -> return (0, 0, 0)
761 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
764 t <- readIORef (svarStopTime ss)
765 gl <- readIORef (svarGainedLostYields yinfo)
768 now <- getTime Monotonic
769 let interval = diffAbsTime64 now startTime
770 return (cnt, gl, interval `div` fromIntegral cnt)
772 let interval = diffAbsTime64 stopTime startTime
773 return (cnt, gl, interval `div` fromIntegral cnt)
774 else return (0, 0, 0)
777 [ "total dispatches = " <> show dispatches
778 , "max workers = " <> show maxWrk
779 , "max outQSize = " <> show maxOq
780 <> (if style == AheadVar
781 then "\nheap max size = " <> show maxHp
784 then "\nmin worker latency = " <> showNanoSecond64 minLat
787 then "\nmax worker latency = " <> showNanoSecond64 maxLat
790 then let lat = avgTime `div` fromIntegral avgCnt
791 in "\navg worker latency = " <> showNanoSecond64 lat
794 then "\nSVar latency = " <> showRelTime64 svarLat
797 then "\nSVar yield count = " <> show svarCnt
799 <> (if svarGainLossCnt > 0
800 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
804{-# NOINLINE dumpSVar #-}
805dumpSVar :: SVar t m a -> IO String
807 (oqList, oqLen) <- readIORef $ outputQueue sv
808 db <- tryReadMVar $ outputDoorBell sv
810 if svarStyle sv == AheadVar
812 (oheap, oheapSeq) <- readIORef $ outputHeap sv
813 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
815 [ "heap length = " <> show (H.size oheap)
816 , "heap seqeunce = " <> show oheapSeq
817 , "work queue length = " <> show (length wq)
818 , "work queue sequence = " <> show wqSeq
822 let style = svarStyle sv
824 if style /= ParallelVar
825 then readIORef $ needDoorBell sv
827 rthread <- readIORef $ workerThreads sv
828 workers <- readIORef $ workerCount sv
829 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
833 "Creator tid = " <> show (svarCreator sv),
834 "style = " <> show (svarStyle sv)
835 , "---------CURRENT STATE-----------"
836 , "outputQueue length computed = " <> show (length oqList)
837 , "outputQueue length maintained = " <> show oqLen
838 -- XXX print the types of events in the outputQueue, first 5
839 , "outputDoorBell = " <> show db
843 [ "needDoorBell = " <> show waiting
844 , "running threads = " <> show rthread
845 -- XXX print the status of first 5 threads
846 , "running thread count = " <> show workers
848 <> "---------STATS-----------\n"
851-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
852-- can keep it on in production to debug problems quickly if and when they
853-- happen, but it may result in unexpected output when threads are left hanging
854-- until they are GCed because the consumer went away.
856{-# NOINLINE mvarExcHandler #-}
857mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
858mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
859 svInfo <- dumpSVar sv
860 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
863{-# NOINLINE stmExcHandler #-}
864stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
865stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
866 svInfo <- dumpSVar sv
867 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
870withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
871withDiagMVar sv label action =
872 if svarInspectMode sv
874 action `catches` [ Handler (mvarExcHandler sv label)
875 , Handler (stmExcHandler sv label)
879------------------------------------------------------------------------------
881------------------------------------------------------------------------------
883-- | A monad that can perform concurrent or parallel IO operations. Streams
884-- that can be composed concurrently require the underlying monad to be
888type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
890-- When we run computations concurrently, we completely isolate the state of
891-- the concurrent computations from the parent computation. The invariant is
892-- that we should never be running two concurrent computations in the same
893-- thread without using the runInIO function. Also, we should never be running
894-- a concurrent computation in the parent thread, otherwise it may affect the
895-- state of the parent which is against the defined semantics of concurrent
897newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }
899captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
900captureMonadState = control $ \run -> run (return $ RunInIO run)
902-- Stolen from the async package. The perf improvement is modest, 2% on a
903-- thread heavy benchmark (parallel composition using noop computations).
904-- A version of forkIO that does not include the outer exception
905-- handler: saves a bit of time when we will be installing our own
907{-# INLINE rawForkIO #-}
908rawForkIO :: IO () -> IO ThreadId
909rawForkIO action = IO $ \ s ->
910 case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)
913doFork :: MonadBaseControl IO m
916 -> (SomeException -> IO ())
918doFork action (RunInIO mrun) exHandler =
920 mask $ \restore -> do
921 tid <- rawForkIO $ catch (restore $ void $ mrun action)
925------------------------------------------------------------------------------
926-- Collecting results from child workers in a streamed fashion
927------------------------------------------------------------------------------
929-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
930-- faster, along with the fields in sv required by send?
931-- XXX make it noinline
933-- XXX we may want to employ an increment and decrement in batches when the
934-- througput is high or when the cost of synchronization is high. For example
935-- if the application is distributed then inc/dec of a shared variable may be
938-- A worker decrements the yield limit before it executes an action. However,
939-- the action may not result in an element being yielded, in that case we have
940-- to increment the yield limit.
942-- Note that we need it to be an Int type so that we have the ability to undo a
943-- decrement that takes it below zero.
944{-# INLINE decrementYieldLimit #-}
945decrementYieldLimit :: SVar t m a -> IO Bool
946decrementYieldLimit sv =
947 case remainingWork sv of
948 Nothing -> return True
950 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
953{-# INLINE incrementYieldLimit #-}
954incrementYieldLimit :: SVar t m a -> IO ()
955incrementYieldLimit sv =
956 case remainingWork sv of
958 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
960-- XXX exception safety of all atomic/MVar operations
962-- TBD Each worker can have their own queue and the consumer can empty one
963-- queue at a time, that way contention can be reduced.
965-- XXX Only yields should be counted in the buffer limit and not the Stop
968{-# INLINE decrementBufferLimit #-}
969decrementBufferLimit :: SVar t m a -> IO ()
970decrementBufferLimit sv =
971 case maxBufferLimit sv of
972 Unlimited -> return ()
974 let ref = pushBufferSpace sv
975 old <- atomicModifyIORefCAS ref $ \x ->
976 (if x >= 1 then x - 1 else x, x)
978 case pushBufferPolicy sv of
979 PushBufferBlock -> blockAndRetry
980 PushBufferDropNew -> do
981 -- We just drop one item and proceed. It is possible
982 -- that by the time we drop the item the consumer
983 -- thread might have run and created space in the
984 -- buffer, but we do not care about that condition.
985 -- This is not pedantically correct but it should be
987 -- XXX we may want to drop only if n == maxBuf
988 -- otherwise we must have space in the buffer and a
989 -- decrement should be possible.
990 block <- atomicModifyIORefCAS (outputQueue sv) $
994 _ : xs -> ((xs, n - 1), False)
995 when block blockAndRetry
996 -- XXX need a dequeue or ring buffer for this
997 PushBufferDropOld -> undefined
1002 let ref = pushBufferSpace sv
1003 liftIO $ takeMVar (pushBufferMVar sv)
1004 old <- atomicModifyIORefCAS ref $ \x ->
1005 (if x >= 1 then x - 1 else x, x)
1006 -- When multiple threads sleep on takeMVar, the first thread would
1007 -- wakeup due to a putMVar by the consumer, but the rest of the threads
1008 -- would have to put back the MVar after taking it and decrementing the
1009 -- buffer count, otherwise all other threads will remain asleep.
1011 then void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
1012 -- We do not put the MVar back in this case, instead we
1013 -- wait for the consumer to put it.
1016{-# INLINE incrementBufferLimit #-}
1017incrementBufferLimit :: SVar t m a -> IO ()
1018incrementBufferLimit sv =
1019 case maxBufferLimit sv of
1020 Unlimited -> return ()
1022 atomicModifyIORefCAS_ (pushBufferSpace sv) (+ 1)
1024 void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
1026{-# INLINE resetBufferLimit #-}
1027resetBufferLimit :: SVar t m a -> IO ()
1028resetBufferLimit sv =
1029 case maxBufferLimit sv of
1030 Unlimited -> return ()
1031 Limited n -> atomicModifyIORefCAS_ (pushBufferSpace sv)
1032 (const (fromIntegral n))
1034-- | This function is used by the producer threads to queue output for the
1035-- consumer thread to consume. Returns whether the queue has more space.
1036send :: SVar t m a -> ChildEvent a -> IO Int
1038 -- XXX can the access to outputQueue and maxBufferLimit be made faster
1040 oldlen <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
1041 ((msg : es, n + 1), n)
1042 when (oldlen <= 0) $ do
1043 -- The wake up must happen only after the store has finished otherwise
1044 -- we can have lost wakeup problems.
1046 -- Since multiple workers can try this at the same time, it is possible
1047 -- that we may put a spurious MVar after the consumer has already seen
1048 -- the output. But that's harmless, at worst it may cause the consumer
1049 -- to read the queue again and find it empty.
1050 -- The important point is that the consumer is guaranteed to receive a
1051 -- doorbell if something was added to the queue after it empties it.
1052 void $ tryPutMVar (outputDoorBell sv) ()
1055workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
1056workerCollectLatency winfo = do
1057 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
1058 cnt1 <- readIORef (workerYieldCount winfo)
1059 let cnt = cnt1 - cnt0
1063 t1 <- getTime Monotonic
1064 let period = fromRelTime64 $ diffAbsTime64 t1 t0
1065 writeIORef (workerLatencyStart winfo) (cnt1, t1)
1066 return $ Just (cnt, period)
1069-- XXX There are a number of gotchas in measuring latencies.
1070-- 1) We measure latencies only when a worker yields a value
1071-- 2) It is possible that a stream calls the stop continuation, in which case
1072-- the worker would not yield a value and we would not account that worker in
1073-- latencies. Even though this case should ideally be accounted we do not
1074-- account it because we cannot or do not distinguish it from the case
1076-- 3) It is possible that a worker returns without yielding anything because it
1077-- never got a chance to pick up work.
1078-- 4) If the system timer resolution is lower than the latency, the latency
1079-- computation turns out to be zero.
1081-- We can fix this if we measure the latencies by counting the work items
1082-- picked rather than based on the outputs yielded.
1083workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
1084workerUpdateLatency yinfo winfo = do
1085 r <- workerCollectLatency winfo
1087 Just (cnt, period) -> do
1088 -- NOTE: On JS platform the timer resolution could be pretty low. When
1089 -- the timer resolution is low, measurement of latencies could be
1090 -- tricky. All the worker latencies will turn out to be zero if they
1091 -- are lower than the resolution. We only take into account those
1092 -- measurements which are more than the timer resolution.
1094 let ref = workerPendingLatency yinfo
1095 (cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
1096 atomicModifyIORefCAS_ ref $
1097 \(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
1098 Nothing -> return ()
1100updateYieldCount :: WorkerInfo -> IO Count
1101updateYieldCount winfo = do
1102 cnt <- readIORef (workerYieldCount winfo)
1104 writeIORef (workerYieldCount winfo) cnt1
1107isBeyondMaxYield :: Count -> WorkerInfo -> Bool
1108isBeyondMaxYield cnt winfo =
1109 let ymax = workerYieldMax winfo
1110 in ymax /= 0 && cnt >= ymax
1112-- XXX we should do rate control periodically based on the total yields rather
1113-- than based on the worker local yields as other workers may have yielded more
1114-- and we should stop based on the aggregate yields. However, latency update
1115-- period can be based on individual worker yields.
1116{-# NOINLINE checkRatePeriodic #-}
1117checkRatePeriodic :: SVar t m a
1122checkRatePeriodic sv yinfo winfo ycnt = do
1123 i <- readIORef (workerPollingInterval yinfo)
1124 -- XXX use generation count to check if the interval has been updated
1125 if i /= 0 && (ycnt `mod` i) == 0
1127 workerUpdateLatency yinfo winfo
1128 -- XXX not required for parallel streams
1129 isBeyondMaxRate sv yinfo
1132-- CAUTION! this also updates the yield count and therefore should be called
1133-- only when we are actually yielding an element.
1134{-# NOINLINE workerRateControl #-}
1135workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
1136workerRateControl sv yinfo winfo = do
1137 cnt <- updateYieldCount winfo
1138 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
1139 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
1141-- XXX we should do rate control here but not latency update in case of ahead
1142-- streams. latency update must be done when we yield directly to outputQueue
1143-- or when we yield to heap.
1145-- returns whether the worker should continue (True) or stop (False).
1146{-# INLINE sendYield #-}
1147sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
1148sendYield sv mwinfo msg = do
1149 oldlen <- send sv msg
1150 let limit = maxBufferLimit sv
1151 bufferSpaceOk <- case limit of
1152 Unlimited -> return True
1154 active <- readIORef (workerCount sv)
1155 return $ (oldlen + 1) < (fromIntegral lim - active)
1159 case yieldRateInfo sv of
1160 Nothing -> return True
1161 Just yinfo -> workerRateControl sv yinfo winfo
1162 Nothing -> return True
1163 return $ bufferSpaceOk && rateLimitOk
1165{-# INLINE workerStopUpdate #-}
1166workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
1167workerStopUpdate winfo info = do
1168 i <- readIORef (workerPollingInterval info)
1169 when (i /= 0) $ workerUpdateLatency info winfo
1171{-# INLINABLE sendStop #-}
1172sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
1173sendStop sv mwinfo = do
1174 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
1175 case (mwinfo, yieldRateInfo sv) of
1176 (Just winfo, Just info) ->
1177 workerStopUpdate winfo info
1180 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
1182-------------------------------------------------------------------------------
1184-------------------------------------------------------------------------------
1186{-# INLINE ringDoorBell #-}
1187ringDoorBell :: SVar t m a -> IO ()
1190 w <- readIORef $ needDoorBell sv
1192 -- Note: the sequence of operations is important for correctness here.
1193 -- We need to set the flag to false strictly before sending the
1194 -- outputDoorBell, otherwise the outputDoorBell may get processed too
1195 -- early and then we may set the flag to False to later making the
1196 -- consumer lose the flag, even without receiving a outputDoorBell.
1197 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1198 void $ tryPutMVar (outputDoorBell sv) ()
1200-------------------------------------------------------------------------------
1202-------------------------------------------------------------------------------
1204-- Note: For purely right associated expressions this queue should have at most
1205-- one element. It grows to more than one when we have left associcated
1206-- expressions. Large left associated compositions can grow this to a
1208{-# INLINE enqueueLIFO #-}
1209enqueueLIFO :: SVar t m a -> IORef [t m a] -> t m a -> IO ()
1210enqueueLIFO sv q m = do
1211 atomicModifyIORefCAS_ q $ \ms -> m : ms
1214-------------------------------------------------------------------------------
1216-------------------------------------------------------------------------------
1218-- XXX we can use the Ahead style sequence/heap mechanism to make the best
1219-- effort to always try to finish the streams on the left side of an expression
1220-- first as long as possible.
1222{-# INLINE enqueueFIFO #-}
1223enqueueFIFO :: SVar t m a -> LinkedQueue (t m a) -> t m a -> IO ()
1224enqueueFIFO sv q m = do
1228-------------------------------------------------------------------------------
1230-------------------------------------------------------------------------------
1232-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
1233-- but always serve them in the same order as they appear in the stream. To
1234-- implement lookahead streams efficiently we assign a sequence number to each
1235-- task when the task is picked up for execution. When the task finishes, the
1236-- output is tagged with the same sequence number and we rearrange the outputs
1237-- in sequence based on that number.
1239-- To explain the mechanism imagine that the current task at the head of the
1240-- stream has a "token" to yield to the outputQueue. The ownership of the token
1241-- is determined by the current sequence number is maintained in outputHeap.
1242-- Sequence number is assigned when a task is queued. When a thread dequeues a
1243-- task it picks up the sequence number as well and when the output is ready it
1244-- uses the sequence number to queue the output to the outputQueue.
1246-- The thread with current sequence number sends the output directly to the
1247-- outputQueue. Other threads push the output to the outputHeap. When the task
1248-- being queued on the heap is a stream of many elements we evaluate only the
1249-- first element and keep the rest of the unevaluated computation in the heap.
1250-- When such a task gets the "token" for outputQueue it evaluates and directly
1251-- yields all the elements to the outputQueue without checking for the
1254-- Note that no two outputs in the heap can have the same sequence numbers and
1255-- therefore we do not need a stable heap. We have also separated the buffer
1256-- for the current task (outputQueue) and the pending tasks (outputHeap) so
1257-- that the pending tasks cannot interfere with the current task. Note that for
1258-- a single task just the outputQueue is enough and for the case of many
1259-- threads just a heap is good enough. However we balance between these two
1260-- cases, so that both are efficient.
1262-- For bigger streams it may make sense to have separate buffers for each
1263-- stream. However, for singleton streams this may become inefficient. However,
1264-- if we do not have separate buffers, then the streams that come later in
1265-- sequence may hog the buffer, hindering the streams that are ahead. For this
1266-- reason we have a single element buffer limitation for the streams being
1267-- executed in advance.
1269-- This scheme works pretty efficiently with less than 40% extra overhead
1270-- compared to the Async streams where we do not have any kind of sequencing of
1271-- the outputs. It is especially devised so that we are most efficient when we
1272-- have short tasks and need just a single thread. Also when a thread yields
1273-- many items it can hold lockfree access to the outputQueue and do it
1276-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
1277-- that they do not hog the resources and hinder the progress of the threads in
1280-- XXX Left associated ahead expressions are expensive. We start a new SVar for
1281-- each left associative expression. The queue is used only for right
1282-- associated expression, we queue the right expression and execute the left.
1283-- Thererefore the queue never has more than one item in it.
1285-- XXX we can fix this. When we queue more than one item on the queue we can
1286-- mark the previously queued item as not-runnable. The not-runnable item is
1287-- not dequeued until the already running one has finished and at that time we
1288-- would also know the exact sequence number of the already queued item.
1290-- we can even run the already queued items but they will have to be sorted in
1291-- layers in the heap. We can use a list of heaps for that.
1292{-# INLINE enqueueAhead #-}
1293enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1294enqueueAhead sv q m = do
1295 atomicModifyIORefCAS_ q $ \ case
1296 ([], n) -> ([m], n + 1) -- increment sequence
1297 _ -> error "not empty"
1300-- enqueue without incrementing the sequence number
1301{-# INLINE reEnqueueAhead #-}
1302reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1303reEnqueueAhead sv q m = do
1304 atomicModifyIORefCAS_ q $ \ case
1305 ([], n) -> ([m], n) -- DO NOT increment sequence
1306 _ -> error "not empty"
1309-- Normally the thread that has the token should never go away. The token gets
1310-- handed over to another thread, but someone or the other has the token at any
1311-- point of time. But if the task that has the token finds that the outputQueue
1312-- is full, in that case it can go away without even handing over the token to
1313-- another thread. In that case it sets the nextSequence number in the heap its
1314-- own sequence number before going away. To handle this case, any task that
1315-- does not have the token tries to dequeue from the heap first before
1316-- dequeuing from the work queue. If it finds that the task at the top of the
1317-- heap is the one that owns the current sequence number then it grabs the
1318-- token and starts with that.
1320-- XXX instead of queueing just the head element and the remaining computation
1321-- on the heap, evaluate as many as we can and place them on the heap. But we
1322-- need to give higher priority to the lower sequence numbers so that lower
1323-- priority tasks do not fill up the heap making higher priority tasks block
1324-- due to full heap. Maybe we can have a weighted space for them in the heap.
1325-- The weight is inversely proportional to the sequence number.
1327-- XXX review for livelock
1329{-# INLINE queueEmptyAhead #-}
1330queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
1331queueEmptyAhead q = liftIO $ do
1332 (xs, _) <- readIORef q
1335{-# INLINE dequeueAhead #-}
1336dequeueAhead :: MonadIO m
1337 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
1338dequeueAhead q = liftIO $
1339 atomicModifyIORefCAS q $ \case
1340 ([], n) -> (([], n), Nothing)
1341 (x : [], n) -> (([], n), Just (x, n))
1342 _ -> error "more than one item on queue"
1344-------------------------------------------------------------------------------
1346-------------------------------------------------------------------------------
1348withIORef :: IORef a -> (a -> IO b) -> IO b
1349withIORef ref f = readIORef ref >>= f
1351atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
1352atomicModifyIORef_ ref f =
1353 atomicModifyIORef ref $ \x -> (f x, ())
1355data HeapDequeueResult t m a =
1358 | Ready (Entry Int (AheadHeapEntry t m a))
1360{-# INLINE dequeueFromHeap #-}
1362 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1363 -> IO (HeapDequeueResult t m a)
1364dequeueFromHeap hpVar =
1365 atomicModifyIORef hpVar $ \pair@(hp, snum) ->
1367 Nothing -> (pair, Clearing)
1371 Just (ent@(Entry seqNo _ev), hp') ->
1373 then ((hp', Nothing), Ready ent)
1374 else assert (seqNo >= n) (pair, Waiting n)
1375 Nothing -> (pair, Waiting n)
1377{-# INLINE dequeueFromHeapSeq #-}
1379 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1381 -> IO (HeapDequeueResult t m a)
1382dequeueFromHeapSeq hpVar i =
1383 atomicModifyIORef hpVar $ \(hp, snum) ->
1388 Just (ent@(Entry seqNo _ev), hp') ->
1390 then ((hp', Nothing), Ready ent)
1391 else assert (seqNo >= i) ((hp, Just i), Waiting i)
1392 Nothing -> ((hp, Just i), Waiting i)
1393 Just _ -> error "dequeueFromHeapSeq: unreachable"
1395heapIsSane :: Maybe Int -> Int -> Bool
1396heapIsSane snum seqNo =
1399 Just n -> seqNo >= n
1401{-# INLINE requeueOnHeapTop #-}
1403 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1404 -> Entry Int (AheadHeapEntry t m a)
1407requeueOnHeapTop hpVar ent seqNo =
1408 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1409 assert (heapIsSane snum seqNo) (H.insert ent hp, Just seqNo)
1411{-# INLINE updateHeapSeq #-}
1413 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1416updateHeapSeq hpVar seqNo =
1417 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1418 assert (heapIsSane snum seqNo) (hp, Just seqNo)
1420-------------------------------------------------------------------------------
1422-------------------------------------------------------------------------------
1424-- XXX To be implemented. Use a linked queue like WAsync and put back the
1425-- remaining computation at the back of the queue instead of the heap, and
1426-- increment the sequence number.
1428-------------------------------------------------------------------------------
1429-- Dispatching workers and tracking them
1430-------------------------------------------------------------------------------
1432-- Thread tracking is needed for two reasons:
1434-- 1) Killing threads on exceptions. Threads may not be left to go away by
1435-- themselves because they may run for significant times before going away or
1436-- worse they may be stuck in IO and never go away.
1438-- 2) To know when all threads are done and the stream has ended.
1440{-# NOINLINE addThread #-}
1441addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1443 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
1445-- This is cheaper than modifyThread because we do not have to send a
1446-- outputDoorBell This can make a difference when more workers are being
1448{-# INLINE delThread #-}
1449delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1451 liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
1453-- If present then delete else add. This takes care of out of order add and
1454-- delete i.e. a delete arriving before we even added a thread.
1455-- This occurs when the forked thread is done even before the 'addThread' right
1456-- after the fork gets a chance to run.
1457{-# INLINE modifyThread #-}
1458modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1459modifyThread sv tid = do
1460 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1462 then let new = S.delete tid old in (new, new)
1463 else let new = S.insert tid old in (new, old)
1464 when (null changed) $
1467 void $ tryPutMVar (outputDoorBell sv) ()
1469-- | This is safe even if we are adding more threads concurrently because if
1470-- a child thread is adding another thread then anyway 'workerThreads' will
1472{-# INLINE allThreadsDone #-}
1473allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1474allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1476{-# NOINLINE handleChildException #-}
1477handleChildException :: SVar t m a -> SomeException -> IO ()
1478handleChildException sv e = do
1480 void $ send sv (ChildStop tid (Just e))
1482{-# NOINLINE recordMaxWorkers #-}
1483recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1484recordMaxWorkers sv = liftIO $ do
1485 active <- readIORef (workerCount sv)
1486 maxWrk <- readIORef (maxWorkers $ svarStats sv)
1487 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1488 modifyIORef (totalDispatches $ svarStats sv) (+1)
1490{-# NOINLINE pushWorker #-}
1491pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1492pushWorker yieldMax sv = do
1493 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1494 when (svarInspectMode sv) $ recordMaxWorkers sv
1495 -- This allocation matters when significant number of workers are being
1496 -- sent. We allocate it only when needed.
1498 case yieldRateInfo sv of
1499 Nothing -> return Nothing
1500 Just _ -> liftIO $ do
1501 cntRef <- newIORef 0
1502 t <- getTime Monotonic
1503 lat <- newIORef (0, t)
1504 return $ Just WorkerInfo
1505 { workerYieldMax = yieldMax
1506 , workerYieldCount = cntRef
1507 , workerLatencyStart = lat
1509 doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
1512-- XXX we can push the workerCount modification in accountThread and use the
1513-- same pushWorker for Parallel case as well.
1515-- | In contrast to pushWorker which always happens only from the consumer
1516-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1517-- producer side. So we need to use a thread safe modification of
1518-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1519-- using a CAS based modification.
1520{-# INLINE pushWorkerPar #-}
1523 => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1524pushWorkerPar sv wloop =
1525 if svarInspectMode sv
1527 else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
1532 {-# NOINLINE forkWithDiag #-}
1534 -- We do not use workerCount in case of ParallelVar but still there is
1535 -- no harm in maintaining it correctly.
1536 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1538 -- This allocation matters when significant number of workers are being
1539 -- sent. We allocate it only when needed. The overhead increases by 4x.
1541 case yieldRateInfo sv of
1542 Nothing -> return Nothing
1543 Just _ -> liftIO $ do
1544 cntRef <- newIORef 0
1545 t <- getTime Monotonic
1546 lat <- newIORef (0, t)
1547 return $ Just WorkerInfo
1548 { workerYieldMax = 0
1549 , workerYieldCount = cntRef
1550 , workerLatencyStart = lat
1553 doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
1557-- True: can dispatch more
1558-- False: cannot dispatch any more
1559dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1560dispatchWorker yieldCount sv = do
1561 let workerLimit = maxWorkerLimit sv
1562 -- XXX in case of Ahead streams we should not send more than one worker
1563 -- when the work queue is done but heap is not done.
1564 done <- liftIO $ isWorkDone sv
1565 -- Note, "done" may not mean that the work is actually finished if there
1566 -- are workers active, because there may be a worker which has not yet
1567 -- queued the leftover work.
1570 qDone <- liftIO $ isQueueDone sv
1571 -- This count may not be accurate as it is decremented by the workers
1572 -- and we have no synchronization with that decrement.
1573 active <- liftIO $ readIORef $ workerCount sv
1576 -- Note that we may deadlock if the previous workers (tasks in the
1577 -- stream) wait/depend on the future workers (tasks in the stream)
1578 -- executing. In that case we should either configure the maxWorker
1579 -- count to higher or use parallel style instead of ahead or async
1581 limit <- case remainingWork sv of
1582 Nothing -> return workerLimit
1584 n <- liftIO $ readIORef ref
1585 case yieldRateInfo sv of
1586 Just _ -> return workerLimit
1590 Unlimited -> Limited (fromIntegral n)
1591 Limited lim -> Limited $ min lim (fromIntegral n)
1593 -- XXX for ahead streams shall we take the heap yields into account
1594 -- for controlling the dispatch? We should not dispatch if the heap
1595 -- has already got the limit covered.
1596 let dispatch = pushWorker yieldCount sv >> return True
1598 Unlimited -> dispatch
1599 -- Note that the use of remainingWork and workerCount is not
1600 -- atomic and the counts may even have changed between reading
1601 -- and using them here, so this is just approximate logic and
1602 -- we cannot rely on it for correctness. We may actually
1603 -- dispatch more workers than required.
1604 Limited lim | lim > fromIntegral active -> dispatch
1607 when (active <= 0) $ pushWorker 0 sv
1611-------------------------------------------------------------------------------
1612-- Dispatch workers with rate control
1613-------------------------------------------------------------------------------
1615-- | This is a magic number and it is overloaded, and used at several places to
1618-- 1. If we have to sleep to slowdown this is the minimum period that we
1619-- accumulate before we sleep. Also, workers do not stop until this much
1620-- sleep time is accumulated.
1621-- 3. Collected latencies are computed and transferred to measured latency
1622-- after a minimum of this period.
1623minThreadDelay :: NanoSecond64
1624minThreadDelay = 1000000
1626-- | Another magic number! When we have to start more workers to cover up a
1627-- number of yields that we are lagging by then we cannot start one worker for
1628-- each yield because that may be a very big number and if the latency of the
1629-- workers is low these number of yields could be very high. We assume that we
1630-- run each extra worker for at least this much time.
1631rateRecoveryTime :: NanoSecond64
1632rateRecoveryTime = 1000000
1634-- We either block, or send one worker with limited yield count or one or more
1635-- workers with unlimited yield count.
1637 = BlockWait NanoSecond64
1638 | PartialWorker Count
1639 | ManyWorkers Int Count
1642-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1652estimateWorkers workerLimit svarYields gainLossYields
1653 svarElapsed wLatency targetLat range =
1654 -- XXX we can have a maxEfficiency combinator as well which runs the
1655 -- producer at the maximal efficiency i.e. the number of workers are chosen
1656 -- such that the latency is minimum or within a range. Or we can call it
1657 -- maxWorkerLatency.
1660 -- How many workers do we need to acheive the required rate?
1662 -- When the workers are IO bound we can increase the throughput by
1663 -- increasing the number of workers as long as the IO device has enough
1664 -- capacity to process all the requests concurrently. If the IO
1665 -- bandwidth is saturated increasing the workers won't help. Also, if
1666 -- the CPU utilization in processing all these requests exceeds the CPU
1667 -- bandwidth, then increasing the number of workers won't help.
1669 -- When the workers are purely CPU bound, increasing the workers beyond
1670 -- the number of CPUs won't help.
1672 -- TODO - measure the CPU and IO requirements of the workers. Have a
1673 -- way to specify the max bandwidth of the underlying IO mechanism and
1674 -- use that to determine the max rate of workers, and also take the CPU
1675 -- bandwidth into account. We can also discover the IO bandwidth if we
1676 -- know that we are not CPU bound, then how much steady state rate are
1677 -- we able to acheive. Design tests for CPU bound and IO bound cases.
1679 -- Calculate how many yields are we ahead or behind to match the exact
1680 -- required rate. Based on that we increase or decrease the effective
1683 -- When the worker latency is lower than required latency we begin with
1684 -- a yield and then wait rather than first waiting and then yielding.
1685 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1686 effectiveYields = svarYields + gainLossYields
1687 deltaYields = fromIntegral targetYields - effectiveYields
1689 -- We recover the deficit by running at a higher/lower rate for a
1690 -- certain amount of time. To keep the effective rate in reasonable
1691 -- limits we use rateRecoveryTime, minLatency and maxLatency.
1692 in if deltaYields > 0
1694 let deltaYieldsFreq :: Double
1696 fromIntegral deltaYields /
1697 fromIntegral rateRecoveryTime
1698 yieldsFreq = 1.0 / fromIntegral targetLat
1699 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1700 requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
1701 adjustedLat = min (max requiredLat (minLatency range))
1703 in assert (adjustedLat > 0) $
1704 if wLatency <= adjustedLat
1705 then PartialWorker deltaYields
1706 else let workers = withLimit $ wLatency `div` adjustedLat
1707 limited = min workers (fromIntegral deltaYields)
1708 in ManyWorkers (fromIntegral limited) deltaYields
1710 let expectedDuration = fromIntegral effectiveYields * targetLat
1711 sleepTime = expectedDuration - svarElapsed
1712 maxSleepTime = maxLatency range - wLatency
1713 s = min sleepTime maxSleepTime
1714 in assert (sleepTime >= 0) $
1715 -- if s is less than 0 it means our maxSleepTime is less
1716 -- than the worker latency.
1717 if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
1722 Limited x -> min n (fromIntegral x)
1724-- | Get the worker latency without resetting workerPendingLatency
1725-- Returns (total yield count, base time, measured latency)
1726-- CAUTION! keep it in sync with collectLatency
1727getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
1728getWorkerLatency yinfo = do
1729 let cur = workerPendingLatency yinfo
1730 col = workerCollectedLatency yinfo
1731 longTerm = svarAllTimeLatency yinfo
1732 measured = workerMeasuredLatency yinfo
1734 (curTotalCount, curCount, curTime) <- readIORef cur
1735 (colTotalCount, colCount, colTime) <- readIORef col
1736 (lcount, ltime) <- readIORef longTerm
1737 prevLat <- readIORef measured
1739 let latCount = colCount + curCount
1740 latTime = colTime + curTime
1741 totalCount = colTotalCount + curTotalCount
1743 if latCount > 0 && latTime > 0
1744 then let lat = latTime `div` fromIntegral latCount
1745 -- XXX Give more weight to new?
1746 in (lat + prevLat) `div` 2
1748 return (lcount + totalCount, ltime, newLat)
1750isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1751isBeyondMaxRate sv yinfo = do
1752 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1753 now <- getTime Monotonic
1754 let duration = fromRelTime64 $ diffAbsTime64 now tstamp
1755 let targetLat = svarLatencyTarget yinfo
1756 gainLoss <- readIORef (svarGainedLostYields yinfo)
1757 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1758 wLatency targetLat (svarLatencyRange yinfo)
1759 cnt <- readIORef $ workerCount sv
1760 return $ case work of
1761 -- XXX set the worker's maxYields or polling interval based on yields
1762 PartialWorker _yields -> cnt > 1
1763 ManyWorkers n _ -> cnt > n
1766-- XXX in case of ahead style stream we need to take the heap size into account
1767-- because we return the workers on the basis of that which causes a condition
1768-- where we keep dispatching and they keep returning. So we must have exactly
1769-- the same logic for not dispatching and for returning.
1772-- True: can dispatch more
1773-- False: full, no more dispatches
1774dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1775dispatchWorkerPaced sv = do
1776 let yinfo = fromJust $ yieldRateInfo sv
1777 (svarYields, svarElapsed, wLatency) <- do
1778 now <- liftIO $ getTime Monotonic
1779 (yieldCount, baseTime, lat) <-
1780 liftIO $ collectLatency sv yinfo False
1781 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
1785 case workerBootstrapLatency yinfo of
1790 return (yieldCount, elapsed, latency)
1793 -- Need to measure the latency with a single worker before we can perform
1797 let workerLimit = maxWorkerLimit sv
1798 let targetLat = svarLatencyTarget yinfo
1799 let range = svarLatencyRange yinfo
1800 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1801 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
1802 wLatency targetLat range
1804 -- XXX we need to take yieldLimit into account here. If we are at the
1805 -- end of the limit as well as the time, we should not be sleeping.
1806 -- If we are not actually planning to dispatch any more workers we need
1807 -- to take that in account.
1810 assert (s >= 0) (return ())
1811 -- XXX note that when we return from here we will block waiting
1812 -- for the result from the existing worker. If that takes too
1813 -- long we won't be able to send another worker until the
1816 -- Sleep only if there are no active workers, otherwise we will
1817 -- defer the output of those. Note we cannot use workerCount
1818 -- here as it is not a reliable way to ensure there are
1819 -- definitely no active workers. When workerCount is 0 we may
1820 -- still have a Stop event waiting in the outputQueue.
1821 done <- allThreadsDone sv
1822 when done $ void $ do
1823 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
1824 liftIO $ threadDelay (fromIntegral us)
1827 PartialWorker yields -> do
1828 assert (yields > 0) (return ())
1829 updateGainedLostYields yinfo yields
1831 done <- allThreadsDone sv
1832 when done $ void $ dispatchWorker yields sv
1834 ManyWorkers netWorkers yields -> do
1835 assert (netWorkers >= 1) (return ())
1836 assert (yields >= 0) (return ())
1837 updateGainedLostYields yinfo yields
1839 let periodRef = workerPollingInterval yinfo
1840 ycnt = max 1 $ yields `div` fromIntegral netWorkers
1841 period = min ycnt (fromIntegral magicMaxBuffer)
1843 old <- liftIO $ readIORef periodRef
1844 when (period < old) $
1845 liftIO $ writeIORef periodRef period
1847 cnt <- liftIO $ readIORef $ workerCount sv
1850 let total = netWorkers - cnt
1851 batch = max 1 $ fromIntegral $
1852 minThreadDelay `div` targetLat
1853 -- XXX stagger the workers over a period?
1854 -- XXX cannot sleep, as that would mean we cannot process
1855 -- the outputs. need to try a different mechanism to
1857 -- when (total > batch) $
1858 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1859 dispatchN (min total batch)
1864 updateGainedLostYields yinfo yields = do
1865 let buf = fromIntegral $ svarRateBuffer yinfo
1866 when (yields /= 0 && abs yields > buf) $ do
1871 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1877 r <- dispatchWorker 0 sv
1879 then dispatchN (n - 1)
1882-------------------------------------------------------------------------------
1883-- Worker dispatch and wait loop
1884-------------------------------------------------------------------------------
1886sendWorkerDelayPaced :: SVar t m a -> IO ()
1887sendWorkerDelayPaced _ = return ()
1889sendWorkerDelay :: SVar t m a -> IO ()
1890sendWorkerDelay _sv =
1891 -- XXX we need a better way to handle this than hardcoded delays. The
1892 -- delays may be different for different systems.
1893 -- If there is a usecase where this is required we can create a combinator
1894 -- to set it as a config in the state.
1897 ncpu <- getNumCapabilities
1900 if (svarStyle sv == AheadVar)
1901 then threadDelay 100
1904 if (svarStyle sv == AheadVar)
1905 then threadDelay 100
1910{-# NOINLINE sendWorkerWait #-}
1913 => (SVar t m a -> IO ())
1914 -> (SVar t m a -> m Bool)
1917sendWorkerWait delay dispatch sv = do
1918 -- Note that we are guaranteed to have at least one outstanding worker when
1919 -- we enter this function. So if we sleep we are guaranteed to be woken up
1920 -- by an outputDoorBell, when the worker exits.
1923 (_, n) <- liftIO $ readIORef (outputQueue sv)
1925 -- The queue may be empty temporarily if the worker has dequeued the
1926 -- work item but has not enqueued the remaining part yet. For the same
1927 -- reason, a worker may come back if it tries to dequeue and finds the
1928 -- queue empty, even though the whole work has not finished yet.
1930 -- If we find that the queue is empty, but it may be empty
1931 -- temporarily, when we checked it. If that's the case we might
1932 -- sleep indefinitely unless the active workers produce some
1933 -- output. We may deadlock specially if the otuput from the active
1934 -- workers depends on the future workers that we may never send.
1935 -- So in case the queue was temporarily empty set a flag to inform
1936 -- the enqueue to send us a doorbell.
1938 -- Note that this is just a best effort mechanism to avoid a
1939 -- deadlock. Deadlocks may still happen if for some weird reason
1940 -- the consuming computation shares an MVar or some other resource
1941 -- with the producing computation and gets blocked on that resource
1942 -- and therefore cannot do any pushworker to add more threads to
1943 -- the producer. In such cases the programmer should use a parallel
1944 -- style so that all the producers are scheduled immediately and
1945 -- unconditionally. We can also use a separate monitor thread to
1946 -- push workers instead of pushing them from the consumer, but then
1947 -- we are no longer using pull based concurrency rate adaptation.
1949 -- XXX update this in the tutorial.
1951 -- Having pending active workers does not mean that we are guaranteed
1952 -- to be woken up if we sleep. In case of Ahead streams, there may be
1953 -- queued items in the heap even though the outputQueue is empty, and
1954 -- we may have active workers which are deadlocked on those items to be
1955 -- processed by the consumer. We should either guarantee that any
1956 -- worker, before returning, clears the heap or we send a worker to
1957 -- clear it. Normally we always send a worker if no output is seen, but
1958 -- if the thread limit is reached or we are using pacing then we may
1959 -- not send a worker. See the concurrentApplication test in the tests,
1960 -- that test case requires at least one yield from the producer to not
1961 -- deadlock, if the last workers output is stuck in the heap then this
1962 -- test fails. This problem can be extended to n threads when the
1963 -- consumer may depend on the evaluation of next n items in the
1966 -- register for the outputDoorBell before we check the queue so that if
1967 -- we sleep because the queue was empty we are guaranteed to get a
1968 -- doorbell on the next enqueue.
1970 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
1971 liftIO storeLoadBarrier
1972 canDoMore <- dispatch sv
1974 -- XXX test for the case when we miss sending a worker when the worker
1975 -- count is more than 1500.
1977 -- XXX Assert here that if the heap is not empty then there is at
1978 -- least one outstanding worker. Otherwise we could be sleeping
1982 then sendWorkerWait delay dispatch sv
1984 liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
1985 $ takeMVar (outputDoorBell sv)
1986 (_, len) <- liftIO $ readIORef (outputQueue sv)
1987 when (len <= 0) $ sendWorkerWait delay dispatch sv
1989-------------------------------------------------------------------------------
1990-- Reading from the workers' output queue/buffer
1991-------------------------------------------------------------------------------
1993{-# INLINE readOutputQRaw #-}
1994readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
1995readOutputQRaw sv = do
1996 (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
1997 when (svarInspectMode sv) $ do
1998 let ref = maxOutQSize $ svarStats sv
1999 oqLen <- readIORef ref
2000 when (len > oqLen) $ writeIORef ref len
2003readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2004readOutputQBounded sv = do
2005 (list, len) <- liftIO $ readOutputQRaw sv
2006 -- When there is no output seen we dispatch more workers to help
2007 -- out if there is work pending in the work queue.
2011 -- send a worker proactively, if needed, even before we start
2012 -- processing the output. This may degrade single processor
2013 -- perf but improves multi-processor, because of more
2021 cnt <- liftIO $ readIORef $ workerCount sv
2022 when (cnt <= 0) $ do
2023 done <- liftIO $ isWorkDone sv
2024 when (not done) (pushWorker 0 sv)
2026 {-# INLINE blockingRead #-}
2028 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
2029 liftIO (fst `fmap` readOutputQRaw sv)
2031readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2032readOutputQPaced sv = do
2033 (list, len) <- liftIO $ readOutputQRaw sv
2037 -- XXX send a worker proactively, if needed, even before we start
2038 -- processing the output.
2039 void $ dispatchWorkerPaced sv
2044 {-# INLINE blockingRead #-}
2046 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
2047 liftIO (fst `fmap` readOutputQRaw sv)
2049postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
2050postProcessBounded sv = do
2051 workersDone <- allThreadsDone sv
2052 -- There may still be work pending even if there are no workers pending
2053 -- because all the workers may return if the outputQueue becomes full. In
2054 -- that case send off a worker to kickstart the work again.
2056 -- Note that isWorkDone can only be safely checked if all workers are done.
2057 -- When some workers are in progress they may have decremented the yield
2058 -- Limit and later ending up incrementing it again. If we look at the yield
2059 -- limit in that window we may falsely say that it is 0 and therefore we
2063 r <- liftIO $ isWorkDone sv
2064 -- Note that we need to guarantee a worker, therefore we cannot just
2065 -- use dispatchWorker which may or may not send a worker.
2066 when (not r) (pushWorker 0 sv)
2067 -- XXX do we need to dispatch many here?
2068 -- void $ dispatchWorker sv
2072postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
2073postProcessPaced sv = do
2074 workersDone <- allThreadsDone sv
2075 -- XXX If during consumption we figure out we are getting delayed then we
2076 -- should trigger dispatch there as well. We should try to check on the
2077 -- workers after consuming every n item from the buffer?
2080 r <- liftIO $ isWorkDone sv
2082 void $ dispatchWorkerPaced sv
2083 -- Note that we need to guarantee a worker since the work is not
2084 -- finished, therefore we cannot just rely on dispatchWorkerPaced
2085 -- which may or may not send a worker.
2086 noWorker <- allThreadsDone sv
2087 when noWorker $ pushWorker 0 sv
2091-------------------------------------------------------------------------------
2093-------------------------------------------------------------------------------
2095getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
2096getYieldRateInfo st = do
2097 -- convert rate in Hertz to latency in Nanoseconds
2098 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
2099 case getStreamRate st of
2100 Just (Rate low goal high buf) ->
2101 let l = rateToLatency goal
2102 minl = rateToLatency high
2103 maxl = rateToLatency low
2104 in mkYieldRateInfo l (LatencyRange minl maxl) buf
2105 Nothing -> return Nothing
2109 mkYieldRateInfo latency latRange buf = do
2110 measured <- newIORef 0
2111 wcur <- newIORef (0,0,0)
2112 wcol <- newIORef (0,0,0)
2113 now <- getTime Monotonic
2114 wlong <- newIORef (0,now)
2115 period <- newIORef 1
2116 gainLoss <- newIORef (Count 0)
2118 return $ Just YieldRateInfo
2119 { svarLatencyTarget = latency
2120 , svarLatencyRange = latRange
2121 , svarRateBuffer = buf
2122 , svarGainedLostYields = gainLoss
2123 , workerBootstrapLatency = getStreamLatency st
2124 , workerPollingInterval = period
2125 , workerMeasuredLatency = measured
2126 , workerPendingLatency = wcur
2127 , workerCollectedLatency = wcol
2128 , svarAllTimeLatency = wlong
2131newSVarStats :: IO SVarStats
2134 maxWrk <- newIORef 0
2138 avgLat <- newIORef (0, NanoSecond64 0)
2139 maxLat <- newIORef (NanoSecond64 0)
2140 minLat <- newIORef (NanoSecond64 0)
2141 stpTime <- newIORef Nothing
2144 { totalDispatches = disp
2145 , maxWorkers = maxWrk
2146 , maxOutQSize = maxOq
2147 , maxHeapSize = maxHs
2148 , maxWorkQSize = maxWq
2149 , avgWorkerLatency = avgLat
2150 , minWorkerLatency = minLat
2151 , maxWorkerLatency = maxLat
2152 , svarStopTime = stpTime
2155-- XXX remove polymorphism in t, inline f
2156getAheadSVar :: MonadAsync m
2158 -> ( IORef ([t m a], Int)
2159 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2166getAheadSVar st f mrun = do
2167 outQ <- newIORef ([], 0)
2168 -- the second component of the tuple is "Nothing" when heap is being
2169 -- cleared, "Just n" when we are expecting sequence number n to arrive
2170 -- before we can start clearing the heap.
2171 outH <- newIORef (H.empty, Just 0)
2172 outQMv <- newEmptyMVar
2173 active <- newIORef 0
2174 wfw <- newIORef False
2175 running <- newIORef S.empty
2176 -- Sequence number is incremented whenever something is queued, therefore,
2177 -- first sequence number would be 0
2178 q <- newIORef ([], -1)
2179 stopMVar <- newMVar ()
2180 yl <- case getYieldLimit st of
2181 Nothing -> return Nothing
2182 Just x -> Just <$> newIORef x
2183 rateInfo <- getYieldRateInfo st
2185 stats <- newSVarStats
2188 let getSVar sv readOutput postProc = SVar
2189 { outputQueue = outQ
2190 , remainingWork = yl
2191 , maxBufferLimit = getMaxBuffer st
2192 , pushBufferSpace = undefined
2193 , pushBufferPolicy = undefined
2194 , pushBufferMVar = undefined
2195 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
2196 , yieldRateInfo = rateInfo
2197 , outputDoorBell = outQMv
2198 , readOutputQ = readOutput sv
2199 , postProcess = postProc sv
2200 , workerThreads = running
2201 , workLoop = f q outH st{streamVar = Just sv} sv
2202 , enqueue = enqueueAhead sv q
2203 , isWorkDone = isWorkDoneAhead sv q outH
2204 , isQueueDone = isQueueDoneAhead sv q
2205 , needDoorBell = wfw
2206 , svarStyle = AheadVar
2207 , svarStopStyle = StopNone
2208 , svarStopBy = undefined
2210 , workerCount = active
2211 , accountThread = delThread sv
2212 , workerStopMVar = stopMVar
2214 , svarInspectMode = getInspectMode st
2216 , aheadWorkQueue = q
2222 case getStreamRate st of
2223 Nothing -> getSVar sv readOutputQBounded postProcessBounded
2224 Just _ -> getSVar sv readOutputQPaced postProcessPaced
2229 {-# INLINE isQueueDoneAhead #-}
2230 isQueueDoneAhead sv q = do
2231 queueDone <- checkEmpty q
2233 case remainingWork sv of
2237 Nothing -> return False
2238 -- XXX note that yieldsDone can only be authoritative only when there
2239 -- are no workers running. If there are active workers they can
2240 -- later increment the yield count and therefore change the result.
2241 return $ yieldsDone || queueDone
2243 {-# INLINE isWorkDoneAhead #-}
2244 isWorkDoneAhead sv q ref = do
2246 (hp, _) <- readIORef ref
2247 return (H.size hp <= 0)
2248 queueDone <- isQueueDoneAhead sv q
2249 return $ heapDone && queueDone
2252 (xs, _) <- readIORef q
2255getParallelSVar :: MonadIO m
2256 => SVarStopStyle -> State t m a -> RunInIO m -> IO (SVar t m a)
2257getParallelSVar ss st mrun = do
2258 outQ <- newIORef ([], 0)
2259 outQMv <- newEmptyMVar
2260 active <- newIORef 0
2261 running <- newIORef S.empty
2262 yl <- case getYieldLimit st of
2263 Nothing -> return Nothing
2264 Just x -> Just <$> newIORef x
2265 rateInfo <- getYieldRateInfo st
2267 case getMaxBuffer st of
2268 Unlimited -> undefined
2269 Limited x -> (fromIntegral x)
2270 remBuf <- newIORef bufLim
2271 pbMVar <- newMVar ()
2273 stats <- newSVarStats
2278 StopBy -> liftIO $ newIORef undefined
2279 _ -> return undefined
2282 SVar { outputQueue = outQ
2283 , remainingWork = yl
2284 , maxBufferLimit = getMaxBuffer st
2285 , pushBufferSpace = remBuf
2286 , pushBufferPolicy = PushBufferBlock
2287 , pushBufferMVar = pbMVar
2288 , maxWorkerLimit = Unlimited
2289 -- Used only for diagnostics
2290 , yieldRateInfo = rateInfo
2291 , outputDoorBell = outQMv
2292 , readOutputQ = readOutputQPar sv
2293 , postProcess = allThreadsDone sv
2294 , workerThreads = running
2295 , workLoop = undefined
2296 , enqueue = undefined
2297 , isWorkDone = undefined
2298 , isQueueDone = undefined
2299 , needDoorBell = undefined
2300 , svarStyle = ParallelVar
2301 , svarStopStyle = ss
2302 , svarStopBy = stopBy
2304 , workerCount = active
2305 , accountThread = modifyThread sv
2306 , workerStopMVar = undefined
2308 , svarInspectMode = getInspectMode st
2310 , aheadWorkQueue = undefined
2311 , outputHeap = undefined
2318 readOutputQPar sv = liftIO $ do
2319 withDiagMVar sv "readOutputQPar: doorbell"
2320 $ takeMVar (outputDoorBell sv)
2321 case yieldRateInfo sv of
2322 Nothing -> return ()
2323 Just yinfo -> void $ collectLatency sv yinfo False
2324 r <- fst `fmap` readOutputQRaw sv
2326 void $ tryTakeMVar (pushBufferMVar sv)
2329 void $ tryPutMVar (pushBufferMVar sv) ()
2332sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
2333sendFirstWorker sv m = do
2334 -- Note: We must have all the work on the queue before sending the
2335 -- pushworker, otherwise the pushworker may exit before we even get a
2337 liftIO $ enqueue sv m
2338 case yieldRateInfo sv of
2339 Nothing -> pushWorker 0 sv
2341 if svarLatencyTarget yinfo == maxBound
2342 then liftIO $ threadDelay maxBound
2343 else pushWorker 1 sv
2346{-# INLINABLE newAheadVar #-}
2347newAheadVar :: MonadAsync m
2350 -> ( IORef ([t m a], Int)
2351 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2357newAheadVar st m wloop = do
2358 mrun <- captureMonadState
2359 sv <- liftIO $ getAheadSVar st wloop mrun
2360 sendFirstWorker sv m
2362{-# INLINABLE newParallelVar #-}
2363newParallelVar :: MonadAsync m
2364 => SVarStopStyle -> State t m a -> m (SVar t m a)
2365newParallelVar ss st = do
2366 mrun <- captureMonadState
2367 liftIO $ getParallelSVar ss st mrun
2369-------------------------------------------------------------------------------
2370-- Write a stream to an SVar
2371-------------------------------------------------------------------------------
2373-- XXX this errors out for Parallel/Ahead SVars
2374-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
2375-- be read back from the SVar using 'fromSVar'.
2376toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
2377toStreamVar sv m = do
2378 liftIO $ enqueue sv m
2379 done <- allThreadsDone sv
2380 -- XXX This is safe only when called from the consumer thread or when no
2381 -- consumer is present. There may be a race if we are not running in the
2383 -- XXX do this only if the work queue is not empty. The work may have been
2384 -- carried out by existing workers.
2386 case yieldRateInfo sv of
2387 Nothing -> pushWorker 0 sv
2388 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