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
30 -- State threaded around the stream
48 , cleanupSVarFromWorker
56 , atomicModifyIORefCAS
74 , HeapDequeueResult(..)
91 , decrementYieldLimitPost
108import Control.Concurrent
109 (ThreadId, myThreadId, threadDelay, throwTo)
110import Control.Concurrent.MVar
111 (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar, tryReadMVar)
112import Control.Exception
113 (SomeException(..), catch, mask, assert, Exception, catches,
114 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
115 BlockedIndefinitelyOnSTM(..))
116import Control.Monad (when)
117import Control.Monad.Catch (MonadThrow)
118import Control.Monad.IO.Class (MonadIO(..))
119import Control.Monad.Trans.Control (MonadBaseControl, control, StM)
121 (casIORef, readForCAS, peekTicket, atomicModifyIORefCAS_,
122 writeBarrier, storeLoadBarrier)
123import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
124import Data.Functor (void)
125import Data.Heap (Heap, Entry(..))
126import Data.Int (Int64)
128 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
129import Data.List ((\\))
130import Data.Maybe (fromJust)
131import Data.Semigroup ((<>))
133import GHC.Conc (ThreadId(..))
135import GHC.IO (IO(..))
136import System.Clock (TimeSpec, Clock(Monotonic), getTime, toNanoSecs)
137import System.IO (hPutStrLn, stderr)
138import Text.Printf (printf)
140import qualified Data.Heap as H
141import qualified Data.Set as S
143-- Always use signed arithmetic to avoid inadvertant overflows of signed values
144-- on conversion when comparing unsigned quantities with signed.
145newtype NanoSecs = NanoSecs Int64
157newtype Count = Count Int64
169------------------------------------------------------------------------------
170-- Parent child thread communication type
171------------------------------------------------------------------------------
173data ThreadAbort = ThreadAbort deriving Show
175instance Exception ThreadAbort
177-- | Events that a child thread may send to a parent thread.
180 | ChildStop ThreadId (Maybe SomeException)
182-- | Sorting out-of-turn outputs in a heap for Ahead style streams
183data AheadHeapEntry (t :: (* -> *) -> * -> *) m a =
186 | AheadEntryStream (t m a)
188------------------------------------------------------------------------------
189-- State threaded around the monad for thread management
190------------------------------------------------------------------------------
192-- | Identify the type of the SVar. Two computations using the same style can
193-- be scheduled on the same SVar.
195 AsyncVar -- depth first concurrent
196 | WAsyncVar -- breadth first concurrent
197 | ParallelVar -- all parallel
198 | AheadVar -- Concurrent look ahead
201-- | An SVar or a Stream Var is a conduit to the output from multiple streams
202-- running concurrently and asynchronously. An SVar can be thought of as an
203-- asynchronous IO handle. We can write any number of streams to an SVar in a
204-- non-blocking manner and then read them back at any time at any pace. The
205-- SVar would run the streams asynchronously and accumulate results. An SVar
206-- may not really execute the stream completely and accumulate all the results.
207-- However, it ensures that the reader can read the results at whatever paces
208-- it wants to read. The SVar monitors and adapts to the consumer's pace.
210-- An SVar is a mini scheduler, it has an associated workLoop that holds the
211-- stream tasks to be picked and run by a pool of worker threads. It has an
212-- associated output queue where the output stream elements are placed by the
213-- worker threads. A outputDoorBell is used by the worker threads to intimate the
214-- consumer thread about availability of new results in the output queue. More
215-- workers are added to the SVar by 'fromStreamVar' on demand if the output
216-- produced is not keeping pace with the consumer. On bounded SVars, workers
217-- block on the output queue to provide throttling of the producer when the
218-- consumer is not pulling fast enough. The number of workers may even get
219-- reduced depending on the consuming pace.
221-- New work is enqueued either at the time of creation of the SVar or as a
222-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
223-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
225-- We measure the individual worker latencies to estimate the number of workers
226-- needed or the amount of time we have to sleep between dispatches to achieve
227-- a particular rate when controlled pace mode it used.
228data WorkerInfo = WorkerInfo
229 { workerYieldMax :: Count -- 0 means unlimited
230 -- total number of yields by the worker till now
231 , workerYieldCount :: IORef Count
232 -- yieldCount at start, timestamp
233 , workerLatencyStart :: IORef (Count, TimeSpec)
237-- | Specifies the stream yield rate in yields per second (@Hertz@).
238-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
239-- allow only as many yields as we have accumulated as per 'rateGoal' since the
240-- start of time. If the consumer or the producer is slower or faster, the
241-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
242-- between the two by increasing or decreasing the pull rate from the producer.
243-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
244-- much as 'rateBuffer'.
246-- 'rateLow' puts a bound on how low the instantaneous rate can go when
247-- recovering the rate gap. In other words, it determines the maximum yield
248-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
249-- rate can go when recovering the rate gap. In other words, it determines the
250-- minimum yield latency. We reduce the latency by increasing concurrency,
251-- therefore we can say that it puts an upper bound on concurrency.
253-- If the 'rateGoal' is 0 or negative the stream never yields a value.
254-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
258 { rateLow :: Double -- ^ The lower rate limit
259 , rateGoal :: Double -- ^ The target rate we want to achieve
260 , rateHigh :: Double -- ^ The upper rate limit
261 , rateBuffer :: Int -- ^ Maximum slack from the goal
264data LatencyRange = LatencyRange
265 { minLatency :: NanoSecs
266 , maxLatency :: NanoSecs
270data YieldRateInfo = YieldRateInfo
271 { svarLatencyTarget :: NanoSecs
272 , svarLatencyRange :: LatencyRange
273 , svarRateBuffer :: Int
274 , svarGainedLostYields :: IORef Count
276 -- Actual latency/througput as seen from the consumer side, we count the
277 -- yields and the time it took to generates those yields. This is used to
278 -- increase or decrease the number of workers needed to achieve the desired
279 -- rate. The idle time of workers is adjusted in this, so that we only
280 -- account for the rate when the consumer actually demands data.
281 -- XXX interval latency is enough, we can move this under diagnostics build
282 , svarAllTimeLatency :: IORef (Count, TimeSpec)
284 -- XXX Worker latency specified by the user to be used before the first
285 -- actual measurement arrives. Not yet implemented
286 , workerBootstrapLatency :: Maybe NanoSecs
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 , workerPollingInterval :: IORef Count
296 -- This is in progress latency stats maintained by the workers which we
297 -- empty into workerCollectedLatency stats at certain intervals - whenever
298 -- we process the stream elements yielded in this period.
299 -- (yieldCount, timeTaken)
300 , workerPendingLatency :: IORef (Count, NanoSecs)
302 -- This is the second level stat which is an accmulation from
303 -- workerPendingLatency stats. We keep accumulating latencies in this
304 -- bucket until we have stats for a sufficient period and then we reset it
305 -- to start collecting for the next period and retain the computed average
306 -- latency for the last period in workerMeasuredLatency.
307 -- (yieldCount, timeTaken)
308 , workerCollectedLatency :: IORef (Count, NanoSecs)
310 -- Latency as measured by workers, aggregated for the last period.
311 , workerMeasuredLatency :: IORef NanoSecs
314data SVarStats = SVarStats {
315 totalDispatches :: IORef Int
316 , maxWorkers :: IORef Int
317 , maxOutQSize :: IORef Int
318 , maxHeapSize :: IORef Int
319 , maxWorkQSize :: IORef Int
320 , avgWorkerLatency :: IORef (Count, NanoSecs)
321 , minWorkerLatency :: IORef NanoSecs
322 , maxWorkerLatency :: IORef NanoSecs
323 , svarStopTime :: IORef (Maybe TimeSpec)
326data Limit = Unlimited | Limited Word deriving Show
328data SVar t m a = SVar
331 svarStyle :: SVarStyle
332 , svarMrun :: RunInIO m
334 -- Shared output queue (events, length)
335 -- XXX For better efficiency we can try a preallocated array type (perhaps
336 -- something like a vector) that allows an O(1) append. That way we will
337 -- avoid constructing and reversing the list. Possibly we can also avoid
338 -- the GC copying overhead. When the size increases we should be able to
339 -- allocate the array in chunks.
340 , outputQueue :: IORef ([ChildEvent a], Int)
341 , outputDoorBell :: MVar () -- signal the consumer about output
342 , readOutputQ :: m [ChildEvent a]
343 , postProcess :: m Bool
345 -- Combined/aggregate parameters
346 , maxWorkerLimit :: Limit
347 , maxBufferLimit :: Limit
348 , remainingWork :: Maybe (IORef Count)
349 , yieldRateInfo :: Maybe YieldRateInfo
351 -- Used only by bounded SVar types
352 , enqueue :: t m a -> IO ()
353 , isWorkDone :: IO Bool
354 , isQueueDone :: IO Bool
355 , needDoorBell :: IORef Bool
356 , workLoop :: Maybe WorkerInfo -> m ()
358 -- Shared, thread tracking
359 , workerThreads :: IORef (Set ThreadId)
360 , workerCount :: IORef Int
361 , accountThread :: ThreadId -> m ()
362 , workerStopMVar :: MVar ()
364 , svarStats :: SVarStats
365 -- to track garbage collection of SVar
366 , svarRef :: Maybe (IORef ())
368 -- Only for diagnostics
369 , svarInspectMode :: Bool
370 , svarCreator :: ThreadId
371 , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))
373 -- Shared work queue (stream, seqNo)
374 , aheadWorkQueue :: IORef ([t m a], Int)
377-------------------------------------------------------------------------------
378-- State for concurrency control
379-------------------------------------------------------------------------------
381-- XXX we can put the resettable fields in a oneShotConfig field and others in
382-- a persistentConfig field. That way reset would be fast and scalable
383-- irrespective of the number of fields.
385-- XXX make all these Limited types and use phantom types to distinguish them
386data State t m a = State
387 { -- one shot configuration, automatically reset for each API call
388 streamVar :: Maybe (SVar t m a)
389 , _yieldLimit :: Maybe Count
391 -- persistent configuration, state that remains valid until changed by
392 -- an explicit setting via a combinator.
393 , _threadsHigh :: Limit
394 , _bufferHigh :: Limit
395 -- XXX these two can be collapsed into a single type
396 , _streamLatency :: Maybe NanoSecs -- bootstrap latency
397 , _maxStreamRate :: Maybe Rate
398 , _inspectMode :: Bool
401-------------------------------------------------------------------------------
402-- State defaults and reset
403-------------------------------------------------------------------------------
405-- A magical value for the buffer size arrived at by running the smallest
406-- possible task and measuring the optimal value of the buffer for that. This
407-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
409magicMaxBuffer :: Word
412defaultMaxThreads, defaultMaxBuffer :: Limit
413defaultMaxThreads = Limited magicMaxBuffer
414defaultMaxBuffer = Limited magicMaxBuffer
416-- The fields prefixed by an _ are not to be accessed or updated directly but
417-- via smart accessor APIs.
418defState :: State t m a
420 { streamVar = Nothing
421 , _yieldLimit = Nothing
422 , _threadsHigh = defaultMaxThreads
423 , _bufferHigh = defaultMaxBuffer
424 , _maxStreamRate = Nothing
425 , _streamLatency = Nothing
426 , _inspectMode = False
429-- XXX if perf gets affected we can have all the Nothing params in a single
430-- structure so that we reset is fast. We can also use rewrite rules such that
431-- reset occurs only in concurrent streams to reduce the impact on serial
433-- We can optimize this so that we clear it only if it is a Just value, it
434-- results in slightly better perf for zip/zipM but the performance of scan
435-- worsens a lot, it does not fuse.
436rstState :: State t m a -> State t m b
438 { streamVar = Nothing
439 , _yieldLimit = Nothing
442-------------------------------------------------------------------------------
443-- Smart get/set routines for State
444-------------------------------------------------------------------------------
446-- Use get/set routines instead of directly accessing the State fields
447setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
448setYieldLimit lim st =
455 else Just (fromIntegral n)
458getYieldLimit :: State t m a -> Maybe Count
459getYieldLimit = _yieldLimit
461setMaxThreads :: Int -> State t m a -> State t m a
467 then defaultMaxThreads
468 else Limited (fromIntegral n)
471getMaxThreads :: State t m a -> Limit
472getMaxThreads = _threadsHigh
474setMaxBuffer :: Int -> State t m a -> State t m a
480 then defaultMaxBuffer
481 else Limited (fromIntegral n)
484getMaxBuffer :: State t m a -> Limit
485getMaxBuffer = _bufferHigh
487setStreamRate :: Maybe Rate -> State t m a -> State t m a
488setStreamRate r st = st { _maxStreamRate = r }
490getStreamRate :: State t m a -> Maybe Rate
491getStreamRate = _maxStreamRate
493setStreamLatency :: Int -> State t m a -> State t m a
494setStreamLatency n st =
495 st { _streamLatency =
498 else Just (fromIntegral n)
501getStreamLatency :: State t m a -> Maybe NanoSecs
502getStreamLatency = _streamLatency
504setInspectMode :: State t m a -> State t m a
505setInspectMode st = st { _inspectMode = True }
507getInspectMode :: State t m a -> Bool
508getInspectMode = _inspectMode
510-------------------------------------------------------------------------------
512-------------------------------------------------------------------------------
514cleanupSVar :: SVar t m a -> IO ()
516 workers <- readIORef (workerThreads sv)
517 Prelude.mapM_ (`throwTo` ThreadAbort)
520cleanupSVarFromWorker :: SVar t m a -> IO ()
521cleanupSVarFromWorker sv = do
522 workers <- readIORef (workerThreads sv)
524 mapM_ (`throwTo` ThreadAbort)
525 (S.toList workers \\ [self])
527-------------------------------------------------------------------------------
528-- Dumping the SVar for debug/diag
529-------------------------------------------------------------------------------
531-- | Convert a number of seconds to a string. The string will consist
532-- of four decimal places, followed by a short description of the time
534secs :: Double -> String
536 | k < 0 = '-' : secs (-k)
537 | k >= 1 = k `with` "s"
538 | k >= 1e-3 = (k*1e3) `with` "ms"
539#ifdef mingw32_HOST_OS
540 | k >= 1e-6 = (k*1e6) `with` "us"
542 | k >= 1e-6 = (k*1e6) `with` "μs"
544 | k >= 1e-9 = (k*1e9) `with` "ns"
545 | k >= 1e-12 = (k*1e12) `with` "ps"
546 | k >= 1e-15 = (k*1e15) `with` "fs"
547 | k >= 1e-18 = (k*1e18) `with` "as"
548 | otherwise = printf "%g s" k
549 where with (t :: Double) (u :: String)
550 | t >= 1e9 = printf "%.4g %s" t u
551 | t >= 1e3 = printf "%.0f %s" t u
552 | t >= 1e2 = printf "%.1f %s" t u
553 | t >= 1e1 = printf "%.2f %s" t u
554 | otherwise = printf "%.3f %s" t u
556-- XXX Code duplicated from collectLatency
557drainLatency :: SVar t m a -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
558drainLatency sv yinfo = do
559 let cur = workerPendingLatency yinfo
560 col = workerCollectedLatency yinfo
561 longTerm = svarAllTimeLatency yinfo
562 measured = workerMeasuredLatency yinfo
564 (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
565 (colCount, colTime) <- readIORef col
566 (lcount, ltime) <- readIORef longTerm
567 prev <- readIORef measured
569 let pendingCount = colCount + count
570 pendingTime = colTime + time
572 lcount' = lcount + pendingCount
573 notUpdated = (lcount', ltime, prev)
575 if (pendingCount > 0)
577 let new = pendingTime `div` (fromIntegral pendingCount)
578 when (svarInspectMode sv) $ do
579 let ss = svarStats sv
580 minLat <- readIORef (minWorkerLatency ss)
581 when (new < minLat || minLat == 0) $
582 writeIORef (minWorkerLatency ss) new
584 maxLat <- readIORef (maxWorkerLatency ss)
585 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
586 modifyIORef (avgWorkerLatency ss) $
587 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
588 -- To avoid minor fluctuations update in batches
589 writeIORef col (0, 0)
590 writeIORef measured new
591 modifyIORef longTerm $ \(_, t) -> (lcount', t)
592 return (lcount', ltime, new)
593 else return notUpdated
595dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
596dumpSVarStats sv ss style = do
597 case yieldRateInfo sv of
600 _ <- liftIO $ drainLatency sv yinfo
603 dispatches <- readIORef $ totalDispatches ss
604 maxWrk <- readIORef $ maxWorkers ss
605 maxOq <- readIORef $ maxOutQSize ss
606 maxHp <- readIORef $ maxHeapSize ss
607 minLat <- readIORef $ minWorkerLatency ss
608 maxLat <- readIORef $ maxWorkerLatency ss
609 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
610 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
611 Nothing -> return (0, 0, 0)
613 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
616 t <- readIORef (svarStopTime ss)
617 gl <- readIORef (svarGainedLostYields yinfo)
620 now <- getTime Monotonic
621 let interval = toNanoSecs (now - startTime)
622 return (cnt, gl, interval `div` fromIntegral cnt)
624 let interval = toNanoSecs (stopTime - startTime)
625 return (cnt, gl, interval `div` fromIntegral cnt)
626 else return (0, 0, 0)
629 [ "total dispatches = " <> show dispatches
630 , "max workers = " <> show maxWrk
631 , "max outQSize = " <> show maxOq
632 <> (if style == AheadVar
633 then "\nheap max size = " <> show maxHp
636 then "\nmin worker latency = "
637 <> secs (fromIntegral minLat * 1e-9)
640 then "\nmax worker latency = "
641 <> secs (fromIntegral maxLat * 1e-9)
644 then let lat = avgTime `div` fromIntegral avgCnt
645 in "\navg worker latency = "
646 <> secs (fromIntegral lat * 1e-9)
649 then "\nSVar latency = "
650 <> secs (fromIntegral svarLat * 1e-9)
653 then "\nSVar yield count = " <> show svarCnt
655 <> (if svarGainLossCnt > 0
656 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
660{-# NOINLINE dumpSVar #-}
661dumpSVar :: SVar t m a -> IO String
663 (oqList, oqLen) <- readIORef $ outputQueue sv
664 db <- tryReadMVar $ outputDoorBell sv
666 if svarStyle sv == AheadVar
668 (oheap, oheapSeq) <- readIORef $ outputHeap sv
669 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
671 [ "heap length = " <> show (H.size oheap)
672 , "heap seqeunce = " <> show oheapSeq
673 , "work queue length = " <> show (length wq)
674 , "work queue sequence = " <> show wqSeq
678 let style = svarStyle sv
680 if style /= ParallelVar
681 then readIORef $ needDoorBell sv
683 rthread <- readIORef $ workerThreads sv
684 workers <- readIORef $ workerCount sv
685 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
689 "Creator tid = " <> show (svarCreator sv),
690 "style = " <> show (svarStyle sv)
691 , "---------CURRENT STATE-----------"
692 , "outputQueue length computed = " <> show (length oqList)
693 , "outputQueue length maintained = " <> show oqLen
694 -- XXX print the types of events in the outputQueue, first 5
695 , "outputDoorBell = " <> show db
699 [ "needDoorBell = " <> show waiting
700 , "running threads = " <> show rthread
701 -- XXX print the status of first 5 threads
702 , "running thread count = " <> show workers
704 <> "---------STATS-----------\n"
707-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
708-- can keep it on in production to debug problems quickly if and when they
709-- happen, but it may result in unexpected output when threads are left hanging
710-- until they are GCed because the consumer went away.
712{-# NOINLINE mvarExcHandler #-}
713mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
714mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
715 svInfo <- dumpSVar sv
716 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
719{-# NOINLINE stmExcHandler #-}
720stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
721stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
722 svInfo <- dumpSVar sv
723 hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
726withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
727withDiagMVar sv label action =
728 if svarInspectMode sv
730 action `catches` [ Handler (mvarExcHandler sv label)
731 , Handler (stmExcHandler sv label)
735-------------------------------------------------------------------------------
737-------------------------------------------------------------------------------
739-- Slightly faster version of CAS. Gained some improvement by avoiding the use
740-- of "evaluate" because we know we do not have exceptions in fn.
741{-# INLINE atomicModifyIORefCAS #-}
742atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
743atomicModifyIORefCAS ref fn = do
744 tkt <- readForCAS ref
750 loop _ 0 = atomicModifyIORef ref fn
752 let (new, result) = fn $ peekTicket old
753 (success, tkt) <- casIORef ref old new
756 else loop tkt (tries - 1)
758{-# INLINE ringDoorBell #-}
759ringDoorBell :: SVar t m a -> IO ()
762 w <- readIORef $ needDoorBell sv
764 -- Note: the sequence of operations is important for correctness here.
765 -- We need to set the flag to false strictly before sending the
766 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
767 -- then we may set the flag to False to later making the consumer lose
768 -- the flag, even without receiving a outputDoorBell.
769 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
770 void $ tryPutMVar (outputDoorBell sv) ()
773------------------------------------------------------------------------------
774-- Spawning threads and collecting result in streamed fashion
775------------------------------------------------------------------------------
777-- | A monad that can perform concurrent or parallel IO operations. Streams
778-- that can be composed concurrently require the underlying monad to be
782type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
784-- When we run computations concurrently, we completely isolate the state of
785-- the concurrent computations from the parent computation. The invariant is
786-- that we should never be running two concurrent computations in the same
787-- thread without using the runInIO function. Also, we should never be running
788-- a concurrent computation in the parent thread, otherwise it may affect the
789-- state of the parent which is against the defined semantics of concurrent
791newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }
793captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
794captureMonadState = control $ \run -> run (return $ RunInIO run)
796-- Stolen from the async package. The perf improvement is modest, 2% on a
797-- thread heavy benchmark (parallel composition using noop computations).
798-- A version of forkIO that does not include the outer exception
799-- handler: saves a bit of time when we will be installing our own
801{-# INLINE rawForkIO #-}
802rawForkIO :: IO () -> IO ThreadId
803rawForkIO action = IO $ \ s ->
804 case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)
807doFork :: MonadBaseControl IO m
810 -> (SomeException -> IO ())
812doFork action (RunInIO mrun) exHandler =
814 mask $ \restore -> do
815 tid <- rawForkIO $ catch (restore $ void $ mrun action)
819-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
820-- faster, along with the fields in sv required by send?
821-- XXX make it noinline
823-- XXX we may want to employ an increment and decrement in batches when the
824-- througput is high or when the cost of synchronization is high. For example
825-- if the application is distributed then inc/dec of a shared variable may be
828-- Note that we need it to be an Int type so that we have the ability to undo a
829-- decrement that takes below zero.
830{-# INLINE decrementYieldLimit #-}
831decrementYieldLimit :: SVar t m a -> IO Bool
832decrementYieldLimit sv =
833 case remainingWork sv of
834 Nothing -> return True
836 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
839-- decrementYieldLimit returns False when the old limit is 0. This one returns
840-- False when the old limit is 1.
841{-# INLINE decrementYieldLimitPost #-}
842decrementYieldLimitPost :: SVar t m a -> IO Bool
843decrementYieldLimitPost sv =
844 case remainingWork sv of
845 Nothing -> return True
847 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
850{-# INLINE incrementYieldLimit #-}
851incrementYieldLimit :: SVar t m a -> IO ()
852incrementYieldLimit sv =
853 case remainingWork sv of
855 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
857-- XXX exception safety of all atomic/MVar operations
859-- TBD Each worker can have their own queue and the consumer can empty one
860-- queue at a time, that way contention can be reduced.
862-- XXX Only yields should be counted in the buffer limit and not the Stop
865-- | This function is used by the producer threads to queue output for the
866-- consumer thread to consume. Returns whether the queue has more space.
867send :: SVar t m a -> ChildEvent a -> IO Bool
869 -- XXX can the access to outputQueue and maxBufferLimit be made faster
871 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
872 ((msg : es, n + 1), n)
874 -- The wake up must happen only after the store has finished otherwise
875 -- we can have lost wakeup problems.
877 -- Since multiple workers can try this at the same time, it is possible
878 -- that we may put a spurious MVar after the consumer has already seen
879 -- the output. But that's harmless, at worst it may cause the consumer
880 -- to read the queue again and find it empty.
881 -- The important point is that the consumer is guaranteed to receive a
882 -- doorbell if something was added to the queue after it empties it.
883 void $ tryPutMVar (outputDoorBell sv) ()
885 -- XXX we should reserve the buffer when we pick up the work from the
886 -- queue, instead of checking it here when it is too late.
887 let limit = maxBufferLimit sv
889 Unlimited -> return True
891 active <- readIORef (workerCount sv)
892 return $ len < (fromIntegral lim - active)
894workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecs))
895workerCollectLatency winfo = do
896 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
897 cnt1 <- readIORef (workerYieldCount winfo)
898 let cnt = cnt1 - cnt0
902 t1 <- getTime Monotonic
903 let period = fromInteger $ toNanoSecs (t1 - t0)
904 writeIORef (workerLatencyStart winfo) (cnt1, t1)
905 return $ Just (cnt, period)
908-- XXX There are a number of gotchas in measuring latencies.
909-- 1) We measure latencies only when a worker yields a value
910-- 2) It is possible that a stream calls the stop continuation, in which case
911-- the worker would not yield a value and we would not account that worker in
912-- latencies. Even though this case should ideally be accounted we do not
913-- account it because we cannot or do not distinguish it from the case
915-- 3) It is possible that a worker returns without yielding anything because it
916-- never got a chance to pick up work.
918-- We can fix this if we measure the latencies by counting the work items
919-- picked rather than based on the outputs yielded.
920workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
921workerUpdateLatency yinfo winfo = do
922 r <- workerCollectLatency winfo
924 Just (cnt, period) -> do
925 let ref = workerPendingLatency yinfo
926 atomicModifyIORefCAS_ ref $ \(n, t) -> (n + cnt, t + period)
929updateYieldCount :: WorkerInfo -> IO Count
930updateYieldCount winfo = do
931 cnt <- readIORef (workerYieldCount winfo)
933 writeIORef (workerYieldCount winfo) cnt1
936isBeyondMaxYield :: Count -> WorkerInfo -> Bool
937isBeyondMaxYield cnt winfo =
938 let ymax = workerYieldMax winfo
939 in ymax /= 0 && cnt >= ymax
941-- XXX we should do rate control periodically based on the total yields rather
942-- than based on the worker local yields as other workers may have yielded more
943-- and we should stop based on the aggregate yields. However, latency update
944-- period can be based on individual worker yields.
945{-# NOINLINE checkRatePeriodic #-}
946checkRatePeriodic :: SVar t m a
951checkRatePeriodic sv yinfo winfo ycnt = do
952 i <- readIORef (workerPollingInterval yinfo)
953 -- XXX use generation count to check if the interval has been updated
954 if i /= 0 && (ycnt `mod` i) == 0
956 workerUpdateLatency yinfo winfo
957 -- XXX not required for parallel streams
958 isBeyondMaxRate sv yinfo
961-- CAUTION! this also updates the yield count and therefore should be called
962-- only when we are actually yielding an element.
963{-# NOINLINE workerRateControl #-}
964workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
965workerRateControl sv yinfo winfo = do
966 cnt <- updateYieldCount winfo
967 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
968 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
970-- XXX we should do rate control here but not latency update in case of ahead
971-- streams. latency update must be done when we yield directly to outputQueue
972-- or when we yield to heap.
973{-# INLINE sendYield #-}
974sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
975sendYield sv mwinfo msg = do
980 case yieldRateInfo sv of
981 Nothing -> return True
982 Just yinfo -> workerRateControl sv yinfo winfo
983 Nothing -> return True
984 return $ r && rateLimitOk
986{-# INLINE workerStopUpdate #-}
987workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
988workerStopUpdate winfo info = do
989 i <- readIORef (workerPollingInterval info)
990 when (i /= 0) $ workerUpdateLatency info winfo
992{-# INLINABLE sendStop #-}
993sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
994sendStop sv mwinfo = do
995 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
996 case (mwinfo, yieldRateInfo sv) of
997 (Just winfo, Just info) ->
998 workerStopUpdate winfo info
1001 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
1003-------------------------------------------------------------------------------
1005-------------------------------------------------------------------------------
1007-- Note: For purely right associated expressions this queue should have at most
1008-- one element. It grows to more than one when we have left associcated
1009-- expressions. Large left associated compositions can grow this to a
1011{-# INLINE enqueueLIFO #-}
1012enqueueLIFO :: SVar t m a -> IORef [t m a] -> t m a -> IO ()
1013enqueueLIFO sv q m = do
1014 atomicModifyIORefCAS_ q $ \ms -> m : ms
1017-------------------------------------------------------------------------------
1019-------------------------------------------------------------------------------
1021-- XXX we can use the Ahead style sequence/heap mechanism to make the best
1022-- effort to always try to finish the streams on the left side of an expression
1023-- first as long as possible.
1025{-# INLINE enqueueFIFO #-}
1026enqueueFIFO :: SVar t m a -> LinkedQueue (t m a) -> t m a -> IO ()
1027enqueueFIFO sv q m = do
1031-------------------------------------------------------------------------------
1033-------------------------------------------------------------------------------
1035-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
1036-- but always serve them in the same order as they appear in the stream. To
1037-- implement lookahead streams efficiently we assign a sequence number to each
1038-- task when the task is picked up for execution. When the task finishes, the
1039-- output is tagged with the same sequence number and we rearrange the outputs
1040-- in sequence based on that number.
1042-- To explain the mechanism imagine that the current task at the head of the
1043-- stream has a "token" to yield to the outputQueue. The ownership of the token
1044-- is determined by the current sequence number is maintained in outputHeap.
1045-- Sequence number is assigned when a task is queued. When a thread dequeues a
1046-- task it picks up the sequence number as well and when the output is ready it
1047-- uses the sequence number to queue the output to the outputQueue.
1049-- The thread with current sequence number sends the output directly to the
1050-- outputQueue. Other threads push the output to the outputHeap. When the task
1051-- being queued on the heap is a stream of many elements we evaluate only the
1052-- first element and keep the rest of the unevaluated computation in the heap.
1053-- When such a task gets the "token" for outputQueue it evaluates and directly
1054-- yields all the elements to the outputQueue without checking for the
1057-- Note that no two outputs in the heap can have the same sequence numbers and
1058-- therefore we do not need a stable heap. We have also separated the buffer
1059-- for the current task (outputQueue) and the pending tasks (outputHeap) so
1060-- that the pending tasks cannot interfere with the current task. Note that for
1061-- a single task just the outputQueue is enough and for the case of many
1062-- threads just a heap is good enough. However we balance between these two
1063-- cases, so that both are efficient.
1065-- For bigger streams it may make sense to have separate buffers for each
1066-- stream. However, for singleton streams this may become inefficient. However,
1067-- if we do not have separate buffers, then the streams that come later in
1068-- sequence may hog the buffer, hindering the streams that are ahead. For this
1069-- reason we have a single element buffer limitation for the streams being
1070-- executed in advance.
1072-- This scheme works pretty efficiently with less than 40% extra overhead
1073-- compared to the Async streams where we do not have any kind of sequencing of
1074-- the outputs. It is especially devised so that we are most efficient when we
1075-- have short tasks and need just a single thread. Also when a thread yields
1076-- many items it can hold lockfree access to the outputQueue and do it
1079-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
1080-- that they do not hog the resources and hinder the progress of the threads in
1083-- XXX Left associated ahead expressions are expensive. We start a new SVar for
1084-- each left associative expression. The queue is used only for right
1085-- associated expression, we queue the right expression and execute the left.
1086-- Thererefore the queue never has more than one item in it.
1088-- XXX we can fix this. When we queue more than one item on the queue we can
1089-- mark the previously queued item as not-runnable. The not-runnable item is
1090-- not dequeued until the already running one has finished and at that time we
1091-- would also know the exact sequence number of the already queued item.
1093-- we can even run the already queued items but they will have to be sorted in
1094-- layers in the heap. We can use a list of heaps for that.
1095{-# INLINE enqueueAhead #-}
1096enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1097enqueueAhead sv q m = do
1098 atomicModifyIORefCAS_ q $ \ case
1099 ([], n) -> ([m], n + 1) -- increment sequence
1100 _ -> error "not empty"
1103-- enqueue without incrementing the sequence number
1104{-# INLINE reEnqueueAhead #-}
1105reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1106reEnqueueAhead sv q m = do
1107 atomicModifyIORefCAS_ q $ \ case
1108 ([], n) -> ([m], n) -- DO NOT increment sequence
1109 _ -> error "not empty"
1112-- Normally the thread that has the token should never go away. The token gets
1113-- handed over to another thread, but someone or the other has the token at any
1114-- point of time. But if the task that has the token finds that the outputQueue
1115-- is full, in that case it can go away without even handing over the token to
1116-- another thread. In that case it sets the nextSequence number in the heap its
1117-- own sequence number before going away. To handle this case, any task that
1118-- does not have the token tries to dequeue from the heap first before
1119-- dequeuing from the work queue. If it finds that the task at the top of the
1120-- heap is the one that owns the current sequence number then it grabs the
1121-- token and starts with that.
1123-- XXX instead of queueing just the head element and the remaining computation
1124-- on the heap, evaluate as many as we can and place them on the heap. But we
1125-- need to give higher priority to the lower sequence numbers so that lower
1126-- priority tasks do not fill up the heap making higher priority tasks block
1127-- due to full heap. Maybe we can have a weighted space for them in the heap.
1128-- The weight is inversely proportional to the sequence number.
1130-- XXX review for livelock
1132{-# INLINE queueEmptyAhead #-}
1133queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
1134queueEmptyAhead q = liftIO $ do
1135 (xs, _) <- readIORef q
1138{-# INLINE dequeueAhead #-}
1139dequeueAhead :: MonadIO m
1140 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
1141dequeueAhead q = liftIO $
1142 atomicModifyIORefCAS q $ \case
1143 ([], n) -> (([], n), Nothing)
1144 (x : [], n) -> (([], n), Just (x, n))
1145 _ -> error "more than one item on queue"
1147-------------------------------------------------------------------------------
1149-------------------------------------------------------------------------------
1151withIORef :: IORef a -> (a -> IO b) -> IO b
1152withIORef ref f = readIORef ref >>= f
1154atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
1155atomicModifyIORef_ ref f =
1156 atomicModifyIORef ref $ \x -> (f x, ())
1158data HeapDequeueResult t m a =
1161 | Ready (Entry Int (AheadHeapEntry t m a))
1163{-# INLINE dequeueFromHeap #-}
1165 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1166 -> IO (HeapDequeueResult t m a)
1167dequeueFromHeap hpVar =
1168 atomicModifyIORef hpVar $ \pair@(hp, snum) ->
1170 Nothing -> (pair, Clearing)
1174 Just (ent@(Entry seqNo _ev), hp') ->
1176 then ((hp', Nothing), Ready ent)
1177 else assert (seqNo >= n) (pair, Waiting n)
1178 Nothing -> (pair, Waiting n)
1180{-# INLINE dequeueFromHeapSeq #-}
1182 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1184 -> IO (HeapDequeueResult t m a)
1185dequeueFromHeapSeq hpVar i =
1186 atomicModifyIORef hpVar $ \(hp, snum) ->
1191 Just (ent@(Entry seqNo _ev), hp') ->
1193 then ((hp', Nothing), Ready ent)
1194 else assert (seqNo >= i) ((hp, Just i), Waiting i)
1195 Nothing -> ((hp, Just i), Waiting i)
1196 Just _ -> error "dequeueFromHeapSeq: unreachable"
1198heapIsSane :: Maybe Int -> Int -> Bool
1199heapIsSane snum seqNo =
1202 Just n -> seqNo >= n
1204{-# INLINE requeueOnHeapTop #-}
1206 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1207 -> Entry Int (AheadHeapEntry t m a)
1210requeueOnHeapTop hpVar ent seqNo =
1211 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1212 assert (heapIsSane snum seqNo) (H.insert ent hp, Just seqNo)
1214{-# INLINE updateHeapSeq #-}
1216 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1219updateHeapSeq hpVar seqNo =
1220 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1221 assert (heapIsSane snum seqNo) (hp, Just seqNo)
1223-------------------------------------------------------------------------------
1225-------------------------------------------------------------------------------
1227-- XXX To be implemented. Use a linked queue like WAsync and put back the
1228-- remaining computation at the back of the queue instead of the heap, and
1229-- increment the sequence number.
1231-- Thread tracking is needed for two reasons:
1233-- 1) Killing threads on exceptions. Threads may not be left to go away by
1234-- themselves because they may run for significant times before going away or
1235-- worse they may be stuck in IO and never go away.
1237-- 2) To know when all threads are done and the stream has ended.
1239{-# NOINLINE addThread #-}
1240addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1242 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
1244-- This is cheaper than modifyThread because we do not have to send a
1245-- outputDoorBell This can make a difference when more workers are being
1247{-# INLINE delThread #-}
1248delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1250 liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
1252-- If present then delete else add. This takes care of out of order add and
1253-- delete i.e. a delete arriving before we even added a thread.
1254-- This occurs when the forked thread is done even before the 'addThread' right
1255-- after the fork gets a chance to run.
1256{-# INLINE modifyThread #-}
1257modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1258modifyThread sv tid = do
1259 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1261 then let new = S.delete tid old in (new, new)
1262 else let new = S.insert tid old in (new, old)
1263 when (null changed) $
1266 void $ tryPutMVar (outputDoorBell sv) ()
1268-- | This is safe even if we are adding more threads concurrently because if
1269-- a child thread is adding another thread then anyway 'workerThreads' will
1271{-# INLINE allThreadsDone #-}
1272allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1273allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1275{-# NOINLINE handleChildException #-}
1276handleChildException :: SVar t m a -> SomeException -> IO ()
1277handleChildException sv e = do
1279 void $ send sv (ChildStop tid (Just e))
1281{-# NOINLINE recordMaxWorkers #-}
1282recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1283recordMaxWorkers sv = liftIO $ do
1284 active <- readIORef (workerCount sv)
1285 maxWrk <- readIORef (maxWorkers $ svarStats sv)
1286 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1287 modifyIORef (totalDispatches $ svarStats sv) (+1)
1289{-# NOINLINE pushWorker #-}
1290pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1291pushWorker yieldMax sv = do
1292 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1293 when (svarInspectMode sv) $ recordMaxWorkers sv
1294 -- This allocation matters when significant number of workers are being
1295 -- sent. We allocate it only when needed.
1297 case yieldRateInfo sv of
1298 Nothing -> return Nothing
1299 Just _ -> liftIO $ do
1300 cntRef <- newIORef 0
1301 t <- getTime Monotonic
1302 lat <- newIORef (0, t)
1303 return $ Just WorkerInfo
1304 { workerYieldMax = yieldMax
1305 , workerYieldCount = cntRef
1306 , workerLatencyStart = lat
1308 doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
1311-- XXX we can push the workerCount modification in accountThread and use the
1312-- same pushWorker for Parallel case as well.
1314-- | In contrast to pushWorker which always happens only from the consumer
1315-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1316-- producer side. So we need to use a thread safe modification of
1317-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1318-- using a CAS based modification.
1319{-# INLINE pushWorkerPar #-}
1322 => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1323pushWorkerPar sv wloop =
1324 if svarInspectMode sv
1326 else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
1331 {-# NOINLINE forkWithDiag #-}
1333 -- We do not use workerCount in case of ParallelVar but still there is
1334 -- no harm in maintaining it correctly.
1335 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1337 -- This allocation matters when significant number of workers are being
1338 -- sent. We allocate it only when needed. The overhead increases by 4x.
1340 case yieldRateInfo sv of
1341 Nothing -> return Nothing
1342 Just _ -> liftIO $ do
1343 cntRef <- newIORef 0
1344 t <- getTime Monotonic
1345 lat <- newIORef (0, t)
1346 return $ Just WorkerInfo
1347 { workerYieldMax = 0
1348 , workerYieldCount = cntRef
1349 , workerLatencyStart = lat
1352 doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
1356-- True: can dispatch more
1357-- False: cannot dispatch any more
1358dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1359dispatchWorker yieldCount sv = do
1360 let workerLimit = maxWorkerLimit sv
1361 -- XXX in case of Ahead streams we should not send more than one worker
1362 -- when the work queue is done but heap is not done.
1363 done <- liftIO $ isWorkDone sv
1364 -- Note, "done" may not mean that the work is actually finished if there
1365 -- are workers active, because there may be a worker which has not yet
1366 -- queued the leftover work.
1369 qDone <- liftIO $ isQueueDone sv
1370 -- Note that the worker count is only decremented during event
1371 -- processing in fromStreamVar and therefore it is safe to read and
1372 -- use it without a lock.
1373 active <- liftIO $ readIORef $ workerCount sv
1376 -- Note that we may deadlock if the previous workers (tasks in the
1377 -- stream) wait/depend on the future workers (tasks in the stream)
1378 -- executing. In that case we should either configure the maxWorker
1379 -- count to higher or use parallel style instead of ahead or async
1381 limit <- case remainingWork sv of
1382 Nothing -> return workerLimit
1384 n <- liftIO $ readIORef ref
1387 Unlimited -> Limited (fromIntegral n)
1388 Limited lim -> Limited $ min lim (fromIntegral n)
1390 -- XXX for ahead streams shall we take the heap yields into account for
1391 -- controlling the dispatch? We should not dispatch if the heap has
1392 -- already got the limit covered.
1393 let dispatch = pushWorker yieldCount sv >> return True
1395 Unlimited -> dispatch
1396 -- Note that the use of remainingWork and workerCount is not
1397 -- atomic and the counts may even have changed between reading and
1398 -- using them here, so this is just approximate logic and we cannot
1399 -- rely on it for correctness. We may actually dispatch more
1400 -- workers than required.
1401 Limited lim | lim > 0 -> dispatch
1404 when (active <= 0) $ pushWorker 0 sv
1408-- | This is a magic number and it is overloaded, and used at several places to
1411-- 1. If we have to sleep to slowdown this is the minimum period that we
1412-- accumulate before we sleep. Also, workers do not stop until this much
1413-- sleep time is accumulated.
1414-- 3. Collected latencies are computed and transferred to measured latency
1415-- after a minimum of this period.
1416minThreadDelay :: NanoSecs
1417minThreadDelay = 10^(6 :: Int)
1419-- | Another magic number! When we have to start more workers to cover up a
1420-- number of yields that we are lagging by then we cannot start one worker for
1421-- each yield because that may be a very big number and if the latency of the
1422-- workers is low these number of yields could be very high. We assume that we
1423-- run each extra worker for at least this much time.
1424rateRecoveryTime :: NanoSecs
1425rateRecoveryTime = 1000000
1427nanoToMicroSecs :: NanoSecs -> Int
1428nanoToMicroSecs s = fromIntegral s `div` 1000
1430-- We either block, or send one worker with limited yield count or one or more
1431-- workers with unlimited yield count.
1433 = BlockWait NanoSecs
1434 | PartialWorker Count
1435 | ManyWorkers Int Count
1438-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1448estimateWorkers workerLimit svarYields gainLossYields
1449 svarElapsed wLatency targetLat range =
1450 -- XXX we can have a maxEfficiency combinator as well which runs the
1451 -- producer at the maximal efficiency i.e. the number of workers are chosen
1452 -- such that the latency is minimum or within a range. Or we can call it
1453 -- maxWorkerLatency.
1456 -- How many workers do we need to acheive the required rate?
1458 -- When the workers are IO bound we can increase the throughput by
1459 -- increasing the number of workers as long as the IO device has enough
1460 -- capacity to process all the requests concurrently. If the IO
1461 -- bandwidth is saturated increasing the workers won't help. Also, if
1462 -- the CPU utilization in processing all these requests exceeds the CPU
1463 -- bandwidth, then increasing the number of workers won't help.
1465 -- When the workers are purely CPU bound, increasing the workers beyond
1466 -- the number of CPUs won't help.
1468 -- TODO - measure the CPU and IO requirements of the workers. Have a
1469 -- way to specify the max bandwidth of the underlying IO mechanism and
1470 -- use that to determine the max rate of workers, and also take the CPU
1471 -- bandwidth into account. We can also discover the IO bandwidth if we
1472 -- know that we are not CPU bound, then how much steady state rate are
1473 -- we able to acheive. Design tests for CPU bound and IO bound cases.
1475 -- Calculate how many yields are we ahead or behind to match the exact
1476 -- required rate. Based on that we increase or decrease the effective
1479 -- When the worker latency is lower than required latency we begin with
1480 -- a yield and then wait rather than first waiting and then yielding.
1481 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1482 effectiveYields = svarYields + gainLossYields
1483 deltaYields = fromIntegral targetYields - effectiveYields
1485 -- We recover the deficit by running at a higher/lower rate for a
1486 -- certain amount of time. To keep the effective rate in reasonable
1487 -- limits we use rateRecoveryTime, minLatency and maxLatency.
1488 in if deltaYields > 0
1490 let deltaYieldsFreq :: Double
1492 fromIntegral deltaYields /
1493 fromIntegral rateRecoveryTime
1494 yieldsFreq = 1.0 / fromIntegral targetLat
1495 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1496 requiredLat = NanoSecs $ round $ 1.0 / totalYieldsFreq
1497 adjustedLat = min (max requiredLat (minLatency range))
1499 in assert (adjustedLat > 0) $
1500 if wLatency <= adjustedLat
1501 then PartialWorker deltaYields
1502 else let workers = withLimit $ wLatency `div` adjustedLat
1503 limited = min workers (fromIntegral deltaYields)
1504 in ManyWorkers (fromIntegral limited) deltaYields
1506 let expectedDuration = fromIntegral effectiveYields * targetLat
1507 sleepTime = expectedDuration - svarElapsed
1508 maxSleepTime = maxLatency range - wLatency
1509 s = min sleepTime maxSleepTime
1510 in assert (sleepTime >= 0) $
1511 -- if s is less than 0 it means our maxSleepTime is less
1512 -- than the worker latency.
1513 if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
1518 Limited x -> min n (fromIntegral x)
1520-- | Get the worker latency without resetting workerPendingLatency
1521-- Returns (total yield count, base time, measured latency)
1522-- CAUTION! keep it in sync with collectLatency
1523getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1524getWorkerLatency yinfo = do
1525 let cur = workerPendingLatency yinfo
1526 col = workerCollectedLatency yinfo
1527 longTerm = svarAllTimeLatency yinfo
1528 measured = workerMeasuredLatency yinfo
1530 (count, time) <- readIORef cur
1531 (colCount, colTime) <- readIORef col
1532 (lcount, ltime) <- readIORef longTerm
1533 prev <- readIORef measured
1535 let pendingCount = colCount + count
1536 pendingTime = colTime + time
1539 then let lat = pendingTime `div` fromIntegral pendingCount
1540 -- XXX Give more weight to new?
1541 in (lat + prev) `div` 2
1543 return (lcount + pendingCount, ltime, new)
1545isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1546isBeyondMaxRate sv yinfo = do
1547 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1548 now <- getTime Monotonic
1549 let duration = fromInteger $ toNanoSecs $ now - tstamp
1550 let targetLat = svarLatencyTarget yinfo
1551 gainLoss <- readIORef (svarGainedLostYields yinfo)
1552 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1553 wLatency targetLat (svarLatencyRange yinfo)
1554 cnt <- readIORef $ workerCount sv
1555 return $ case work of
1556 -- XXX set the worker's maxYields or polling interval based on yields
1557 PartialWorker _yields -> cnt > 1
1558 ManyWorkers n _ -> cnt > n
1561-- Every once in a while workers update the latencies and check the yield rate.
1562-- They return if we are above the expected yield rate. If we check too often
1563-- it may impact performance, if we check less often we may have a stale
1564-- picture. We update every minThreadDelay but we translate that into a yield
1565-- count based on latency so that the checking overhead is little.
1567-- XXX use a generation count to indicate that the value is updated. If the
1568-- value is updated an existing worker must check it again on the next yield.
1569-- Otherwise it is possible that we may keep updating it and because of the mod
1570-- worker keeps skipping it.
1571updateWorkerPollingInterval :: YieldRateInfo -> NanoSecs -> IO ()
1572updateWorkerPollingInterval yinfo latency = do
1573 let periodRef = workerPollingInterval yinfo
1574 cnt = max 1 $ minThreadDelay `div` latency
1575 period = min cnt (fromIntegral magicMaxBuffer)
1577 writeIORef periodRef (fromIntegral period)
1579-- Returns a triple, (1) yield count since last collection, (2) the base time
1580-- when we started counting, (3) average latency in the last measurement
1581-- period. The former two are used for accurate measurement of the going rate
1582-- whereas the average is used for future estimates e.g. how many workers
1583-- should be maintained to maintain the rate.
1584-- CAUTION! keep it in sync with getWorkerLatency
1585collectLatency :: SVar t m a -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1586collectLatency sv yinfo = do
1587 let cur = workerPendingLatency yinfo
1588 col = workerCollectedLatency yinfo
1589 longTerm = svarAllTimeLatency yinfo
1590 measured = workerMeasuredLatency yinfo
1592 (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
1593 (colCount, colTime) <- readIORef col
1594 (lcount, ltime) <- readIORef longTerm
1595 prev <- readIORef measured
1597 let pendingCount = colCount + count
1598 pendingTime = colTime + time
1600 lcount' = lcount + pendingCount
1601 tripleWith lat = (lcount', ltime, lat)
1605 let new = pendingTime `div` (fromIntegral pendingCount)
1606 when (svarInspectMode sv) $ do
1607 let ss = svarStats sv
1608 minLat <- readIORef (minWorkerLatency ss)
1609 when (new < minLat || minLat == 0) $
1610 writeIORef (minWorkerLatency ss) new
1612 maxLat <- readIORef (maxWorkerLatency ss)
1613 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
1614 -- When we have collected a significant sized batch we compute the new
1615 -- latency using that batch and return the new latency, otherwise we
1616 -- return the previous latency derived from the previous batch.
1617 if (pendingCount > fromIntegral magicMaxBuffer)
1618 || (pendingTime > minThreadDelay)
1619 || (let r = fromIntegral new / fromIntegral prev :: Double
1620 in prev > 0 && (r > 2 || r < 0.5))
1623 when (svarInspectMode sv) $ do
1624 let ss = svarStats sv
1625 modifyIORef (avgWorkerLatency ss) $
1626 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
1627 updateWorkerPollingInterval yinfo (max new prev)
1628 writeIORef col (0, 0)
1629 writeIORef measured ((prev + new) `div` 2)
1630 modifyIORef longTerm $ \(_, t) -> (lcount', t)
1631 return $ tripleWith new
1633 writeIORef col (pendingCount, pendingTime)
1634 return $ tripleWith prev
1635 else return $ tripleWith prev
1637-- XXX in case of ahead style stream we need to take the heap size into account
1638-- because we return the workers on the basis of that which causes a condition
1639-- where we keep dispatching and they keep returning. So we must have exactly
1640-- the same logic for not dispatching and for returning.
1643-- True: can dispatch more
1644-- False: full, no more dispatches
1645dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1646dispatchWorkerPaced sv = do
1647 let yinfo = fromJust $ yieldRateInfo sv
1648 (svarYields, svarElapsed, wLatency) <- do
1649 now <- liftIO $ getTime Monotonic
1650 (yieldCount, baseTime, lat) <-
1651 liftIO $ collectLatency sv yinfo
1652 let elapsed = fromInteger $ toNanoSecs $ now - baseTime
1656 case workerBootstrapLatency yinfo of
1661 return (yieldCount, elapsed, latency)
1664 -- Need to measure the latency with a single worker before we can perform
1668 let workerLimit = maxWorkerLimit sv
1669 let targetLat = svarLatencyTarget yinfo
1670 let range = svarLatencyRange yinfo
1671 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1672 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
1673 wLatency targetLat range
1675 -- XXX we need to take yieldLimit into account here. If we are at the
1676 -- end of the limit as well as the time, we should not be sleeping.
1677 -- If we are not actually planning to dispatch any more workers we need
1678 -- to take that in account.
1681 assert (s >= 0) (return ())
1682 -- XXX note that when we return from here we will block waiting
1683 -- for the result from the existing worker. If that takes too
1684 -- long we won't be able to send another worker until the
1687 -- Sleep only if there are no active workers, otherwise we will
1688 -- defer the output of those. Note we cannot use workerCount
1689 -- here as it is not a reliable way to ensure there are
1690 -- definitely no active workers. When workerCount is 0 we may
1691 -- still have a Stop event waiting in the outputQueue.
1692 done <- allThreadsDone sv
1693 when done $ void $ do
1694 liftIO $ threadDelay $ nanoToMicroSecs s
1697 PartialWorker yields -> do
1698 assert (yields > 0) (return ())
1699 updateGainedLostYields yinfo yields
1701 done <- allThreadsDone sv
1702 when done $ void $ dispatchWorker yields sv
1704 ManyWorkers netWorkers yields -> do
1705 assert (netWorkers >= 1) (return ())
1706 assert (yields >= 0) (return ())
1707 updateGainedLostYields yinfo yields
1709 let periodRef = workerPollingInterval yinfo
1710 ycnt = max 1 $ yields `div` fromIntegral netWorkers
1711 period = min ycnt (fromIntegral magicMaxBuffer)
1713 old <- liftIO $ readIORef periodRef
1714 when (period < old) $
1715 liftIO $ writeIORef periodRef period
1717 cnt <- liftIO $ readIORef $ workerCount sv
1720 let total = netWorkers - cnt
1721 batch = max 1 $ fromIntegral $
1722 minThreadDelay `div` targetLat
1723 -- XXX stagger the workers over a period?
1724 -- XXX cannot sleep, as that would mean we cannot process the
1725 -- outputs. need to try a different mechanism to stagger.
1726 -- when (total > batch) $
1727 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1728 dispatchN (min total batch)
1733 updateGainedLostYields yinfo yields = do
1734 let buf = fromIntegral $ svarRateBuffer yinfo
1735 when (yields /= 0 && abs yields > buf) $ do
1740 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1746 r <- dispatchWorker 0 sv
1748 then dispatchN (n - 1)
1751sendWorkerDelayPaced :: SVar t m a -> IO ()
1752sendWorkerDelayPaced _ = return ()
1754sendWorkerDelay :: SVar t m a -> IO ()
1755sendWorkerDelay _sv =
1756 -- XXX we need a better way to handle this than hardcoded delays. The
1757 -- delays may be different for different systems.
1758 -- If there is a usecase where this is required we can create a combinator
1759 -- to set it as a config in the state.
1762 ncpu <- getNumCapabilities
1765 if (svarStyle sv == AheadVar)
1766 then threadDelay 100
1769 if (svarStyle sv == AheadVar)
1770 then threadDelay 100
1775{-# NOINLINE sendWorkerWait #-}
1778 => (SVar t m a -> IO ())
1779 -> (SVar t m a -> m Bool)
1782sendWorkerWait delay dispatch sv = do
1783 -- Note that we are guaranteed to have at least one outstanding worker when
1784 -- we enter this function. So if we sleep we are guaranteed to be woken up
1785 -- by an outputDoorBell, when the worker exits.
1788 (_, n) <- liftIO $ readIORef (outputQueue sv)
1790 -- The queue may be empty temporarily if the worker has dequeued the
1791 -- work item but has not enqueued the remaining part yet. For the same
1792 -- reason, a worker may come back if it tries to dequeue and finds the
1793 -- queue empty, even though the whole work has not finished yet.
1795 -- If we find that the queue is empty, but it may be empty
1796 -- temporarily, when we checked it. If that's the case we might
1797 -- sleep indefinitely unless the active workers produce some
1798 -- output. We may deadlock specially if the otuput from the active
1799 -- workers depends on the future workers that we may never send.
1800 -- So in case the queue was temporarily empty set a flag to inform
1801 -- the enqueue to send us a doorbell.
1803 -- Note that this is just a best effort mechanism to avoid a
1804 -- deadlock. Deadlocks may still happen if for some weird reason
1805 -- the consuming computation shares an MVar or some other resource
1806 -- with the producing computation and gets blocked on that resource
1807 -- and therefore cannot do any pushworker to add more threads to
1808 -- the producer. In such cases the programmer should use a parallel
1809 -- style so that all the producers are scheduled immediately and
1810 -- unconditionally. We can also use a separate monitor thread to
1811 -- push workers instead of pushing them from the consumer, but then
1812 -- we are no longer using pull based concurrency rate adaptation.
1814 -- XXX update this in the tutorial.
1816 -- Having pending active workers does not mean that we are guaranteed
1817 -- to be woken up if we sleep. In case of Ahead streams, there may be
1818 -- queued items in the heap even though the outputQueue is empty, and
1819 -- we may have active workers which are deadlocked on those items to be
1820 -- processed by the consumer. We should either guarantee that any
1821 -- worker, before returning, clears the heap or we send a worker to clear
1822 -- it. Normally we always send a worker if no output is seen, but if
1823 -- the thread limit is reached or we are using pacing then we may not
1824 -- send a worker. See the concurrentApplication test in the tests, that
1825 -- test case requires at least one yield from the producer to not
1826 -- deadlock, if the last workers output is stuck in the heap then this
1827 -- test fails. This problem can be extended to n threads when the
1828 -- consumer may depend on the evaluation of next n items in the
1831 -- register for the outputDoorBell before we check the queue so that if we
1832 -- sleep because the queue was empty we are guaranteed to get a
1833 -- doorbell on the next enqueue.
1835 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
1836 liftIO storeLoadBarrier
1837 canDoMore <- dispatch sv
1839 -- XXX test for the case when we miss sending a worker when the worker
1840 -- count is more than 1500.
1842 -- XXX Assert here that if the heap is not empty then there is at
1843 -- least one outstanding worker. Otherwise we could be sleeping
1847 then sendWorkerWait delay dispatch sv
1849 liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
1850 $ takeMVar (outputDoorBell sv)
1851 (_, len) <- liftIO $ readIORef (outputQueue sv)
1852 when (len <= 0) $ sendWorkerWait delay dispatch sv
1854{-# INLINE readOutputQRaw #-}
1855readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
1856readOutputQRaw sv = do
1857 (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
1858 when (svarInspectMode sv) $ do
1859 let ref = maxOutQSize $ svarStats sv
1860 oqLen <- readIORef ref
1861 when (len > oqLen) $ writeIORef ref len
1864readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1865readOutputQBounded sv = do
1866 (list, len) <- liftIO $ readOutputQRaw sv
1867 -- When there is no output seen we dispatch more workers to help
1868 -- out if there is work pending in the work queue.
1872 -- send a worker proactively, if needed, even before we start
1873 -- processing the output. This may degrade single processor
1874 -- perf but improves multi-processor, because of more
1882 cnt <- liftIO $ readIORef $ workerCount sv
1883 when (cnt <= 0) $ do
1884 done <- liftIO $ isWorkDone sv
1885 when (not done) (pushWorker 0 sv)
1887 {-# INLINE blockingRead #-}
1889 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
1890 liftIO (fst `fmap` readOutputQRaw sv)
1892readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1893readOutputQPaced sv = do
1894 (list, len) <- liftIO $ readOutputQRaw sv
1898 -- XXX send a worker proactively, if needed, even before we start
1899 -- processing the output.
1900 void $ dispatchWorkerPaced sv
1905 {-# INLINE blockingRead #-}
1907 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
1908 liftIO (fst `fmap` readOutputQRaw sv)
1910postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
1911postProcessBounded sv = do
1912 workersDone <- allThreadsDone sv
1913 -- There may still be work pending even if there are no workers pending
1914 -- because all the workers may return if the outputQueue becomes full. In
1915 -- that case send off a worker to kickstart the work again.
1917 -- Note that isWorkDone can only be safely checked if all workers are done.
1918 -- When some workers are in progress they may have decremented the yield
1919 -- Limit and later ending up incrementing it again. If we look at the yield
1920 -- limit in that window we may falsely say that it is 0 and therefore we
1924 r <- liftIO $ isWorkDone sv
1925 -- Note that we need to guarantee a worker, therefore we cannot just
1926 -- use dispatchWorker which may or may not send a worker.
1927 when (not r) (pushWorker 0 sv)
1928 -- XXX do we need to dispatch many here?
1929 -- void $ dispatchWorker sv
1933postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
1934postProcessPaced sv = do
1935 workersDone <- allThreadsDone sv
1936 -- XXX If during consumption we figure out we are getting delayed then we
1937 -- should trigger dispatch there as well. We should try to check on the
1938 -- workers after consuming every n item from the buffer?
1941 r <- liftIO $ isWorkDone sv
1943 void $ dispatchWorkerPaced sv
1944 -- Note that we need to guarantee a worker since the work is not
1945 -- finished, therefore we cannot just rely on dispatchWorkerPaced
1946 -- which may or may not send a worker.
1947 noWorker <- allThreadsDone sv
1948 when noWorker $ pushWorker 0 sv
1952getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
1953getYieldRateInfo st = do
1954 -- convert rate in Hertz to latency in Nanoseconds
1955 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
1956 case getStreamRate st of
1957 Just (Rate low goal high buf) ->
1958 let l = rateToLatency goal
1959 minl = rateToLatency high
1960 maxl = rateToLatency low
1961 in mkYieldRateInfo l (LatencyRange minl maxl) buf
1962 Nothing -> return Nothing
1966 mkYieldRateInfo latency latRange buf = do
1967 measured <- newIORef 0
1968 wcur <- newIORef (0,0)
1969 wcol <- newIORef (0,0)
1970 now <- getTime Monotonic
1971 wlong <- newIORef (0,now)
1972 period <- newIORef 1
1973 gainLoss <- newIORef (Count 0)
1975 return $ Just YieldRateInfo
1976 { svarLatencyTarget = latency
1977 , svarLatencyRange = latRange
1978 , svarRateBuffer = buf
1979 , svarGainedLostYields = gainLoss
1980 , workerBootstrapLatency = getStreamLatency st
1981 , workerPollingInterval = period
1982 , workerMeasuredLatency = measured
1983 , workerPendingLatency = wcur
1984 , workerCollectedLatency = wcol
1985 , svarAllTimeLatency = wlong
1988newSVarStats :: IO SVarStats
1991 maxWrk <- newIORef 0
1995 avgLat <- newIORef (0, NanoSecs 0)
1996 maxLat <- newIORef (NanoSecs 0)
1997 minLat <- newIORef (NanoSecs 0)
1998 stpTime <- newIORef Nothing
2001 { totalDispatches = disp
2002 , maxWorkers = maxWrk
2003 , maxOutQSize = maxOq
2004 , maxHeapSize = maxHs
2005 , maxWorkQSize = maxWq
2006 , avgWorkerLatency = avgLat
2007 , minWorkerLatency = minLat
2008 , maxWorkerLatency = maxLat
2009 , svarStopTime = stpTime
2012getAheadSVar :: MonadAsync m
2014 -> ( IORef ([t m a], Int)
2015 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2022getAheadSVar st f mrun = do
2023 outQ <- newIORef ([], 0)
2024 -- the second component of the tuple is "Nothing" when heap is being
2025 -- cleared, "Just n" when we are expecting sequence number n to arrive
2026 -- before we can start clearing the heap.
2027 outH <- newIORef (H.empty, Just 0)
2028 outQMv <- newEmptyMVar
2029 active <- newIORef 0
2030 wfw <- newIORef False
2031 running <- newIORef S.empty
2032 -- Sequence number is incremented whenever something is queued, therefore,
2033 -- first sequence number would be 0
2034 q <- newIORef ([], -1)
2035 stopMVar <- newMVar ()
2036 yl <- case getYieldLimit st of
2037 Nothing -> return Nothing
2038 Just x -> Just <$> newIORef x
2039 rateInfo <- getYieldRateInfo st
2041 stats <- newSVarStats
2044 let getSVar sv readOutput postProc = SVar
2045 { outputQueue = outQ
2046 , remainingWork = yl
2047 , maxBufferLimit = getMaxBuffer st
2048 , maxWorkerLimit = getMaxThreads st
2049 , yieldRateInfo = rateInfo
2050 , outputDoorBell = outQMv
2051 , readOutputQ = readOutput sv
2052 , postProcess = postProc sv
2053 , workerThreads = running
2054 , workLoop = f q outH st{streamVar = Just sv} sv
2055 , enqueue = enqueueAhead sv q
2056 , isWorkDone = isWorkDoneAhead sv q outH
2057 , isQueueDone = isQueueDoneAhead sv q
2058 , needDoorBell = wfw
2059 , svarStyle = AheadVar
2061 , workerCount = active
2062 , accountThread = delThread sv
2063 , workerStopMVar = stopMVar
2065 , svarInspectMode = getInspectMode st
2067 , aheadWorkQueue = q
2073 case getStreamRate st of
2074 Nothing -> getSVar sv readOutputQBounded postProcessBounded
2075 Just _ -> getSVar sv readOutputQPaced postProcessPaced
2080 {-# INLINE isQueueDoneAhead #-}
2081 isQueueDoneAhead sv q = do
2082 queueDone <- checkEmpty q
2084 case remainingWork sv of
2088 Nothing -> return False
2089 -- XXX note that yieldsDone can only be authoritative only when there
2090 -- are no workers running. If there are active workers they can
2091 -- later increment the yield count and therefore change the result.
2092 return $ yieldsDone || queueDone
2094 {-# INLINE isWorkDoneAhead #-}
2095 isWorkDoneAhead sv q ref = do
2097 (hp, _) <- readIORef ref
2098 return (H.size hp <= 0)
2099 queueDone <- isQueueDoneAhead sv q
2100 return $ heapDone && queueDone
2103 (xs, _) <- readIORef q
2106getParallelSVar :: MonadIO m => State t m a -> RunInIO m -> IO (SVar t m a)
2107getParallelSVar st mrun = do
2108 outQ <- newIORef ([], 0)
2109 outQMv <- newEmptyMVar
2110 active <- newIORef 0
2111 running <- newIORef S.empty
2112 yl <- case getYieldLimit st of
2113 Nothing -> return Nothing
2114 Just x -> Just <$> newIORef x
2115 rateInfo <- getYieldRateInfo st
2117 stats <- newSVarStats
2121 SVar { outputQueue = outQ
2122 , remainingWork = yl
2123 , maxBufferLimit = Unlimited
2124 , maxWorkerLimit = Unlimited
2125 -- Used only for diagnostics
2126 , yieldRateInfo = rateInfo
2127 , outputDoorBell = outQMv
2128 , readOutputQ = readOutputQPar sv
2129 , postProcess = allThreadsDone sv
2130 , workerThreads = running
2131 , workLoop = undefined
2132 , enqueue = undefined
2133 , isWorkDone = undefined
2134 , isQueueDone = undefined
2135 , needDoorBell = undefined
2136 , svarStyle = ParallelVar
2138 , workerCount = active
2139 , accountThread = modifyThread sv
2140 , workerStopMVar = undefined
2142 , svarInspectMode = getInspectMode st
2144 , aheadWorkQueue = undefined
2145 , outputHeap = undefined
2152 readOutputQPar sv = liftIO $ do
2153 withDiagMVar sv "readOutputQPar: doorbell"
2154 $ takeMVar (outputDoorBell sv)
2155 case yieldRateInfo sv of
2156 Nothing -> return ()
2157 Just yinfo -> void $ collectLatency sv yinfo
2158 fst `fmap` readOutputQRaw sv
2160sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
2161sendFirstWorker sv m = do
2162 -- Note: We must have all the work on the queue before sending the
2163 -- pushworker, otherwise the pushworker may exit before we even get a
2165 liftIO $ enqueue sv m
2166 case yieldRateInfo sv of
2167 Nothing -> pushWorker 0 sv
2169 if svarLatencyTarget yinfo == maxBound
2170 then liftIO $ threadDelay maxBound
2171 else pushWorker 1 sv
2174{-# INLINABLE newAheadVar #-}
2175newAheadVar :: MonadAsync m
2178 -> ( IORef ([t m a], Int)
2179 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2185newAheadVar st m wloop = do
2186 mrun <- captureMonadState
2187 sv <- liftIO $ getAheadSVar st wloop mrun
2188 sendFirstWorker sv m
2190{-# INLINABLE newParallelVar #-}
2191newParallelVar :: MonadAsync m => State t m a -> m (SVar t m a)
2192newParallelVar st = do
2193 mrun <- captureMonadState
2194 liftIO $ getParallelSVar st mrun
2196-- XXX this errors out for Parallel/Ahead SVars
2197-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
2198-- be read back from the SVar using 'fromSVar'.
2199toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
2200toStreamVar sv m = do
2201 liftIO $ enqueue sv m
2202 done <- allThreadsDone sv
2203 -- XXX This is safe only when called from the consumer thread or when no
2204 -- consumer is present. There may be a race if we are not running in the
2206 -- XXX do this only if the work queue is not empty. The work may have been
2207 -- carried out by existing workers.
2209 case yieldRateInfo sv of
2210 Nothing -> pushWorker 0 sv
2211 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