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 ScopedTypeVariables #-}
12{-# LANGUAGE UnboxedTuples #-}
15-- Module : Streamly.SVar
16-- Copyright : (c) 2017 Harendra Kumar
19-- Maintainer : harendra.kumar@gmail.com
20-- Stability : experimental
24#ifdef DIAGNOSTICS_VERBOSE
34 -- State threaded around the stream
50 , cleanupSVarFromWorker
56 , atomicModifyIORefCAS
74 , HeapDequeueResult(..)
89 , decrementYieldLimitPost
108import Control.Concurrent
109 (ThreadId, myThreadId, threadDelay, throwTo)
110import Control.Concurrent.MVar
111 (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar)
112import Control.Exception (SomeException(..), catch, mask, assert, Exception)
113import Control.Monad (when)
114import Control.Monad.Catch (MonadThrow)
115import Control.Monad.IO.Class (MonadIO(..))
116import Control.Monad.Trans.Control (MonadBaseControl, control)
118 (casIORef, readForCAS, peekTicket, atomicModifyIORefCAS_,
119 writeBarrier, storeLoadBarrier)
120import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
121import Data.Functor (void)
122import Data.Heap (Heap, Entry(..))
123import Data.Int (Int64)
125 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
126import Data.List ((\\))
127import Data.Maybe (fromJust)
129import GHC.Conc (ThreadId(..))
131import GHC.IO (IO(..))
132import System.Clock (TimeSpec, Clock(Monotonic), getTime, toNanoSecs)
134import qualified Data.Heap as H
135import qualified Data.Set as S
137-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
138-- can keep it on in production to debug problems quickly if and when they
139-- happen, but it may result in unexpected output when threads are left hanging
140-- until they are GCed because the consumer went away.
143import Control.Concurrent.MVar (tryTakeMVar)
144import Control.Exception
145 (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
146 BlockedIndefinitelyOnSTM(..))
147import System.IO (hPutStrLn, stderr)
148import Text.Printf (printf)
151-- Always use signed arithmetic to avoid inadvertant overflows of signed values
152-- on conversion when comparing unsigned quantities with signed.
153newtype NanoSecs = NanoSecs Int64
165newtype Count = Count Int64
177------------------------------------------------------------------------------
178-- Parent child thread communication type
179------------------------------------------------------------------------------
181data ThreadAbort = ThreadAbort deriving Show
183instance Exception ThreadAbort
185-- | Events that a child thread may send to a parent thread.
188 | ChildStop ThreadId (Maybe SomeException)
190-- | Sorting out-of-turn outputs in a heap for Ahead style streams
191data AheadHeapEntry (t :: (* -> *) -> * -> *) m a =
193 | AheadEntryStream (t m a)
195------------------------------------------------------------------------------
196-- State threaded around the monad for thread management
197------------------------------------------------------------------------------
199-- | Identify the type of the SVar. Two computations using the same style can
200-- be scheduled on the same SVar.
202 AsyncVar -- depth first concurrent
203 | WAsyncVar -- breadth first concurrent
204 | ParallelVar -- all parallel
205 | AheadVar -- Concurrent look ahead
208-- | An SVar or a Stream Var is a conduit to the output from multiple streams
209-- running concurrently and asynchronously. An SVar can be thought of as an
210-- asynchronous IO handle. We can write any number of streams to an SVar in a
211-- non-blocking manner and then read them back at any time at any pace. The
212-- SVar would run the streams asynchronously and accumulate results. An SVar
213-- may not really execute the stream completely and accumulate all the results.
214-- However, it ensures that the reader can read the results at whatever paces
215-- it wants to read. The SVar monitors and adapts to the consumer's pace.
217-- An SVar is a mini scheduler, it has an associated workLoop that holds the
218-- stream tasks to be picked and run by a pool of worker threads. It has an
219-- associated output queue where the output stream elements are placed by the
220-- worker threads. A outputDoorBell is used by the worker threads to intimate the
221-- consumer thread about availability of new results in the output queue. More
222-- workers are added to the SVar by 'fromStreamVar' on demand if the output
223-- produced is not keeping pace with the consumer. On bounded SVars, workers
224-- block on the output queue to provide throttling of the producer when the
225-- consumer is not pulling fast enough. The number of workers may even get
226-- reduced depending on the consuming pace.
228-- New work is enqueued either at the time of creation of the SVar or as a
229-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
230-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
232-- We measure the individual worker latencies to estimate the number of workers
233-- needed or the amount of time we have to sleep between dispatches to achieve
234-- a particular rate when controlled pace mode it used.
235data WorkerInfo = WorkerInfo
236 { workerYieldMax :: Count -- 0 means unlimited
237 -- total number of yields by the worker till now
238 , workerYieldCount :: IORef Count
239 -- yieldCount at start, timestamp
240 , workerLatencyStart :: IORef (Count, TimeSpec)
244-- | Specifies the stream yield rate in yields per second (@Hertz@).
245-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
246-- allow only as many yields as we have accumulated as per 'rateGoal' since the
247-- start of time. If the consumer or the producer is slower or faster, the
248-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
249-- between the two by increasing or decreasing the pull rate from the producer.
250-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
251-- much as 'rateBuffer'.
253-- 'rateLow' puts a bound on how low the instantaneous rate can go when
254-- recovering the rate gap. In other words, it determines the maximum yield
255-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
256-- rate can go when recovering the rate gap. In other words, it determines the
257-- minimum yield latency. We reduce the latency by increasing concurrency,
258-- therefore we can say that it puts an upper bound on concurrency.
260-- If the 'rateGoal' is 0 or negative the stream never yields a value.
261-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
265 { rateLow :: Double -- ^ The lower rate limit
266 , rateGoal :: Double -- ^ The target rate we want to achieve
267 , rateHigh :: Double -- ^ The upper rate limit
268 , rateBuffer :: Int -- ^ Maximum slack from the goal
271data LatencyRange = LatencyRange
272 { minLatency :: NanoSecs
273 , maxLatency :: NanoSecs
277data YieldRateInfo = YieldRateInfo
278 { svarLatencyTarget :: NanoSecs
279 , svarLatencyRange :: LatencyRange
280 , svarRateBuffer :: Int
281 , svarGainedLostYields :: IORef Count
283 -- Actual latency/througput as seen from the consumer side, we count the
284 -- yields and the time it took to generates those yields. This is used to
285 -- increase or decrease the number of workers needed to achieve the desired
286 -- rate. The idle time of workers is adjusted in this, so that we only
287 -- account for the rate when the consumer actually demands data.
288 -- XXX interval latency is enough, we can move this under diagnostics build
289 , svarAllTimeLatency :: IORef (Count, TimeSpec)
291 -- XXX Worker latency specified by the user to be used before the first
292 -- actual measurement arrives. Not yet implemented
293 , workerBootstrapLatency :: Maybe NanoSecs
295 -- After how many yields the worker should update the latency information.
296 -- If the latency is high, this count is kept lower and vice-versa. XXX If
297 -- the latency suddenly becomes too high this count may remain too high for
298 -- long time, in such cases the consumer can change it.
299 -- 0 means no latency computation
300 -- XXX this is derivable from workerMeasuredLatency, can be removed.
301 , workerPollingInterval :: IORef Count
303 -- This is in progress latency stats maintained by the workers which we
304 -- empty into workerCollectedLatency stats at certain intervals - whenever
305 -- we process the stream elements yielded in this period.
306 -- (yieldCount, timeTaken)
307 , workerPendingLatency :: IORef (Count, NanoSecs)
309 -- This is the second level stat which is an accmulation from
310 -- workerPendingLatency stats. We keep accumulating latencies in this
311 -- bucket until we have stats for a sufficient period and then we reset it
312 -- to start collecting for the next period and retain the computed average
313 -- latency for the last period in workerMeasuredLatency.
314 -- (yieldCount, timeTaken)
315 , workerCollectedLatency :: IORef (Count, NanoSecs)
317 -- Latency as measured by workers, aggregated for the last period.
318 , workerMeasuredLatency :: IORef NanoSecs
321data SVarStats = SVarStats {
322 totalDispatches :: IORef Int
323 , maxWorkers :: IORef Int
324 , maxOutQSize :: IORef Int
325 , maxHeapSize :: IORef Int
326 , maxWorkQSize :: IORef Int
327 , avgWorkerLatency :: IORef (Count, NanoSecs)
328 , minWorkerLatency :: IORef NanoSecs
329 , maxWorkerLatency :: IORef NanoSecs
330 , svarStopTime :: IORef (Maybe TimeSpec)
333data Limit = Unlimited | Limited Word deriving Show
335data SVar t m a = SVar
338 svarStyle :: SVarStyle
340 -- Shared output queue (events, length)
341 -- XXX For better efficiency we can try a preallocated array type (perhaps
342 -- something like a vector) that allows an O(1) append. That way we will
343 -- avoid constructing and reversing the list. Possibly we can also avoid
344 -- the GC copying overhead. When the size increases we should be able to
345 -- allocate the array in chunks.
346 , outputQueue :: IORef ([ChildEvent a], Int)
347 , outputDoorBell :: MVar () -- signal the consumer about output
348 , readOutputQ :: m [ChildEvent a]
349 , postProcess :: m Bool
351 -- Combined/aggregate parameters
352 , maxWorkerLimit :: Limit
353 , maxBufferLimit :: Limit
354 , remainingWork :: Maybe (IORef Count)
355 , yieldRateInfo :: Maybe YieldRateInfo
357 -- Used only by bounded SVar types
358 , enqueue :: t m a -> IO ()
359 , isWorkDone :: IO Bool
360 , isQueueDone :: IO Bool
361 , needDoorBell :: IORef Bool
362 , workLoop :: Maybe WorkerInfo -> m ()
364 -- Shared, thread tracking
365 , workerThreads :: IORef (Set ThreadId)
366 , workerCount :: IORef Int
367 , accountThread :: ThreadId -> m ()
368 , workerStopMVar :: MVar ()
370 , svarStats :: SVarStats
371 -- to track garbage collection of SVar
372 , svarRef :: Maybe (IORef ())
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)
382-------------------------------------------------------------------------------
383-- State for concurrency control
384-------------------------------------------------------------------------------
386-- XXX we can put the resettable fields in a oneShotConfig field and others in
387-- a persistentConfig field. That way reset would be fast and scalable
388-- irrespective of the number of fields.
390-- XXX make all these Limited types and use phantom types to distinguish them
391data State t m a = State
392 { -- one shot configuration, automatically reset for each API call
393 streamVar :: Maybe (SVar t m a)
394 , _yieldLimit :: Maybe Count
396 -- persistent configuration, state that remains valid until changed by
397 -- an explicit setting via a combinator.
398 , _threadsHigh :: Limit
399 , _bufferHigh :: Limit
400 -- XXX these two can be collapsed into a single type
401 , _streamLatency :: Maybe NanoSecs -- bootstrap latency
402 , _maxStreamRate :: Maybe Rate
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
432-- XXX if perf gets affected we can have all the Nothing params in a single
433-- structure so that we reset is fast. We can also use rewrite rules such that
434-- reset occurs only in concurrent streams to reduce the impact on serial
436-- We can optimize this so that we clear it only if it is a Just value, it
437-- results in slightly better perf for zip/zipM but the performance of scan
438-- worsens a lot, it does not fuse.
439rstState :: State t m a -> State t m b
441 { streamVar = Nothing
442 , _yieldLimit = Nothing
445-------------------------------------------------------------------------------
446-- Smart get/set routines for State
447-------------------------------------------------------------------------------
449-- Use get/set routines instead of directly accessing the State fields
450setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
451setYieldLimit lim st =
458 else Just (fromIntegral n)
461getYieldLimit :: State t m a -> Maybe Count
462getYieldLimit = _yieldLimit
464setMaxThreads :: Int -> State t m a -> State t m a
470 then defaultMaxThreads
471 else Limited (fromIntegral n)
474getMaxThreads :: State t m a -> Limit
475getMaxThreads = _threadsHigh
477setMaxBuffer :: Int -> State t m a -> State t m a
483 then defaultMaxBuffer
484 else Limited (fromIntegral n)
487getMaxBuffer :: State t m a -> Limit
488getMaxBuffer = _bufferHigh
490setStreamRate :: Maybe Rate -> State t m a -> State t m a
491setStreamRate r st = st { _maxStreamRate = r }
493getStreamRate :: State t m a -> Maybe Rate
494getStreamRate = _maxStreamRate
496setStreamLatency :: Int -> State t m a -> State t m a
497setStreamLatency n st =
498 st { _streamLatency =
503 else Just (fromIntegral n)
506getStreamLatency :: State t m a -> Maybe NanoSecs
507getStreamLatency = _streamLatency
509-------------------------------------------------------------------------------
511-------------------------------------------------------------------------------
513cleanupSVar :: SVar t m a -> IO ()
515 workers <- readIORef (workerThreads sv)
516 Prelude.mapM_ (\tid -> throwTo tid ThreadAbort)
519cleanupSVarFromWorker :: SVar t m a -> IO ()
520cleanupSVarFromWorker sv = do
521 workers <- readIORef (workerThreads sv)
523 mapM_ (\tid -> throwTo tid ThreadAbort)
524 (S.toList workers \\ [self])
526-------------------------------------------------------------------------------
527-- Dumping the SVar for debug/diag
528-------------------------------------------------------------------------------
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 :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
558drainLatency _ss 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)
579 minLat <- readIORef (minWorkerLatency _ss)
580 when (new < minLat || minLat == 0) $
581 writeIORef (minWorkerLatency _ss) new
583 maxLat <- readIORef (maxWorkerLatency _ss)
584 when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
586 -- To avoid minor fluctuations update in batches
587 writeIORef col (0, 0)
588 writeIORef measured new
590 modifyIORef (avgWorkerLatency _ss) $
591 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
593 modifyIORef longTerm $ \(_, t) -> (lcount', t)
594 return (lcount', ltime, new)
595 else return notUpdated
597dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
598dumpSVarStats sv ss style = do
599 case yieldRateInfo sv of
602 _ <- liftIO $ drainLatency (svarStats sv) yinfo
605 dispatches <- readIORef $ totalDispatches ss
606 maxWrk <- readIORef $ maxWorkers ss
607 maxOq <- readIORef $ maxOutQSize ss
608 maxHp <- readIORef $ maxHeapSize ss
609 minLat <- readIORef $ minWorkerLatency ss
610 maxLat <- readIORef $ maxWorkerLatency ss
611 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
612 (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
613 Nothing -> return (0, 0, 0)
615 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
618 t <- readIORef (svarStopTime ss)
619 gl <- readIORef (svarGainedLostYields yinfo)
622 now <- getTime Monotonic
623 let interval = toNanoSecs (now - startTime)
624 return $ (cnt, gl, interval `div` fromIntegral cnt)
626 let interval = toNanoSecs (stopTime - startTime)
627 return $ (cnt, gl, interval `div` fromIntegral cnt)
628 else return (0, 0, 0)
631 [ "total dispatches = " ++ show dispatches
632 , "max workers = " ++ show maxWrk
633 , "max outQSize = " ++ show maxOq
634 ++ (if style == AheadVar
635 then "\nheap max size = " ++ show maxHp
638 then "\nmin worker latency = "
639 ++ secs (fromIntegral minLat * 1e-9)
642 then "\nmax worker latency = "
643 ++ secs (fromIntegral maxLat * 1e-9)
646 then let lat = avgTime `div` fromIntegral avgCnt
647 in "\navg worker latency = "
648 ++ secs (fromIntegral lat * 1e-9)
651 then "\nSVar latency = "
652 ++ secs (fromIntegral svarLat * 1e-9)
655 then "\nSVar yield count = " ++ show svarCnt
657 ++ (if svarGainLossCnt > 0
658 then "\nSVar gain/loss yield count = " ++ show svarGainLossCnt
662{-# NOINLINE dumpSVar #-}
663dumpSVar :: SVar t m a -> IO String
665 (oqList, oqLen) <- readIORef $ outputQueue sv
666 db <- tryTakeMVar $ outputDoorBell sv
668 if svarStyle sv == AheadVar
670 (oheap, oheapSeq) <- readIORef $ outputHeap sv
671 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
673 [ "heap length = " ++ show (H.size oheap)
674 , "heap seqeunce = " ++ show oheapSeq
675 , "work queue length = " ++ show (length wq)
676 , "work queue sequence = " ++ show wqSeq
680 let style = svarStyle sv
682 if style /= ParallelVar
683 then readIORef $ needDoorBell sv
685 rthread <- readIORef $ workerThreads sv
686 workers <- readIORef $ workerCount sv
687 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
690 [ "Creator tid = " ++ show (svarCreator sv)
691 , "style = " ++ show (svarStyle sv)
692 , "---------CURRENT STATE-----------"
693 , "outputQueue length computed = " ++ show (length oqList)
694 , "outputQueue length maintained = " ++ show oqLen
695 -- XXX print the types of events in the outputQueue, first 5
696 , "outputDoorBell = " ++ show db
698 ++ aheadDump ++ unlines
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{-# NOINLINE mvarExcHandler #-}
708mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
709mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
710 svInfo <- dumpSVar sv
711 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo
714{-# NOINLINE stmExcHandler #-}
715stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
716stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
717 svInfo <- dumpSVar sv
718 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo
721withDBGMVar :: SVar t m a -> String -> IO () -> IO ()
722withDBGMVar sv label action =
723 action `catches` [ Handler (mvarExcHandler sv label)
724 , Handler (stmExcHandler sv label)
727withDBGMVar :: SVar t m a -> String -> IO () -> IO ()
728withDBGMVar _ _ action = action
731-------------------------------------------------------------------------------
733-------------------------------------------------------------------------------
735-- Slightly faster version of CAS. Gained some improvement by avoiding the use
736-- of "evaluate" because we know we do not have exceptions in fn.
737{-# INLINE atomicModifyIORefCAS #-}
738atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
739atomicModifyIORefCAS ref fn = do
740 tkt <- readForCAS ref
746 loop _ 0 = atomicModifyIORef ref fn
748 let (new, result) = fn $ peekTicket old
749 (success, tkt) <- casIORef ref old new
752 else loop tkt (tries - 1)
754------------------------------------------------------------------------------
755-- Spawning threads and collecting result in streamed fashion
756------------------------------------------------------------------------------
758-- | A monad that can perform concurrent or parallel IO operations. Streams
759-- that can be composed concurrently require the underlying monad to be
763type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
765-- Stolen from the async package. The perf improvement is modest, 2% on a
766-- thread heavy benchmark (parallel composition using noop computations).
767-- A version of forkIO that does not include the outer exception
768-- handler: saves a bit of time when we will be installing our own
770{-# INLINE rawForkIO #-}
771rawForkIO :: IO () -> IO ThreadId
772rawForkIO action = IO $ \ s ->
773 case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
776doFork :: MonadBaseControl IO m
778 -> (SomeException -> IO ())
780doFork action exHandler =
781 control $ \runInIO ->
782 mask $ \restore -> do
783 tid <- rawForkIO $ catch (restore $ void $ runInIO action)
787-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
788-- faster, along with the fields in sv required by send?
789-- XXX make it noinline
791-- XXX we may want to employ an increment and decrement in batches when the
792-- througput is high or when the cost of synchronization is high. For example
793-- if the application is distributed then inc/dec of a shared variable may be
796-- Note that we need it to be an Int type so that we have the ability to undo a
797-- decrement that takes below zero.
798{-# INLINE decrementYieldLimit #-}
799decrementYieldLimit :: SVar t m a -> IO Bool
800decrementYieldLimit sv =
801 case remainingWork sv of
802 Nothing -> return True
804 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
807-- decrementYieldLimit returns False when the old limit is 0. This one returns
808-- False when the old limit is 1.
809{-# INLINE decrementYieldLimitPost #-}
810decrementYieldLimitPost :: SVar t m a -> IO Bool
811decrementYieldLimitPost sv =
812 case remainingWork sv of
813 Nothing -> return True
815 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
818{-# INLINE incrementYieldLimit #-}
819incrementYieldLimit :: SVar t m a -> IO ()
820incrementYieldLimit sv =
821 case remainingWork sv of
823 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
825-- XXX exception safety of all atomic/MVar operations
827-- TBD Each worker can have their own queue and the consumer can empty one
828-- queue at a time, that way contention can be reduced.
830-- XXX Only yields should be counted in the buffer limit and not the Stop
833-- | This function is used by the producer threads to queue output for the
834-- consumer thread to consume. Returns whether the queue has more space.
835send :: SVar t m a -> ChildEvent a -> IO Bool
837 -- XXX can the access to outputQueue and maxBufferLimit be made faster
839 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
840 ((msg : es, n + 1), n)
842 -- The wake up must happen only after the store has finished otherwise
843 -- we can have lost wakeup problems.
845 -- Since multiple workers can try this at the same time, it is possible
846 -- that we may put a spurious MVar after the consumer has already seen
847 -- the output. But that's harmless, at worst it may cause the consumer
848 -- to read the queue again and find it empty.
849 -- The important point is that the consumer is guaranteed to receive a
850 -- doorbell if something was added to the queue after it empties it.
851 void $ tryPutMVar (outputDoorBell sv) ()
853 -- XXX we should reserve the buffer when we pick up the work from the
854 -- queue, instead of checking it here when it is too late.
855 let limit = maxBufferLimit sv
857 Unlimited -> return True
859 active <- readIORef (workerCount sv)
860 return $ len < ((fromIntegral lim) - active)
862workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecs))
863workerCollectLatency winfo = do
864 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
865 cnt1 <- readIORef (workerYieldCount winfo)
866 let cnt = cnt1 - cnt0
870 t1 <- getTime Monotonic
871 let period = fromInteger $ toNanoSecs (t1 - t0)
872 writeIORef (workerLatencyStart winfo) (cnt1, t1)
873 return $ Just (cnt, period)
876-- XXX There are a number of gotchas in measuring latencies.
877-- 1) We measure latencies only when a worker yields a value
878-- 2) It is possible that a stream calls the stop continuation, in which case
879-- the worker would not yield a value and we would not account that worker in
880-- latencies. Even though this case should ideally be accounted we do not
881-- account it because we cannot or do not distinguish it from the case
883-- 3) It is possible that a worker returns without yielding anything because it
884-- never got a chance to pick up work.
886-- We can fix this if we measure the latencies by counting the work items
887-- picked rather than based on the outputs yielded.
888workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
889workerUpdateLatency yinfo winfo = do
890 r <- workerCollectLatency winfo
892 Just (cnt, period) -> do
893 let ref = workerPendingLatency yinfo
894 atomicModifyIORefCAS_ ref $ \(n, t) -> (n + cnt, t + period)
897updateYieldCount :: WorkerInfo -> IO Count
898updateYieldCount winfo = do
899 cnt <- readIORef (workerYieldCount winfo)
901 writeIORef (workerYieldCount winfo) cnt1
904isBeyondMaxYield :: Count -> WorkerInfo -> Bool
905isBeyondMaxYield cnt winfo =
906 let ymax = workerYieldMax winfo
907 in ymax /= 0 && cnt >= ymax
909-- XXX we should do rate control periodically based on the total yields rather
910-- than based on the worker local yields as other workers may have yielded more
911-- and we should stop based on the aggregate yields. However, latency update
912-- period can be based on individual worker yields.
913{-# NOINLINE checkRatePeriodic #-}
914checkRatePeriodic :: SVar t m a
919checkRatePeriodic sv yinfo winfo ycnt = do
920 i <- readIORef (workerPollingInterval yinfo)
921 -- XXX use generation count to check if the interval has been updated
922 if (i /= 0 && (ycnt `mod` i) == 0)
924 workerUpdateLatency yinfo winfo
925 -- XXX not required for parallel streams
926 isBeyondMaxRate sv yinfo
929-- CAUTION! this also updates the yield count and therefore should be called
930-- only when we are actually yielding an element.
931{-# NOINLINE workerRateControl #-}
932workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
933workerRateControl sv yinfo winfo = do
934 cnt <- updateYieldCount winfo
935 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
936 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
938-- XXX we should do rate control here but not latency update in case of ahead
939-- streams. latency update must be done when we yield directly to outputQueue
940-- or when we yield to heap.
941{-# INLINE sendYield #-}
942sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
943sendYield sv mwinfo msg = do
948 case yieldRateInfo sv of
949 Nothing -> return True
950 Just yinfo -> workerRateControl sv yinfo winfo
951 Nothing -> return True
952 return $ r && rateLimitOk
954{-# INLINE workerStopUpdate #-}
955workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
956workerStopUpdate winfo info = do
957 i <- readIORef (workerPollingInterval info)
958 when (i /= 0) $ workerUpdateLatency info winfo
960{-# INLINABLE sendStop #-}
961sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
962sendStop sv mwinfo = do
963 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
966 case yieldRateInfo sv of
968 Just info -> workerStopUpdate winfo info
970 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
972-------------------------------------------------------------------------------
974-------------------------------------------------------------------------------
976-- Note: For purely right associated expressions this queue should have at most
977-- one element. It grows to more than one when we have left associcated
978-- expressions. Large left associated compositions can grow this to a
980{-# INLINE enqueueLIFO #-}
981enqueueLIFO :: SVar t m a -> IORef [t m a] -> t m a -> IO ()
982enqueueLIFO sv q m = do
983 atomicModifyIORefCAS_ q $ \ms -> m : ms
985 w <- readIORef $ needDoorBell sv
987 -- Note: the sequence of operations is important for correctness here.
988 -- We need to set the flag to false strictly before sending the
989 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
990 -- then we may set the flag to False to later making the consumer lose
991 -- the flag, even without receiving a outputDoorBell.
992 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
993 void $ tryPutMVar (outputDoorBell sv) ()
995-------------------------------------------------------------------------------
997-------------------------------------------------------------------------------
999-- XXX we can use the Ahead style sequence/heap mechanism to make the best
1000-- effort to always try to finish the streams on the left side of an expression
1001-- first as long as possible.
1003{-# INLINE enqueueFIFO #-}
1004enqueueFIFO :: SVar t m a -> LinkedQueue (t m a) -> t m a -> IO ()
1005enqueueFIFO sv q m = do
1008 w <- readIORef $ needDoorBell sv
1010 -- Note: the sequence of operations is important for correctness here.
1011 -- We need to set the flag to false strictly before sending the
1012 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
1013 -- then we may set the flag to False to later making the consumer lose
1014 -- the flag, even without receiving a outputDoorBell.
1015 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1016 void $ tryPutMVar (outputDoorBell sv) ()
1018-------------------------------------------------------------------------------
1020-------------------------------------------------------------------------------
1022-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
1023-- but always serve them in the same order as they appear in the stream. To
1024-- implement lookahead streams efficiently we assign a sequence number to each
1025-- task when the task is picked up for execution. When the task finishes, the
1026-- output is tagged with the same sequence number and we rearrange the outputs
1027-- in sequence based on that number.
1029-- To explain the mechanism imagine that the current task at the head of the
1030-- stream has a "token" to yield to the outputQueue. The ownership of the token
1031-- is determined by the current sequence number is maintained in outputHeap.
1032-- Sequence number is assigned when a task is queued. When a thread dequeues a
1033-- task it picks up the sequence number as well and when the output is ready it
1034-- uses the sequence number to queue the output to the outputQueue.
1036-- The thread with current sequence number sends the output directly to the
1037-- outputQueue. Other threads push the output to the outputHeap. When the task
1038-- being queued on the heap is a stream of many elements we evaluate only the
1039-- first element and keep the rest of the unevaluated computation in the heap.
1040-- When such a task gets the "token" for outputQueue it evaluates and directly
1041-- yields all the elements to the outputQueue without checking for the
1044-- Note that no two outputs in the heap can have the same sequence numbers and
1045-- therefore we do not need a stable heap. We have also separated the buffer
1046-- for the current task (outputQueue) and the pending tasks (outputHeap) so
1047-- that the pending tasks cannot interfere with the current task. Note that for
1048-- a single task just the outputQueue is enough and for the case of many
1049-- threads just a heap is good enough. However we balance between these two
1050-- cases, so that both are efficient.
1052-- For bigger streams it may make sense to have separate buffers for each
1053-- stream. However, for singleton streams this may become inefficient. However,
1054-- if we do not have separate buffers, then the streams that come later in
1055-- sequence may hog the buffer, hindering the streams that are ahead. For this
1056-- reason we have a single element buffer limitation for the streams being
1057-- executed in advance.
1059-- This scheme works pretty efficiently with less than 40% extra overhead
1060-- compared to the Async streams where we do not have any kind of sequencing of
1061-- the outputs. It is especially devised so that we are most efficient when we
1062-- have short tasks and need just a single thread. Also when a thread yields
1063-- many items it can hold lockfree access to the outputQueue and do it
1066-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
1067-- that they do not hog the resources and hinder the progress of the threads in
1070-- XXX Left associated ahead expressions are expensive. We start a new SVar for
1071-- each left associative expression. The queue is used only for right
1072-- associated expression, we queue the right expression and execute the left.
1073-- Thererefore the queue never has more than one item in it.
1075-- XXX we can fix this. When we queue more than one item on the queue we can
1076-- mark the previously queued item as not-runnable. The not-runnable item is
1077-- not dequeued until the already running one has finished and at that time we
1078-- would also know the exact sequence number of the already queued item.
1080-- we can even run the already queued items but they will have to be sorted in
1081-- layers in the heap. We can use a list of heaps for that.
1082{-# INLINE enqueueAhead #-}
1083enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1084enqueueAhead sv q m = do
1085 atomicModifyIORefCAS_ q $ \ case
1086 ([], n) -> ([m], n + 1) -- increment sequence
1087 _ -> error "not empty"
1089 w <- readIORef $ needDoorBell sv
1091 -- Note: the sequence of operations is important for correctness here.
1092 -- We need to set the flag to false strictly before sending the
1093 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
1094 -- then we may set the flag to False to later making the consumer lose
1095 -- the flag, even without receiving a outputDoorBell.
1096 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1097 void $ tryPutMVar (outputDoorBell sv) ()
1099-- enqueue without incrementing the sequence number
1100{-# INLINE reEnqueueAhead #-}
1101reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
1102reEnqueueAhead sv q m = do
1103 atomicModifyIORefCAS_ q $ \ case
1104 ([], n) -> ([m], n) -- DO NOT increment sequence
1105 _ -> error "not empty"
1107 w <- readIORef $ needDoorBell sv
1109 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1110 void $ tryPutMVar (outputDoorBell sv) ()
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 $ do
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') | seqNo == n ->
1175 ((hp', Nothing), Ready ent)
1176 _ -> (pair, Waiting n)
1178{-# INLINE dequeueFromHeapSeq #-}
1180 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1182 -> IO (HeapDequeueResult t m a)
1183dequeueFromHeapSeq hpVar i =
1184 atomicModifyIORef hpVar $ \(hp, snum) ->
1189 Just (ent@(Entry seqNo _ev), hp') | seqNo == i ->
1190 ((hp', Nothing), Ready ent)
1191 _ -> ((hp, Just i), Waiting i)
1192 Just _ -> error "dequeueFromHeapSeq: unreachable"
1194{-# INLINE requeueOnHeapTop #-}
1196 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1197 -> Entry Int (AheadHeapEntry t m a)
1200requeueOnHeapTop hpVar ent seqNo =
1201 atomicModifyIORef_ hpVar $ \(hp, _) -> (H.insert ent hp, Just seqNo)
1203{-# INLINE updateHeapSeq #-}
1205 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1208updateHeapSeq hpVar seqNo =
1209 atomicModifyIORef_ hpVar $ \(hp, _) -> (hp, Just seqNo)
1211-------------------------------------------------------------------------------
1213-------------------------------------------------------------------------------
1215-- XXX To be implemented. Use a linked queue like WAsync and put back the
1216-- remaining computation at the back of the queue instead of the heap, and
1217-- increment the sequence number.
1219-- Thread tracking is needed for two reasons:
1221-- 1) Killing threads on exceptions. Threads may not be left to go away by
1222-- themselves because they may run for significant times before going away or
1223-- worse they may be stuck in IO and never go away.
1225-- 2) To know when all threads are done and the stream has ended.
1227{-# NOINLINE addThread #-}
1228addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1230 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
1232-- This is cheaper than modifyThread because we do not have to send a
1233-- outputDoorBell This can make a difference when more workers are being
1235{-# INLINE delThread #-}
1236delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1238 liftIO $ modifyIORef (workerThreads sv) $ (\s -> S.delete tid s)
1240-- If present then delete else add. This takes care of out of order add and
1241-- delete i.e. a delete arriving before we even added a thread.
1242-- This occurs when the forked thread is done even before the 'addThread' right
1243-- after the fork gets a chance to run.
1244{-# INLINE modifyThread #-}
1245modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1246modifyThread sv tid = do
1247 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1248 if (S.member tid old)
1249 then let new = (S.delete tid old) in (new, new)
1250 else let new = (S.insert tid old) in (new, old)
1254 void $ tryPutMVar (outputDoorBell sv) ()
1257-- | This is safe even if we are adding more threads concurrently because if
1258-- a child thread is adding another thread then anyway 'workerThreads' will
1260{-# INLINE allThreadsDone #-}
1261allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1262allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1264{-# NOINLINE handleChildException #-}
1265handleChildException :: SVar t m a -> SomeException -> IO ()
1266handleChildException sv e = do
1268 void $ send sv (ChildStop tid (Just e))
1271recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1272recordMaxWorkers sv = liftIO $ do
1273 active <- readIORef (workerCount sv)
1274 maxWrk <- readIORef (maxWorkers $ svarStats sv)
1275 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1276 modifyIORef (totalDispatches $ svarStats sv) (+1)
1279{-# NOINLINE pushWorker #-}
1280pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1281pushWorker yieldMax sv = do
1282 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1286 -- This allocation matters when significant number of workers are being
1287 -- sent. We allocate it only when needed.
1289 case yieldRateInfo sv of
1290 Nothing -> return Nothing
1291 Just _ -> liftIO $ do
1292 cntRef <- newIORef 0
1293 t <- getTime Monotonic
1294 lat <- newIORef (0, t)
1295 return $ Just $ WorkerInfo
1296 { workerYieldMax = yieldMax
1297 , workerYieldCount = cntRef
1298 , workerLatencyStart = lat
1300 doFork (workLoop sv winfo) (handleChildException sv) >>= addThread sv
1302-- XXX we can push the workerCount modification in accountThread and use the
1303-- same pushWorker for Parallel case as well.
1305-- | In contrast to pushWorker which always happens only from the consumer
1306-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1307-- producer side. So we need to use a thread safe modification of
1308-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1309-- using a CAS based modification.
1310{-# NOINLINE pushWorkerPar #-}
1311pushWorkerPar :: MonadAsync m => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1312pushWorkerPar sv wloop = do
1313 -- We do not use workerCount in case of ParallelVar but still there is no
1314 -- harm in maintaining it correctly.
1316 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1318 -- This allocation matters when significant number of workers are being
1319 -- sent. We allocate it only when needed. The overhead increases by 4x.
1321 case yieldRateInfo sv of
1322 Nothing -> return Nothing
1323 Just _ -> liftIO $ do
1324 cntRef <- newIORef 0
1325 t <- getTime Monotonic
1326 lat <- newIORef (0, t)
1327 return $ Just $ WorkerInfo
1328 { workerYieldMax = 0
1329 , workerYieldCount = cntRef
1330 , workerLatencyStart = lat
1333 doFork (wloop winfo) (handleChildException sv) >>= modifyThread sv
1335 doFork (wloop Nothing) (handleChildException sv) >>= modifyThread sv
1339-- True: can dispatch more
1340-- False: cannot dispatch any more
1341dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1342dispatchWorker yieldCount sv = do
1343 let workerLimit = maxWorkerLimit sv
1344 -- XXX in case of Ahead streams we should not send more than one worker
1345 -- when the work queue is done but heap is not done.
1346 done <- liftIO $ isWorkDone sv
1347 -- Note, "done" may not mean that the work is actually finished if there
1348 -- are workers active, because there may be a worker which has not yet
1349 -- queued the leftover work.
1352 qDone <- liftIO $ isQueueDone sv
1353 -- Note that the worker count is only decremented during event
1354 -- processing in fromStreamVar and therefore it is safe to read and
1355 -- use it without a lock.
1356 active <- liftIO $ readIORef $ workerCount sv
1359 -- Note that we may deadlock if the previous workers (tasks in the
1360 -- stream) wait/depend on the future workers (tasks in the stream)
1361 -- executing. In that case we should either configure the maxWorker
1362 -- count to higher or use parallel style instead of ahead or async
1364 limit <- case remainingWork sv of
1365 Nothing -> return workerLimit
1367 n <- liftIO $ readIORef ref
1370 Unlimited -> Limited (fromIntegral n)
1371 Limited lim -> Limited $ min lim (fromIntegral n)
1373 -- XXX for ahead streams shall we take the heap yields into account for
1374 -- controlling the dispatch? We should not dispatch if the heap has
1375 -- already got the limit covered.
1376 let dispatch = pushWorker yieldCount sv >> return True
1378 Unlimited -> dispatch
1379 -- Note that the use of remainingWork and workerCount is not
1380 -- atomic and the counts may even have changed between reading and
1381 -- using them here, so this is just approximate logic and we cannot
1382 -- rely on it for correctness. We may actually dispatch more
1383 -- workers than required.
1384 Limited lim | lim > 0 -> dispatch
1387 when (active <= 0) $ pushWorker 0 sv
1391-- | This is a magic number and it is overloaded, and used at several places to
1394-- 1. If we have to sleep to slowdown this is the minimum period that we
1395-- accumulate before we sleep. Also, workers do not stop until this much
1396-- sleep time is accumulated.
1397-- 3. Collected latencies are computed and transferred to measured latency
1398-- after a minimum of this period.
1399minThreadDelay :: NanoSecs
1400minThreadDelay = 10^(6 :: Int)
1402-- | Another magic number! When we have to start more workers to cover up a
1403-- number of yields that we are lagging by then we cannot start one worker for
1404-- each yield because that may be a very big number and if the latency of the
1405-- workers is low these number of yields could be very high. We assume that we
1406-- run each extra worker for at least this much time.
1407rateRecoveryTime :: NanoSecs
1408rateRecoveryTime = 1000000
1410nanoToMicroSecs :: NanoSecs -> Int
1411nanoToMicroSecs s = (fromIntegral s) `div` 1000
1413-- We either block, or send one worker with limited yield count or one or more
1414-- workers with unlimited yield count.
1416 = BlockWait NanoSecs
1417 | PartialWorker Count
1418 | ManyWorkers Int Count
1421-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1431estimateWorkers workerLimit svarYields gainLossYields
1432 svarElapsed wLatency targetLat range =
1433 -- XXX we can have a maxEfficiency combinator as well which runs the
1434 -- producer at the maximal efficiency i.e. the number of workers are chosen
1435 -- such that the latency is minimum or within a range. Or we can call it
1436 -- maxWorkerLatency.
1439 -- How many workers do we need to acheive the required rate?
1441 -- When the workers are IO bound we can increase the throughput by
1442 -- increasing the number of workers as long as the IO device has enough
1443 -- capacity to process all the requests concurrently. If the IO
1444 -- bandwidth is saturated increasing the workers won't help. Also, if
1445 -- the CPU utilization in processing all these requests exceeds the CPU
1446 -- bandwidth, then increasing the number of workers won't help.
1448 -- When the workers are purely CPU bound, increasing the workers beyond
1449 -- the number of CPUs won't help.
1451 -- TODO - measure the CPU and IO requirements of the workers. Have a
1452 -- way to specify the max bandwidth of the underlying IO mechanism and
1453 -- use that to determine the max rate of workers, and also take the CPU
1454 -- bandwidth into account. We can also discover the IO bandwidth if we
1455 -- know that we are not CPU bound, then how much steady state rate are
1456 -- we able to acheive. Design tests for CPU bound and IO bound cases.
1458 -- Calculate how many yields are we ahead or behind to match the exact
1459 -- required rate. Based on that we increase or decrease the effective
1462 -- When the worker latency is lower than required latency we begin with
1463 -- a yield and then wait rather than first waiting and then yielding.
1464 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1465 effectiveYields = svarYields + gainLossYields
1466 deltaYields = fromIntegral targetYields - effectiveYields
1468 -- We recover the deficit by running at a higher/lower rate for a
1469 -- certain amount of time. To keep the effective rate in reasonable
1470 -- limits we use rateRecoveryTime, minLatency and maxLatency.
1471 in if deltaYields > 0
1473 let deltaYieldsFreq :: Double
1475 fromIntegral deltaYields /
1476 fromIntegral rateRecoveryTime
1477 yieldsFreq = 1.0 / fromIntegral targetLat
1478 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1479 requiredLat = NanoSecs $ round $ 1.0 / totalYieldsFreq
1480 adjustedLat = min (max requiredLat (minLatency range))
1482 in assert (adjustedLat > 0) $
1483 if wLatency <= adjustedLat
1484 then PartialWorker deltaYields
1485 else let workers = withLimit $ wLatency `div` adjustedLat
1486 limited = min workers (fromIntegral deltaYields)
1487 in ManyWorkers (fromIntegral limited) deltaYields
1489 let expectedDuration = fromIntegral effectiveYields * targetLat
1490 sleepTime = expectedDuration - svarElapsed
1491 maxSleepTime = maxLatency range - wLatency
1492 s = min sleepTime maxSleepTime
1493 in assert (sleepTime >= 0) $
1494 -- if s is less than 0 it means our maxSleepTime is less
1495 -- than the worker latency.
1496 if (s > 0) then BlockWait s else ManyWorkers 1 (Count 0)
1501 Limited x -> min n (fromIntegral x)
1503-- | Get the worker latency without resetting workerPendingLatency
1504-- Returns (total yield count, base time, measured latency)
1505-- CAUTION! keep it in sync with collectLatency
1506getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1507getWorkerLatency yinfo = do
1508 let cur = workerPendingLatency yinfo
1509 col = workerCollectedLatency yinfo
1510 longTerm = svarAllTimeLatency yinfo
1511 measured = workerMeasuredLatency yinfo
1513 (count, time) <- readIORef cur
1514 (colCount, colTime) <- readIORef col
1515 (lcount, ltime) <- readIORef longTerm
1516 prev <- readIORef measured
1518 let pendingCount = colCount + count
1519 pendingTime = colTime + time
1522 then let lat = pendingTime `div` (fromIntegral pendingCount)
1523 -- XXX Give more weight to new?
1524 in (lat + prev) `div` 2
1526 return (lcount + pendingCount, ltime, new)
1528isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1529isBeyondMaxRate sv yinfo = do
1530 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1531 now <- getTime Monotonic
1532 let duration = fromInteger $ toNanoSecs $ now - tstamp
1533 let targetLat = svarLatencyTarget yinfo
1534 gainLoss <- readIORef (svarGainedLostYields yinfo)
1535 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1536 wLatency targetLat (svarLatencyRange yinfo)
1537 cnt <- readIORef $ workerCount sv
1538 return $ case work of
1539 -- XXX set the worker's maxYields or polling interval based on yields
1540 PartialWorker _yields -> cnt > 1
1541 ManyWorkers n _ -> cnt > n
1544-- Every once in a while workers update the latencies and check the yield rate.
1545-- They return if we are above the expected yield rate. If we check too often
1546-- it may impact performance, if we check less often we may have a stale
1547-- picture. We update every minThreadDelay but we translate that into a yield
1548-- count based on latency so that the checking overhead is little.
1550-- XXX use a generation count to indicate that the value is updated. If the
1551-- value is updated an existing worker must check it again on the next yield.
1552-- Otherwise it is possible that we may keep updating it and because of the mod
1553-- worker keeps skipping it.
1554updateWorkerPollingInterval :: YieldRateInfo -> NanoSecs -> IO ()
1555updateWorkerPollingInterval yinfo latency = do
1556 let periodRef = workerPollingInterval yinfo
1557 cnt = max 1 $ minThreadDelay `div` latency
1558 period = min cnt (fromIntegral magicMaxBuffer)
1560 writeIORef periodRef (fromIntegral period)
1562-- Returns a triple, (1) yield count since last collection, (2) the base time
1563-- when we started counting, (3) average latency in the last measurement
1564-- period. The former two are used for accurate measurement of the going rate
1565-- whereas the average is used for future estimates e.g. how many workers
1566-- should be maintained to maintain the rate.
1567-- CAUTION! keep it in sync with getWorkerLatency
1568collectLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1569collectLatency _ss yinfo = do
1570 let cur = workerPendingLatency yinfo
1571 col = workerCollectedLatency yinfo
1572 longTerm = svarAllTimeLatency yinfo
1573 measured = workerMeasuredLatency yinfo
1575 (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
1576 (colCount, colTime) <- readIORef col
1577 (lcount, ltime) <- readIORef longTerm
1578 prev <- readIORef measured
1580 let pendingCount = colCount + count
1581 pendingTime = colTime + time
1583 lcount' = lcount + pendingCount
1584 tripleWith lat = (lcount', ltime, lat)
1586 if (pendingCount > 0)
1588 let new = pendingTime `div` (fromIntegral pendingCount)
1590 minLat <- readIORef (minWorkerLatency _ss)
1591 when (new < minLat || minLat == 0) $
1592 writeIORef (minWorkerLatency _ss) new
1594 maxLat <- readIORef (maxWorkerLatency _ss)
1595 when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
1597 -- When we have collected a significant sized batch we compute the new
1598 -- latency using that batch and return the new latency, otherwise we
1599 -- return the previous latency derived from the previous batch.
1600 if (pendingCount > fromIntegral magicMaxBuffer)
1601 || (pendingTime > minThreadDelay)
1602 || (let r = (fromIntegral new) / (fromIntegral prev) :: Double
1603 in prev > 0 && (r > 2 || r < 0.5))
1606 updateWorkerPollingInterval yinfo (max new prev)
1607 writeIORef col (0, 0)
1608 writeIORef measured ((prev + new) `div` 2)
1610 modifyIORef (avgWorkerLatency _ss) $
1611 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
1613 modifyIORef longTerm $ \(_, t) -> (lcount', t)
1614 return $ tripleWith new
1616 writeIORef col (pendingCount, pendingTime)
1617 return $ tripleWith prev
1618 else return $ tripleWith prev
1620-- XXX in case of ahead style stream we need to take the heap size into account
1621-- because we return the workers on the basis of that which causes a condition
1622-- where we keep dispatching and they keep returning. So we must have exactly
1623-- the same logic for not dispatching and for returning.
1626-- True: can dispatch more
1627-- False: full, no more dispatches
1628dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1629dispatchWorkerPaced sv = do
1630 let yinfo = fromJust $ yieldRateInfo sv
1631 (svarYields, svarElapsed, wLatency) <- do
1632 now <- liftIO $ getTime Monotonic
1633 (yieldCount, baseTime, lat) <-
1634 liftIO $ collectLatency (svarStats sv) yinfo
1635 let elapsed = fromInteger $ toNanoSecs $ now - baseTime
1639 case workerBootstrapLatency yinfo of
1644 return (yieldCount, elapsed, latency)
1647 -- Need to measure the latency with a single worker before we can perform
1651 let workerLimit = maxWorkerLimit sv
1652 let targetLat = svarLatencyTarget yinfo
1653 let range = svarLatencyRange yinfo
1654 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1655 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
1656 wLatency targetLat range
1658 -- XXX we need to take yieldLimit into account here. If we are at the
1659 -- end of the limit as well as the time, we should not be sleeping.
1660 -- If we are not actually planning to dispatch any more workers we need
1661 -- to take that in account.
1664 assert (s >= 0) (return ())
1665 -- XXX note that when we return from here we will block waiting
1666 -- for the result from the existing worker. If that takes too
1667 -- long we won't be able to send another worker until the
1670 -- Sleep only if there are no active workers, otherwise we will
1671 -- defer the output of those. Note we cannot use workerCount
1672 -- here as it is not a reliable way to ensure there are
1673 -- definitely no active workers. When workerCount is 0 we may
1674 -- still have a Stop event waiting in the outputQueue.
1675 done <- allThreadsDone sv
1676 when done $ void $ do
1677 liftIO $ threadDelay $ nanoToMicroSecs s
1680 PartialWorker yields -> do
1681 assert (yields > 0) (return ())
1682 updateGainedLostYields yinfo yields
1684 done <- allThreadsDone sv
1685 when done $ void $ dispatchWorker yields sv
1687 ManyWorkers netWorkers yields -> do
1688 assert (netWorkers >= 1) (return ())
1689 assert (yields >= 0) (return ())
1690 updateGainedLostYields yinfo yields
1692 let periodRef = workerPollingInterval yinfo
1693 ycnt = max 1 $ yields `div` fromIntegral netWorkers
1694 period = min ycnt (fromIntegral magicMaxBuffer)
1696 old <- liftIO $ readIORef periodRef
1697 when (period < old) $
1698 liftIO $ writeIORef periodRef period
1700 cnt <- liftIO $ readIORef $ workerCount sv
1701 if (cnt < netWorkers)
1703 let total = netWorkers - cnt
1704 batch = max 1 $ fromIntegral $
1705 minThreadDelay `div` targetLat
1706 r <- dispatchN (min total batch)
1707 -- XXX stagger the workers over a period?
1708 -- XXX cannot sleep, as that would mean we cannot process the
1709 -- outputs. need to try a different mechanism to stagger.
1710 -- when (total > batch) $
1711 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1717 updateGainedLostYields yinfo yields = do
1718 let buf = fromIntegral $ svarRateBuffer yinfo
1719 when (yields /= 0 && abs yields > buf) $ do
1724 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1730 r <- dispatchWorker 0 sv
1732 then dispatchN (n - 1)
1735sendWorkerDelayPaced :: SVar t m a -> IO ()
1736sendWorkerDelayPaced _ = return ()
1738sendWorkerDelay :: SVar t m a -> IO ()
1739sendWorkerDelay _sv = do
1740 -- XXX we need a better way to handle this than hardcoded delays. The
1741 -- delays may be different for different systems.
1742 -- If there is a usecase where this is required we can create a combinator
1743 -- to set it as a config in the state.
1745 ncpu <- getNumCapabilities
1748 if (svarStyle sv == AheadVar)
1749 then threadDelay 100
1752 if (svarStyle sv == AheadVar)
1753 then threadDelay 100
1758{-# NOINLINE sendWorkerWait #-}
1761 => (SVar t m a -> IO ())
1762 -> (SVar t m a -> m Bool)
1765sendWorkerWait delay dispatch sv = do
1766 -- Note that we are guaranteed to have at least one outstanding worker when
1767 -- we enter this function. So if we sleep we are guaranteed to be woken up
1768 -- by an outputDoorBell, when the worker exits.
1771 (_, n) <- liftIO $ readIORef (outputQueue sv)
1773 -- The queue may be empty temporarily if the worker has dequeued the
1774 -- work item but has not enqueued the remaining part yet. For the same
1775 -- reason, a worker may come back if it tries to dequeue and finds the
1776 -- queue empty, even though the whole work has not finished yet.
1778 -- If we find that the queue is empty, but it may be empty
1779 -- temporarily, when we checked it. If that's the case we might
1780 -- sleep indefinitely unless the active workers produce some
1781 -- output. We may deadlock specially if the otuput from the active
1782 -- workers depends on the future workers that we may never send.
1783 -- So in case the queue was temporarily empty set a flag to inform
1784 -- the enqueue to send us a doorbell.
1786 -- Note that this is just a best effort mechanism to avoid a
1787 -- deadlock. Deadlocks may still happen if for some weird reason
1788 -- the consuming computation shares an MVar or some other resource
1789 -- with the producing computation and gets blocked on that resource
1790 -- and therefore cannot do any pushworker to add more threads to
1791 -- the producer. In such cases the programmer should use a parallel
1792 -- style so that all the producers are scheduled immediately and
1793 -- unconditionally. We can also use a separate monitor thread to
1794 -- push workers instead of pushing them from the consumer, but then
1795 -- we are no longer using pull based concurrency rate adaptation.
1797 -- XXX update this in the tutorial.
1799 -- Having pending active workers does not mean that we are guaranteed
1800 -- to be woken up if we sleep. In case of Ahead streams, there may be
1801 -- queued items in the heap even though the outputQueue is empty, and
1802 -- we may have active workers which are deadlocked on those items to be
1803 -- processed by the consumer. We should either guarantee that any
1804 -- worker, before returning, clears the heap or we send a worker to clear
1805 -- it. Normally we always send a worker if no output is seen, but if
1806 -- the thread limit is reached or we are using pacing then we may not
1807 -- send a worker. See the concurrentApplication test in the tests, that
1808 -- test case requires at least one yield from the producer to not
1809 -- deadlock, if the last workers output is stuck in the heap then this
1810 -- test fails. This problem can be extended to n threads when the
1811 -- consumer may depend on the evaluation of next n items in the
1814 -- register for the outputDoorBell before we check the queue so that if we
1815 -- sleep because the queue was empty we are guaranteed to get a
1816 -- doorbell on the next enqueue.
1818 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
1819 liftIO $ storeLoadBarrier
1820 canDoMore <- dispatch sv
1822 -- XXX test for the case when we miss sending a worker when the worker
1823 -- count is more than 1500.
1825 -- XXX Assert here that if the heap is not empty then there is at
1826 -- least one outstanding worker. Otherwise we could be sleeping
1830 then sendWorkerWait delay dispatch sv
1832 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
1833 $ takeMVar (outputDoorBell sv)
1834 (_, len) <- liftIO $ readIORef (outputQueue sv)
1835 when (len <= 0) $ sendWorkerWait delay dispatch sv
1837{-# INLINE readOutputQRaw #-}
1838readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
1839readOutputQRaw sv = do
1840 (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
1842 oqLen <- readIORef (maxOutQSize $ svarStats sv)
1843 when (len > oqLen) $ writeIORef (maxOutQSize $ svarStats sv) len
1847readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1848readOutputQBounded sv = do
1849 (list, len) <- liftIO $ readOutputQRaw sv
1850 -- When there is no output seen we dispatch more workers to help
1851 -- out if there is work pending in the work queue.
1855 -- send a worker proactively, if needed, even before we start
1856 -- processing the output. This may degrade single processor
1857 -- perf but improves multi-processor, because of more
1865 cnt <- liftIO $ readIORef $ workerCount sv
1866 when (cnt <= 0) $ do
1867 done <- liftIO $ isWorkDone sv
1868 when (not done) $ pushWorker 0 sv
1870 {-# INLINE blockingRead #-}
1872 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
1873 liftIO $ (readOutputQRaw sv >>= return . fst)
1875readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1876readOutputQPaced sv = do
1877 (list, len) <- liftIO $ readOutputQRaw sv
1881 -- XXX send a worker proactively, if needed, even before we start
1882 -- processing the output.
1883 void $ dispatchWorkerPaced sv
1888 {-# INLINE blockingRead #-}
1890 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
1891 liftIO $ (readOutputQRaw sv >>= return . fst)
1893postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
1894postProcessBounded sv = do
1895 workersDone <- allThreadsDone sv
1896 -- There may still be work pending even if there are no workers pending
1897 -- because all the workers may return if the outputQueue becomes full. In
1898 -- that case send off a worker to kickstart the work again.
1900 -- Note that isWorkDone can only be safely checked if all workers are done.
1901 -- When some workers are in progress they may have decremented the yield
1902 -- Limit and later ending up incrementing it again. If we look at the yield
1903 -- limit in that window we may falsely say that it is 0 and therefore we
1907 r <- liftIO $ isWorkDone sv
1908 -- Note that we need to guarantee a worker, therefore we cannot just
1909 -- use dispatchWorker which may or may not send a worker.
1910 when (not r) $ pushWorker 0 sv
1911 -- XXX do we need to dispatch many here?
1912 -- void $ dispatchWorker sv
1916postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
1917postProcessPaced sv = do
1918 workersDone <- allThreadsDone sv
1919 -- XXX If during consumption we figure out we are getting delayed then we
1920 -- should trigger dispatch there as well. We should try to check on the
1921 -- workers after consuming every n item from the buffer?
1924 r <- liftIO $ isWorkDone sv
1926 void $ dispatchWorkerPaced sv
1927 -- Note that we need to guarantee a worker since the work is not
1928 -- finished, therefore we cannot just rely on dispatchWorkerPaced
1929 -- which may or may not send a worker.
1930 noWorker <- allThreadsDone sv
1931 when noWorker $ pushWorker 0 sv
1935getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
1936getYieldRateInfo st = do
1937 -- convert rate in Hertz to latency in Nanoseconds
1938 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
1939 case getStreamRate st of
1940 Just (Rate low goal high buf) ->
1941 let l = rateToLatency goal
1942 minl = rateToLatency high
1943 maxl = rateToLatency low
1944 in mkYieldRateInfo l (LatencyRange minl maxl) buf
1945 Nothing -> return Nothing
1949 mkYieldRateInfo latency latRange buf = do
1950 measured <- newIORef 0
1951 wcur <- newIORef (0,0)
1952 wcol <- newIORef (0,0)
1953 now <- getTime Monotonic
1954 wlong <- newIORef (0,now)
1955 period <- newIORef 1
1956 gainLoss <- newIORef (Count 0)
1958 return $ Just YieldRateInfo
1959 { svarLatencyTarget = latency
1960 , svarLatencyRange = latRange
1961 , svarRateBuffer = buf
1962 , svarGainedLostYields = gainLoss
1963 , workerBootstrapLatency = getStreamLatency st
1964 , workerPollingInterval = period
1965 , workerMeasuredLatency = measured
1966 , workerPendingLatency = wcur
1967 , workerCollectedLatency = wcol
1968 , svarAllTimeLatency = wlong
1971getAheadSVar :: MonadAsync m
1973 -> ( IORef ([t m a], Int)
1974 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
1980getAheadSVar st f = do
1981 outQ <- newIORef ([], 0)
1982 -- the second component of the tuple is "Nothing" when heap is being
1983 -- cleared, "Just n" when we are expecting sequence number n to arrive
1984 -- before we can start clearing the heap.
1985 outH <- newIORef (H.empty, Just 0)
1986 outQMv <- newEmptyMVar
1987 active <- newIORef 0
1988 wfw <- newIORef False
1989 running <- newIORef S.empty
1990 q <- newIORef ([], -1)
1991 stopMVar <- newMVar ()
1992 yl <- case getYieldLimit st of
1993 Nothing -> return Nothing
1994 Just x -> Just <$> newIORef x
1995 rateInfo <- getYieldRateInfo st
1998 maxWrk <- newIORef 0
2002 avgLat <- newIORef (0, NanoSecs 0)
2003 maxLat <- newIORef (NanoSecs 0)
2004 minLat <- newIORef (NanoSecs 0)
2005 stpTime <- newIORef Nothing
2010 let getSVar sv readOutput postProc = SVar
2011 { outputQueue = outQ
2012 , remainingWork = yl
2013 , maxBufferLimit = getMaxBuffer st
2014 , maxWorkerLimit = getMaxThreads st
2015 , yieldRateInfo = rateInfo
2016 , outputDoorBell = outQMv
2017 , readOutputQ = readOutput sv
2018 , postProcess = postProc sv
2019 , workerThreads = running
2020 , workLoop = f q outH st{streamVar = Just sv} sv
2021 , enqueue = enqueueAhead sv q
2022 , isWorkDone = isWorkDoneAhead sv q outH
2023 , isQueueDone = isQueueDoneAhead sv q
2024 , needDoorBell = wfw
2025 , svarStyle = AheadVar
2026 , workerCount = active
2027 , accountThread = delThread sv
2028 , workerStopMVar = stopMVar
2032 , aheadWorkQueue = q
2035 , svarStats = SVarStats
2036 { totalDispatches = disp
2037 , maxWorkers = maxWrk
2038 , maxOutQSize = maxOq
2039 , maxHeapSize = maxHs
2040 , maxWorkQSize = maxWq
2041 , avgWorkerLatency = avgLat
2042 , minWorkerLatency = minLat
2043 , maxWorkerLatency = maxLat
2044 , svarStopTime = stpTime
2049 case getStreamRate st of
2050 Nothing -> getSVar sv readOutputQBounded postProcessBounded
2051 Just _ -> getSVar sv readOutputQPaced postProcessPaced
2056 {-# INLINE isQueueDoneAhead #-}
2057 isQueueDoneAhead sv q = do
2058 queueDone <- checkEmpty q
2060 case remainingWork sv of
2064 Nothing -> return False
2065 -- XXX note that yieldsDone can only be authoritative only when there
2066 -- are no workers running. If there are active workers they can
2067 -- later increment the yield count and therefore change the result.
2068 return $ yieldsDone || queueDone
2070 {-# INLINE isWorkDoneAhead #-}
2071 isWorkDoneAhead sv q ref = do
2073 (hp, _) <- readIORef ref
2074 return (H.size hp <= 0)
2075 queueDone <- isQueueDoneAhead sv q
2076 return $ heapDone && queueDone
2079 (xs, _) <- readIORef q
2082getParallelSVar :: MonadIO m => State t m a -> IO (SVar t m a)
2083getParallelSVar st = do
2084 outQ <- newIORef ([], 0)
2085 outQMv <- newEmptyMVar
2086 active <- newIORef 0
2087 running <- newIORef S.empty
2088 yl <- case getYieldLimit st of
2089 Nothing -> return Nothing
2090 Just x -> Just <$> newIORef x
2091 rateInfo <- getYieldRateInfo st
2094 maxWrk <- newIORef 0
2098 avgLat <- newIORef (0, NanoSecs 0)
2099 maxLat <- newIORef (NanoSecs 0)
2100 minLat <- newIORef (NanoSecs 0)
2101 stpTime <- newIORef Nothing
2107 SVar { outputQueue = outQ
2108 , remainingWork = yl
2109 , maxBufferLimit = Unlimited
2110 , maxWorkerLimit = Unlimited
2111 -- Used only for diagnostics
2112 , yieldRateInfo = rateInfo
2113 , outputDoorBell = outQMv
2114 , readOutputQ = readOutputQPar sv
2115 , postProcess = allThreadsDone sv
2116 , workerThreads = running
2117 , workLoop = undefined
2118 , enqueue = undefined
2119 , isWorkDone = undefined
2120 , isQueueDone = undefined
2121 , needDoorBell = undefined
2122 , svarStyle = ParallelVar
2123 , workerCount = active
2124 , accountThread = modifyThread sv
2125 , workerStopMVar = undefined
2129 , aheadWorkQueue = undefined
2130 , outputHeap = undefined
2132 , svarStats = SVarStats
2133 { totalDispatches = disp
2134 , maxWorkers = maxWrk
2135 , maxOutQSize = maxOq
2136 , maxHeapSize = maxHs
2137 , maxWorkQSize = maxWq
2138 , avgWorkerLatency = avgLat
2139 , minWorkerLatency = minLat
2140 , maxWorkerLatency = maxLat
2141 , svarStopTime = stpTime
2148 readOutputQPar sv = liftIO $ do
2149 withDBGMVar sv "readOutputQPar: doorbell" $ takeMVar (outputDoorBell sv)
2150 case yieldRateInfo sv of
2151 Nothing -> return ()
2152 Just yinfo -> void $ collectLatency (svarStats sv) yinfo
2153 readOutputQRaw sv >>= return . fst
2155sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
2156sendFirstWorker sv m = do
2157 -- Note: We must have all the work on the queue before sending the
2158 -- pushworker, otherwise the pushworker may exit before we even get a
2160 liftIO $ enqueue sv m
2161 case yieldRateInfo sv of
2162 Nothing -> pushWorker 0 sv
2164 if svarLatencyTarget yinfo == maxBound
2165 then liftIO $ threadDelay maxBound
2166 else pushWorker 1 sv
2169{-# INLINABLE newAheadVar #-}
2170newAheadVar :: MonadAsync m
2173 -> ( IORef ([t m a], Int)
2174 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2180newAheadVar st m wloop = do
2181 sv <- liftIO $ getAheadSVar st wloop
2182 sendFirstWorker sv m
2184{-# INLINABLE newParallelVar #-}
2185newParallelVar :: MonadAsync m => State t m a -> m (SVar t m a)
2186newParallelVar st = liftIO $ getParallelSVar st
2188-- XXX this errors out for Parallel/Ahead SVars
2189-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
2190-- be read back from the SVar using 'fromSVar'.
2191toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
2192toStreamVar sv m = do
2193 liftIO $ (enqueue sv) m
2194 done <- allThreadsDone sv
2195 -- XXX This is safe only when called from the consumer thread or when no
2196 -- consumer is present. There may be a race if we are not running in the
2198 -- XXX do this only if the work queue is not empty. The work may have been
2199 -- carried out by existing workers.
2201 case yieldRateInfo sv of
2202 Nothing -> pushWorker 0 sv
2203 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