1{-# LANGUAGE UnboxedTuples #-}
4-- Module : Streamly.Internal.Data.SVar
5-- Copyright : (c) 2017 Harendra Kumar
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
12module Streamly.Internal.Data.SVar
19 -- State threaded around the stream
38 , cleanupSVarFromWorker
61 , handleChildException
67 , HeapDequeueResult(..)
85 , decrementBufferLimit
86 , incrementBufferLimit
108import Control.Concurrent
109 (ThreadId, myThreadId, threadDelay, throwTo, forkIO, killThread)
110import Control.Concurrent.MVar
111 (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar,
113import Control.Exception
114 (SomeException(..), catch, mask, assert, Exception, catches,
115 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
116 BlockedIndefinitelyOnSTM(..))
117import Control.Monad (when)
118import Control.Monad.Catch (MonadThrow)
119import Control.Monad.IO.Class (MonadIO(..))
120import Control.Monad.Trans.Control
121 (MonadBaseControl, control, StM, liftBaseDiscard)
122import Streamly.Internal.Data.Atomics
123 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
125import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
126import Data.Functor (void)
127import Data.Heap (Heap, Entry(..))
128import Data.Int (Int64)
129import Data.Kind (Type)
131 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
132import Data.Maybe (fromJust, fromMaybe)
133#if __GLASGOW_HASKELL__ < 808
134import Data.Semigroup ((<>))
137import GHC.Conc (ThreadId(..))
139import GHC.IO (IO(..))
140import System.IO (hPutStrLn, stderr)
141import System.Mem.Weak (addFinalizer)
143import Streamly.Internal.Data.Time.Clock (Clock(..), getTime)
144import Streamly.Internal.Data.Time.Units
145 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
146 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
148import qualified Data.Heap as H
149import qualified Data.Set as S
151newtype Count = Count Int64
163------------------------------------------------------------------------------
164-- Parent child thread communication type
165------------------------------------------------------------------------------
167data ThreadAbort = ThreadAbort deriving Show
169instance Exception ThreadAbort
171-- | Events that a child thread may send to a parent thread.
174 | ChildStop ThreadId (Maybe SomeException)
176-- | Sorting out-of-turn outputs in a heap for Ahead style streams
177data AheadHeapEntry (t :: (Type -> Type) -> Type -> Type) m a =
180 | AheadEntryStream (RunInIO m, t m a)
183------------------------------------------------------------------------------
184-- State threaded around the monad for thread management
185------------------------------------------------------------------------------
187-- | Identify the type of the SVar. Two computations using the same style can
188-- be scheduled on the same SVar.
190 AsyncVar -- depth first concurrent
191 | WAsyncVar -- breadth first concurrent
192 | ParallelVar -- all parallel
193 | AheadVar -- Concurrent look ahead
196-- | An SVar or a Stream Var is a conduit to the output from multiple streams
197-- running concurrently and asynchronously. An SVar can be thought of as an
198-- asynchronous IO handle. We can write any number of streams to an SVar in a
199-- non-blocking manner and then read them back at any time at any pace. The
200-- SVar would run the streams asynchronously and accumulate results. An SVar
201-- may not really execute the stream completely and accumulate all the results.
202-- However, it ensures that the reader can read the results at whatever paces
203-- it wants to read. The SVar monitors and adapts to the consumer's pace.
205-- An SVar is a mini scheduler, it has an associated workLoop that holds the
206-- stream tasks to be picked and run by a pool of worker threads. It has an
207-- associated output queue where the output stream elements are placed by the
208-- worker threads. A outputDoorBell is used by the worker threads to intimate the
209-- consumer thread about availability of new results in the output queue. More
210-- workers are added to the SVar by 'fromStreamVar' on demand if the output
211-- produced is not keeping pace with the consumer. On bounded SVars, workers
212-- block on the output queue to provide throttling of the producer when the
213-- consumer is not pulling fast enough. The number of workers may even get
214-- reduced depending on the consuming pace.
216-- New work is enqueued either at the time of creation of the SVar or as a
217-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
218-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
220-- We measure the individual worker latencies to estimate the number of workers
221-- needed or the amount of time we have to sleep between dispatches to achieve
222-- a particular rate when controlled pace mode it used.
223data WorkerInfo = WorkerInfo
224 { workerYieldMax :: Count -- 0 means unlimited
225 -- total number of yields by the worker till now
226 , workerYieldCount :: IORef Count
227 -- yieldCount at start, timestamp
228 , workerLatencyStart :: IORef (Count, AbsTime)
232-- | Specifies the stream yield rate in yields per second (@Hertz@).
233-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
234-- allow only as many yields as we have accumulated as per 'rateGoal' since the
235-- start of time. If the consumer or the producer is slower or faster, the
236-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
237-- between the two by increasing or decreasing the pull rate from the producer.
238-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
239-- much as 'rateBuffer'.
241-- 'rateLow' puts a bound on how low the instantaneous rate can go when
242-- recovering the rate gap. In other words, it determines the maximum yield
243-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
244-- rate can go when recovering the rate gap. In other words, it determines the
245-- minimum yield latency. We reduce the latency by increasing concurrency,
246-- therefore we can say that it puts an upper bound on concurrency.
248-- If the 'rateGoal' is 0 or negative the stream never yields a value.
249-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
253 { rateLow :: Double -- ^ The lower rate limit
254 , rateGoal :: Double -- ^ The target rate we want to achieve
255 , rateHigh :: Double -- ^ The upper rate limit
256 , rateBuffer :: Int -- ^ Maximum slack from the goal
259data LatencyRange = LatencyRange
260 { minLatency :: NanoSecond64
261 , maxLatency :: NanoSecond64
265data YieldRateInfo = YieldRateInfo
266 { svarLatencyTarget :: NanoSecond64
267 , svarLatencyRange :: LatencyRange
268 , svarRateBuffer :: Int
270 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
271 -- read by the worker threads
272 , svarGainedLostYields :: IORef Count
274 -- Actual latency/througput as seen from the consumer side, we count the
275 -- yields and the time it took to generates those yields. This is used to
276 -- increase or decrease the number of workers needed to achieve the desired
277 -- rate. The idle time of workers is adjusted in this, so that we only
278 -- account for the rate when the consumer actually demands data.
279 -- XXX interval latency is enough, we can move this under diagnostics build
280 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
281 -- read by the worker threads
282 , svarAllTimeLatency :: IORef (Count, AbsTime)
284 -- XXX Worker latency specified by the user to be used before the first
285 -- actual measurement arrives. Not yet implemented
286 , workerBootstrapLatency :: Maybe NanoSecond64
288 -- After how many yields the worker should update the latency information.
289 -- If the latency is high, this count is kept lower and vice-versa. XXX If
290 -- the latency suddenly becomes too high this count may remain too high for
291 -- long time, in such cases the consumer can change it.
292 -- 0 means no latency computation
293 -- XXX this is derivable from workerMeasuredLatency, can be removed.
294 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
295 -- read by the worker threads
296 , workerPollingInterval :: IORef Count
298 -- This is in progress latency stats maintained by the workers which we
299 -- empty into workerCollectedLatency stats at certain intervals - whenever
300 -- we process the stream elements yielded in this period. The first count
301 -- is all yields, the second count is only those yields for which the
302 -- latency was measured to be non-zero (note that if the timer resolution
303 -- is low the measured latency may be zero e.g. on JS platform).
304 -- [LOCKING] Locked access. Modified by the consumer thread as well as
305 -- worker threads. Workers modify it periodically based on
306 -- workerPollingInterval and not on every yield to reduce the locking
308 -- (allYieldCount, yieldCount, timeTaken)
309 , workerPendingLatency :: IORef (Count, Count, NanoSecond64)
311 -- This is the second level stat which is an accmulation from
312 -- workerPendingLatency stats. We keep accumulating latencies in this
313 -- bucket until we have stats for a sufficient period and then we reset it
314 -- to start collecting for the next period and retain the computed average
315 -- latency for the last period in workerMeasuredLatency.
316 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
317 -- read by the worker threads
318 -- (allYieldCount, yieldCount, timeTaken)
319 , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)
321 -- Latency as measured by workers, aggregated for the last period.
322 -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
323 -- read by the worker threads
324 , workerMeasuredLatency :: IORef NanoSecond64
327data SVarStats = SVarStats {
328 totalDispatches :: IORef Int
329 , maxWorkers :: IORef Int
330 , maxOutQSize :: IORef Int
331 , maxHeapSize :: IORef Int
332 , maxWorkQSize :: IORef Int
333 , avgWorkerLatency :: IORef (Count, NanoSecond64)
334 , minWorkerLatency :: IORef NanoSecond64
335 , maxWorkerLatency :: IORef NanoSecond64
336 , svarStopTime :: IORef (Maybe AbsTime)
339-- This is essentially a 'Maybe Word' type
340data Limit = Unlimited | Limited Word deriving Show
342instance Eq Limit where
343 Unlimited == Unlimited = True
344 Unlimited == Limited _ = False
345 Limited _ == Unlimited = False
346 Limited x == Limited y = x == y
348instance Ord Limit where
349 Unlimited <= Unlimited = True
350 Unlimited <= Limited _ = False
351 Limited _ <= Unlimited = True
352 Limited x <= Limited y = x <= y
354-- When to stop the composed stream.
356 StopNone -- stops only when all streams are finished
357 | StopAny -- stop when any stream finishes
358 | StopBy -- stop when a specific stream finishes
361-- | Buffering policy for persistent push workers (in ParallelT). In a pull
362-- style SVar (in AsyncT, AheadT etc.), the consumer side dispatches workers on
363-- demand, workers terminate if the buffer is full or if the consumer is not
364-- cosuming fast enough. In a push style SVar, a worker is dispatched only
365-- once, workers are persistent and keep pushing work to the consumer via a
366-- bounded buffer. If the buffer becomes full the worker either blocks, or it
367-- can drop an item from the buffer to make space.
369-- Pull style SVars are useful in lazy stream evaluation whereas push style
370-- SVars are useful in strict left Folds.
372-- XXX Maybe we can separate the implementation in two different types instead
373-- of using a common SVar type.
375data PushBufferPolicy =
376 PushBufferDropNew -- drop the latest element and continue
377 | PushBufferDropOld -- drop the oldest element and continue
378 | PushBufferBlock -- block the thread until space
381-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have
382-- references to the original SVar stored in several functions which will keep
383-- pointing to the original data and the new updates won't reflect there.
384-- Any updateable parts must be kept in mutable references (IORef).
386data SVar t m a = SVar
389 svarStyle :: SVarStyle
390 , svarMrun :: RunInIO m
391 , svarStopStyle :: SVarStopStyle
392 , svarStopBy :: IORef ThreadId
394 -- Shared output queue (events, length)
395 -- XXX For better efficiency we can try a preallocated array type (perhaps
396 -- something like a vector) that allows an O(1) append. That way we will
397 -- avoid constructing and reversing the list. Possibly we can also avoid
398 -- the GC copying overhead. When the size increases we should be able to
399 -- allocate the array in chunks.
401 -- [LOCKING] Frequent locked access. This is updated by workers on each
402 -- yield and once in a while read by the consumer thread. This could have
403 -- big locking overhead if the number of workers is high.
405 -- XXX We can use a per-CPU data structure to reduce the locking overhead.
406 -- However, a per-cpu structure cannot guarantee the exact sequence in
407 -- which the elements were added, though that may not be important.
408 , outputQueue :: IORef ([ChildEvent a], Int)
410 -- [LOCKING] Infrequent MVar. Used when the outputQ transitions from empty
411 -- to non-empty, or a work item is queued by a worker to the work queue and
412 -- needDoorBell is set by the consumer.
413 , outputDoorBell :: MVar () -- signal the consumer about output
414 , readOutputQ :: m [ChildEvent a]
415 , postProcess :: m Bool
417 -- channel to send events from the consumer to the worker. Used to send
418 -- exceptions from a fold driver to the fold computation running as a
419 -- consumer thread in the concurrent fold cases. Currently only one event
420 -- is sent by the fold so we do not really need a queue for it.
421 , outputQueueFromConsumer :: IORef ([ChildEvent a], Int)
422 , outputDoorBellFromConsumer :: MVar ()
424 -- Combined/aggregate parameters
425 -- This is truncated to maxBufferLimit if set to more than that. Otherwise
426 -- potentially each worker may yield one value to the buffer in the worst
427 -- case exceeding the requested buffer size.
428 , maxWorkerLimit :: Limit
429 , maxBufferLimit :: Limit
430 -- These two are valid and used only when maxBufferLimit is Limited.
431 , pushBufferSpace :: IORef Count
432 , pushBufferPolicy :: PushBufferPolicy
433 -- [LOCKING] The consumer puts this MVar after emptying the buffer, workers
434 -- block on it when the buffer becomes full. No overhead unless the buffer
436 , pushBufferMVar :: MVar ()
438 -- [LOCKING] Read only access by consumer when dispatching a worker.
439 -- Decremented by workers when picking work and undo decrement if the
440 -- worker does not yield a value.
441 , remainingWork :: Maybe (IORef Count)
442 , yieldRateInfo :: Maybe YieldRateInfo
444 -- Used only by bounded SVar types
445 , enqueue :: (RunInIO m, t m a) -> IO ()
446 , isWorkDone :: IO Bool
447 , isQueueDone :: IO Bool
448 , needDoorBell :: IORef Bool
449 , workLoop :: Maybe WorkerInfo -> m ()
451 -- Shared, thread tracking
452 -- [LOCKING] Updated unlocked only by consumer thread in case of
453 -- Async/Ahead style SVars. Updated locked by worker threads in case of
455 , workerThreads :: IORef (Set ThreadId)
456 -- [LOCKING] Updated locked by consumer thread when dispatching a worker
457 -- and by the worker threads when the thread stops. This is read unsafely
458 -- at several places where we want to rely on an approximate value.
459 , workerCount :: IORef Int
460 , accountThread :: ThreadId -> m ()
461 , workerStopMVar :: MVar ()
463 , svarStats :: SVarStats
464 -- to track garbage collection of SVar
465 , svarRef :: Maybe (IORef ())
467 -- Only for diagnostics
468 , svarInspectMode :: Bool
469 , svarCreator :: ThreadId
470 , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))
472 -- Shared work queue (stream, seqNo)
473 , aheadWorkQueue :: IORef ([t m a], Int)
476-------------------------------------------------------------------------------
477-- State for concurrency control
478-------------------------------------------------------------------------------
480-- XXX we can put the resettable fields in a oneShotConfig field and others in
481-- a persistentConfig field. That way reset would be fast and scalable
482-- irrespective of the number of fields.
484-- XXX make all these Limited types and use phantom types to distinguish them
485data State t m a = State
486 { -- one shot configuration, automatically reset for each API call
487 streamVar :: Maybe (SVar t m a)
488 , _yieldLimit :: Maybe Count
490 -- persistent configuration, state that remains valid until changed by
491 -- an explicit setting via a combinator.
492 , _threadsHigh :: Limit
493 , _bufferHigh :: Limit
494 -- XXX these two can be collapsed into a single type
495 , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency
496 , _maxStreamRate :: Maybe Rate
497 , _inspectMode :: Bool
500-------------------------------------------------------------------------------
501-- State defaults and reset
502-------------------------------------------------------------------------------
504-- A magical value for the buffer size arrived at by running the smallest
505-- possible task and measuring the optimal value of the buffer for that. This
506-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
508magicMaxBuffer :: Word
511defaultMaxThreads, defaultMaxBuffer :: Limit
512defaultMaxThreads = Limited magicMaxBuffer
513defaultMaxBuffer = Limited magicMaxBuffer
515-- The fields prefixed by an _ are not to be accessed or updated directly but
516-- via smart accessor APIs.
517defState :: State t m a
519 { streamVar = Nothing
520 , _yieldLimit = Nothing
521 , _threadsHigh = defaultMaxThreads
522 , _bufferHigh = defaultMaxBuffer
523 , _maxStreamRate = Nothing
524 , _streamLatency = Nothing
525 , _inspectMode = False
528-- XXX if perf gets affected we can have all the Nothing params in a single
529-- structure so that we reset is fast. We can also use rewrite rules such that
530-- reset occurs only in concurrent streams to reduce the impact on serial
532-- We can optimize this so that we clear it only if it is a Just value, it
533-- results in slightly better perf for zip/zipM but the performance of scan
534-- worsens a lot, it does not fuse.
536-- XXX This has a side effect of clearing the SVar and yieldLimit, therefore it
537-- should not be used to convert from the same type to the same type, unless
538-- you want to clear the SVar. For clearing the SVar you should be using the
539-- appropriate unStream functions instead.
541-- | Adapt the stream state from one type to another.
542adaptState :: State t m a -> State t n b
544 { streamVar = Nothing
545 , _yieldLimit = Nothing
548-------------------------------------------------------------------------------
549-- Smart get/set routines for State
550-------------------------------------------------------------------------------
552-- Use get/set routines instead of directly accessing the State fields
553setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
554setYieldLimit lim st =
561 else Just (fromIntegral n)
564getYieldLimit :: State t m a -> Maybe Count
565getYieldLimit = _yieldLimit
567setMaxThreads :: Int -> State t m a -> State t m a
573 then defaultMaxThreads
574 else Limited (fromIntegral n)
577getMaxThreads :: State t m a -> Limit
578getMaxThreads = _threadsHigh
580setMaxBuffer :: Int -> State t m a -> State t m a
586 then defaultMaxBuffer
587 else Limited (fromIntegral n)
590getMaxBuffer :: State t m a -> Limit
591getMaxBuffer = _bufferHigh
593setStreamRate :: Maybe Rate -> State t m a -> State t m a
594setStreamRate r st = st { _maxStreamRate = r }
596getStreamRate :: State t m a -> Maybe Rate
597getStreamRate = _maxStreamRate
599setStreamLatency :: Int -> State t m a -> State t m a
600setStreamLatency n st =
601 st { _streamLatency =
604 else Just (fromIntegral n)
607getStreamLatency :: State t m a -> Maybe NanoSecond64
608getStreamLatency = _streamLatency
610setInspectMode :: State t m a -> State t m a
611setInspectMode st = st { _inspectMode = True }
613getInspectMode :: State t m a -> Bool
614getInspectMode = _inspectMode
616-------------------------------------------------------------------------------
618-------------------------------------------------------------------------------
620cleanupSVar :: SVar t m a -> IO ()
622 workers <- readIORef (workerThreads sv)
623 Prelude.mapM_ (`throwTo` ThreadAbort)
626cleanupSVarFromWorker :: SVar t m a -> IO ()
627cleanupSVarFromWorker sv = do
628 workers <- readIORef (workerThreads sv)
630 Prelude.mapM_ (`throwTo` ThreadAbort)
631 (Prelude.filter (/= self) $ S.toList workers)
633-------------------------------------------------------------------------------
634-- Worker latency data collection
635-------------------------------------------------------------------------------
637-- Every once in a while workers update the latencies and check the yield rate.
638-- They return if we are above the expected yield rate. If we check too often
639-- it may impact performance, if we check less often we may have a stale
640-- picture. We update every minThreadDelay but we translate that into a yield
641-- count based on latency so that the checking overhead is little.
643-- XXX use a generation count to indicate that the value is updated. If the
644-- value is updated an existing worker must check it again on the next yield.
645-- Otherwise it is possible that we may keep updating it and because of the mod
646-- worker keeps skipping it.
647updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
648updateWorkerPollingInterval yinfo latency = do
649 let periodRef = workerPollingInterval yinfo
650 cnt = max 1 $ minThreadDelay `div` latency
651 period = min cnt (fromIntegral magicMaxBuffer)
653 writeIORef periodRef (fromIntegral period)
655{-# INLINE recordMinMaxLatency #-}
656recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()
657recordMinMaxLatency sv new = do
658 let ss = svarStats sv
659 minLat <- readIORef (minWorkerLatency ss)
660 when (new < minLat || minLat == 0) $
661 writeIORef (minWorkerLatency ss) new
663 maxLat <- readIORef (maxWorkerLatency ss)
664 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
666recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()
667recordAvgLatency sv (count, time) = do
668 let ss = svarStats sv
669 modifyIORef (avgWorkerLatency ss) $
670 \(cnt, t) -> (cnt + count, t + time)
672-- Pour the pending latency stats into a collection bucket
673{-# INLINE collectWorkerPendingLatency #-}
674collectWorkerPendingLatency
675 :: IORef (Count, Count, NanoSecond64)
676 -> IORef (Count, Count, NanoSecond64)
677 -> IO (Count, Maybe (Count, NanoSecond64))
678collectWorkerPendingLatency cur col = do
679 (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
681 (fcnt, cnt, t) <- readIORef col
682 let totalCount = fcnt + fcount
683 latCount = cnt + count
685 writeIORef col (totalCount, latCount, latTime)
687 assert (latCount == 0 || latTime /= 0) (return ())
689 if latCount > 0 && latTime > 0
690 then Just (latCount, latTime)
692 return (totalCount, latPair)
694{-# INLINE shouldUseCollectedBatch #-}
695shouldUseCollectedBatch
701shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
702 let r = fromIntegral newLat / fromIntegral prevLat :: Double
703 in (collectedYields > fromIntegral magicMaxBuffer)
704 || (collectedTime > minThreadDelay)
705 || (prevLat > 0 && (r > 2 || r < 0.5))
708-- Returns a triple, (1) yield count since last collection, (2) the base time
709-- when we started counting, (3) average latency in the last measurement
710-- period. The former two are used for accurate measurement of the going rate
711-- whereas the average is used for future estimates e.g. how many workers
712-- should be maintained to maintain the rate.
713-- CAUTION! keep it in sync with getWorkerLatency
714collectLatency :: SVar t m a
717 -> IO (Count, AbsTime, NanoSecond64)
718collectLatency sv yinfo drain = do
719 let cur = workerPendingLatency yinfo
720 col = workerCollectedLatency yinfo
721 longTerm = svarAllTimeLatency yinfo
722 measured = workerMeasuredLatency yinfo
724 (newCount, newLatPair) <- collectWorkerPendingLatency cur col
725 (lcount, ltime) <- readIORef longTerm
726 prevLat <- readIORef measured
728 let newLcount = lcount + newCount
729 retWith lat = return (newLcount, ltime, lat)
732 Nothing -> retWith prevLat
733 Just (count, time) -> do
734 let newLat = time `div` fromIntegral count
735 when (svarInspectMode sv) $ recordMinMaxLatency sv newLat
736 -- When we have collected a significant sized batch we compute the
737 -- new latency using that batch and return the new latency,
738 -- otherwise we return the previous latency derived from the
740 if shouldUseCollectedBatch newCount time newLat prevLat || drain
742 -- XXX make this NOINLINE?
743 updateWorkerPollingInterval yinfo (max newLat prevLat)
744 when (svarInspectMode sv) $ recordAvgLatency sv (count, time)
745 writeIORef col (0, 0, 0)
746 writeIORef measured ((prevLat + newLat) `div` 2)
747 modifyIORef longTerm $ \(_, t) -> (newLcount, t)
751-------------------------------------------------------------------------------
752-- Dumping the SVar for debug/diag
753-------------------------------------------------------------------------------
755dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
756dumpSVarStats sv ss style = do
757 case yieldRateInfo sv of
760 _ <- liftIO $ collectLatency sv yinfo True
763 dispatches <- readIORef $ totalDispatches ss
764 maxWrk <- readIORef $ maxWorkers ss
765 maxOq <- readIORef $ maxOutQSize ss
766 maxHp <- readIORef $ maxHeapSize ss
767 minLat <- readIORef $ minWorkerLatency ss
768 maxLat <- readIORef $ maxWorkerLatency ss
769 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
770 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
771 Nothing -> return (0, 0, 0)
773 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
776 t <- readIORef (svarStopTime ss)
777 gl <- readIORef (svarGainedLostYields yinfo)
780 now <- getTime Monotonic
781 let interval = diffAbsTime64 now startTime
782 return (cnt, gl, interval `div` fromIntegral cnt)
784 let interval = diffAbsTime64 stopTime startTime
785 return (cnt, gl, interval `div` fromIntegral cnt)
786 else return (0, 0, 0)
789 [ "total dispatches = " <> show dispatches
790 , "max workers = " <> show maxWrk
791 , "max outQSize = " <> show maxOq
792 <> (if style == AheadVar
793 then "\nheap max size = " <> show maxHp
796 then "\nmin worker latency = " <> showNanoSecond64 minLat
799 then "\nmax worker latency = " <> showNanoSecond64 maxLat
802 then let lat = avgTime `div` fromIntegral avgCnt
803 in "\navg worker latency = " <> showNanoSecond64 lat
806 then "\nSVar latency = " <> showRelTime64 svarLat
809 then "\nSVar yield count = " <> show svarCnt
811 <> (if svarGainLossCnt > 0
812 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
816{-# NOINLINE dumpSVar #-}
817dumpSVar :: SVar t m a -> IO String
819 (oqList, oqLen) <- readIORef $ outputQueue sv
820 db <- tryReadMVar $ outputDoorBell sv
822 if svarStyle sv == AheadVar
824 (oheap, oheapSeq) <- readIORef $ outputHeap sv
825 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
827 [ "heap length = " <> show (H.size oheap)
828 , "heap seqeunce = " <> show oheapSeq
829 , "work queue length = " <> show (length wq)
830 , "work queue sequence = " <> show wqSeq
834 let style = svarStyle sv
836 if style /= ParallelVar
837 then readIORef $ needDoorBell sv
839 rthread <- readIORef $ workerThreads sv
840 workers <- readIORef $ workerCount sv
841 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
845 "Creator tid = " <> show (svarCreator sv),
846 "style = " <> show (svarStyle sv)
847 , "---------CURRENT STATE-----------"
848 , "outputQueue length computed = " <> show (length oqList)
849 , "outputQueue length maintained = " <> show oqLen
850 -- XXX print the types of events in the outputQueue, first 5
851 , "outputDoorBell = " <> show db
855 [ "needDoorBell = " <> show waiting
856 , "running threads = " <> show rthread
857 -- XXX print the status of first 5 threads
858 , "running thread count = " <> show workers
860 <> "---------STATS-----------\n"
863printSVar :: SVar t m a -> String -> IO ()
865 svInfo <- dumpSVar sv
866 hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo
868-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
869-- can keep it on in production to debug problems quickly if and when they
870-- happen, but it may result in unexpected output when threads are left hanging
871-- until they are GCed because the consumer went away.
873{-# NOINLINE mvarExcHandler #-}
874mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
875mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
876 svInfo <- dumpSVar sv
877 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
880{-# NOINLINE stmExcHandler #-}
881stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
882stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
883 svInfo <- dumpSVar sv
884 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
887withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
888withDiagMVar sv label action =
889 if svarInspectMode sv
891 action `catches` [ Handler (mvarExcHandler sv label)
892 , Handler (stmExcHandler sv label)
896------------------------------------------------------------------------------
898------------------------------------------------------------------------------
900-- | A monad that can perform concurrent or parallel IO operations. Streams
901-- that can be composed concurrently require the underlying monad to be
905type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
907-- When we run computations concurrently, we completely isolate the state of
908-- the concurrent computations from the parent computation. The invariant is
909-- that we should never be running two concurrent computations in the same
910-- thread without using the runInIO function. Also, we should never be running
911-- a concurrent computation in the parent thread, otherwise it may affect the
912-- state of the parent which is against the defined semantics of concurrent
914newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }
916captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
917captureMonadState = control $ \run -> run (return $ RunInIO run)
919-- Stolen from the async package. The perf improvement is modest, 2% on a
920-- thread heavy benchmark (parallel composition using noop computations).
921-- A version of forkIO that does not include the outer exception
922-- handler: saves a bit of time when we will be installing our own
924{-# INLINE rawForkIO #-}
925rawForkIO :: IO () -> IO ThreadId
926rawForkIO action = IO $ \ s ->
927 case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)
930doFork :: MonadBaseControl IO m
933 -> (SomeException -> IO ())
935doFork action (RunInIO mrun) exHandler =
937 mask $ \restore -> do
938 tid <- rawForkIO $ catch (restore $ void $ mrun action)
942{-# INLINABLE fork #-}
943fork :: MonadBaseControl IO m => m () -> m ThreadId
944fork = liftBaseDiscard forkIO
946-- | Fork a thread that is automatically killed as soon as the reference to the
947-- returned threadId is garbage collected.
949{-# INLINABLE forkManaged #-}
950forkManaged :: (MonadIO m, MonadBaseControl IO m) => m () -> m ThreadId
951forkManaged action = do
953 liftIO $ addFinalizer tid (killThread tid)
956------------------------------------------------------------------------------
957-- Collecting results from child workers in a streamed fashion
958------------------------------------------------------------------------------
960-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
961-- faster, along with the fields in sv required by send?
962-- XXX make it noinline
964-- XXX we may want to employ an increment and decrement in batches when the
965-- througput is high or when the cost of synchronization is high. For example
966-- if the application is distributed then inc/dec of a shared variable may be
969-- A worker decrements the yield limit before it executes an action. However,
970-- the action may not result in an element being yielded, in that case we have
971-- to increment the yield limit.
973-- Note that we need it to be an Int type so that we have the ability to undo a
974-- decrement that takes it below zero.
975{-# INLINE decrementYieldLimit #-}
976decrementYieldLimit :: SVar t m a -> IO Bool
977decrementYieldLimit sv =
978 case remainingWork sv of
979 Nothing -> return True
981 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
984{-# INLINE incrementYieldLimit #-}
985incrementYieldLimit :: SVar t m a -> IO ()
986incrementYieldLimit sv =
987 case remainingWork sv of
989 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
991-- XXX exception safety of all atomic/MVar operations
993-- TBD Each worker can have their own queue and the consumer can empty one
994-- queue at a time, that way contention can be reduced.
996-- XXX Only yields should be counted in the buffer limit and not the Stop
999{-# INLINE decrementBufferLimit #-}
1000decrementBufferLimit :: SVar t m a -> IO ()
1001decrementBufferLimit sv =
1002 case maxBufferLimit sv of
1003 Unlimited -> return ()
1005 let ref = pushBufferSpace sv
1006 old <- atomicModifyIORefCAS ref $ \x ->
1007 (if x >= 1 then x - 1 else x, x)
1009 case pushBufferPolicy sv of
1010 PushBufferBlock -> blockAndRetry
1011 PushBufferDropNew -> do
1012 -- We just drop one item and proceed. It is possible
1013 -- that by the time we drop the item the consumer
1014 -- thread might have run and created space in the
1015 -- buffer, but we do not care about that condition.
1016 -- This is not pedantically correct but it should be
1017 -- fine in practice.
1018 -- XXX we may want to drop only if n == maxBuf
1019 -- otherwise we must have space in the buffer and a
1020 -- decrement should be possible.
1021 block <- atomicModifyIORefCAS (outputQueue sv) $
1024 [] -> (([],n), True)
1025 _ : xs -> ((xs, n - 1), False)
1026 when block blockAndRetry
1027 -- XXX need a dequeue or ring buffer for this
1028 PushBufferDropOld -> undefined
1033 let ref = pushBufferSpace sv
1034 liftIO $ takeMVar (pushBufferMVar sv)
1035 old <- atomicModifyIORefCAS ref $ \x ->
1036 (if x >= 1 then x - 1 else x, x)
1037 -- When multiple threads sleep on takeMVar, the first thread would
1038 -- wakeup due to a putMVar by the consumer, but the rest of the threads
1039 -- would have to put back the MVar after taking it and decrementing the
1040 -- buffer count, otherwise all other threads will remain asleep.
1042 then void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
1043 -- We do not put the MVar back in this case, instead we
1044 -- wait for the consumer to put it.
1047{-# INLINE incrementBufferLimit #-}
1048incrementBufferLimit :: SVar t m a -> IO ()
1049incrementBufferLimit sv =
1050 case maxBufferLimit sv of
1051 Unlimited -> return ()
1053 atomicModifyIORefCAS_ (pushBufferSpace sv) (+ 1)
1055 void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
1057{-# INLINE resetBufferLimit #-}
1058resetBufferLimit :: SVar t m a -> IO ()
1059resetBufferLimit sv =
1060 case maxBufferLimit sv of
1061 Unlimited -> return ()
1062 Limited n -> atomicModifyIORefCAS_ (pushBufferSpace sv)
1063 (const (fromIntegral n))
1065{-# INLINE sendWithDoorBell #-}
1067 IORef ([ChildEvent a], Int) -> MVar () -> ChildEvent a -> IO Int
1068sendWithDoorBell q bell msg = do
1069 -- XXX can the access to outputQueue be made faster somehow?
1070 oldlen <- atomicModifyIORefCAS q $ \(es, n) ->
1071 ((msg : es, n + 1), n)
1072 when (oldlen <= 0) $ do
1073 -- The wake up must happen only after the store has finished otherwise
1074 -- we can have lost wakeup problems.
1076 -- Since multiple workers can try this at the same time, it is possible
1077 -- that we may put a spurious MVar after the consumer has already seen
1078 -- the output. But that's harmless, at worst it may cause the consumer
1079 -- to read the queue again and find it empty.
1080 -- The important point is that the consumer is guaranteed to receive a
1081 -- doorbell if something was added to the queue after it empties it.
1082 void $ tryPutMVar bell ()
1085-- | This function is used by the producer threads to queue output for the
1086-- consumer thread to consume. Returns whether the queue has more space.
1087send :: SVar t m a -> ChildEvent a -> IO Int
1088send sv = sendWithDoorBell (outputQueue sv) (outputDoorBell sv)
1090-- There is no bound implemented on the buffer, this is assumed to be low
1092sendToProducer :: SVar t m a -> ChildEvent a -> IO Int
1093sendToProducer sv msg = do
1094 -- In case the producer stream is blocked on pushing to the fold buffer
1095 -- then wake it up so that it can check for the stop event or exception
1096 -- being sent to it otherwise we will be deadlocked.
1097 void $ tryPutMVar (pushBufferMVar sv) ()
1098 sendWithDoorBell (outputQueueFromConsumer sv)
1099 (outputDoorBellFromConsumer sv) msg
1101-- {-# NOINLINE sendStopToProducer #-}
1102sendStopToProducer :: MonadIO m => SVar t m a -> m ()
1103sendStopToProducer sv = liftIO $ do
1105 void $ sendToProducer sv (ChildStop tid Nothing)
1107workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
1108workerCollectLatency winfo = do
1109 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
1110 cnt1 <- readIORef (workerYieldCount winfo)
1111 let cnt = cnt1 - cnt0
1115 t1 <- getTime Monotonic
1116 let period = fromRelTime64 $ diffAbsTime64 t1 t0
1117 writeIORef (workerLatencyStart winfo) (cnt1, t1)
1118 return $ Just (cnt, period)
1121-- XXX There are a number of gotchas in measuring latencies.
1122-- 1) We measure latencies only when a worker yields a value
1123-- 2) It is possible that a stream calls the stop continuation, in which case
1124-- the worker would not yield a value and we would not account that worker in
1125-- latencies. Even though this case should ideally be accounted we do not
1126-- account it because we cannot or do not distinguish it from the case
1128-- 3) It is possible that a worker returns without yielding anything because it
1129-- never got a chance to pick up work.
1130-- 4) If the system timer resolution is lower than the latency, the latency
1131-- computation turns out to be zero.
1133-- We can fix this if we measure the latencies by counting the work items
1134-- picked rather than based on the outputs yielded.
1135workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
1136workerUpdateLatency yinfo winfo = do
1137 r <- workerCollectLatency winfo
1139 Just (cnt, period) -> do
1140 -- NOTE: On JS platform the timer resolution could be pretty low. When
1141 -- the timer resolution is low, measurement of latencies could be
1142 -- tricky. All the worker latencies will turn out to be zero if they
1143 -- are lower than the resolution. We only take into account those
1144 -- measurements which are more than the timer resolution.
1146 let ref = workerPendingLatency yinfo
1147 (cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
1148 atomicModifyIORefCAS_ ref $
1149 \(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
1150 Nothing -> return ()
1152updateYieldCount :: WorkerInfo -> IO Count
1153updateYieldCount winfo = do
1154 cnt <- readIORef (workerYieldCount winfo)
1156 writeIORef (workerYieldCount winfo) cnt1
1159isBeyondMaxYield :: Count -> WorkerInfo -> Bool
1160isBeyondMaxYield cnt winfo =
1161 let ymax = workerYieldMax winfo
1162 in ymax /= 0 && cnt >= ymax
1164-- XXX we should do rate control periodically based on the total yields rather
1165-- than based on the worker local yields as other workers may have yielded more
1166-- and we should stop based on the aggregate yields. However, latency update
1167-- period can be based on individual worker yields.
1168{-# NOINLINE checkRatePeriodic #-}
1169checkRatePeriodic :: SVar t m a
1174checkRatePeriodic sv yinfo winfo ycnt = do
1175 i <- readIORef (workerPollingInterval yinfo)
1176 -- XXX use generation count to check if the interval has been updated
1177 if i /= 0 && (ycnt `mod` i) == 0
1179 workerUpdateLatency yinfo winfo
1180 -- XXX not required for parallel streams
1181 isBeyondMaxRate sv yinfo
1184-- CAUTION! this also updates the yield count and therefore should be called
1185-- only when we are actually yielding an element.
1186{-# NOINLINE workerRateControl #-}
1187workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
1188workerRateControl sv yinfo winfo = do
1189 cnt <- updateYieldCount winfo
1190 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
1191 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
1193-- XXX we should do rate control here but not latency update in case of ahead
1194-- streams. latency update must be done when we yield directly to outputQueue
1195-- or when we yield to heap.
1197-- returns whether the worker should continue (True) or stop (False).
1198{-# INLINE sendYield #-}
1199sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
1200sendYield sv mwinfo msg = do
1201 oldlen <- send sv msg
1202 let limit = maxBufferLimit sv
1203 bufferSpaceOk <- case limit of
1204 Unlimited -> return True
1206 active <- readIORef (workerCount sv)
1207 return $ (oldlen + 1) < (fromIntegral lim - active)
1211 case yieldRateInfo sv of
1212 Nothing -> return True
1213 Just yinfo -> workerRateControl sv yinfo winfo
1214 Nothing -> return True
1215 return $ bufferSpaceOk && rateLimitOk
1217{-# INLINE workerStopUpdate #-}
1218workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
1219workerStopUpdate winfo info = do
1220 i <- readIORef (workerPollingInterval info)
1221 when (i /= 0) $ workerUpdateLatency info winfo
1223{-# INLINABLE sendStop #-}
1224sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
1225sendStop sv mwinfo = do
1226 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
1227 case (mwinfo, yieldRateInfo sv) of
1228 (Just winfo, Just info) ->
1229 workerStopUpdate winfo info
1232 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
1234-------------------------------------------------------------------------------
1236-------------------------------------------------------------------------------
1238{-# INLINE ringDoorBell #-}
1239ringDoorBell :: SVar t m a -> IO ()
1242 w <- readIORef $ needDoorBell sv
1244 -- Note: the sequence of operations is important for correctness here.
1245 -- We need to set the flag to false strictly before sending the
1246 -- outputDoorBell, otherwise the outputDoorBell may get processed too
1247 -- early and then we may set the flag to False to later making the
1248 -- consumer lose the flag, even without receiving a outputDoorBell.
1249 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1250 void $ tryPutMVar (outputDoorBell sv) ()
1252-------------------------------------------------------------------------------
1254-------------------------------------------------------------------------------
1256-- Note: For purely right associated expressions this queue should have at most
1257-- one element. It grows to more than one when we have left associcated
1258-- expressions. Large left associated compositions can grow this to a
1260{-# INLINE enqueueLIFO #-}
1262 SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
1263enqueueLIFO sv q m = do
1264 atomicModifyIORefCAS_ q $ \ms -> m : ms
1267-------------------------------------------------------------------------------
1269-------------------------------------------------------------------------------
1271-- XXX we can use the Ahead style sequence/heap mechanism to make the best
1272-- effort to always try to finish the streams on the left side of an expression
1273-- first as long as possible.
1275{-# INLINE enqueueFIFO #-}
1278 -> LinkedQueue (RunInIO m, t m a)
1279 -> (RunInIO m, t m a)
1281enqueueFIFO sv q m = do
1285-------------------------------------------------------------------------------
1287-------------------------------------------------------------------------------
1289-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
1290-- but always serve them in the same order as they appear in the stream. To
1291-- implement lookahead streams efficiently we assign a sequence number to each
1292-- task when the task is picked up for execution. When the task finishes, the
1293-- output is tagged with the same sequence number and we rearrange the outputs
1294-- in sequence based on that number.
1296-- To explain the mechanism imagine that the current task at the head of the
1297-- stream has a "token" to yield to the outputQueue. The ownership of the token
1298-- is determined by the current sequence number is maintained in outputHeap.
1299-- Sequence number is assigned when a task is queued. When a thread dequeues a
1300-- task it picks up the sequence number as well and when the output is ready it
1301-- uses the sequence number to queue the output to the outputQueue.
1303-- The thread with current sequence number sends the output directly to the
1304-- outputQueue. Other threads push the output to the outputHeap. When the task
1305-- being queued on the heap is a stream of many elements we evaluate only the
1306-- first element and keep the rest of the unevaluated computation in the heap.
1307-- When such a task gets the "token" for outputQueue it evaluates and directly
1308-- yields all the elements to the outputQueue without checking for the
1311-- Note that no two outputs in the heap can have the same sequence numbers and
1312-- therefore we do not need a stable heap. We have also separated the buffer
1313-- for the current task (outputQueue) and the pending tasks (outputHeap) so
1314-- that the pending tasks cannot interfere with the current task. Note that for
1315-- a single task just the outputQueue is enough and for the case of many
1316-- threads just a heap is good enough. However we balance between these two
1317-- cases, so that both are efficient.
1319-- For bigger streams it may make sense to have separate buffers for each
1320-- stream. However, for singleton streams this may become inefficient. However,
1321-- if we do not have separate buffers, then the streams that come later in
1322-- sequence may hog the buffer, hindering the streams that are ahead. For this
1323-- reason we have a single element buffer limitation for the streams being
1324-- executed in advance.
1326-- This scheme works pretty efficiently with less than 40% extra overhead
1327-- compared to the Async streams where we do not have any kind of sequencing of
1328-- the outputs. It is especially devised so that we are most efficient when we
1329-- have short tasks and need just a single thread. Also when a thread yields
1330-- many items it can hold lockfree access to the outputQueue and do it
1333-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
1334-- that they do not hog the resources and hinder the progress of the threads in
1337-- XXX Left associated ahead expressions are expensive. We start a new SVar for
1338-- each left associative expression. The queue is used only for right
1339-- associated expression, we queue the right expression and execute the left.
1340-- Thererefore the queue never has more than one item in it.
1342-- XXX we can fix this. When we queue more than one item on the queue we can
1343-- mark the previously queued item as not-runnable. The not-runnable item is
1344-- not dequeued until the already running one has finished and at that time we
1345-- would also know the exact sequence number of the already queued item.
1347-- we can even run the already queued items but they will have to be sorted in
1348-- layers in the heap. We can use a list of heaps for that.
1349{-# INLINE enqueueAhead #-}
1350enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> (RunInIO m, t m a) -> IO ()
1351enqueueAhead sv q m = do
1352 atomicModifyIORefCAS_ q $ \ case
1353 ([], n) -> ([snd m], n + 1) -- increment sequence
1354 _ -> error "enqueueAhead: queue is not empty"
1357-- enqueue without incrementing the sequence number
1358{-# INLINE reEnqueueAhead #-}
1359reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1360reEnqueueAhead sv q m = do
1361 atomicModifyIORefCAS_ q $ \ case
1362 ([], n) -> ([m], n) -- DO NOT increment sequence
1363 _ -> error "reEnqueueAhead: queue is not empty"
1366-- Normally the thread that has the token should never go away. The token gets
1367-- handed over to another thread, but someone or the other has the token at any
1368-- point of time. But if the task that has the token finds that the outputQueue
1369-- is full, in that case it can go away without even handing over the token to
1370-- another thread. In that case it sets the nextSequence number in the heap its
1371-- own sequence number before going away. To handle this case, any task that
1372-- does not have the token tries to dequeue from the heap first before
1373-- dequeuing from the work queue. If it finds that the task at the top of the
1374-- heap is the one that owns the current sequence number then it grabs the
1375-- token and starts with that.
1377-- XXX instead of queueing just the head element and the remaining computation
1378-- on the heap, evaluate as many as we can and place them on the heap. But we
1379-- need to give higher priority to the lower sequence numbers so that lower
1380-- priority tasks do not fill up the heap making higher priority tasks block
1381-- due to full heap. Maybe we can have a weighted space for them in the heap.
1382-- The weight is inversely proportional to the sequence number.
1384-- XXX review for livelock
1386{-# INLINE queueEmptyAhead #-}
1387queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
1388queueEmptyAhead q = liftIO $ do
1389 (xs, _) <- readIORef q
1392{-# INLINE dequeueAhead #-}
1393dequeueAhead :: MonadIO m
1394 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
1395dequeueAhead q = liftIO $
1396 atomicModifyIORefCAS q $ \case
1397 ([], n) -> (([], n), Nothing)
1398 (x : [], n) -> (([], n), Just (x, n))
1399 _ -> error "more than one item on queue"
1401-------------------------------------------------------------------------------
1403-------------------------------------------------------------------------------
1405withIORef :: IORef a -> (a -> IO b) -> IO b
1406withIORef ref f = readIORef ref >>= f
1408atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
1409atomicModifyIORef_ ref f =
1410 atomicModifyIORef ref $ \x -> (f x, ())
1412data HeapDequeueResult t m a =
1415 | Ready (Entry Int (AheadHeapEntry t m a))
1417{-# INLINE dequeueFromHeap #-}
1419 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1420 -> IO (HeapDequeueResult t m a)
1421dequeueFromHeap hpVar =
1422 atomicModifyIORef hpVar $ \pair@(hp, snum) ->
1424 Nothing -> (pair, Clearing)
1428 Just (ent@(Entry seqNo _ev), hp') ->
1430 then ((hp', Nothing), Ready ent)
1431 else assert (seqNo >= n) (pair, Waiting n)
1432 Nothing -> (pair, Waiting n)
1434{-# INLINE dequeueFromHeapSeq #-}
1436 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1438 -> IO (HeapDequeueResult t m a)
1439dequeueFromHeapSeq hpVar i =
1440 atomicModifyIORef hpVar $ \(hp, snum) ->
1445 Just (ent@(Entry seqNo _ev), hp') ->
1447 then ((hp', Nothing), Ready ent)
1448 else assert (seqNo >= i) ((hp, Just i), Waiting i)
1449 Nothing -> ((hp, Just i), Waiting i)
1450 Just _ -> error "dequeueFromHeapSeq: unreachable"
1452heapIsSane :: Maybe Int -> Int -> Bool
1453heapIsSane snum seqNo =
1456 Just n -> seqNo >= n
1458{-# INLINE requeueOnHeapTop #-}
1460 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1461 -> Entry Int (AheadHeapEntry t m a)
1464requeueOnHeapTop hpVar ent seqNo =
1465 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1466 assert (heapIsSane snum seqNo) (H.insert ent hp, Just seqNo)
1468{-# INLINE updateHeapSeq #-}
1470 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1473updateHeapSeq hpVar seqNo =
1474 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1475 assert (heapIsSane snum seqNo) (hp, Just seqNo)
1477-------------------------------------------------------------------------------
1479-------------------------------------------------------------------------------
1481-- XXX To be implemented. Use a linked queue like WAsync and put back the
1482-- remaining computation at the back of the queue instead of the heap, and
1483-- increment the sequence number.
1485-------------------------------------------------------------------------------
1486-- Dispatching workers and tracking them
1487-------------------------------------------------------------------------------
1489-- Thread tracking is needed for two reasons:
1491-- 1) Killing threads on exceptions. Threads may not be left to go away by
1492-- themselves because they may run for significant times before going away or
1493-- worse they may be stuck in IO and never go away.
1495-- 2) To know when all threads are done and the stream has ended.
1497{-# NOINLINE addThread #-}
1498addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1500 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
1502-- This is cheaper than modifyThread because we do not have to send a
1503-- outputDoorBell This can make a difference when more workers are being
1505{-# INLINE delThread #-}
1506delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1508 liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
1510-- If present then delete else add. This takes care of out of order add and
1511-- delete i.e. a delete arriving before we even added a thread.
1512-- This occurs when the forked thread is done even before the 'addThread' right
1513-- after the fork gets a chance to run.
1514{-# INLINE modifyThread #-}
1515modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1516modifyThread sv tid = do
1517 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1519 then let new = S.delete tid old in (new, new)
1520 else let new = S.insert tid old in (new, old)
1521 when (null changed) $
1524 void $ tryPutMVar (outputDoorBell sv) ()
1526-- | This is safe even if we are adding more threads concurrently because if
1527-- a child thread is adding another thread then anyway 'workerThreads' will
1529{-# INLINE allThreadsDone #-}
1530allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1531allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1533{-# NOINLINE handleChildException #-}
1534handleChildException :: SVar t m a -> SomeException -> IO ()
1535handleChildException sv e = do
1537 void $ send sv (ChildStop tid (Just e))
1539{-# NOINLINE handleFoldException #-}
1540handleFoldException :: SVar t m a -> SomeException -> IO ()
1541handleFoldException sv e = do
1543 void $ sendToProducer sv (ChildStop tid (Just e))
1545{-# NOINLINE recordMaxWorkers #-}
1546recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1547recordMaxWorkers sv = liftIO $ do
1548 active <- readIORef (workerCount sv)
1549 maxWrk <- readIORef (maxWorkers $ svarStats sv)
1550 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1551 modifyIORef (totalDispatches $ svarStats sv) (+1)
1553{-# NOINLINE pushWorker #-}
1554pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1555pushWorker yieldMax sv = do
1556 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1557 when (svarInspectMode sv) $ recordMaxWorkers sv
1558 -- This allocation matters when significant number of workers are being
1559 -- sent. We allocate it only when needed.
1561 case yieldRateInfo sv of
1562 Nothing -> return Nothing
1563 Just _ -> liftIO $ do
1564 cntRef <- newIORef 0
1565 t <- getTime Monotonic
1566 lat <- newIORef (0, t)
1567 return $ Just WorkerInfo
1568 { workerYieldMax = yieldMax
1569 , workerYieldCount = cntRef
1570 , workerLatencyStart = lat
1572 doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
1575-- XXX we can push the workerCount modification in accountThread and use the
1576-- same pushWorker for Parallel case as well.
1578-- | In contrast to pushWorker which always happens only from the consumer
1579-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1580-- producer side. So we need to use a thread safe modification of
1581-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1582-- using a CAS based modification.
1583{-# INLINE pushWorkerPar #-}
1586 => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1587pushWorkerPar sv wloop =
1588 if svarInspectMode sv
1590 else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
1595 {-# NOINLINE forkWithDiag #-}
1597 -- We do not use workerCount in case of ParallelVar but still there is
1598 -- no harm in maintaining it correctly.
1599 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1601 -- This allocation matters when significant number of workers are being
1602 -- sent. We allocate it only when needed. The overhead increases by 4x.
1604 case yieldRateInfo sv of
1605 Nothing -> return Nothing
1606 Just _ -> liftIO $ do
1607 cntRef <- newIORef 0
1608 t <- getTime Monotonic
1609 lat <- newIORef (0, t)
1610 return $ Just WorkerInfo
1611 { workerYieldMax = 0
1612 , workerYieldCount = cntRef
1613 , workerLatencyStart = lat
1616 doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
1620-- True: can dispatch more
1621-- False: cannot dispatch any more
1622dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1623dispatchWorker yieldCount sv = do
1624 let workerLimit = maxWorkerLimit sv
1625 -- XXX in case of Ahead streams we should not send more than one worker
1626 -- when the work queue is done but heap is not done.
1627 done <- liftIO $ isWorkDone sv
1628 -- Note, "done" may not mean that the work is actually finished if there
1629 -- are workers active, because there may be a worker which has not yet
1630 -- queued the leftover work.
1633 qDone <- liftIO $ isQueueDone sv
1634 -- This count may not be accurate as it is decremented by the workers
1635 -- and we have no synchronization with that decrement.
1636 active <- liftIO $ readIORef $ workerCount sv
1639 -- Note that we may deadlock if the previous workers (tasks in the
1640 -- stream) wait/depend on the future workers (tasks in the stream)
1641 -- executing. In that case we should either configure the maxWorker
1642 -- count to higher or use parallel style instead of ahead or async
1644 limit <- case remainingWork sv of
1645 Nothing -> return workerLimit
1647 n <- liftIO $ readIORef ref
1648 case yieldRateInfo sv of
1649 Just _ -> return workerLimit
1653 Unlimited -> Limited (fromIntegral n)
1654 Limited lim -> Limited $ min lim (fromIntegral n)
1656 -- XXX for ahead streams shall we take the heap yields into account
1657 -- for controlling the dispatch? We should not dispatch if the heap
1658 -- has already got the limit covered.
1659 let dispatch = pushWorker yieldCount sv >> return True
1661 Unlimited -> dispatch
1662 -- Note that the use of remainingWork and workerCount is not
1663 -- atomic and the counts may even have changed between reading
1664 -- and using them here, so this is just approximate logic and
1665 -- we cannot rely on it for correctness. We may actually
1666 -- dispatch more workers than required.
1667 Limited lim | lim > fromIntegral active -> dispatch
1670 when (active <= 0) $ pushWorker 0 sv
1674-------------------------------------------------------------------------------
1675-- Dispatch workers with rate control
1676-------------------------------------------------------------------------------
1678-- | This is a magic number and it is overloaded, and used at several places to
1681-- 1. If we have to sleep to slowdown this is the minimum period that we
1682-- accumulate before we sleep. Also, workers do not stop until this much
1683-- sleep time is accumulated.
1684-- 3. Collected latencies are computed and transferred to measured latency
1685-- after a minimum of this period.
1686minThreadDelay :: NanoSecond64
1687minThreadDelay = 1000000
1689-- | Another magic number! When we have to start more workers to cover up a
1690-- number of yields that we are lagging by then we cannot start one worker for
1691-- each yield because that may be a very big number and if the latency of the
1692-- workers is low these number of yields could be very high. We assume that we
1693-- run each extra worker for at least this much time.
1694rateRecoveryTime :: NanoSecond64
1695rateRecoveryTime = 1000000
1697-- We either block, or send one worker with limited yield count or one or more
1698-- workers with unlimited yield count.
1700 = BlockWait NanoSecond64
1701 | PartialWorker Count
1702 | ManyWorkers Int Count
1705-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1715estimateWorkers workerLimit svarYields gainLossYields
1716 svarElapsed wLatency targetLat range =
1717 -- XXX we can have a maxEfficiency combinator as well which runs the
1718 -- producer at the maximal efficiency i.e. the number of workers are chosen
1719 -- such that the latency is minimum or within a range. Or we can call it
1720 -- maxWorkerLatency.
1723 -- How many workers do we need to achieve the required rate?
1725 -- When the workers are IO bound we can increase the throughput by
1726 -- increasing the number of workers as long as the IO device has enough
1727 -- capacity to process all the requests concurrently. If the IO
1728 -- bandwidth is saturated increasing the workers won't help. Also, if
1729 -- the CPU utilization in processing all these requests exceeds the CPU
1730 -- bandwidth, then increasing the number of workers won't help.
1732 -- When the workers are purely CPU bound, increasing the workers beyond
1733 -- the number of CPUs won't help.
1735 -- TODO - measure the CPU and IO requirements of the workers. Have a
1736 -- way to specify the max bandwidth of the underlying IO mechanism and
1737 -- use that to determine the max rate of workers, and also take the CPU
1738 -- bandwidth into account. We can also discover the IO bandwidth if we
1739 -- know that we are not CPU bound, then how much steady state rate are
1740 -- we able to achieve. Design tests for CPU bound and IO bound cases.
1742 -- Calculate how many yields are we ahead or behind to match the exact
1743 -- required rate. Based on that we increase or decrease the effective
1746 -- When the worker latency is lower than required latency we begin with
1747 -- a yield and then wait rather than first waiting and then yielding.
1748 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1749 effectiveYields = svarYields + gainLossYields
1750 deltaYields = fromIntegral targetYields - effectiveYields
1752 -- We recover the deficit by running at a higher/lower rate for a
1753 -- certain amount of time. To keep the effective rate in reasonable
1754 -- limits we use rateRecoveryTime, minLatency and maxLatency.
1755 in if deltaYields > 0
1757 let deltaYieldsFreq :: Double
1759 fromIntegral deltaYields /
1760 fromIntegral rateRecoveryTime
1761 yieldsFreq = 1.0 / fromIntegral targetLat
1762 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1763 requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
1764 adjustedLat = min (max requiredLat (minLatency range))
1766 in assert (adjustedLat > 0) $
1767 if wLatency <= adjustedLat
1768 then PartialWorker deltaYields
1769 else let workers = withLimit $ wLatency `div` adjustedLat
1770 limited = min workers (fromIntegral deltaYields)
1771 in ManyWorkers (fromIntegral limited) deltaYields
1773 let expectedDuration = fromIntegral effectiveYields * targetLat
1774 sleepTime = expectedDuration - svarElapsed
1775 maxSleepTime = maxLatency range - wLatency
1776 s = min sleepTime maxSleepTime
1777 in assert (sleepTime >= 0) $
1778 -- if s is less than 0 it means our maxSleepTime is less
1779 -- than the worker latency.
1780 if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
1785 Limited x -> min n (fromIntegral x)
1787-- | Get the worker latency without resetting workerPendingLatency
1788-- Returns (total yield count, base time, measured latency)
1789-- CAUTION! keep it in sync with collectLatency
1790getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
1791getWorkerLatency yinfo = do
1792 let cur = workerPendingLatency yinfo
1793 col = workerCollectedLatency yinfo
1794 longTerm = svarAllTimeLatency yinfo
1795 measured = workerMeasuredLatency yinfo
1797 (curTotalCount, curCount, curTime) <- readIORef cur
1798 (colTotalCount, colCount, colTime) <- readIORef col
1799 (lcount, ltime) <- readIORef longTerm
1800 prevLat <- readIORef measured
1802 let latCount = colCount + curCount
1803 latTime = colTime + curTime
1804 totalCount = colTotalCount + curTotalCount
1806 if latCount > 0 && latTime > 0
1807 then let lat = latTime `div` fromIntegral latCount
1808 -- XXX Give more weight to new?
1809 in (lat + prevLat) `div` 2
1811 return (lcount + totalCount, ltime, newLat)
1813isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1814isBeyondMaxRate sv yinfo = do
1815 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1816 now <- getTime Monotonic
1817 let duration = fromRelTime64 $ diffAbsTime64 now tstamp
1818 let targetLat = svarLatencyTarget yinfo
1819 gainLoss <- readIORef (svarGainedLostYields yinfo)
1820 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1821 wLatency targetLat (svarLatencyRange yinfo)
1822 cnt <- readIORef $ workerCount sv
1823 return $ case work of
1824 -- XXX set the worker's maxYields or polling interval based on yields
1825 PartialWorker _yields -> cnt > 1
1826 ManyWorkers n _ -> cnt > n
1829-- XXX in case of ahead style stream we need to take the heap size into account
1830-- because we return the workers on the basis of that which causes a condition
1831-- where we keep dispatching and they keep returning. So we must have exactly
1832-- the same logic for not dispatching and for returning.
1835-- True: can dispatch more
1836-- False: full, no more dispatches
1837dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1838dispatchWorkerPaced sv = do
1839 let yinfo = fromJust $ yieldRateInfo sv
1840 (svarYields, svarElapsed, wLatency) <- do
1841 now <- liftIO $ getTime Monotonic
1842 (yieldCount, baseTime, lat) <-
1843 liftIO $ collectLatency sv yinfo False
1844 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
1847 then fromMaybe lat (workerBootstrapLatency yinfo)
1850 return (yieldCount, elapsed, latency)
1853 -- Need to measure the latency with a single worker before we can perform
1857 let workerLimit = maxWorkerLimit sv
1858 let targetLat = svarLatencyTarget yinfo
1859 let range = svarLatencyRange yinfo
1860 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1861 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
1862 wLatency targetLat range
1864 -- XXX we need to take yieldLimit into account here. If we are at the
1865 -- end of the limit as well as the time, we should not be sleeping.
1866 -- If we are not actually planning to dispatch any more workers we need
1867 -- to take that in account.
1870 assert (s >= 0) (return ())
1871 -- XXX note that when we return from here we will block waiting
1872 -- for the result from the existing worker. If that takes too
1873 -- long we won't be able to send another worker until the
1876 -- Sleep only if there are no active workers, otherwise we will
1877 -- defer the output of those. Note we cannot use workerCount
1878 -- here as it is not a reliable way to ensure there are
1879 -- definitely no active workers. When workerCount is 0 we may
1880 -- still have a Stop event waiting in the outputQueue.
1881 done <- allThreadsDone sv
1882 when done $ void $ do
1883 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
1884 liftIO $ threadDelay (fromIntegral us)
1887 PartialWorker yields -> do
1888 assert (yields > 0) (return ())
1889 updateGainedLostYields yinfo yields
1891 done <- allThreadsDone sv
1892 when done $ void $ dispatchWorker yields sv
1894 ManyWorkers netWorkers yields -> do
1895 assert (netWorkers >= 1) (return ())
1896 assert (yields >= 0) (return ())
1897 updateGainedLostYields yinfo yields
1899 let periodRef = workerPollingInterval yinfo
1900 ycnt = max 1 $ yields `div` fromIntegral netWorkers
1901 period = min ycnt (fromIntegral magicMaxBuffer)
1903 old <- liftIO $ readIORef periodRef
1904 when (period < old) $
1905 liftIO $ writeIORef periodRef period
1907 cnt <- liftIO $ readIORef $ workerCount sv
1910 let total = netWorkers - cnt
1911 batch = max 1 $ fromIntegral $
1912 minThreadDelay `div` targetLat
1913 -- XXX stagger the workers over a period?
1914 -- XXX cannot sleep, as that would mean we cannot process
1915 -- the outputs. need to try a different mechanism to
1917 -- when (total > batch) $
1918 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1919 dispatchN (min total batch)
1924 updateGainedLostYields yinfo yields = do
1925 let buf = fromIntegral $ svarRateBuffer yinfo
1926 when (yields /= 0 && abs yields > buf) $ do
1931 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1937 r <- dispatchWorker 0 sv
1939 then dispatchN (n - 1)
1942-------------------------------------------------------------------------------
1943-- Worker dispatch and wait loop
1944-------------------------------------------------------------------------------
1946sendWorkerDelayPaced :: SVar t m a -> IO ()
1947sendWorkerDelayPaced _ = return ()
1949sendWorkerDelay :: SVar t m a -> IO ()
1950sendWorkerDelay _sv =
1951 -- XXX we need a better way to handle this than hardcoded delays. The
1952 -- delays may be different for different systems.
1953 -- If there is a usecase where this is required we can create a combinator
1954 -- to set it as a config in the state.
1957 ncpu <- getNumCapabilities
1960 if (svarStyle sv == AheadVar)
1961 then threadDelay 100
1964 if (svarStyle sv == AheadVar)
1965 then threadDelay 100
1970{-# NOINLINE sendWorkerWait #-}
1973 => (SVar t m a -> IO ())
1974 -> (SVar t m a -> m Bool)
1977sendWorkerWait delay dispatch sv = do
1978 -- Note that we are guaranteed to have at least one outstanding worker when
1979 -- we enter this function. So if we sleep we are guaranteed to be woken up
1980 -- by an outputDoorBell, when the worker exits.
1983 (_, n) <- liftIO $ readIORef (outputQueue sv)
1985 -- The queue may be empty temporarily if the worker has dequeued the
1986 -- work item but has not enqueued the remaining part yet. For the same
1987 -- reason, a worker may come back if it tries to dequeue and finds the
1988 -- queue empty, even though the whole work has not finished yet.
1990 -- If we find that the queue is empty, but it may be empty
1991 -- temporarily, when we checked it. If that's the case we might
1992 -- sleep indefinitely unless the active workers produce some
1993 -- output. We may deadlock specially if the otuput from the active
1994 -- workers depends on the future workers that we may never send.
1995 -- So in case the queue was temporarily empty set a flag to inform
1996 -- the enqueue to send us a doorbell.
1998 -- Note that this is just a best effort mechanism to avoid a
1999 -- deadlock. Deadlocks may still happen if for some weird reason
2000 -- the consuming computation shares an MVar or some other resource
2001 -- with the producing computation and gets blocked on that resource
2002 -- and therefore cannot do any pushworker to add more threads to
2003 -- the producer. In such cases the programmer should use a parallel
2004 -- style so that all the producers are scheduled immediately and
2005 -- unconditionally. We can also use a separate monitor thread to
2006 -- push workers instead of pushing them from the consumer, but then
2007 -- we are no longer using pull based concurrency rate adaptation.
2009 -- XXX update this in the tutorial.
2011 -- Having pending active workers does not mean that we are guaranteed
2012 -- to be woken up if we sleep. In case of Ahead streams, there may be
2013 -- queued items in the heap even though the outputQueue is empty, and
2014 -- we may have active workers which are deadlocked on those items to be
2015 -- processed by the consumer. We should either guarantee that any
2016 -- worker, before returning, clears the heap or we send a worker to
2017 -- clear it. Normally we always send a worker if no output is seen, but
2018 -- if the thread limit is reached or we are using pacing then we may
2019 -- not send a worker. See the concurrentApplication test in the tests,
2020 -- that test case requires at least one yield from the producer to not
2021 -- deadlock, if the last workers output is stuck in the heap then this
2022 -- test fails. This problem can be extended to n threads when the
2023 -- consumer may depend on the evaluation of next n items in the
2026 -- register for the outputDoorBell before we check the queue so that if
2027 -- we sleep because the queue was empty we are guaranteed to get a
2028 -- doorbell on the next enqueue.
2030 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
2031 liftIO storeLoadBarrier
2032 canDoMore <- dispatch sv
2034 -- XXX test for the case when we miss sending a worker when the worker
2035 -- count is more than 1500.
2037 -- XXX Assert here that if the heap is not empty then there is at
2038 -- least one outstanding worker. Otherwise we could be sleeping
2042 then sendWorkerWait delay dispatch sv
2044 liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
2045 $ takeMVar (outputDoorBell sv)
2046 (_, len) <- liftIO $ readIORef (outputQueue sv)
2047 when (len <= 0) $ sendWorkerWait delay dispatch sv
2049-------------------------------------------------------------------------------
2050-- Reading from the workers' output queue/buffer
2051-------------------------------------------------------------------------------
2053{-# INLINE readOutputQBasic #-}
2054readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)
2055readOutputQBasic q = atomicModifyIORefCAS q $ \x -> (([],0), x)
2057{-# INLINE readOutputQRaw #-}
2058readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
2059readOutputQRaw sv = do
2060 (list, len) <- readOutputQBasic (outputQueue sv)
2061 when (svarInspectMode sv) $ do
2062 let ref = maxOutQSize $ svarStats sv
2063 oqLen <- readIORef ref
2064 when (len > oqLen) $ writeIORef ref len
2067readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2068readOutputQBounded sv = do
2069 (list, len) <- liftIO $ readOutputQRaw sv
2070 -- When there is no output seen we dispatch more workers to help
2071 -- out if there is work pending in the work queue.
2075 -- send a worker proactively, if needed, even before we start
2076 -- processing the output. This may degrade single processor
2077 -- perf but improves multi-processor, because of more
2085 cnt <- liftIO $ readIORef $ workerCount sv
2086 when (cnt <= 0) $ do
2087 done <- liftIO $ isWorkDone sv
2088 when (not done) (pushWorker 0 sv)
2090 {-# INLINE blockingRead #-}
2092 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
2093 liftIO (fst `fmap` readOutputQRaw sv)
2095readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2096readOutputQPaced sv = do
2097 (list, len) <- liftIO $ readOutputQRaw sv
2101 -- XXX send a worker proactively, if needed, even before we start
2102 -- processing the output.
2103 void $ dispatchWorkerPaced sv
2108 {-# INLINE blockingRead #-}
2110 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
2111 liftIO (fst `fmap` readOutputQRaw sv)
2113postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
2114postProcessBounded sv = do
2115 workersDone <- allThreadsDone sv
2116 -- There may still be work pending even if there are no workers pending
2117 -- because all the workers may return if the outputQueue becomes full. In
2118 -- that case send off a worker to kickstart the work again.
2120 -- Note that isWorkDone can only be safely checked if all workers are done.
2121 -- When some workers are in progress they may have decremented the yield
2122 -- Limit and later ending up incrementing it again. If we look at the yield
2123 -- limit in that window we may falsely say that it is 0 and therefore we
2127 r <- liftIO $ isWorkDone sv
2128 -- Note that we need to guarantee a worker, therefore we cannot just
2129 -- use dispatchWorker which may or may not send a worker.
2130 when (not r) (pushWorker 0 sv)
2131 -- XXX do we need to dispatch many here?
2132 -- void $ dispatchWorker sv
2136postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
2137postProcessPaced sv = do
2138 workersDone <- allThreadsDone sv
2139 -- XXX If during consumption we figure out we are getting delayed then we
2140 -- should trigger dispatch there as well. We should try to check on the
2141 -- workers after consuming every n item from the buffer?
2144 r <- liftIO $ isWorkDone sv
2146 void $ dispatchWorkerPaced sv
2147 -- Note that we need to guarantee a worker since the work is not
2148 -- finished, therefore we cannot just rely on dispatchWorkerPaced
2149 -- which may or may not send a worker.
2150 noWorker <- allThreadsDone sv
2151 when noWorker $ pushWorker 0 sv
2155-------------------------------------------------------------------------------
2157-------------------------------------------------------------------------------
2159getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
2160getYieldRateInfo st = do
2161 -- convert rate in Hertz to latency in Nanoseconds
2162 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
2163 case getStreamRate st of
2164 Just (Rate low goal high buf) ->
2165 let l = rateToLatency goal
2166 minl = rateToLatency high
2167 maxl = rateToLatency low
2168 in mkYieldRateInfo l (LatencyRange minl maxl) buf
2169 Nothing -> return Nothing
2173 mkYieldRateInfo latency latRange buf = do
2174 measured <- newIORef 0
2175 wcur <- newIORef (0,0,0)
2176 wcol <- newIORef (0,0,0)
2177 now <- getTime Monotonic
2178 wlong <- newIORef (0,now)
2179 period <- newIORef 1
2180 gainLoss <- newIORef (Count 0)
2182 return $ Just YieldRateInfo
2183 { svarLatencyTarget = latency
2184 , svarLatencyRange = latRange
2185 , svarRateBuffer = buf
2186 , svarGainedLostYields = gainLoss
2187 , workerBootstrapLatency = getStreamLatency st
2188 , workerPollingInterval = period
2189 , workerMeasuredLatency = measured
2190 , workerPendingLatency = wcur
2191 , workerCollectedLatency = wcol
2192 , svarAllTimeLatency = wlong
2195newSVarStats :: IO SVarStats
2198 maxWrk <- newIORef 0
2202 avgLat <- newIORef (0, NanoSecond64 0)
2203 maxLat <- newIORef (NanoSecond64 0)
2204 minLat <- newIORef (NanoSecond64 0)
2205 stpTime <- newIORef Nothing
2208 { totalDispatches = disp
2209 , maxWorkers = maxWrk
2210 , maxOutQSize = maxOq
2211 , maxHeapSize = maxHs
2212 , maxWorkQSize = maxWq
2213 , avgWorkerLatency = avgLat
2214 , minWorkerLatency = minLat
2215 , maxWorkerLatency = maxLat
2216 , svarStopTime = stpTime
2219-- XXX remove polymorphism in t, inline f
2220getAheadSVar :: MonadAsync m
2222 -> ( IORef ([t m a], Int)
2223 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2230getAheadSVar st f mrun = do
2231 outQ <- newIORef ([], 0)
2232 -- the second component of the tuple is "Nothing" when heap is being
2233 -- cleared, "Just n" when we are expecting sequence number n to arrive
2234 -- before we can start clearing the heap.
2235 outH <- newIORef (H.empty, Just 0)
2236 outQMv <- newEmptyMVar
2237 active <- newIORef 0
2238 wfw <- newIORef False
2239 running <- newIORef S.empty
2240 -- Sequence number is incremented whenever something is queued, therefore,
2241 -- first sequence number would be 0
2242 q <- newIORef ([], -1)
2243 stopMVar <- newMVar ()
2244 yl <- case getYieldLimit st of
2245 Nothing -> return Nothing
2246 Just x -> Just <$> newIORef x
2247 rateInfo <- getYieldRateInfo st
2249 stats <- newSVarStats
2252 let getSVar sv readOutput postProc = SVar
2253 { outputQueue = outQ
2254 , outputQueueFromConsumer = undefined
2255 , remainingWork = yl
2256 , maxBufferLimit = getMaxBuffer st
2257 , pushBufferSpace = undefined
2258 , pushBufferPolicy = undefined
2259 , pushBufferMVar = undefined
2260 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
2261 , yieldRateInfo = rateInfo
2262 , outputDoorBell = outQMv
2263 , outputDoorBellFromConsumer = undefined
2264 , readOutputQ = readOutput sv
2265 , postProcess = postProc sv
2266 , workerThreads = running
2267 , workLoop = f q outH st{streamVar = Just sv} sv
2268 , enqueue = enqueueAhead sv q
2269 , isWorkDone = isWorkDoneAhead sv q outH
2270 , isQueueDone = isQueueDoneAhead sv q
2271 , needDoorBell = wfw
2272 , svarStyle = AheadVar
2273 , svarStopStyle = StopNone
2274 , svarStopBy = undefined
2276 , workerCount = active
2277 , accountThread = delThread sv
2278 , workerStopMVar = stopMVar
2280 , svarInspectMode = getInspectMode st
2282 , aheadWorkQueue = q
2288 case getStreamRate st of
2289 Nothing -> getSVar sv readOutputQBounded postProcessBounded
2290 Just _ -> getSVar sv readOutputQPaced postProcessPaced
2295 {-# INLINE isQueueDoneAhead #-}
2296 isQueueDoneAhead sv q = do
2297 queueDone <- checkEmpty q
2299 case remainingWork sv of
2303 Nothing -> return False
2304 -- XXX note that yieldsDone can only be authoritative only when there
2305 -- are no workers running. If there are active workers they can
2306 -- later increment the yield count and therefore change the result.
2307 return $ yieldsDone || queueDone
2309 {-# INLINE isWorkDoneAhead #-}
2310 isWorkDoneAhead sv q ref = do
2312 (hp, _) <- readIORef ref
2313 return (H.size hp <= 0)
2314 queueDone <- isQueueDoneAhead sv q
2315 return $ heapDone && queueDone
2318 (xs, _) <- readIORef q
2321getParallelSVar :: MonadIO m
2322 => SVarStopStyle -> State t m a -> RunInIO m -> IO (SVar t m a)
2323getParallelSVar ss st mrun = do
2324 outQ <- newIORef ([], 0)
2325 outQRev <- newIORef ([], 0)
2326 outQMv <- newEmptyMVar
2327 outQMvRev <- newEmptyMVar
2328 active <- newIORef 0
2329 running <- newIORef S.empty
2330 yl <- case getYieldLimit st of
2331 Nothing -> return Nothing
2332 Just x -> Just <$> newIORef x
2333 rateInfo <- getYieldRateInfo st
2335 case getMaxBuffer st of
2336 Unlimited -> undefined
2337 Limited x -> fromIntegral x
2338 remBuf <- newIORef bufLim
2339 pbMVar <- newMVar ()
2341 stats <- newSVarStats
2346 StopBy -> liftIO $ newIORef undefined
2347 _ -> return undefined
2350 SVar { outputQueue = outQ
2351 , outputQueueFromConsumer = outQRev
2352 , remainingWork = yl
2353 , maxBufferLimit = getMaxBuffer st
2354 , pushBufferSpace = remBuf
2355 , pushBufferPolicy = PushBufferBlock
2356 , pushBufferMVar = pbMVar
2357 , maxWorkerLimit = Unlimited
2358 -- Used only for diagnostics
2359 , yieldRateInfo = rateInfo
2360 , outputDoorBell = outQMv
2361 , outputDoorBellFromConsumer = outQMvRev
2362 , readOutputQ = readOutputQPar sv
2363 , postProcess = allThreadsDone sv
2364 , workerThreads = running
2365 , workLoop = undefined
2366 , enqueue = undefined
2367 , isWorkDone = undefined
2368 , isQueueDone = undefined
2369 , needDoorBell = undefined
2370 , svarStyle = ParallelVar
2371 , svarStopStyle = ss
2372 , svarStopBy = stopBy
2374 , workerCount = active
2375 , accountThread = modifyThread sv
2376 , workerStopMVar = undefined
2378 , svarInspectMode = getInspectMode st
2380 , aheadWorkQueue = undefined
2381 , outputHeap = undefined
2388 readOutputQPar sv = liftIO $ do
2389 withDiagMVar sv "readOutputQPar: doorbell"
2390 $ takeMVar (outputDoorBell sv)
2391 case yieldRateInfo sv of
2392 Nothing -> return ()
2393 Just yinfo -> void $ collectLatency sv yinfo False
2394 r <- fst `fmap` readOutputQRaw sv
2396 void $ tryTakeMVar (pushBufferMVar sv)
2399 void $ tryPutMVar (pushBufferMVar sv) ()
2402sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
2403sendFirstWorker sv m = do
2404 -- Note: We must have all the work on the queue before sending the
2405 -- pushworker, otherwise the pushworker may exit before we even get a
2407 runIn <- captureMonadState
2408 liftIO $ enqueue sv (runIn, m)
2409 case yieldRateInfo sv of
2410 Nothing -> pushWorker 0 sv
2412 if svarLatencyTarget yinfo == maxBound
2413 then liftIO $ threadDelay maxBound
2414 else pushWorker 1 sv
2417{-# INLINABLE newAheadVar #-}
2418newAheadVar :: MonadAsync m
2421 -> ( IORef ([t m a], Int)
2422 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2428newAheadVar st m wloop = do
2429 mrun <- captureMonadState
2430 sv <- liftIO $ getAheadSVar st wloop mrun
2431 sendFirstWorker sv m
2433{-# INLINABLE newParallelVar #-}
2434newParallelVar :: MonadAsync m
2435 => SVarStopStyle -> State t m a -> m (SVar t m a)
2436newParallelVar ss st = do
2437 mrun <- captureMonadState
2438 liftIO $ getParallelSVar ss st mrun
2440-------------------------------------------------------------------------------
2441-- Write a stream to an SVar
2442-------------------------------------------------------------------------------
2444-- XXX this errors out for Parallel/Ahead SVars
2445-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
2446-- be read back from the SVar using 'fromSVar'.
2447toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
2448toStreamVar sv m = do
2449 runIn <- captureMonadState
2450 liftIO $ enqueue sv (runIn, m)
2451 done <- allThreadsDone sv
2452 -- XXX This is safe only when called from the consumer thread or when no
2453 -- consumer is present. There may be a race if we are not running in the
2455 -- XXX do this only if the work queue is not empty. The work may have been
2456 -- carried out by existing workers.
2458 case yieldRateInfo sv of
2459 Nothing -> pushWorker 0 sv
2460 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