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(..)
91 , decrementYieldLimitPost
107import Control.Concurrent
108 (ThreadId, myThreadId, threadDelay, throwTo)
109import Control.Concurrent.MVar
110 (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar, tryReadMVar)
111import Control.Exception
112 (SomeException(..), catch, mask, assert, Exception, catches,
113 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
114 BlockedIndefinitelyOnSTM(..))
115import Control.Monad (when)
116import Control.Monad.Catch (MonadThrow)
117import Control.Monad.IO.Class (MonadIO(..))
118import Control.Monad.Trans.Control (MonadBaseControl, control, StM)
119import Streamly.Atomics
120 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
122import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
123import Data.Functor (void)
124import Data.Heap (Heap, Entry(..))
125import Data.Int (Int64)
127 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
128import Data.List ((\\))
129import Data.Maybe (fromJust)
130import Data.Semigroup ((<>))
132import GHC.Conc (ThreadId(..))
134import GHC.IO (IO(..))
135import Streamly.Time.Clock (Clock(..), getTime)
136import Streamly.Time.Units
137 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
138 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
139import System.IO (hPutStrLn, stderr)
141import qualified Data.Heap as H
142import qualified Data.Set as S
144newtype Count = Count Int64
156------------------------------------------------------------------------------
157-- Parent child thread communication type
158------------------------------------------------------------------------------
160data ThreadAbort = ThreadAbort deriving Show
162instance Exception ThreadAbort
164-- | Events that a child thread may send to a parent thread.
167 | ChildStop ThreadId (Maybe SomeException)
169-- | Sorting out-of-turn outputs in a heap for Ahead style streams
170data AheadHeapEntry (t :: (* -> *) -> * -> *) m a =
173 | AheadEntryStream (t m a)
175------------------------------------------------------------------------------
176-- State threaded around the monad for thread management
177------------------------------------------------------------------------------
179-- | Identify the type of the SVar. Two computations using the same style can
180-- be scheduled on the same SVar.
182 AsyncVar -- depth first concurrent
183 | WAsyncVar -- breadth first concurrent
184 | ParallelVar -- all parallel
185 | AheadVar -- Concurrent look ahead
188-- | An SVar or a Stream Var is a conduit to the output from multiple streams
189-- running concurrently and asynchronously. An SVar can be thought of as an
190-- asynchronous IO handle. We can write any number of streams to an SVar in a
191-- non-blocking manner and then read them back at any time at any pace. The
192-- SVar would run the streams asynchronously and accumulate results. An SVar
193-- may not really execute the stream completely and accumulate all the results.
194-- However, it ensures that the reader can read the results at whatever paces
195-- it wants to read. The SVar monitors and adapts to the consumer's pace.
197-- An SVar is a mini scheduler, it has an associated workLoop that holds the
198-- stream tasks to be picked and run by a pool of worker threads. It has an
199-- associated output queue where the output stream elements are placed by the
200-- worker threads. A outputDoorBell is used by the worker threads to intimate the
201-- consumer thread about availability of new results in the output queue. More
202-- workers are added to the SVar by 'fromStreamVar' on demand if the output
203-- produced is not keeping pace with the consumer. On bounded SVars, workers
204-- block on the output queue to provide throttling of the producer when the
205-- consumer is not pulling fast enough. The number of workers may even get
206-- reduced depending on the consuming pace.
208-- New work is enqueued either at the time of creation of the SVar or as a
209-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
210-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
212-- We measure the individual worker latencies to estimate the number of workers
213-- needed or the amount of time we have to sleep between dispatches to achieve
214-- a particular rate when controlled pace mode it used.
215data WorkerInfo = WorkerInfo
216 { workerYieldMax :: Count -- 0 means unlimited
217 -- total number of yields by the worker till now
218 , workerYieldCount :: IORef Count
219 -- yieldCount at start, timestamp
220 , workerLatencyStart :: IORef (Count, AbsTime)
224-- | Specifies the stream yield rate in yields per second (@Hertz@).
225-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
226-- allow only as many yields as we have accumulated as per 'rateGoal' since the
227-- start of time. If the consumer or the producer is slower or faster, the
228-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
229-- between the two by increasing or decreasing the pull rate from the producer.
230-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
231-- much as 'rateBuffer'.
233-- 'rateLow' puts a bound on how low the instantaneous rate can go when
234-- recovering the rate gap. In other words, it determines the maximum yield
235-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
236-- rate can go when recovering the rate gap. In other words, it determines the
237-- minimum yield latency. We reduce the latency by increasing concurrency,
238-- therefore we can say that it puts an upper bound on concurrency.
240-- If the 'rateGoal' is 0 or negative the stream never yields a value.
241-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
245 { rateLow :: Double -- ^ The lower rate limit
246 , rateGoal :: Double -- ^ The target rate we want to achieve
247 , rateHigh :: Double -- ^ The upper rate limit
248 , rateBuffer :: Int -- ^ Maximum slack from the goal
251data LatencyRange = LatencyRange
252 { minLatency :: NanoSecond64
253 , maxLatency :: NanoSecond64
257data YieldRateInfo = YieldRateInfo
258 { svarLatencyTarget :: NanoSecond64
259 , svarLatencyRange :: LatencyRange
260 , svarRateBuffer :: Int
261 , svarGainedLostYields :: IORef Count
263 -- Actual latency/througput as seen from the consumer side, we count the
264 -- yields and the time it took to generates those yields. This is used to
265 -- increase or decrease the number of workers needed to achieve the desired
266 -- rate. The idle time of workers is adjusted in this, so that we only
267 -- account for the rate when the consumer actually demands data.
268 -- XXX interval latency is enough, we can move this under diagnostics build
269 , svarAllTimeLatency :: IORef (Count, AbsTime)
271 -- XXX Worker latency specified by the user to be used before the first
272 -- actual measurement arrives. Not yet implemented
273 , workerBootstrapLatency :: Maybe NanoSecond64
275 -- After how many yields the worker should update the latency information.
276 -- If the latency is high, this count is kept lower and vice-versa. XXX If
277 -- the latency suddenly becomes too high this count may remain too high for
278 -- long time, in such cases the consumer can change it.
279 -- 0 means no latency computation
280 -- XXX this is derivable from workerMeasuredLatency, can be removed.
281 , workerPollingInterval :: IORef Count
283 -- This is in progress latency stats maintained by the workers which we
284 -- empty into workerCollectedLatency stats at certain intervals - whenever
285 -- we process the stream elements yielded in this period. The first count
286 -- is all yields, the second count is only those yields for which the
287 -- latency was measured to be non-zero (note that if the timer resolution
288 -- is low the measured latency may be zero e.g. on JS platform).
289 -- (allYieldCount, yieldCount, timeTaken)
290 , workerPendingLatency :: IORef (Count, Count, NanoSecond64)
292 -- This is the second level stat which is an accmulation from
293 -- workerPendingLatency stats. We keep accumulating latencies in this
294 -- bucket until we have stats for a sufficient period and then we reset it
295 -- to start collecting for the next period and retain the computed average
296 -- latency for the last period in workerMeasuredLatency.
297 -- (allYieldCount, yieldCount, timeTaken)
298 , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)
300 -- Latency as measured by workers, aggregated for the last period.
301 , workerMeasuredLatency :: IORef NanoSecond64
304data SVarStats = SVarStats {
305 totalDispatches :: IORef Int
306 , maxWorkers :: IORef Int
307 , maxOutQSize :: IORef Int
308 , maxHeapSize :: IORef Int
309 , maxWorkQSize :: IORef Int
310 , avgWorkerLatency :: IORef (Count, NanoSecond64)
311 , minWorkerLatency :: IORef NanoSecond64
312 , maxWorkerLatency :: IORef NanoSecond64
313 , svarStopTime :: IORef (Maybe AbsTime)
316data Limit = Unlimited | Limited Word deriving Show
318-- When to stop the composed stream.
320 StopNone -- stops only when all streams are finished
321 | StopAny -- stop when any stream finishes
322 | StopBy -- stop when a specific stream finishes
325-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have
326-- references to the original SVar stored in several functions which will keep
327-- pointing to the original data and the new updates won't reflect there.
328-- Any updateable parts must be kept in mutable references (IORef).
330data SVar t m a = SVar
333 svarStyle :: SVarStyle
334 , svarMrun :: RunInIO m
335 , svarStopStyle :: SVarStopStyle
336 , svarStopBy :: IORef ThreadId
338 -- Shared output queue (events, length)
339 -- XXX For better efficiency we can try a preallocated array type (perhaps
340 -- something like a vector) that allows an O(1) append. That way we will
341 -- avoid constructing and reversing the list. Possibly we can also avoid
342 -- the GC copying overhead. When the size increases we should be able to
343 -- allocate the array in chunks.
344 , outputQueue :: IORef ([ChildEvent a], Int)
345 , outputDoorBell :: MVar () -- signal the consumer about output
346 , readOutputQ :: m [ChildEvent a]
347 , postProcess :: m Bool
349 -- Combined/aggregate parameters
350 , maxWorkerLimit :: Limit
351 , maxBufferLimit :: Limit
352 , remainingWork :: Maybe (IORef Count)
353 , yieldRateInfo :: Maybe YieldRateInfo
355 -- Used only by bounded SVar types
356 , enqueue :: t m a -> IO ()
357 , isWorkDone :: IO Bool
358 , isQueueDone :: IO Bool
359 , needDoorBell :: IORef Bool
360 , workLoop :: Maybe WorkerInfo -> m ()
362 -- Shared, thread tracking
363 , workerThreads :: IORef (Set ThreadId)
364 , workerCount :: IORef Int
365 , accountThread :: ThreadId -> m ()
366 , workerStopMVar :: MVar ()
368 , svarStats :: SVarStats
369 -- to track garbage collection of SVar
370 , svarRef :: Maybe (IORef ())
372 -- Only for diagnostics
373 , svarInspectMode :: Bool
374 , svarCreator :: ThreadId
375 , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))
377 -- Shared work queue (stream, seqNo)
378 , aheadWorkQueue :: IORef ([t m a], Int)
381-------------------------------------------------------------------------------
382-- State for concurrency control
383-------------------------------------------------------------------------------
385-- XXX we can put the resettable fields in a oneShotConfig field and others in
386-- a persistentConfig field. That way reset would be fast and scalable
387-- irrespective of the number of fields.
389-- XXX make all these Limited types and use phantom types to distinguish them
390data State t m a = State
391 { -- one shot configuration, automatically reset for each API call
392 streamVar :: Maybe (SVar t m a)
393 , _yieldLimit :: Maybe Count
395 -- persistent configuration, state that remains valid until changed by
396 -- an explicit setting via a combinator.
397 , _threadsHigh :: Limit
398 , _bufferHigh :: Limit
399 -- XXX these two can be collapsed into a single type
400 , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency
401 , _maxStreamRate :: Maybe Rate
402 , _inspectMode :: Bool
405-------------------------------------------------------------------------------
406-- State defaults and reset
407-------------------------------------------------------------------------------
409-- A magical value for the buffer size arrived at by running the smallest
410-- possible task and measuring the optimal value of the buffer for that. This
411-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
413magicMaxBuffer :: Word
416defaultMaxThreads, defaultMaxBuffer :: Limit
417defaultMaxThreads = Limited magicMaxBuffer
418defaultMaxBuffer = Limited magicMaxBuffer
420-- The fields prefixed by an _ are not to be accessed or updated directly but
421-- via smart accessor APIs.
422defState :: State t m a
424 { streamVar = Nothing
425 , _yieldLimit = Nothing
426 , _threadsHigh = defaultMaxThreads
427 , _bufferHigh = defaultMaxBuffer
428 , _maxStreamRate = Nothing
429 , _streamLatency = Nothing
430 , _inspectMode = False
433-- XXX if perf gets affected we can have all the Nothing params in a single
434-- structure so that we reset is fast. We can also use rewrite rules such that
435-- reset occurs only in concurrent streams to reduce the impact on serial
437-- We can optimize this so that we clear it only if it is a Just value, it
438-- results in slightly better perf for zip/zipM but the performance of scan
439-- worsens a lot, it does not fuse.
441-- XXX This has a side effect of clearing the SVar and yieldLimit, therefore it
442-- should not be used to convert from the same type to the same type, unless
443-- you want to clear the SVar. For clearing the SVar you should be using the
444-- appropriate unStream functions instead.
446-- | Adapt the stream state from one type to another.
447adaptState :: State t m a -> State t m b
449 { streamVar = Nothing
450 , _yieldLimit = Nothing
453-------------------------------------------------------------------------------
454-- Smart get/set routines for State
455-------------------------------------------------------------------------------
457-- Use get/set routines instead of directly accessing the State fields
458setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
459setYieldLimit lim st =
466 else Just (fromIntegral n)
469getYieldLimit :: State t m a -> Maybe Count
470getYieldLimit = _yieldLimit
472setMaxThreads :: Int -> State t m a -> State t m a
478 then defaultMaxThreads
479 else Limited (fromIntegral n)
482getMaxThreads :: State t m a -> Limit
483getMaxThreads = _threadsHigh
485setMaxBuffer :: Int -> State t m a -> State t m a
491 then defaultMaxBuffer
492 else Limited (fromIntegral n)
495getMaxBuffer :: State t m a -> Limit
496getMaxBuffer = _bufferHigh
498setStreamRate :: Maybe Rate -> State t m a -> State t m a
499setStreamRate r st = st { _maxStreamRate = r }
501getStreamRate :: State t m a -> Maybe Rate
502getStreamRate = _maxStreamRate
504setStreamLatency :: Int -> State t m a -> State t m a
505setStreamLatency n st =
506 st { _streamLatency =
509 else Just (fromIntegral n)
512getStreamLatency :: State t m a -> Maybe NanoSecond64
513getStreamLatency = _streamLatency
515setInspectMode :: State t m a -> State t m a
516setInspectMode st = st { _inspectMode = True }
518getInspectMode :: State t m a -> Bool
519getInspectMode = _inspectMode
521-------------------------------------------------------------------------------
523-------------------------------------------------------------------------------
525cleanupSVar :: SVar t m a -> IO ()
527 workers <- readIORef (workerThreads sv)
528 Prelude.mapM_ (`throwTo` ThreadAbort)
531cleanupSVarFromWorker :: SVar t m a -> IO ()
532cleanupSVarFromWorker sv = do
533 workers <- readIORef (workerThreads sv)
535 mapM_ (`throwTo` ThreadAbort)
536 (S.toList workers \\ [self])
538-------------------------------------------------------------------------------
539-- Worker latency data collection
540-------------------------------------------------------------------------------
542-- Every once in a while workers update the latencies and check the yield rate.
543-- They return if we are above the expected yield rate. If we check too often
544-- it may impact performance, if we check less often we may have a stale
545-- picture. We update every minThreadDelay but we translate that into a yield
546-- count based on latency so that the checking overhead is little.
548-- XXX use a generation count to indicate that the value is updated. If the
549-- value is updated an existing worker must check it again on the next yield.
550-- Otherwise it is possible that we may keep updating it and because of the mod
551-- worker keeps skipping it.
552updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
553updateWorkerPollingInterval yinfo latency = do
554 let periodRef = workerPollingInterval yinfo
555 cnt = max 1 $ minThreadDelay `div` latency
556 period = min cnt (fromIntegral magicMaxBuffer)
558 writeIORef periodRef (fromIntegral period)
560{-# INLINE recordMinMaxLatency #-}
561recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()
562recordMinMaxLatency sv new = do
563 let ss = svarStats sv
564 minLat <- readIORef (minWorkerLatency ss)
565 when (new < minLat || minLat == 0) $
566 writeIORef (minWorkerLatency ss) new
568 maxLat <- readIORef (maxWorkerLatency ss)
569 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
571recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()
572recordAvgLatency sv (count, time) = do
573 let ss = svarStats sv
574 modifyIORef (avgWorkerLatency ss) $
575 \(cnt, t) -> (cnt + count, t + time)
577-- Pour the pending latency stats into a collection bucket
578{-# INLINE collectWorkerPendingLatency #-}
579collectWorkerPendingLatency
580 :: IORef (Count, Count, NanoSecond64)
581 -> IORef (Count, Count, NanoSecond64)
582 -> IO (Count, Maybe (Count, NanoSecond64))
583collectWorkerPendingLatency cur col = do
584 (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
586 (fcnt, cnt, t) <- readIORef col
587 let totalCount = fcnt + fcount
588 latCount = cnt + count
590 writeIORef col (totalCount, latCount, latTime)
592 assert (latCount == 0 || latTime /= 0) (return ())
594 if latCount > 0 && latTime > 0
595 then Just $ (latCount, latTime)
597 return (totalCount, latPair)
599{-# INLINE shouldUseCollectedBatch #-}
600shouldUseCollectedBatch
606shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
607 let r = fromIntegral newLat / fromIntegral prevLat :: Double
608 in (collectedYields > fromIntegral magicMaxBuffer)
609 || (collectedTime > minThreadDelay)
610 || (prevLat > 0 && (r > 2 || r < 0.5))
613-- Returns a triple, (1) yield count since last collection, (2) the base time
614-- when we started counting, (3) average latency in the last measurement
615-- period. The former two are used for accurate measurement of the going rate
616-- whereas the average is used for future estimates e.g. how many workers
617-- should be maintained to maintain the rate.
618-- CAUTION! keep it in sync with getWorkerLatency
619collectLatency :: SVar t m a
622 -> IO (Count, AbsTime, NanoSecond64)
623collectLatency sv yinfo drain = do
624 let cur = workerPendingLatency yinfo
625 col = workerCollectedLatency yinfo
626 longTerm = svarAllTimeLatency yinfo
627 measured = workerMeasuredLatency yinfo
629 (newCount, newLatPair) <- collectWorkerPendingLatency cur col
630 (lcount, ltime) <- readIORef longTerm
631 prevLat <- readIORef measured
633 let newLcount = lcount + newCount
634 retWith lat = return (newLcount, ltime, lat)
637 Nothing -> retWith prevLat
638 Just (count, time) -> do
639 let newLat = time `div` (fromIntegral count)
640 when (svarInspectMode sv) $ recordMinMaxLatency sv newLat
641 -- When we have collected a significant sized batch we compute the
642 -- new latency using that batch and return the new latency,
643 -- otherwise we return the previous latency derived from the
645 if shouldUseCollectedBatch newCount time newLat prevLat || drain
647 -- XXX make this NOINLINE?
648 updateWorkerPollingInterval yinfo (max newLat prevLat)
649 when (svarInspectMode sv) $ recordAvgLatency sv (count, time)
650 writeIORef col (0, 0, 0)
651 writeIORef measured ((prevLat + newLat) `div` 2)
652 modifyIORef longTerm $ \(_, t) -> (newLcount, t)
656-------------------------------------------------------------------------------
657-- Dumping the SVar for debug/diag
658-------------------------------------------------------------------------------
660dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
661dumpSVarStats sv ss style = do
662 case yieldRateInfo sv of
665 _ <- liftIO $ collectLatency sv yinfo True
668 dispatches <- readIORef $ totalDispatches ss
669 maxWrk <- readIORef $ maxWorkers ss
670 maxOq <- readIORef $ maxOutQSize ss
671 maxHp <- readIORef $ maxHeapSize ss
672 minLat <- readIORef $ minWorkerLatency ss
673 maxLat <- readIORef $ maxWorkerLatency ss
674 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
675 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
676 Nothing -> return (0, 0, 0)
678 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
681 t <- readIORef (svarStopTime ss)
682 gl <- readIORef (svarGainedLostYields yinfo)
685 now <- getTime Monotonic
686 let interval = diffAbsTime64 now startTime
687 return (cnt, gl, interval `div` fromIntegral cnt)
689 let interval = diffAbsTime64 stopTime startTime
690 return (cnt, gl, interval `div` fromIntegral cnt)
691 else return (0, 0, 0)
694 [ "total dispatches = " <> show dispatches
695 , "max workers = " <> show maxWrk
696 , "max outQSize = " <> show maxOq
697 <> (if style == AheadVar
698 then "\nheap max size = " <> show maxHp
701 then "\nmin worker latency = " <> showNanoSecond64 minLat
704 then "\nmax worker latency = " <> showNanoSecond64 maxLat
707 then let lat = avgTime `div` fromIntegral avgCnt
708 in "\navg worker latency = " <> showNanoSecond64 lat
711 then "\nSVar latency = " <> showRelTime64 svarLat
714 then "\nSVar yield count = " <> show svarCnt
716 <> (if svarGainLossCnt > 0
717 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
721{-# NOINLINE dumpSVar #-}
722dumpSVar :: SVar t m a -> IO String
724 (oqList, oqLen) <- readIORef $ outputQueue sv
725 db <- tryReadMVar $ outputDoorBell sv
727 if svarStyle sv == AheadVar
729 (oheap, oheapSeq) <- readIORef $ outputHeap sv
730 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
732 [ "heap length = " <> show (H.size oheap)
733 , "heap seqeunce = " <> show oheapSeq
734 , "work queue length = " <> show (length wq)
735 , "work queue sequence = " <> show wqSeq
739 let style = svarStyle sv
741 if style /= ParallelVar
742 then readIORef $ needDoorBell sv
744 rthread <- readIORef $ workerThreads sv
745 workers <- readIORef $ workerCount sv
746 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
750 "Creator tid = " <> show (svarCreator sv),
751 "style = " <> show (svarStyle sv)
752 , "---------CURRENT STATE-----------"
753 , "outputQueue length computed = " <> show (length oqList)
754 , "outputQueue length maintained = " <> show oqLen
755 -- XXX print the types of events in the outputQueue, first 5
756 , "outputDoorBell = " <> show db
760 [ "needDoorBell = " <> show waiting
761 , "running threads = " <> show rthread
762 -- XXX print the status of first 5 threads
763 , "running thread count = " <> show workers
765 <> "---------STATS-----------\n"
768-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
769-- can keep it on in production to debug problems quickly if and when they
770-- happen, but it may result in unexpected output when threads are left hanging
771-- until they are GCed because the consumer went away.
773{-# NOINLINE mvarExcHandler #-}
774mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
775mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
776 svInfo <- dumpSVar sv
777 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
780{-# NOINLINE stmExcHandler #-}
781stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
782stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
783 svInfo <- dumpSVar sv
784 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
787withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
788withDiagMVar sv label action =
789 if svarInspectMode sv
791 action `catches` [ Handler (mvarExcHandler sv label)
792 , Handler (stmExcHandler sv label)
796------------------------------------------------------------------------------
798------------------------------------------------------------------------------
800-- | A monad that can perform concurrent or parallel IO operations. Streams
801-- that can be composed concurrently require the underlying monad to be
805type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
807-- When we run computations concurrently, we completely isolate the state of
808-- the concurrent computations from the parent computation. The invariant is
809-- that we should never be running two concurrent computations in the same
810-- thread without using the runInIO function. Also, we should never be running
811-- a concurrent computation in the parent thread, otherwise it may affect the
812-- state of the parent which is against the defined semantics of concurrent
814newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }
816captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
817captureMonadState = control $ \run -> run (return $ RunInIO run)
819-- Stolen from the async package. The perf improvement is modest, 2% on a
820-- thread heavy benchmark (parallel composition using noop computations).
821-- A version of forkIO that does not include the outer exception
822-- handler: saves a bit of time when we will be installing our own
824{-# INLINE rawForkIO #-}
825rawForkIO :: IO () -> IO ThreadId
826rawForkIO action = IO $ \ s ->
827 case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)
830doFork :: MonadBaseControl IO m
833 -> (SomeException -> IO ())
835doFork action (RunInIO mrun) exHandler =
837 mask $ \restore -> do
838 tid <- rawForkIO $ catch (restore $ void $ mrun action)
842------------------------------------------------------------------------------
843-- Collecting results from child workers in a streamed fashion
844------------------------------------------------------------------------------
846-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
847-- faster, along with the fields in sv required by send?
848-- XXX make it noinline
850-- XXX we may want to employ an increment and decrement in batches when the
851-- througput is high or when the cost of synchronization is high. For example
852-- if the application is distributed then inc/dec of a shared variable may be
855-- Note that we need it to be an Int type so that we have the ability to undo a
856-- decrement that takes below zero.
857{-# INLINE decrementYieldLimit #-}
858decrementYieldLimit :: SVar t m a -> IO Bool
859decrementYieldLimit sv =
860 case remainingWork sv of
861 Nothing -> return True
863 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
866-- decrementYieldLimit returns False when the old limit is 0. This one returns
867-- False when the old limit is 1.
868{-# INLINE decrementYieldLimitPost #-}
869decrementYieldLimitPost :: SVar t m a -> IO Bool
870decrementYieldLimitPost sv =
871 case remainingWork sv of
872 Nothing -> return True
874 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
877{-# INLINE incrementYieldLimit #-}
878incrementYieldLimit :: SVar t m a -> IO ()
879incrementYieldLimit sv =
880 case remainingWork sv of
882 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
884-- XXX exception safety of all atomic/MVar operations
886-- TBD Each worker can have their own queue and the consumer can empty one
887-- queue at a time, that way contention can be reduced.
889-- XXX Only yields should be counted in the buffer limit and not the Stop
892-- | This function is used by the producer threads to queue output for the
893-- consumer thread to consume. Returns whether the queue has more space.
894send :: SVar t m a -> ChildEvent a -> IO Bool
896 -- XXX can the access to outputQueue and maxBufferLimit be made faster
898 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
899 ((msg : es, n + 1), n)
901 -- The wake up must happen only after the store has finished otherwise
902 -- we can have lost wakeup problems.
904 -- Since multiple workers can try this at the same time, it is possible
905 -- that we may put a spurious MVar after the consumer has already seen
906 -- the output. But that's harmless, at worst it may cause the consumer
907 -- to read the queue again and find it empty.
908 -- The important point is that the consumer is guaranteed to receive a
909 -- doorbell if something was added to the queue after it empties it.
910 void $ tryPutMVar (outputDoorBell sv) ()
912 -- XXX we should reserve the buffer when we pick up the work from the
913 -- queue, instead of checking it here when it is too late.
914 let limit = maxBufferLimit sv
916 Unlimited -> return True
918 active <- readIORef (workerCount sv)
919 return $ len < (fromIntegral lim - active)
921workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
922workerCollectLatency winfo = do
923 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
924 cnt1 <- readIORef (workerYieldCount winfo)
925 let cnt = cnt1 - cnt0
929 t1 <- getTime Monotonic
930 let period = fromRelTime64 $ diffAbsTime64 t1 t0
931 writeIORef (workerLatencyStart winfo) (cnt1, t1)
932 return $ Just (cnt, period)
935-- XXX There are a number of gotchas in measuring latencies.
936-- 1) We measure latencies only when a worker yields a value
937-- 2) It is possible that a stream calls the stop continuation, in which case
938-- the worker would not yield a value and we would not account that worker in
939-- latencies. Even though this case should ideally be accounted we do not
940-- account it because we cannot or do not distinguish it from the case
942-- 3) It is possible that a worker returns without yielding anything because it
943-- never got a chance to pick up work.
944-- 4) If the system timer resolution is lower than the latency, the latency
945-- computation turns out to be zero.
947-- We can fix this if we measure the latencies by counting the work items
948-- picked rather than based on the outputs yielded.
949workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
950workerUpdateLatency yinfo winfo = do
951 r <- workerCollectLatency winfo
953 Just (cnt, period) -> do
954 -- NOTE: On JS platform the timer resolution could be pretty low. When
955 -- the timer resolution is low, measurement of latencies could be
956 -- tricky. All the worker latencies will turn out to be zero if they
957 -- are lower than the resolution. We only take into account those
958 -- measurements which are more than the timer resolution.
960 let ref = workerPendingLatency yinfo
961 (cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
962 atomicModifyIORefCAS_ ref $
963 \(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
966updateYieldCount :: WorkerInfo -> IO Count
967updateYieldCount winfo = do
968 cnt <- readIORef (workerYieldCount winfo)
970 writeIORef (workerYieldCount winfo) cnt1
973isBeyondMaxYield :: Count -> WorkerInfo -> Bool
974isBeyondMaxYield cnt winfo =
975 let ymax = workerYieldMax winfo
976 in ymax /= 0 && cnt >= ymax
978-- XXX we should do rate control periodically based on the total yields rather
979-- than based on the worker local yields as other workers may have yielded more
980-- and we should stop based on the aggregate yields. However, latency update
981-- period can be based on individual worker yields.
982{-# NOINLINE checkRatePeriodic #-}
983checkRatePeriodic :: SVar t m a
988checkRatePeriodic sv yinfo winfo ycnt = do
989 i <- readIORef (workerPollingInterval yinfo)
990 -- XXX use generation count to check if the interval has been updated
991 if i /= 0 && (ycnt `mod` i) == 0
993 workerUpdateLatency yinfo winfo
994 -- XXX not required for parallel streams
995 isBeyondMaxRate sv yinfo
998-- CAUTION! this also updates the yield count and therefore should be called
999-- only when we are actually yielding an element.
1000{-# NOINLINE workerRateControl #-}
1001workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
1002workerRateControl sv yinfo winfo = do
1003 cnt <- updateYieldCount winfo
1004 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
1005 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
1007-- XXX we should do rate control here but not latency update in case of ahead
1008-- streams. latency update must be done when we yield directly to outputQueue
1009-- or when we yield to heap.
1010{-# INLINE sendYield #-}
1011sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
1012sendYield sv mwinfo msg = do
1017 case yieldRateInfo sv of
1018 Nothing -> return True
1019 Just yinfo -> workerRateControl sv yinfo winfo
1020 Nothing -> return True
1021 return $ r && rateLimitOk
1023{-# INLINE workerStopUpdate #-}
1024workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
1025workerStopUpdate winfo info = do
1026 i <- readIORef (workerPollingInterval info)
1027 when (i /= 0) $ workerUpdateLatency info winfo
1029{-# INLINABLE sendStop #-}
1030sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
1031sendStop sv mwinfo = do
1032 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
1033 case (mwinfo, yieldRateInfo sv) of
1034 (Just winfo, Just info) ->
1035 workerStopUpdate winfo info
1038 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
1040-------------------------------------------------------------------------------
1042-------------------------------------------------------------------------------
1044{-# INLINE ringDoorBell #-}
1045ringDoorBell :: SVar t m a -> IO ()
1048 w <- readIORef $ needDoorBell sv
1050 -- Note: the sequence of operations is important for correctness here.
1051 -- We need to set the flag to false strictly before sending the
1052 -- outputDoorBell, otherwise the outputDoorBell may get processed too
1053 -- early and then we may set the flag to False to later making the
1054 -- consumer lose the flag, even without receiving a outputDoorBell.
1055 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1056 void $ tryPutMVar (outputDoorBell sv) ()
1058-------------------------------------------------------------------------------
1060-------------------------------------------------------------------------------
1062-- Note: For purely right associated expressions this queue should have at most
1063-- one element. It grows to more than one when we have left associcated
1064-- expressions. Large left associated compositions can grow this to a
1066{-# INLINE enqueueLIFO #-}
1067enqueueLIFO :: SVar t m a -> IORef [t m a] -> t m a -> IO ()
1068enqueueLIFO sv q m = do
1069 atomicModifyIORefCAS_ q $ \ms -> m : ms
1072-------------------------------------------------------------------------------
1074-------------------------------------------------------------------------------
1076-- XXX we can use the Ahead style sequence/heap mechanism to make the best
1077-- effort to always try to finish the streams on the left side of an expression
1078-- first as long as possible.
1080{-# INLINE enqueueFIFO #-}
1081enqueueFIFO :: SVar t m a -> LinkedQueue (t m a) -> t m a -> IO ()
1082enqueueFIFO sv q m = do
1086-------------------------------------------------------------------------------
1088-------------------------------------------------------------------------------
1090-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
1091-- but always serve them in the same order as they appear in the stream. To
1092-- implement lookahead streams efficiently we assign a sequence number to each
1093-- task when the task is picked up for execution. When the task finishes, the
1094-- output is tagged with the same sequence number and we rearrange the outputs
1095-- in sequence based on that number.
1097-- To explain the mechanism imagine that the current task at the head of the
1098-- stream has a "token" to yield to the outputQueue. The ownership of the token
1099-- is determined by the current sequence number is maintained in outputHeap.
1100-- Sequence number is assigned when a task is queued. When a thread dequeues a
1101-- task it picks up the sequence number as well and when the output is ready it
1102-- uses the sequence number to queue the output to the outputQueue.
1104-- The thread with current sequence number sends the output directly to the
1105-- outputQueue. Other threads push the output to the outputHeap. When the task
1106-- being queued on the heap is a stream of many elements we evaluate only the
1107-- first element and keep the rest of the unevaluated computation in the heap.
1108-- When such a task gets the "token" for outputQueue it evaluates and directly
1109-- yields all the elements to the outputQueue without checking for the
1112-- Note that no two outputs in the heap can have the same sequence numbers and
1113-- therefore we do not need a stable heap. We have also separated the buffer
1114-- for the current task (outputQueue) and the pending tasks (outputHeap) so
1115-- that the pending tasks cannot interfere with the current task. Note that for
1116-- a single task just the outputQueue is enough and for the case of many
1117-- threads just a heap is good enough. However we balance between these two
1118-- cases, so that both are efficient.
1120-- For bigger streams it may make sense to have separate buffers for each
1121-- stream. However, for singleton streams this may become inefficient. However,
1122-- if we do not have separate buffers, then the streams that come later in
1123-- sequence may hog the buffer, hindering the streams that are ahead. For this
1124-- reason we have a single element buffer limitation for the streams being
1125-- executed in advance.
1127-- This scheme works pretty efficiently with less than 40% extra overhead
1128-- compared to the Async streams where we do not have any kind of sequencing of
1129-- the outputs. It is especially devised so that we are most efficient when we
1130-- have short tasks and need just a single thread. Also when a thread yields
1131-- many items it can hold lockfree access to the outputQueue and do it
1134-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
1135-- that they do not hog the resources and hinder the progress of the threads in
1138-- XXX Left associated ahead expressions are expensive. We start a new SVar for
1139-- each left associative expression. The queue is used only for right
1140-- associated expression, we queue the right expression and execute the left.
1141-- Thererefore the queue never has more than one item in it.
1143-- XXX we can fix this. When we queue more than one item on the queue we can
1144-- mark the previously queued item as not-runnable. The not-runnable item is
1145-- not dequeued until the already running one has finished and at that time we
1146-- would also know the exact sequence number of the already queued item.
1148-- we can even run the already queued items but they will have to be sorted in
1149-- layers in the heap. We can use a list of heaps for that.
1150{-# INLINE enqueueAhead #-}
1151enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1152enqueueAhead sv q m = do
1153 atomicModifyIORefCAS_ q $ \ case
1154 ([], n) -> ([m], n + 1) -- increment sequence
1155 _ -> error "not empty"
1158-- enqueue without incrementing the sequence number
1159{-# INLINE reEnqueueAhead #-}
1160reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1161reEnqueueAhead sv q m = do
1162 atomicModifyIORefCAS_ q $ \ case
1163 ([], n) -> ([m], n) -- DO NOT increment sequence
1164 _ -> error "not empty"
1167-- Normally the thread that has the token should never go away. The token gets
1168-- handed over to another thread, but someone or the other has the token at any
1169-- point of time. But if the task that has the token finds that the outputQueue
1170-- is full, in that case it can go away without even handing over the token to
1171-- another thread. In that case it sets the nextSequence number in the heap its
1172-- own sequence number before going away. To handle this case, any task that
1173-- does not have the token tries to dequeue from the heap first before
1174-- dequeuing from the work queue. If it finds that the task at the top of the
1175-- heap is the one that owns the current sequence number then it grabs the
1176-- token and starts with that.
1178-- XXX instead of queueing just the head element and the remaining computation
1179-- on the heap, evaluate as many as we can and place them on the heap. But we
1180-- need to give higher priority to the lower sequence numbers so that lower
1181-- priority tasks do not fill up the heap making higher priority tasks block
1182-- due to full heap. Maybe we can have a weighted space for them in the heap.
1183-- The weight is inversely proportional to the sequence number.
1185-- XXX review for livelock
1187{-# INLINE queueEmptyAhead #-}
1188queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
1189queueEmptyAhead q = liftIO $ do
1190 (xs, _) <- readIORef q
1193{-# INLINE dequeueAhead #-}
1194dequeueAhead :: MonadIO m
1195 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
1196dequeueAhead q = liftIO $
1197 atomicModifyIORefCAS q $ \case
1198 ([], n) -> (([], n), Nothing)
1199 (x : [], n) -> (([], n), Just (x, n))
1200 _ -> error "more than one item on queue"
1202-------------------------------------------------------------------------------
1204-------------------------------------------------------------------------------
1206withIORef :: IORef a -> (a -> IO b) -> IO b
1207withIORef ref f = readIORef ref >>= f
1209atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
1210atomicModifyIORef_ ref f =
1211 atomicModifyIORef ref $ \x -> (f x, ())
1213data HeapDequeueResult t m a =
1216 | Ready (Entry Int (AheadHeapEntry t m a))
1218{-# INLINE dequeueFromHeap #-}
1220 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1221 -> IO (HeapDequeueResult t m a)
1222dequeueFromHeap hpVar =
1223 atomicModifyIORef hpVar $ \pair@(hp, snum) ->
1225 Nothing -> (pair, Clearing)
1229 Just (ent@(Entry seqNo _ev), hp') ->
1231 then ((hp', Nothing), Ready ent)
1232 else assert (seqNo >= n) (pair, Waiting n)
1233 Nothing -> (pair, Waiting n)
1235{-# INLINE dequeueFromHeapSeq #-}
1237 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1239 -> IO (HeapDequeueResult t m a)
1240dequeueFromHeapSeq hpVar i =
1241 atomicModifyIORef hpVar $ \(hp, snum) ->
1246 Just (ent@(Entry seqNo _ev), hp') ->
1248 then ((hp', Nothing), Ready ent)
1249 else assert (seqNo >= i) ((hp, Just i), Waiting i)
1250 Nothing -> ((hp, Just i), Waiting i)
1251 Just _ -> error "dequeueFromHeapSeq: unreachable"
1253heapIsSane :: Maybe Int -> Int -> Bool
1254heapIsSane snum seqNo =
1257 Just n -> seqNo >= n
1259{-# INLINE requeueOnHeapTop #-}
1261 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1262 -> Entry Int (AheadHeapEntry t m a)
1265requeueOnHeapTop hpVar ent seqNo =
1266 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1267 assert (heapIsSane snum seqNo) (H.insert ent hp, Just seqNo)
1269{-# INLINE updateHeapSeq #-}
1271 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1274updateHeapSeq hpVar seqNo =
1275 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1276 assert (heapIsSane snum seqNo) (hp, Just seqNo)
1278-------------------------------------------------------------------------------
1280-------------------------------------------------------------------------------
1282-- XXX To be implemented. Use a linked queue like WAsync and put back the
1283-- remaining computation at the back of the queue instead of the heap, and
1284-- increment the sequence number.
1286-------------------------------------------------------------------------------
1287-- Dispatching workers and tracking them
1288-------------------------------------------------------------------------------
1290-- Thread tracking is needed for two reasons:
1292-- 1) Killing threads on exceptions. Threads may not be left to go away by
1293-- themselves because they may run for significant times before going away or
1294-- worse they may be stuck in IO and never go away.
1296-- 2) To know when all threads are done and the stream has ended.
1298{-# NOINLINE addThread #-}
1299addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1301 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
1303-- This is cheaper than modifyThread because we do not have to send a
1304-- outputDoorBell This can make a difference when more workers are being
1306{-# INLINE delThread #-}
1307delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1309 liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
1311-- If present then delete else add. This takes care of out of order add and
1312-- delete i.e. a delete arriving before we even added a thread.
1313-- This occurs when the forked thread is done even before the 'addThread' right
1314-- after the fork gets a chance to run.
1315{-# INLINE modifyThread #-}
1316modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1317modifyThread sv tid = do
1318 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1320 then let new = S.delete tid old in (new, new)
1321 else let new = S.insert tid old in (new, old)
1322 when (null changed) $
1325 void $ tryPutMVar (outputDoorBell sv) ()
1327-- | This is safe even if we are adding more threads concurrently because if
1328-- a child thread is adding another thread then anyway 'workerThreads' will
1330{-# INLINE allThreadsDone #-}
1331allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1332allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1334{-# NOINLINE handleChildException #-}
1335handleChildException :: SVar t m a -> SomeException -> IO ()
1336handleChildException sv e = do
1338 void $ send sv (ChildStop tid (Just e))
1340{-# NOINLINE recordMaxWorkers #-}
1341recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1342recordMaxWorkers sv = liftIO $ do
1343 active <- readIORef (workerCount sv)
1344 maxWrk <- readIORef (maxWorkers $ svarStats sv)
1345 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1346 modifyIORef (totalDispatches $ svarStats sv) (+1)
1348{-# NOINLINE pushWorker #-}
1349pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1350pushWorker yieldMax sv = do
1351 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1352 when (svarInspectMode sv) $ recordMaxWorkers sv
1353 -- This allocation matters when significant number of workers are being
1354 -- sent. We allocate it only when needed.
1356 case yieldRateInfo sv of
1357 Nothing -> return Nothing
1358 Just _ -> liftIO $ do
1359 cntRef <- newIORef 0
1360 t <- getTime Monotonic
1361 lat <- newIORef (0, t)
1362 return $ Just WorkerInfo
1363 { workerYieldMax = yieldMax
1364 , workerYieldCount = cntRef
1365 , workerLatencyStart = lat
1367 doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
1370-- XXX we can push the workerCount modification in accountThread and use the
1371-- same pushWorker for Parallel case as well.
1373-- | In contrast to pushWorker which always happens only from the consumer
1374-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1375-- producer side. So we need to use a thread safe modification of
1376-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1377-- using a CAS based modification.
1378{-# INLINE pushWorkerPar #-}
1381 => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1382pushWorkerPar sv wloop =
1383 if svarInspectMode sv
1385 else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
1390 {-# NOINLINE forkWithDiag #-}
1392 -- We do not use workerCount in case of ParallelVar but still there is
1393 -- no harm in maintaining it correctly.
1394 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1396 -- This allocation matters when significant number of workers are being
1397 -- sent. We allocate it only when needed. The overhead increases by 4x.
1399 case yieldRateInfo sv of
1400 Nothing -> return Nothing
1401 Just _ -> liftIO $ do
1402 cntRef <- newIORef 0
1403 t <- getTime Monotonic
1404 lat <- newIORef (0, t)
1405 return $ Just WorkerInfo
1406 { workerYieldMax = 0
1407 , workerYieldCount = cntRef
1408 , workerLatencyStart = lat
1411 doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
1415-- True: can dispatch more
1416-- False: cannot dispatch any more
1417dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1418dispatchWorker yieldCount sv = do
1419 let workerLimit = maxWorkerLimit sv
1420 -- XXX in case of Ahead streams we should not send more than one worker
1421 -- when the work queue is done but heap is not done.
1422 done <- liftIO $ isWorkDone sv
1423 -- Note, "done" may not mean that the work is actually finished if there
1424 -- are workers active, because there may be a worker which has not yet
1425 -- queued the leftover work.
1428 qDone <- liftIO $ isQueueDone sv
1429 -- Note that the worker count is only decremented during event
1430 -- processing in fromStreamVar and therefore it is safe to read and
1431 -- use it without a lock.
1432 active <- liftIO $ readIORef $ workerCount sv
1435 -- Note that we may deadlock if the previous workers (tasks in the
1436 -- stream) wait/depend on the future workers (tasks in the stream)
1437 -- executing. In that case we should either configure the maxWorker
1438 -- count to higher or use parallel style instead of ahead or async
1440 limit <- case remainingWork sv of
1441 Nothing -> return workerLimit
1443 n <- liftIO $ readIORef ref
1444 case yieldRateInfo sv of
1445 Just _ -> return workerLimit
1449 Unlimited -> Limited (fromIntegral n)
1450 Limited lim -> Limited $ min lim (fromIntegral n)
1452 -- XXX for ahead streams shall we take the heap yields into account
1453 -- for controlling the dispatch? We should not dispatch if the heap
1454 -- has already got the limit covered.
1455 let dispatch = pushWorker yieldCount sv >> return True
1457 Unlimited -> dispatch
1458 -- Note that the use of remainingWork and workerCount is not
1459 -- atomic and the counts may even have changed between reading
1460 -- and using them here, so this is just approximate logic and
1461 -- we cannot rely on it for correctness. We may actually
1462 -- dispatch more workers than required.
1463 Limited lim | lim > fromIntegral active -> dispatch
1466 when (active <= 0) $ pushWorker 0 sv
1470-------------------------------------------------------------------------------
1471-- Dispatch workers with rate control
1472-------------------------------------------------------------------------------
1474-- | This is a magic number and it is overloaded, and used at several places to
1477-- 1. If we have to sleep to slowdown this is the minimum period that we
1478-- accumulate before we sleep. Also, workers do not stop until this much
1479-- sleep time is accumulated.
1480-- 3. Collected latencies are computed and transferred to measured latency
1481-- after a minimum of this period.
1482minThreadDelay :: NanoSecond64
1483minThreadDelay = 1000000
1485-- | Another magic number! When we have to start more workers to cover up a
1486-- number of yields that we are lagging by then we cannot start one worker for
1487-- each yield because that may be a very big number and if the latency of the
1488-- workers is low these number of yields could be very high. We assume that we
1489-- run each extra worker for at least this much time.
1490rateRecoveryTime :: NanoSecond64
1491rateRecoveryTime = 1000000
1493-- We either block, or send one worker with limited yield count or one or more
1494-- workers with unlimited yield count.
1496 = BlockWait NanoSecond64
1497 | PartialWorker Count
1498 | ManyWorkers Int Count
1501-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1511estimateWorkers workerLimit svarYields gainLossYields
1512 svarElapsed wLatency targetLat range =
1513 -- XXX we can have a maxEfficiency combinator as well which runs the
1514 -- producer at the maximal efficiency i.e. the number of workers are chosen
1515 -- such that the latency is minimum or within a range. Or we can call it
1516 -- maxWorkerLatency.
1519 -- How many workers do we need to acheive the required rate?
1521 -- When the workers are IO bound we can increase the throughput by
1522 -- increasing the number of workers as long as the IO device has enough
1523 -- capacity to process all the requests concurrently. If the IO
1524 -- bandwidth is saturated increasing the workers won't help. Also, if
1525 -- the CPU utilization in processing all these requests exceeds the CPU
1526 -- bandwidth, then increasing the number of workers won't help.
1528 -- When the workers are purely CPU bound, increasing the workers beyond
1529 -- the number of CPUs won't help.
1531 -- TODO - measure the CPU and IO requirements of the workers. Have a
1532 -- way to specify the max bandwidth of the underlying IO mechanism and
1533 -- use that to determine the max rate of workers, and also take the CPU
1534 -- bandwidth into account. We can also discover the IO bandwidth if we
1535 -- know that we are not CPU bound, then how much steady state rate are
1536 -- we able to acheive. Design tests for CPU bound and IO bound cases.
1538 -- Calculate how many yields are we ahead or behind to match the exact
1539 -- required rate. Based on that we increase or decrease the effective
1542 -- When the worker latency is lower than required latency we begin with
1543 -- a yield and then wait rather than first waiting and then yielding.
1544 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1545 effectiveYields = svarYields + gainLossYields
1546 deltaYields = fromIntegral targetYields - effectiveYields
1548 -- We recover the deficit by running at a higher/lower rate for a
1549 -- certain amount of time. To keep the effective rate in reasonable
1550 -- limits we use rateRecoveryTime, minLatency and maxLatency.
1551 in if deltaYields > 0
1553 let deltaYieldsFreq :: Double
1555 fromIntegral deltaYields /
1556 fromIntegral rateRecoveryTime
1557 yieldsFreq = 1.0 / fromIntegral targetLat
1558 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1559 requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
1560 adjustedLat = min (max requiredLat (minLatency range))
1562 in assert (adjustedLat > 0) $
1563 if wLatency <= adjustedLat
1564 then PartialWorker deltaYields
1565 else let workers = withLimit $ wLatency `div` adjustedLat
1566 limited = min workers (fromIntegral deltaYields)
1567 in ManyWorkers (fromIntegral limited) deltaYields
1569 let expectedDuration = fromIntegral effectiveYields * targetLat
1570 sleepTime = expectedDuration - svarElapsed
1571 maxSleepTime = maxLatency range - wLatency
1572 s = min sleepTime maxSleepTime
1573 in assert (sleepTime >= 0) $
1574 -- if s is less than 0 it means our maxSleepTime is less
1575 -- than the worker latency.
1576 if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
1581 Limited x -> min n (fromIntegral x)
1583-- | Get the worker latency without resetting workerPendingLatency
1584-- Returns (total yield count, base time, measured latency)
1585-- CAUTION! keep it in sync with collectLatency
1586getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
1587getWorkerLatency yinfo = do
1588 let cur = workerPendingLatency yinfo
1589 col = workerCollectedLatency yinfo
1590 longTerm = svarAllTimeLatency yinfo
1591 measured = workerMeasuredLatency yinfo
1593 (curTotalCount, curCount, curTime) <- readIORef cur
1594 (colTotalCount, colCount, colTime) <- readIORef col
1595 (lcount, ltime) <- readIORef longTerm
1596 prevLat <- readIORef measured
1598 let latCount = colCount + curCount
1599 latTime = colTime + curTime
1600 totalCount = colTotalCount + curTotalCount
1602 if latCount > 0 && latTime > 0
1603 then let lat = latTime `div` fromIntegral latCount
1604 -- XXX Give more weight to new?
1605 in (lat + prevLat) `div` 2
1607 return (lcount + totalCount, ltime, newLat)
1609isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1610isBeyondMaxRate sv yinfo = do
1611 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1612 now <- getTime Monotonic
1613 let duration = fromRelTime64 $ diffAbsTime64 now tstamp
1614 let targetLat = svarLatencyTarget yinfo
1615 gainLoss <- readIORef (svarGainedLostYields yinfo)
1616 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1617 wLatency targetLat (svarLatencyRange yinfo)
1618 cnt <- readIORef $ workerCount sv
1619 return $ case work of
1620 -- XXX set the worker's maxYields or polling interval based on yields
1621 PartialWorker _yields -> cnt > 1
1622 ManyWorkers n _ -> cnt > n
1625-- XXX in case of ahead style stream we need to take the heap size into account
1626-- because we return the workers on the basis of that which causes a condition
1627-- where we keep dispatching and they keep returning. So we must have exactly
1628-- the same logic for not dispatching and for returning.
1631-- True: can dispatch more
1632-- False: full, no more dispatches
1633dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1634dispatchWorkerPaced sv = do
1635 let yinfo = fromJust $ yieldRateInfo sv
1636 (svarYields, svarElapsed, wLatency) <- do
1637 now <- liftIO $ getTime Monotonic
1638 (yieldCount, baseTime, lat) <-
1639 liftIO $ collectLatency sv yinfo False
1640 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
1644 case workerBootstrapLatency yinfo of
1649 return (yieldCount, elapsed, latency)
1652 -- Need to measure the latency with a single worker before we can perform
1656 let workerLimit = maxWorkerLimit sv
1657 let targetLat = svarLatencyTarget yinfo
1658 let range = svarLatencyRange yinfo
1659 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1660 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
1661 wLatency targetLat range
1663 -- XXX we need to take yieldLimit into account here. If we are at the
1664 -- end of the limit as well as the time, we should not be sleeping.
1665 -- If we are not actually planning to dispatch any more workers we need
1666 -- to take that in account.
1669 assert (s >= 0) (return ())
1670 -- XXX note that when we return from here we will block waiting
1671 -- for the result from the existing worker. If that takes too
1672 -- long we won't be able to send another worker until the
1675 -- Sleep only if there are no active workers, otherwise we will
1676 -- defer the output of those. Note we cannot use workerCount
1677 -- here as it is not a reliable way to ensure there are
1678 -- definitely no active workers. When workerCount is 0 we may
1679 -- still have a Stop event waiting in the outputQueue.
1680 done <- allThreadsDone sv
1681 when done $ void $ do
1682 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
1683 liftIO $ threadDelay (fromIntegral us)
1686 PartialWorker yields -> do
1687 assert (yields > 0) (return ())
1688 updateGainedLostYields yinfo yields
1690 done <- allThreadsDone sv
1691 when done $ void $ dispatchWorker yields sv
1693 ManyWorkers netWorkers yields -> do
1694 assert (netWorkers >= 1) (return ())
1695 assert (yields >= 0) (return ())
1696 updateGainedLostYields yinfo yields
1698 let periodRef = workerPollingInterval yinfo
1699 ycnt = max 1 $ yields `div` fromIntegral netWorkers
1700 period = min ycnt (fromIntegral magicMaxBuffer)
1702 old <- liftIO $ readIORef periodRef
1703 when (period < old) $
1704 liftIO $ writeIORef periodRef period
1706 cnt <- liftIO $ readIORef $ workerCount sv
1709 let total = netWorkers - cnt
1710 batch = max 1 $ fromIntegral $
1711 minThreadDelay `div` targetLat
1712 -- XXX stagger the workers over a period?
1713 -- XXX cannot sleep, as that would mean we cannot process
1714 -- the outputs. need to try a different mechanism to
1716 -- when (total > batch) $
1717 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1718 dispatchN (min total batch)
1723 updateGainedLostYields yinfo yields = do
1724 let buf = fromIntegral $ svarRateBuffer yinfo
1725 when (yields /= 0 && abs yields > buf) $ do
1730 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1736 r <- dispatchWorker 0 sv
1738 then dispatchN (n - 1)
1741-------------------------------------------------------------------------------
1742-- Worker dispatch and wait loop
1743-------------------------------------------------------------------------------
1745sendWorkerDelayPaced :: SVar t m a -> IO ()
1746sendWorkerDelayPaced _ = return ()
1748sendWorkerDelay :: SVar t m a -> IO ()
1749sendWorkerDelay _sv =
1750 -- XXX we need a better way to handle this than hardcoded delays. The
1751 -- delays may be different for different systems.
1752 -- If there is a usecase where this is required we can create a combinator
1753 -- to set it as a config in the state.
1756 ncpu <- getNumCapabilities
1759 if (svarStyle sv == AheadVar)
1760 then threadDelay 100
1763 if (svarStyle sv == AheadVar)
1764 then threadDelay 100
1769{-# NOINLINE sendWorkerWait #-}
1772 => (SVar t m a -> IO ())
1773 -> (SVar t m a -> m Bool)
1776sendWorkerWait delay dispatch sv = do
1777 -- Note that we are guaranteed to have at least one outstanding worker when
1778 -- we enter this function. So if we sleep we are guaranteed to be woken up
1779 -- by an outputDoorBell, when the worker exits.
1782 (_, n) <- liftIO $ readIORef (outputQueue sv)
1784 -- The queue may be empty temporarily if the worker has dequeued the
1785 -- work item but has not enqueued the remaining part yet. For the same
1786 -- reason, a worker may come back if it tries to dequeue and finds the
1787 -- queue empty, even though the whole work has not finished yet.
1789 -- If we find that the queue is empty, but it may be empty
1790 -- temporarily, when we checked it. If that's the case we might
1791 -- sleep indefinitely unless the active workers produce some
1792 -- output. We may deadlock specially if the otuput from the active
1793 -- workers depends on the future workers that we may never send.
1794 -- So in case the queue was temporarily empty set a flag to inform
1795 -- the enqueue to send us a doorbell.
1797 -- Note that this is just a best effort mechanism to avoid a
1798 -- deadlock. Deadlocks may still happen if for some weird reason
1799 -- the consuming computation shares an MVar or some other resource
1800 -- with the producing computation and gets blocked on that resource
1801 -- and therefore cannot do any pushworker to add more threads to
1802 -- the producer. In such cases the programmer should use a parallel
1803 -- style so that all the producers are scheduled immediately and
1804 -- unconditionally. We can also use a separate monitor thread to
1805 -- push workers instead of pushing them from the consumer, but then
1806 -- we are no longer using pull based concurrency rate adaptation.
1808 -- XXX update this in the tutorial.
1810 -- Having pending active workers does not mean that we are guaranteed
1811 -- to be woken up if we sleep. In case of Ahead streams, there may be
1812 -- queued items in the heap even though the outputQueue is empty, and
1813 -- we may have active workers which are deadlocked on those items to be
1814 -- processed by the consumer. We should either guarantee that any
1815 -- worker, before returning, clears the heap or we send a worker to
1816 -- clear it. Normally we always send a worker if no output is seen, but
1817 -- if the thread limit is reached or we are using pacing then we may
1818 -- not send a worker. See the concurrentApplication test in the tests,
1819 -- that test case requires at least one yield from the producer to not
1820 -- deadlock, if the last workers output is stuck in the heap then this
1821 -- test fails. This problem can be extended to n threads when the
1822 -- consumer may depend on the evaluation of next n items in the
1825 -- register for the outputDoorBell before we check the queue so that if
1826 -- we sleep because the queue was empty we are guaranteed to get a
1827 -- doorbell on the next enqueue.
1829 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
1830 liftIO storeLoadBarrier
1831 canDoMore <- dispatch sv
1833 -- XXX test for the case when we miss sending a worker when the worker
1834 -- count is more than 1500.
1836 -- XXX Assert here that if the heap is not empty then there is at
1837 -- least one outstanding worker. Otherwise we could be sleeping
1841 then sendWorkerWait delay dispatch sv
1843 liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
1844 $ takeMVar (outputDoorBell sv)
1845 (_, len) <- liftIO $ readIORef (outputQueue sv)
1846 when (len <= 0) $ sendWorkerWait delay dispatch sv
1848-------------------------------------------------------------------------------
1849-- Reading from the workers' output queue/buffer
1850-------------------------------------------------------------------------------
1852{-# INLINE readOutputQRaw #-}
1853readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
1854readOutputQRaw sv = do
1855 (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
1856 when (svarInspectMode sv) $ do
1857 let ref = maxOutQSize $ svarStats sv
1858 oqLen <- readIORef ref
1859 when (len > oqLen) $ writeIORef ref len
1862readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1863readOutputQBounded sv = do
1864 (list, len) <- liftIO $ readOutputQRaw sv
1865 -- When there is no output seen we dispatch more workers to help
1866 -- out if there is work pending in the work queue.
1870 -- send a worker proactively, if needed, even before we start
1871 -- processing the output. This may degrade single processor
1872 -- perf but improves multi-processor, because of more
1880 cnt <- liftIO $ readIORef $ workerCount sv
1881 when (cnt <= 0) $ do
1882 done <- liftIO $ isWorkDone sv
1883 when (not done) (pushWorker 0 sv)
1885 {-# INLINE blockingRead #-}
1887 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
1888 liftIO (fst `fmap` readOutputQRaw sv)
1890readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1891readOutputQPaced sv = do
1892 (list, len) <- liftIO $ readOutputQRaw sv
1896 -- XXX send a worker proactively, if needed, even before we start
1897 -- processing the output.
1898 void $ dispatchWorkerPaced sv
1903 {-# INLINE blockingRead #-}
1905 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
1906 liftIO (fst `fmap` readOutputQRaw sv)
1908postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
1909postProcessBounded sv = do
1910 workersDone <- allThreadsDone sv
1911 -- There may still be work pending even if there are no workers pending
1912 -- because all the workers may return if the outputQueue becomes full. In
1913 -- that case send off a worker to kickstart the work again.
1915 -- Note that isWorkDone can only be safely checked if all workers are done.
1916 -- When some workers are in progress they may have decremented the yield
1917 -- Limit and later ending up incrementing it again. If we look at the yield
1918 -- limit in that window we may falsely say that it is 0 and therefore we
1922 r <- liftIO $ isWorkDone sv
1923 -- Note that we need to guarantee a worker, therefore we cannot just
1924 -- use dispatchWorker which may or may not send a worker.
1925 when (not r) (pushWorker 0 sv)
1926 -- XXX do we need to dispatch many here?
1927 -- void $ dispatchWorker sv
1931postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
1932postProcessPaced sv = do
1933 workersDone <- allThreadsDone sv
1934 -- XXX If during consumption we figure out we are getting delayed then we
1935 -- should trigger dispatch there as well. We should try to check on the
1936 -- workers after consuming every n item from the buffer?
1939 r <- liftIO $ isWorkDone sv
1941 void $ dispatchWorkerPaced sv
1942 -- Note that we need to guarantee a worker since the work is not
1943 -- finished, therefore we cannot just rely on dispatchWorkerPaced
1944 -- which may or may not send a worker.
1945 noWorker <- allThreadsDone sv
1946 when noWorker $ pushWorker 0 sv
1950-------------------------------------------------------------------------------
1952-------------------------------------------------------------------------------
1954getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
1955getYieldRateInfo st = do
1956 -- convert rate in Hertz to latency in Nanoseconds
1957 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
1958 case getStreamRate st of
1959 Just (Rate low goal high buf) ->
1960 let l = rateToLatency goal
1961 minl = rateToLatency high
1962 maxl = rateToLatency low
1963 in mkYieldRateInfo l (LatencyRange minl maxl) buf
1964 Nothing -> return Nothing
1968 mkYieldRateInfo latency latRange buf = do
1969 measured <- newIORef 0
1970 wcur <- newIORef (0,0,0)
1971 wcol <- newIORef (0,0,0)
1972 now <- getTime Monotonic
1973 wlong <- newIORef (0,now)
1974 period <- newIORef 1
1975 gainLoss <- newIORef (Count 0)
1977 return $ Just YieldRateInfo
1978 { svarLatencyTarget = latency
1979 , svarLatencyRange = latRange
1980 , svarRateBuffer = buf
1981 , svarGainedLostYields = gainLoss
1982 , workerBootstrapLatency = getStreamLatency st
1983 , workerPollingInterval = period
1984 , workerMeasuredLatency = measured
1985 , workerPendingLatency = wcur
1986 , workerCollectedLatency = wcol
1987 , svarAllTimeLatency = wlong
1990newSVarStats :: IO SVarStats
1993 maxWrk <- newIORef 0
1997 avgLat <- newIORef (0, NanoSecond64 0)
1998 maxLat <- newIORef (NanoSecond64 0)
1999 minLat <- newIORef (NanoSecond64 0)
2000 stpTime <- newIORef Nothing
2003 { totalDispatches = disp
2004 , maxWorkers = maxWrk
2005 , maxOutQSize = maxOq
2006 , maxHeapSize = maxHs
2007 , maxWorkQSize = maxWq
2008 , avgWorkerLatency = avgLat
2009 , minWorkerLatency = minLat
2010 , maxWorkerLatency = maxLat
2011 , svarStopTime = stpTime
2014-- XXX remove polymorphism in t, inline f
2015getAheadSVar :: MonadAsync m
2017 -> ( IORef ([t m a], Int)
2018 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2025getAheadSVar st f mrun = do
2026 outQ <- newIORef ([], 0)
2027 -- the second component of the tuple is "Nothing" when heap is being
2028 -- cleared, "Just n" when we are expecting sequence number n to arrive
2029 -- before we can start clearing the heap.
2030 outH <- newIORef (H.empty, Just 0)
2031 outQMv <- newEmptyMVar
2032 active <- newIORef 0
2033 wfw <- newIORef False
2034 running <- newIORef S.empty
2035 -- Sequence number is incremented whenever something is queued, therefore,
2036 -- first sequence number would be 0
2037 q <- newIORef ([], -1)
2038 stopMVar <- newMVar ()
2039 yl <- case getYieldLimit st of
2040 Nothing -> return Nothing
2041 Just x -> Just <$> newIORef x
2042 rateInfo <- getYieldRateInfo st
2044 stats <- newSVarStats
2047 let getSVar sv readOutput postProc = SVar
2048 { outputQueue = outQ
2049 , remainingWork = yl
2050 , maxBufferLimit = getMaxBuffer st
2051 , maxWorkerLimit = getMaxThreads st
2052 , yieldRateInfo = rateInfo
2053 , outputDoorBell = outQMv
2054 , readOutputQ = readOutput sv
2055 , postProcess = postProc sv
2056 , workerThreads = running
2057 , workLoop = f q outH st{streamVar = Just sv} sv
2058 , enqueue = enqueueAhead sv q
2059 , isWorkDone = isWorkDoneAhead sv q outH
2060 , isQueueDone = isQueueDoneAhead sv q
2061 , needDoorBell = wfw
2062 , svarStyle = AheadVar
2063 , svarStopStyle = StopNone
2064 , svarStopBy = undefined
2066 , workerCount = active
2067 , accountThread = delThread sv
2068 , workerStopMVar = stopMVar
2070 , svarInspectMode = getInspectMode st
2072 , aheadWorkQueue = q
2078 case getStreamRate st of
2079 Nothing -> getSVar sv readOutputQBounded postProcessBounded
2080 Just _ -> getSVar sv readOutputQPaced postProcessPaced
2085 {-# INLINE isQueueDoneAhead #-}
2086 isQueueDoneAhead sv q = do
2087 queueDone <- checkEmpty q
2089 case remainingWork sv of
2093 Nothing -> return False
2094 -- XXX note that yieldsDone can only be authoritative only when there
2095 -- are no workers running. If there are active workers they can
2096 -- later increment the yield count and therefore change the result.
2097 return $ yieldsDone || queueDone
2099 {-# INLINE isWorkDoneAhead #-}
2100 isWorkDoneAhead sv q ref = do
2102 (hp, _) <- readIORef ref
2103 return (H.size hp <= 0)
2104 queueDone <- isQueueDoneAhead sv q
2105 return $ heapDone && queueDone
2108 (xs, _) <- readIORef q
2111getParallelSVar :: MonadIO m
2112 => SVarStopStyle -> State t m a -> RunInIO m -> IO (SVar t m a)
2113getParallelSVar ss st mrun = do
2114 outQ <- newIORef ([], 0)
2115 outQMv <- newEmptyMVar
2116 active <- newIORef 0
2117 running <- newIORef S.empty
2118 yl <- case getYieldLimit st of
2119 Nothing -> return Nothing
2120 Just x -> Just <$> newIORef x
2121 rateInfo <- getYieldRateInfo st
2123 stats <- newSVarStats
2128 StopBy -> liftIO $ newIORef undefined
2129 _ -> return undefined
2132 SVar { outputQueue = outQ
2133 , remainingWork = yl
2134 , maxBufferLimit = Unlimited
2135 , maxWorkerLimit = Unlimited
2136 -- Used only for diagnostics
2137 , yieldRateInfo = rateInfo
2138 , outputDoorBell = outQMv
2139 , readOutputQ = readOutputQPar sv
2140 , postProcess = allThreadsDone sv
2141 , workerThreads = running
2142 , workLoop = undefined
2143 , enqueue = undefined
2144 , isWorkDone = undefined
2145 , isQueueDone = undefined
2146 , needDoorBell = undefined
2147 , svarStyle = ParallelVar
2148 , svarStopStyle = ss
2149 , svarStopBy = stopBy
2151 , workerCount = active
2152 , accountThread = modifyThread sv
2153 , workerStopMVar = undefined
2155 , svarInspectMode = getInspectMode st
2157 , aheadWorkQueue = undefined
2158 , outputHeap = undefined
2165 readOutputQPar sv = liftIO $ do
2166 withDiagMVar sv "readOutputQPar: doorbell"
2167 $ takeMVar (outputDoorBell sv)
2168 case yieldRateInfo sv of
2169 Nothing -> return ()
2170 Just yinfo -> void $ collectLatency sv yinfo False
2171 fst `fmap` readOutputQRaw sv
2173sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
2174sendFirstWorker sv m = do
2175 -- Note: We must have all the work on the queue before sending the
2176 -- pushworker, otherwise the pushworker may exit before we even get a
2178 liftIO $ enqueue sv m
2179 case yieldRateInfo sv of
2180 Nothing -> pushWorker 0 sv
2182 if svarLatencyTarget yinfo == maxBound
2183 then liftIO $ threadDelay maxBound
2184 else pushWorker 1 sv
2187{-# INLINABLE newAheadVar #-}
2188newAheadVar :: MonadAsync m
2191 -> ( IORef ([t m a], Int)
2192 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2198newAheadVar st m wloop = do
2199 mrun <- captureMonadState
2200 sv <- liftIO $ getAheadSVar st wloop mrun
2201 sendFirstWorker sv m
2203{-# INLINABLE newParallelVar #-}
2204newParallelVar :: MonadAsync m
2205 => SVarStopStyle -> State t m a -> m (SVar t m a)
2206newParallelVar ss st = do
2207 mrun <- captureMonadState
2208 liftIO $ getParallelSVar ss st mrun
2210-------------------------------------------------------------------------------
2211-- Write a stream to an SVar
2212-------------------------------------------------------------------------------
2214-- XXX this errors out for Parallel/Ahead SVars
2215-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
2216-- be read back from the SVar using 'fromSVar'.
2217toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
2218toStreamVar sv m = do
2219 liftIO $ enqueue sv m
2220 done <- allThreadsDone sv
2221 -- XXX This is safe only when called from the consumer thread or when no
2222 -- consumer is present. There may be a race if we are not running in the
2224 -- XXX do this only if the work queue is not empty. The work may have been
2225 -- carried out by existing workers.
2227 case yieldRateInfo sv of
2228 Nothing -> pushWorker 0 sv
2229 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