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 UnboxedTuples #-}
14-- Module : Streamly.SVar
15-- Copyright : (c) 2017 Harendra Kumar
18-- Maintainer : harendra.kumar@gmail.com
19-- Stability : experimental
23#ifdef DIAGNOSTICS_VERBOSE
33 -- State threaded around the stream
52 , atomicModifyIORefCAS
95import Control.Concurrent
96 (ThreadId, myThreadId, threadDelay, getNumCapabilities)
97import Control.Concurrent.MVar
98 (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar)
99import Control.Exception (SomeException(..), catch, mask, assert)
100import Control.Monad (when)
101import Control.Monad.Catch (MonadThrow)
102import Control.Monad.IO.Class (MonadIO(..))
103import Control.Monad.Trans.Control (MonadBaseControl, control)
105 (casIORef, readForCAS, peekTicket, atomicModifyIORefCAS_,
106 writeBarrier, storeLoadBarrier)
107import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
108import Data.Functor (void)
109import Data.Heap (Heap, Entry(..))
110import Data.Int (Int64)
112 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
113import Data.Maybe (fromJust)
115import Data.Word (Word64)
116import GHC.Conc (ThreadId(..))
118import GHC.IO (IO(..))
120 (TimeSpec, Clock(Monotonic), getTime, toNanoSecs, fromNanoSecs)
122import qualified Data.Heap as H
123import qualified Data.Set as S
125-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
126-- can keep it on in production to debug problems quickly if and when they
127-- happen, but it may result in unexpected output when threads are left hanging
128-- until they are GCed because the consumer went away.
131import Control.Concurrent.MVar (tryTakeMVar)
132import Control.Exception
133 (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
134 BlockedIndefinitelyOnSTM(..))
135import System.IO (hPutStrLn, stderr)
138newtype NanoSecs = NanoSecs Word64
150newtype Count = Count Word64
162------------------------------------------------------------------------------
163-- Parent child thread communication type
164------------------------------------------------------------------------------
166-- | Events that a child thread may send to a parent thread.
169 | ChildStop ThreadId (Maybe SomeException)
171-- | Sorting out-of-turn outputs in a heap for Ahead style streams
172data AheadHeapEntry (t :: (* -> *) -> * -> *) m a =
174 | AheadEntryStream (t m a)
176------------------------------------------------------------------------------
177-- State threaded around the monad for thread management
178------------------------------------------------------------------------------
180-- | Identify the type of the SVar. Two computations using the same style can
181-- be scheduled on the same SVar.
183 AsyncVar -- depth first concurrent
184 | WAsyncVar -- breadth first concurrent
185 | ParallelVar -- all parallel
186 | AheadVar -- Concurrent look ahead
189-- | An SVar or a Stream Var is a conduit to the output from multiple streams
190-- running concurrently and asynchronously. An SVar can be thought of as an
191-- asynchronous IO handle. We can write any number of streams to an SVar in a
192-- non-blocking manner and then read them back at any time at any pace. The
193-- SVar would run the streams asynchronously and accumulate results. An SVar
194-- may not really execute the stream completely and accumulate all the results.
195-- However, it ensures that the reader can read the results at whatever paces
196-- it wants to read. The SVar monitors and adapts to the consumer's pace.
198-- An SVar is a mini scheduler, it has an associated workLoop that holds the
199-- stream tasks to be picked and run by a pool of worker threads. It has an
200-- associated output queue where the output stream elements are placed by the
201-- worker threads. A outputDoorBell is used by the worker threads to intimate the
202-- consumer thread about availability of new results in the output queue. More
203-- workers are added to the SVar by 'fromStreamVar' on demand if the output
204-- produced is not keeping pace with the consumer. On bounded SVars, workers
205-- block on the output queue to provide throttling of the producer when the
206-- consumer is not pulling fast enough. The number of workers may even get
207-- reduced depending on the consuming pace.
209-- New work is enqueued either at the time of creation of the SVar or as a
210-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
211-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
213-- We measure the individual worker latencies to estimate the number of workers
214-- needed or the amount of time we have to sleep between dispatches to achieve
215-- a particular rate when controlled pace mode it used.
216data WorkerInfo = WorkerInfo
217 { workerYieldMax :: Count -- 0 means unlimited
218 -- total number of yields by the worker till now
219 , workerYieldCount :: IORef Count
220 -- yieldCount at start, timestamp
221 , workerLatencyStart :: IORef (Count, TimeSpec)
225data YieldRateInfo = YieldRateInfo
226 { expectedYieldLatency :: NanoSecs
228 -- XXX Worker latency specified by the user to be used before the first
229 -- actual measurement arrives. Not yet implemented
230 , workerBootstrapLatency :: Maybe NanoSecs
232 -- After how many yields the worker should update the latency information.
233 -- If the latency is high, this count is kept lower and vice-versa. XXX If
234 -- the latency suddenly becomes too high this count may remain too high for
235 -- long time, in such cases the consumer can change it.
236 -- 0 means no latency computation
237 -- XXX this is derivable from workerMeasuredLatency, can be removed.
238 , workerLatencyPeriod :: IORef Count
240 -- This is in progress latency stats maintained by the workers which we
241 -- empty into workerCollectedLatency stats at certain intervals - whenever
242 -- we process the stream elements yielded in this period.
243 -- (yieldCount, timeTaken)
244 , workerCurrentLatency :: IORef (Count, NanoSecs)
246 -- This is the second level stat which is an accmulation from
247 -- workerCurrentLatency stats. We keep accumulating latencies in this
248 -- bucket until we have stats for a sufficient period and then we reset it
249 -- to start collecting for the next period and retain the computed average
250 -- latency for the last period in workerMeasuredLatency.
251 -- (yieldCount, timeTaken)
252 , workerCollectedLatency :: IORef (Count, NanoSecs)
254 -- Latency as measured by workers, aggregated for the last period.
255 , workerMeasuredLatency :: IORef NanoSecs
257 -- Actual latency/througput as seen from the consumer side, we count the
258 -- yields and the time it took to generates those yields. This is used to
259 -- increase or decrease the number of workers needed to achieve the desired
260 -- rate. The idle time of workers is adjusted in this, so that we only
261 -- account for the rate when the consumer actually demands data.
262 , workerLongTermLatency :: IORef (Count, TimeSpec)
264 -- When the last worker exits it updates this timestamp. We use this to
265 -- adjust workerLongTermLatency so that we do not account the idle time in
266 -- that. XXX Note that not all workers stop at the same time therefore
267 -- there is a certain inaccuracy in this mechanism.
268 , workerStopTimeStamp :: IORef TimeSpec
271data SVarStats = SVarStats {
272 totalDispatches :: IORef Int
273 , maxWorkers :: IORef Int
274 , maxOutQSize :: IORef Int
275 , maxHeapSize :: IORef Int
276 , maxWorkQSize :: IORef Int
277 , avgWorkerLatency :: IORef (Count, NanoSecs)
278 , minWorkerLatency :: IORef NanoSecs
279 , maxWorkerLatency :: IORef NanoSecs
280 , svarStopTime :: IORef (Maybe TimeSpec)
283data Limit = Unlimited | Limited Word deriving Show
285data SVar t m a = SVar
288 svarStyle :: SVarStyle
290 -- Shared output queue (events, length)
291 , outputQueue :: IORef ([ChildEvent a], Int)
292 , outputDoorBell :: MVar () -- signal the consumer about output
293 , readOutputQ :: m [ChildEvent a]
294 , postProcess :: m Bool
296 -- Combined/aggregate parameters
297 , maxWorkerLimit :: Limit
298 , maxBufferLimit :: Limit
299 , remainingYields :: Maybe (IORef Count)
300 , yieldRateInfo :: Maybe YieldRateInfo
302 -- Used only by bounded SVar types
303 , enqueue :: t m a -> IO ()
304 , isWorkDone :: IO Bool
305 , needDoorBell :: IORef Bool
306 , workLoop :: WorkerInfo -> m ()
308 -- Shared, thread tracking
309 , workerThreads :: IORef (Set ThreadId)
310 , workerCount :: IORef Int
311 , accountThread :: ThreadId -> m ()
312 , workerStopMVar :: MVar ()
314 , svarStats :: SVarStats
316 -- to track garbage collection of SVar
317 , svarRef :: Maybe (IORef ())
318 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)
319 -- Shared work queue (stream, seqNo)
320 , aheadWorkQueue :: IORef ([t m a], Int)
324-------------------------------------------------------------------------------
325-- State for concurrency control
326-------------------------------------------------------------------------------
328-- XXX we can put the resettable fields in a oneShotConfig field and others in
329-- a persistentConfig field. That way reset would be fast and scalable
330-- irrespective of the number of fields.
332-- XXX make all these Limited types and use phantom types to distinguish them
333data State t m a = State
334 { -- one shot configuration, automatically reset for each API call
335 streamVar :: Maybe (SVar t m a)
336 , _yieldLimit :: Maybe Count
338 -- persistent configuration, state that remains valid until changed by
339 -- an explicit setting via a combinator.
340 , _threadsHigh :: Limit
341 , _bufferHigh :: Limit
342 , _streamLatency :: Maybe NanoSecs
343 , _maxStreamRate :: Maybe Double -- yields per second
346-------------------------------------------------------------------------------
347-- State defaults and reset
348-------------------------------------------------------------------------------
350-- A magical value for the buffer size arrived at by running the smallest
351-- possible task and measuring the optimal value of the buffer for that. This
352-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
354magicMaxBuffer :: Word
357defaultMaxThreads, defaultMaxBuffer :: Limit
358defaultMaxThreads = Limited magicMaxBuffer
359defaultMaxBuffer = Limited magicMaxBuffer
361-- The fields prefixed by an _ are not to be accessed or updated directly but
362-- via smart accessor APIs.
363defState :: State t m a
365 { streamVar = Nothing
366 , _yieldLimit = Nothing
367 , _threadsHigh = defaultMaxThreads
368 , _bufferHigh = defaultMaxBuffer
369 , _maxStreamRate = Nothing
370 , _streamLatency = Nothing
373-- XXX if perf gets affected we can have all the Nothing params in a single
374-- structure so that we reset is fast. We can also use rewrite rules such that
375-- reset occurs only in concurrent streams to reduce the impact on serial
377-- We can optimize this so that we clear it only if it is a Just value, it
378-- results in slightly better perf for zip/zipM but the performance of scan
379-- worsens a lot, it does not fuse.
380rstState :: State t m a -> State t m b
382 { streamVar = Nothing
383 , _yieldLimit = Nothing
386-------------------------------------------------------------------------------
387-- Smart get/set routines for State
388-------------------------------------------------------------------------------
390-- Use get/set routines instead of directly accessing the State fields
391setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
392setYieldLimit lim st =
399 else Just (fromIntegral n)
402getYieldLimit :: State t m a -> Maybe Count
403getYieldLimit = _yieldLimit
405setMaxThreads :: Int -> State t m a -> State t m a
411 then defaultMaxThreads
412 else Limited (fromIntegral n)
415getMaxThreads :: State t m a -> Limit
416getMaxThreads = _threadsHigh
418setMaxBuffer :: Int -> State t m a -> State t m a
424 then defaultMaxBuffer
425 else Limited (fromIntegral n)
428getMaxBuffer :: State t m a -> Limit
429getMaxBuffer = _bufferHigh
431setMaxStreamRate :: Double -> State t m a -> State t m a
432setMaxStreamRate n st =
433 st { _maxStreamRate =
441getMaxStreamRate :: State t m a -> Maybe Double
442getMaxStreamRate = _maxStreamRate
444setStreamLatency :: Int -> State t m a -> State t m a
445setStreamLatency n st =
446 st { _streamLatency =
451 else Just (fromIntegral n)
454getStreamLatency :: State t m a -> Maybe NanoSecs
455getStreamLatency = _streamLatency
457-------------------------------------------------------------------------------
458-- Dumping the SVar for debug/diag
459-------------------------------------------------------------------------------
462dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
463dumpSVarStats sv ss style = do
464 case yieldRateInfo sv of
467 _ <- liftIO $ collectLatency (svarStats sv) yinfo
470 dispatches <- readIORef $ totalDispatches ss
471 maxWrk <- readIORef $ maxWorkers ss
472 maxOq <- readIORef $ maxOutQSize ss
473 maxHp <- readIORef $ maxHeapSize ss
474 minLat <- readIORef $ minWorkerLatency ss
475 maxLat <- readIORef $ maxWorkerLatency ss
476 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
477 svarLat <- case yieldRateInfo sv of
480 (cnt, startTime) <- readIORef $ workerLongTermLatency yinfo
483 t <- readIORef (svarStopTime ss)
486 now <- getTime Monotonic
487 let interval = toNanoSecs (now - startTime)
488 return $ interval `div` fromIntegral cnt
490 let interval = toNanoSecs (stopTime - startTime)
491 return $ interval `div` fromIntegral cnt
495 [ "total dispatches = " ++ show dispatches
496 , "max workers = " ++ show maxWrk
497 , "max outQSize = " ++ show maxOq
498 ++ (if style == AheadVar
499 then "\nheap max size = " ++ show maxHp
502 then "\nmin worker latency = " ++ show minLat
505 then "\nmax worker latency = " ++ show maxLat
508 then "\navg worker latency = "
509 ++ show (avgTime `div` fromIntegral avgCnt)
512 then "\nSVar latency = " ++ show svarLat
516{-# NOINLINE dumpSVar #-}
517dumpSVar :: SVar t m a -> IO String
520 (oqList, oqLen) <- readIORef $ outputQueue sv
521 db <- tryTakeMVar $ outputDoorBell sv
523 if svarStyle sv == AheadVar
525 (oheap, oheapSeq) <- readIORef $ outputHeap sv
526 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
528 [ "heap length = " ++ show (H.size oheap)
529 , "heap seqeunce = " ++ show oheapSeq
530 , "work queue length = " ++ show (length wq)
531 , "work queue sequence = " ++ show wqSeq
535 let style = svarStyle sv
537 if style /= ParallelVar
538 then readIORef $ needDoorBell sv
540 rthread <- readIORef $ workerThreads sv
541 workers <- readIORef $ workerCount sv
542 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
545 [ "tid = " ++ show tid
546 , "style = " ++ show (svarStyle sv)
547 , "---------CURRENT STATE-----------"
548 , "outputQueue length computed = " ++ show (length oqList)
549 , "outputQueue length maintained = " ++ show oqLen
550 , "outputDoorBell = " ++ show db
552 ++ aheadDump ++ unlines
553 [ "needDoorBell = " ++ show waiting
554 , "running threads = " ++ show rthread
555 , "running thread count = " ++ show workers
557 ++ "---------STATS-----------\n"
560{-# NOINLINE mvarExcHandler #-}
561mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
562mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
563 svInfo <- dumpSVar sv
564 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo
567{-# NOINLINE stmExcHandler #-}
568stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
569stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
570 svInfo <- dumpSVar sv
571 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo
574withDBGMVar :: SVar t m a -> String -> IO () -> IO ()
575withDBGMVar sv label action =
576 action `catches` [ Handler (mvarExcHandler sv label)
577 , Handler (stmExcHandler sv label)
580withDBGMVar :: SVar t m a -> String -> IO () -> IO ()
581withDBGMVar _ _ action = action
584-------------------------------------------------------------------------------
586-------------------------------------------------------------------------------
588-- Slightly faster version of CAS. Gained some improvement by avoiding the use
589-- of "evaluate" because we know we do not have exceptions in fn.
590{-# INLINE atomicModifyIORefCAS #-}
591atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
592atomicModifyIORefCAS ref fn = do
593 tkt <- readForCAS ref
599 loop _ 0 = atomicModifyIORef ref fn
601 let (new, result) = fn $ peekTicket old
602 (success, tkt) <- casIORef ref old new
605 else loop tkt (tries - 1)
607------------------------------------------------------------------------------
608-- Spawning threads and collecting result in streamed fashion
609------------------------------------------------------------------------------
611-- | A monad that can perform concurrent or parallel IO operations. Streams
612-- that can be composed concurrently require the underlying monad to be
616type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
618-- Stolen from the async package. The perf improvement is modest, 2% on a
619-- thread heavy benchmark (parallel composition using noop computations).
620-- A version of forkIO that does not include the outer exception
621-- handler: saves a bit of time when we will be installing our own
623{-# INLINE rawForkIO #-}
624rawForkIO :: IO () -> IO ThreadId
625rawForkIO action = IO $ \ s ->
626 case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
629doFork :: MonadBaseControl IO m
631 -> (SomeException -> IO ())
633doFork action exHandler =
634 control $ \runInIO ->
635 mask $ \restore -> do
636 tid <- rawForkIO $ catch (restore $ void $ runInIO action)
640-- XXX Can we make access to remainingYields and yieldRateInfo fields in sv
641-- faster, along with the fields in sv required by send?
642-- XXX make it noinline
644-- XXX we may want to employ an increment and decrement in batches when the
645-- througput is high or when the cost of synchronization is high. For example
646-- if the application is distributed then inc/dec of a shared variable may be
648{-# INLINE decrementYieldLimit #-}
649decrementYieldLimit :: SVar t m a -> IO Bool
650decrementYieldLimit sv =
651 case remainingYields sv of
652 Nothing -> return True
654 r <- atomicModifyIORefCAS ref $ \x ->
655 (if x >= 1 then x - 1 else 0, x)
658-- XXX increment is not safe when the yield limit was already 0
659-- and therefore was not decremented. We need yield limit to be
660-- int and negative if we want this to work.
661{-# INLINE incrementYieldLimit #-}
662incrementYieldLimit :: SVar t m a -> IO ()
663incrementYieldLimit sv =
664 case remainingYields sv of
666 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
668-- XXX exception safety of all atomic/MVar operations
670-- TBD Each worker can have their own queue and the consumer can empty one
671-- queue at a time, that way contention can be reduced.
673-- XXX Only yields should be counted in the buffer limit and not the Stop
676-- | This function is used by the producer threads to queue output for the
677-- consumer thread to consume. Returns whether the queue has more space.
678send :: SVar t m a -> ChildEvent a -> IO Bool
680 -- XXX can the access to outputQueue and maxBufferLimit be made faster
682 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
683 ((msg : es, n + 1), n)
685 -- The wake up must happen only after the store has finished otherwise
686 -- we can have lost wakeup problems.
688 -- Since multiple workers can try this at the same time, it is possible
689 -- that we may put a spurious MVar after the consumer has already seen
690 -- the output. But that's harmless, at worst it may cause the consumer
691 -- to read the queue again and find it empty.
692 -- The important point is that the consumer is guaranteed to receive a
693 -- doorbell if something was added to the queue after it empties it.
694 void $ tryPutMVar (outputDoorBell sv) ()
696 -- XXX we should reserve the buffer when we pick up the work from the
697 -- queue, instead of checking it here when it is too late.
698 let limit = maxBufferLimit sv
700 Unlimited -> return True
702 active <- readIORef (workerCount sv)
703 return $ len < ((fromIntegral lim) - active)
705-- XXX We assume that a worker always yields a value. If we can have
706-- workers that return without yielding anything our computations to
707-- determine the number of workers may be off.
708workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
709workerUpdateLatency yinfo winfo = do
710 cnt1 <- readIORef (workerYieldCount winfo)
711 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
712 t1 <- getTime Monotonic
713 writeIORef (workerLatencyStart winfo) (cnt1, t1)
714 let period = fromInteger $ toNanoSecs (t1 - t0)
715 let ref = workerCurrentLatency yinfo
716 atomicModifyIORefCAS ref $ \(ycnt, ytime) ->
717 ((ycnt + cnt1 - cnt0, ytime + period), ())
719updateYieldCount :: WorkerInfo -> IO Count
720updateYieldCount winfo = do
721 cnt <- readIORef (workerYieldCount winfo)
723 writeIORef (workerYieldCount winfo) cnt1
726isBeyondMaxYield :: Count -> WorkerInfo -> Bool
727isBeyondMaxYield cnt winfo =
728 let ymax = workerYieldMax winfo
729 in ymax /= 0 && cnt >= ymax
731-- XXX we should do rate control periodically based on the total yields rather
732-- than based on the worker local yields as other workers may have yielded more
733-- and we should stop based on the aggregate yields. However, latency update
734-- period can be based on individual worker yields.
735{-# NOINLINE checkRatePeriodic #-}
736checkRatePeriodic :: SVar t m a
741checkRatePeriodic sv yinfo winfo ycnt = do
742 i <- readIORef (workerLatencyPeriod yinfo)
743 if (i /= 0 && (ycnt `mod` i) == 0)
745 workerUpdateLatency yinfo winfo
746 isBeyondMaxRate sv yinfo
749-- CAUTION! this also updates the yield count and therefore should be called
750-- only when we are actually yielding an element.
751{-# NOINLINE workerRateControl #-}
752workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
753workerRateControl sv yinfo winfo = do
754 cnt <- updateYieldCount winfo
755 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
756 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
758-- XXX we should do rate control here but not latency update in case of ahead
759-- streams. latency update must be done when we yield directly to outputQueue
760-- or when we yield to heap.
761{-# INLINE sendYield #-}
762sendYield :: SVar t m a -> WorkerInfo -> ChildEvent a -> IO Bool
763sendYield sv winfo msg = do
766 case yieldRateInfo sv of
767 Nothing -> return True
768 Just yinfo -> workerRateControl sv yinfo winfo
769 return $ r && rateLimitOk
771{-# INLINE workerStopUpdate #-}
772workerStopUpdate :: WorkerInfo -> YieldRateInfo -> Int -> IO ()
773workerStopUpdate winfo info n = do
774 i <- readIORef (workerLatencyPeriod info)
775 when (i /= 0) $ workerUpdateLatency info winfo
777 t <- getTime Monotonic
778 writeIORef (workerStopTimeStamp info) t
780{-# INLINABLE sendStop #-}
781sendStop :: SVar t m a -> WorkerInfo -> IO ()
782sendStop sv winfo = do
783 n <- liftIO $ atomicModifyIORefCAS (workerCount sv) $ \n -> (n - 1, n)
784 case yieldRateInfo sv of
786 Just info -> workerStopUpdate winfo info n
787 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
789-------------------------------------------------------------------------------
791-------------------------------------------------------------------------------
793-- Note: For purely right associated expressions this queue should have at most
794-- one element. It grows to more than one when we have left associcated
795-- expressions. Large left associated compositions can grow this to a
797{-# INLINE enqueueLIFO #-}
798enqueueLIFO :: SVar t m a -> IORef [t m a] -> t m a -> IO ()
799enqueueLIFO sv q m = do
800 atomicModifyIORefCAS_ q $ \ms -> m : ms
802 w <- readIORef $ needDoorBell sv
804 -- Note: the sequence of operations is important for correctness here.
805 -- We need to set the flag to false strictly before sending the
806 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
807 -- then we may set the flag to False to later making the consumer lose
808 -- the flag, even without receiving a outputDoorBell.
809 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
810 void $ tryPutMVar (outputDoorBell sv) ()
812-------------------------------------------------------------------------------
814-------------------------------------------------------------------------------
816-- XXX we can use the Ahead style sequence/heap mechanism to make the best
817-- effort to always try to finish the streams on the left side of an expression
818-- first as long as possible.
820{-# INLINE enqueueFIFO #-}
821enqueueFIFO :: SVar t m a -> LinkedQueue (t m a) -> t m a -> IO ()
822enqueueFIFO sv q m = do
825 w <- readIORef $ needDoorBell sv
827 -- Note: the sequence of operations is important for correctness here.
828 -- We need to set the flag to false strictly before sending the
829 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
830 -- then we may set the flag to False to later making the consumer lose
831 -- the flag, even without receiving a outputDoorBell.
832 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
833 void $ tryPutMVar (outputDoorBell sv) ()
835-------------------------------------------------------------------------------
837-------------------------------------------------------------------------------
839-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
840-- but always serve them in the same order as they appear in the stream. To
841-- implement lookahead streams efficiently we assign a sequence number to each
842-- task when the task is picked up for execution. When the task finishes, the
843-- output is tagged with the same sequence number and we rearrange the outputs
844-- in sequence based on that number.
846-- To explain the mechanism imagine that the current task at the head of the
847-- stream has a "token" to yield to the outputQueue. The ownership of the token
848-- is determined by the current sequence number is maintained in outputHeap.
849-- Sequence number is assigned when a task is queued. When a thread dequeues a
850-- task it picks up the sequence number as well and when the output is ready it
851-- uses the sequence number to queue the output to the outputQueue.
853-- The thread with current sequence number sends the output directly to the
854-- outputQueue. Other threads push the output to the outputHeap. When the task
855-- being queued on the heap is a stream of many elements we evaluate only the
856-- first element and keep the rest of the unevaluated computation in the heap.
857-- When such a task gets the "token" for outputQueue it evaluates and directly
858-- yields all the elements to the outputQueue without checking for the
861-- Note that no two outputs in the heap can have the same sequence numbers and
862-- therefore we do not need a stable heap. We have also separated the buffer
863-- for the current task (outputQueue) and the pending tasks (outputHeap) so
864-- that the pending tasks cannot interfere with the current task. Note that for
865-- a single task just the outputQueue is enough and for the case of many
866-- threads just a heap is good enough. However we balance between these two
867-- cases, so that both are efficient.
869-- For bigger streams it may make sense to have separate buffers for each
870-- stream. However, for singleton streams this may become inefficient. However,
871-- if we do not have separate buffers, then the streams that come later in
872-- sequence may hog the buffer, hindering the streams that are ahead. For this
873-- reason we have a single element buffer limitation for the streams being
874-- executed in advance.
876-- This scheme works pretty efficiently with less than 40% extra overhead
877-- compared to the Async streams where we do not have any kind of sequencing of
878-- the outputs. It is especially devised so that we are most efficient when we
879-- have short tasks and need just a single thread. Also when a thread yields
880-- many items it can hold lockfree access to the outputQueue and do it
883-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
884-- that they do not hog the resources and hinder the progress of the threads in
887-- XXX Left associated ahead expressions are expensive. We start a new SVar for
888-- each left associative expression. The queue is used only for right
889-- associated expression, we queue the right expression and execute the left.
890-- Thererefore the queue never has more than one item in it.
892-- XXX we can fix this. When we queue more than one item on the queue we can
893-- mark the previously queued item as not-runnable. The not-runnable item is
894-- not dequeued until the already running one has finished and at that time we
895-- would also know the exact sequence number of the already queued item.
897-- we can even run the already queued items but they will have to be sorted in
898-- layers in the heap. We can use a list of heaps for that.
899{-# INLINE enqueueAhead #-}
900enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
901enqueueAhead sv q m = do
902 atomicModifyIORefCAS_ q $ \ case
903 ([], n) -> ([m], n + 1) -- increment sequence
904 _ -> error "not empty"
906 w <- readIORef $ needDoorBell sv
908 -- Note: the sequence of operations is important for correctness here.
909 -- We need to set the flag to false strictly before sending the
910 -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
911 -- then we may set the flag to False to later making the consumer lose
912 -- the flag, even without receiving a outputDoorBell.
913 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
914 void $ tryPutMVar (outputDoorBell sv) ()
916-- enqueue without incrementing the sequence number
917{-# INLINE reEnqueueAhead #-}
918reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
919reEnqueueAhead sv q m = do
920 atomicModifyIORefCAS_ q $ \ case
921 ([], n) -> ([m], n) -- DO NOT increment sequence
922 _ -> error "not empty"
924 w <- readIORef $ needDoorBell sv
926 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
927 void $ tryPutMVar (outputDoorBell sv) ()
929-- Normally the thread that has the token should never go away. The token gets
930-- handed over to another thread, but someone or the other has the token at any
931-- point of time. But if the task that has the token finds that the outputQueue
932-- is full, in that case it can go away without even handing over the token to
933-- another thread. In that case it sets the nextSequence number in the heap its
934-- own sequence number before going away. To handle this case, any task that
935-- does not have the token tries to dequeue from the heap first before
936-- dequeuing from the work queue. If it finds that the task at the top of the
937-- heap is the one that owns the current sequence number then it grabs the
938-- token and starts with that.
940-- XXX instead of queueing just the head element and the remaining computation
941-- on the heap, evaluate as many as we can and place them on the heap. But we
942-- need to give higher priority to the lower sequence numbers so that lower
943-- priority tasks do not fill up the heap making higher priority tasks block
944-- due to full heap. Maybe we can have a weighted space for them in the heap.
945-- The weight is inversely proportional to the sequence number.
947-- XXX review for livelock
949{-# INLINE queueEmptyAhead #-}
950queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
951queueEmptyAhead q = liftIO $ do
952 (xs, _) <- readIORef q
955{-# INLINE dequeueAhead #-}
956dequeueAhead :: MonadIO m
957 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
958dequeueAhead q = liftIO $ do
959 atomicModifyIORefCAS q $ \case
960 ([], n) -> (([], n), Nothing)
961 (x : [], n) -> (([], n), Just (x, n))
962 _ -> error "more than one item on queue"
964{-# INLINE dequeueFromHeap #-}
966 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
967 -> IO (Maybe (Entry Int (AheadHeapEntry t m a)))
968dequeueFromHeap hpRef = do
969 atomicModifyIORef hpRef $ \hp@(h, snum) -> do
972 Nothing -> (hp, Nothing)
973 Just (ent@(Entry seqNo _ev), hp') ->
975 then ((hp', seqNo), Just ent)
978-------------------------------------------------------------------------------
980-------------------------------------------------------------------------------
982-- XXX To be implemented. Use a linked queue like WAsync and put back the
983-- remaining computation at the back of the queue instead of the heap, and
984-- increment the sequence number.
986-- Thread tracking is needed for two reasons:
988-- 1) Killing threads on exceptions. Threads may not be left to go away by
989-- themselves because they may run for significant times before going away or
990-- worse they may be stuck in IO and never go away.
992-- 2) To know when all threads are done and the stream has ended.
994{-# NOINLINE addThread #-}
995addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
997 liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
999-- This is cheaper than modifyThread because we do not have to send a
1000-- outputDoorBell This can make a difference when more workers are being
1002{-# INLINE delThread #-}
1003delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1005 liftIO $ modifyIORef (workerThreads sv) $ (\s -> S.delete tid s)
1007-- If present then delete else add. This takes care of out of order add and
1008-- delete i.e. a delete arriving before we even added a thread.
1009-- This occurs when the forked thread is done even before the 'addThread' right
1010-- after the fork gets a chance to run.
1011{-# INLINE modifyThread #-}
1012modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1013modifyThread sv tid = do
1014 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1015 if (S.member tid old)
1016 then let new = (S.delete tid old) in (new, new)
1017 else let new = (S.insert tid old) in (new, old)
1021 void $ tryPutMVar (outputDoorBell sv) ()
1024-- | This is safe even if we are adding more threads concurrently because if
1025-- a child thread is adding another thread then anyway 'workerThreads' will
1027{-# INLINE allThreadsDone #-}
1028allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1029allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1031{-# NOINLINE handleChildException #-}
1032handleChildException :: SVar t m a -> SomeException -> IO ()
1033handleChildException sv e = do
1035 void $ send sv (ChildStop tid (Just e))
1038recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1039recordMaxWorkers sv = liftIO $ do
1040 active <- readIORef (workerCount sv)
1041 maxWrk <- readIORef (maxWorkers $ svarStats sv)
1042 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1043 modifyIORef (totalDispatches $ svarStats sv) (+1)
1046{-# NOINLINE pushWorker #-}
1047pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1048pushWorker yieldMax sv = do
1049 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1053 -- XXX we can make this allocation conditional, it might matter when
1054 -- significant number of workers are being sent.
1056 cntRef <- liftIO $ newIORef 0
1057 t <- liftIO $ getTime Monotonic
1058 lat <- liftIO $ newIORef (0, t)
1060 { workerYieldMax = yieldMax
1061 , workerYieldCount = cntRef
1062 , workerLatencyStart = lat
1064 doFork (workLoop sv winfo) (handleChildException sv) >>= addThread sv
1066-- XXX we can push the workerCount modification in accountThread and use the
1067-- same pushWorker for Parallel case as well.
1069-- | In contrast to pushWorker which always happens only from the consumer
1070-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1071-- producer side. So we need to use a thread safe modification of
1072-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1073-- using a CAS based modification.
1074{-# NOINLINE pushWorkerPar #-}
1075pushWorkerPar :: MonadAsync m => SVar t m a -> m () -> m ()
1076pushWorkerPar sv wloop = do
1077 -- We do not use workerCount in case of ParallelVar but still there is no
1078 -- harm in maintaining it correctly.
1080 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1083 doFork wloop (handleChildException sv) >>= modifyThread sv
1086-- True: can dispatch more
1087-- False: cannot dispatch any more
1088dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1089dispatchWorker yieldCount sv = do
1090 let workerLimit = maxWorkerLimit sv
1091 -- XXX in case of Ahead streams we should not send more than one worker
1092 -- when the work queue is done but heap is not done.
1093 done <- liftIO $ isWorkDone sv
1096 -- Note that the worker count is only decremented during event
1097 -- processing in fromStreamVar and therefore it is safe to read and
1098 -- use it without a lock.
1099 active <- liftIO $ readIORef $ workerCount sv
1100 -- Note that we may deadlock if the previous workers (tasks in the
1101 -- stream) wait/depend on the future workers (tasks in the stream)
1102 -- executing. In that case we should either configure the maxWorker
1103 -- count to higher or use parallel style instead of ahead or async
1105 limit <- case remainingYields sv of
1106 Nothing -> return workerLimit
1108 n <- liftIO $ readIORef x
1111 Unlimited -> Limited (fromIntegral n)
1112 Limited lim -> Limited $ min lim (fromIntegral n)
1114 -- XXX for ahead streams shall we take the heap yields into account for
1115 -- controlling the dispatch? We should not dispatch if the heap has
1116 -- already got the limit covered.
1117 let dispatch = pushWorker yieldCount sv >> return True
1119 Unlimited -> dispatch
1120 -- Note that the use of remainingYields and workerCount is not
1121 -- atomic and the counts may even have changed between reading and
1122 -- using them here, so this is just approximate logic and we cannot
1123 -- rely on it for correctness. We may actually dispatch more
1124 -- workers than required.
1125 Limited lim | active < (fromIntegral lim) -> dispatch
1129-- | This is a magic number and it is overloaded, and used at several places to
1132-- 1. If we have to sleep to slowdown this is the minimum period that we
1133-- accumulate before we sleep. Also, workers do not stop until this much
1134-- sleep time is accumulated.
1135-- 3. Collected latencies are computed and transferred to measured latency
1136-- after a minimum of this period.
1137minThreadDelay :: NanoSecs
1138minThreadDelay = 10^(6 :: Int)
1140-- | Another magic number! When we have to start more workers to cover up a
1141-- number of yields that we are lagging by then we cannot start one worker for
1142-- each yield because that may be a very big number and if the latency of the
1143-- workers is low these number of yields could be very high. We use this number
1144-- as the duration in which we want cover any lag. In other words we accumulate
1145-- the lag (lagging number of yields) such that it covers this much duration
1146-- for a single worker to produce those yields, before we start an extra
1148coverUpLatency :: NanoSecs
1149coverUpLatency = 1000000
1151nanoToMicroSecs :: NanoSecs -> Int
1152nanoToMicroSecs s = (fromIntegral s) `div` 1000
1154-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1156 :: Limit -> Count -> NanoSecs -> NanoSecs -> NanoSecs -> Int
1157countDispatchWorkers workerLimit count duration wLatency expectedLat =
1158 -- Calculate how many workers do we need to maintain the rate.
1160 -- XXX Increasing the workers may reduce the overall latency per
1161 -- yield of the workers, sometimes quite a bit. We can keep a
1162 -- latency histogram for 1,2,4..2^20 workers so that we can decide
1163 -- which bucket is optimal and decide the number of workers based
1166 -- XXX put a cap on extraworkers? Should not be more than nWorkers.
1167 -- Bridge the gap slowly rather than in a burst in one go.
1169 -- XXX we can have a maxEfficiency combinator as well which runs the
1170 -- producer at the maximal efficiency i.e. the number of workers are chosen
1171 -- such that the latency is minimum or within a range. Or we can call it
1172 -- maxWorkerLatency.
1174 let nWorkers = (fromIntegral wLatency) / (fromIntegral expectedLat)
1175 -- Calculate how many yields are we ahead or behind to match the exact
1176 -- required rate. Based on that we increase or decrease the effective
1178 d = fromIntegral duration
1179 e = fromIntegral expectedLat
1180 n = max 1 (fromIntegral coverUpLatency / fromIntegral wLatency)
1182 extraYields = (d / e) - (fromIntegral count)
1183 extraWorkers = extraYields / n
1184 requiredWorkers = round $ nWorkers + extraWorkers
1185 in case workerLimit of
1186 Unlimited -> requiredWorkers
1187 Limited x -> min requiredWorkers (fromIntegral x)
1189-- | Get the worker latency without resetting workerCurrentLatency
1190-- Returns (total yield count, base time, measured latency)
1191getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1192getWorkerLatency yinfo = do
1193 let cur = workerCurrentLatency yinfo
1194 col = workerCollectedLatency yinfo
1195 longTerm = workerLongTermLatency yinfo
1196 measured = workerMeasuredLatency yinfo
1198 (count, time) <- readIORef cur
1199 (colCount, colTime) <- readIORef col
1200 (lcount, ltime) <- readIORef longTerm
1201 prev <- readIORef measured
1203 let pendingCount = colCount + count
1204 pendingTime = colTime + time
1205 -- XXX Take the average of previous and the latest
1208 then ((pendingTime `div` (fromIntegral pendingCount)) + prev)
1211 return (lcount + pendingCount, ltime, new)
1213isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1214isBeyondMaxRate sv yinfo = do
1215 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1216 now <- getTime Monotonic
1217 let duration = fromInteger $ toNanoSecs $ now - tstamp
1218 let expectedLat = expectedYieldLatency yinfo
1219 let netWorkers = countDispatchWorkers (maxWorkerLimit sv) count duration
1220 wLatency expectedLat
1221 cnt <- readIORef $ workerCount sv
1224 let sleepTime = ((fromIntegral count) * expectedLat) - duration
1225 if sleepTime >= minThreadDelay
1228 else if cnt > netWorkers
1232useCollectedBatch :: YieldRateInfo
1234 -> IORef (Count, NanoSecs)
1237useCollectedBatch yinfo latency colRef measRef = do
1238 let periodRef = workerLatencyPeriod yinfo
1239 cnt = max 1 $ minThreadDelay `div` latency
1240 period = min cnt (fromIntegral magicMaxBuffer)
1242 -- XXX we should use some weight for the old latency as well in arriving at
1243 -- the new latency, we have the counts, we can use that for some weighting.
1244 writeIORef periodRef (fromIntegral period)
1245 writeIORef colRef (0, 0)
1246 writeIORef measRef latency
1248-- Returns a triple, (1) yield count since last collection, (2) the base time
1249-- when we started counting, (3) average latency in the last measurement
1250-- period. The former two are used for accurate measurement of the going rate
1251-- whereas the average is used for future estimates e.g. how many workers
1252-- should be maintained to maintain the rate.
1253collectLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1254collectLatency _ss yinfo = do
1255 let cur = workerCurrentLatency yinfo
1256 col = workerCollectedLatency yinfo
1257 longTerm = workerLongTermLatency yinfo
1258 measured = workerMeasuredLatency yinfo
1260 (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
1261 (colCount, colTime) <- readIORef col
1262 (lcount, ltime) <- readIORef longTerm
1263 prev <- readIORef measured
1265 let pendingCount = colCount + count
1266 pendingTime = colTime + time
1268 lcount' = lcount + pendingCount
1269 notUpdated = (lcount', ltime, prev)
1271 if (pendingCount > 0)
1273 let new = pendingTime `div` (fromIntegral pendingCount)
1275 minLat <- readIORef (minWorkerLatency _ss)
1276 when (new < minLat || minLat == 0) $
1277 writeIORef (minWorkerLatency _ss) new
1279 maxLat <- readIORef (maxWorkerLatency _ss)
1280 when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
1282 -- To avoid minor fluctuations update in batches
1283 if (pendingCount > fromIntegral magicMaxBuffer)
1284 || (pendingTime > minThreadDelay)
1285 || (let r = (fromIntegral new) / (fromIntegral prev) :: Double
1286 in prev > 0 && (r > 2 || r < 0.5))
1289 useCollectedBatch yinfo new col measured
1291 modifyIORef (avgWorkerLatency _ss) $
1292 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
1294 modifyIORef longTerm $ \(_, t) -> (lcount', t)
1295 return (lcount', ltime, new)
1297 writeIORef col (pendingCount, pendingTime)
1299 else return notUpdated
1301adjustLatencies :: YieldRateInfo -> IO ()
1302adjustLatencies yinfo = do
1303 let longTerm = workerLongTermLatency yinfo
1304 t0 <- readIORef (workerStopTimeStamp yinfo)
1305 t1 <- getTime Monotonic
1306 modifyIORef longTerm $ \(c, t) -> (c, t + t1 - t0)
1309-- XXX in case of ahead style stream we need to take the heap size into account
1310-- because we return the workers on the basis of that which causes a condition
1311-- where we keep dispatching and they keep returning. So we must have exactly
1312-- the same logic for not dispatching and for returning.
1315-- True: can dispatch more
1316-- False: full, no more dispatches
1317dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1318dispatchWorkerPaced sv = do
1319 let workerLimit = maxWorkerLimit sv
1320 let yinfo = fromJust $ yieldRateInfo sv
1321 (count, tstamp, wLatency) <- liftIO $ collectLatency (svarStats sv) yinfo
1322 -- XXX use workerBootstrapLatency
1324 -- Need to measure the latency with a single worker before we can perform
1328 now <- liftIO $ getTime Monotonic
1329 let duration = fromInteger $ toNanoSecs $ now - tstamp
1330 let expectedLat = expectedYieldLatency yinfo
1331 let netWorkers = countDispatchWorkers workerLimit count duration
1332 wLatency expectedLat
1334 -- XXX we need to take yieldLimit into account here. If we are at the
1335 -- end of the limit as well as the time, we should not be sleeping.
1336 -- If we are not actually planning to dispatch any more workers we need
1337 -- to take that in account.
1340 let sleepTime = ((fromIntegral count) * expectedLat) - duration
1342 when (sleepTime >= minThreadDelay) $ do
1343 liftIO $ threadDelay $ nanoToMicroSecs sleepTime
1345 cnt <- liftIO $ readIORef $ workerCount sv
1346 when (cnt < 1) $ void $ do
1347 let n = countDispatchYields (expectedLat - wLatency)
1348 in dispatchWorker (fromIntegral (max n 1)) sv
1351 -- XXX stagger the workers over a period
1352 cnt <- liftIO $ readIORef $ workerCount sv
1353 if (cnt < netWorkers)
1354 then dispatchWorker 0 sv
1359 countDispatchYields :: NanoSecs -> Int
1360 countDispatchYields idleTime =
1361 assert (idleTime >= 0) $
1362 -- we do not want to sleep for less than minThreadDelay
1363 let maxYieldsPerWorker = fromIntegral magicMaxBuffer
1364 in if idleTime < minThreadDelay
1367 then maxYieldsPerWorker
1369 let n = (minThreadDelay + idleTime) `div` idleTime
1370 in min (fromIntegral n) maxYieldsPerWorker
1373sendWorkerDelayPaced :: SVar t m a -> IO ()
1374sendWorkerDelayPaced _ = return ()
1376sendWorkerDelay :: SVar t m a -> IO ()
1377sendWorkerDelay sv = do
1378 -- XXX we need a better way to handle this than hardcoded delays. The
1379 -- delays may be different for different systems.
1380 ncpu <- getNumCapabilities
1383 if (svarStyle sv == AheadVar)
1384 then threadDelay 100
1387 if (svarStyle sv == AheadVar)
1388 then threadDelay 100
1391{-# NOINLINE sendWorkerWait #-}
1394 => (SVar t m a -> IO ())
1395 -> (SVar t m a -> m Bool)
1398sendWorkerWait delay dispatch sv = do
1399 -- Note that we are guaranteed to have at least one outstanding worker when
1400 -- we enter this function. So if we sleep we are guaranteed to be woken up
1401 -- by a outputDoorBell, when the worker exits.
1404 (_, n) <- liftIO $ readIORef (outputQueue sv)
1406 -- The queue may be empty temporarily if the worker has dequeued the
1407 -- work item but has not enqueued the remaining part yet. For the same
1408 -- reason, a worker may come back if it tries to dequeue and finds the
1409 -- queue empty, even though the whole work has not finished yet.
1411 -- If we find that the queue is empty, but it may be empty
1412 -- temporarily, when we checked it. If that's the case we might
1413 -- sleep indefinitely unless the active workers produce some
1414 -- output. We may deadlock specially if the otuput from the active
1415 -- workers depends on the future workers that we may never send.
1416 -- So in case the queue was temporarily empty set a flag to inform
1417 -- the enqueue to send us a doorbell.
1419 -- Note that this is just a best effort mechanism to avoid a
1420 -- deadlock. Deadlocks may still happen if for some weird reason
1421 -- the consuming computation shares an MVar or some other resource
1422 -- with the producing computation and gets blocked on that resource
1423 -- and therefore cannot do any pushworker to add more threads to
1424 -- the producer. In such cases the programmer should use a parallel
1425 -- style so that all the producers are scheduled immediately and
1426 -- unconditionally. We can also use a separate monitor thread to
1427 -- push workers instead of pushing them from the consumer, but then
1428 -- we are no longer using pull based concurrency rate adaptation.
1430 -- XXX update this in the tutorial.
1432 -- Having pending active workers does not mean that we are guaranteed
1433 -- to be woken up if we sleep. In case of Ahead streams, there may be
1434 -- queued items in the heap even though the outputQueue is empty, and
1435 -- we may have active workers which are deadlocked on those items to be
1436 -- processed by the consumer. We should either guarantee that any
1437 -- worker, before returning, clears the heap or we send a worker to clear
1438 -- it. Normally we always send a worker if no output is seen, but if
1439 -- the thread limit is reached or we are using pacing then we may not
1440 -- send a worker. See the concurrentApplication test in the tests, that
1441 -- test case requires at least one yield from the producer to not
1442 -- deadlock, if the last workers output is stuck in the heap then this
1443 -- test fails. This problem can be extended to n threads when the
1444 -- consumer may depend on the evaluation of next n items in the
1447 -- register for the outputDoorBell before we check the queue so that if we
1448 -- sleep because the queue was empty we are guaranteed to get a
1449 -- doorbell on the next enqueue.
1451 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
1452 liftIO $ storeLoadBarrier
1453 canDoMore <- dispatch sv
1455 -- XXX test for the case when we miss sending a worker when the worker
1456 -- count is more than 1500.
1458 -- XXX Assert here that if the heap is not empty then there is at
1459 -- least one outstanding worker. Otherwise we could be sleeping
1463 then sendWorkerWait delay dispatch sv
1465 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
1466 $ takeMVar (outputDoorBell sv)
1467 (_, len) <- liftIO $ readIORef (outputQueue sv)
1468 when (len <= 0) $ sendWorkerWait delay dispatch sv
1470{-# INLINE readOutputQRaw #-}
1471readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
1472readOutputQRaw sv = do
1473 (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
1475 oqLen <- readIORef (maxOutQSize $ svarStats sv)
1476 when (len > oqLen) $ writeIORef (maxOutQSize $ svarStats sv) len
1480readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1481readOutputQBounded sv = do
1482 (list, len) <- liftIO $ readOutputQRaw sv
1483 -- When there is no output seen we dispatch more workers to help
1484 -- out if there is work pending in the work queue.
1488 -- send a worker proactively, if needed, even before we start
1489 -- processing the output. This may degrade single processor
1490 -- perf but improves multi-processor, because of more
1498 cnt <- liftIO $ readIORef $ workerCount sv
1499 when (cnt <= 0) $ do
1500 done <- liftIO $ isWorkDone sv
1501 when (not done) $ pushWorker 0 sv
1503 {-# INLINE blockingRead #-}
1505 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
1506 liftIO $ (readOutputQRaw sv >>= return . fst)
1508readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1509readOutputQPaced sv = do
1510 (list, len) <- liftIO $ readOutputQRaw sv
1514 -- XXX send a worker proactively, if needed, even before we start
1515 -- processing the output.
1516 void $ dispatchWorkerPaced sv
1521 {-# INLINE blockingRead #-}
1523 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
1524 liftIO $ (readOutputQRaw sv >>= return . fst)
1526postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
1527postProcessBounded sv = do
1528 workersDone <- allThreadsDone sv
1529 -- There may still be work pending even if there are no workers pending
1530 -- because all the workers may return if the outputQueue becomes full. In
1531 -- that case send off a worker to kickstart the work again.
1534 r <- liftIO $ isWorkDone sv
1535 -- XXX do we need to dispatch many here?
1536 when (not r) $ pushWorker 0 sv
1537 -- void $ dispatchWorker sv
1541postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
1542postProcessPaced sv = do
1543 workersDone <- allThreadsDone sv
1544 -- XXX If during consumption we figure out we are getting delayed then we
1545 -- should trigger dispatch there as well. We should try to check on the
1546 -- workers after consuming every n item from the buffer?
1549 r <- liftIO $ isWorkDone sv
1551 let yinfo = fromJust $ yieldRateInfo sv
1552 liftIO $ adjustLatencies yinfo
1553 void $ dispatchWorkerPaced sv
1557getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
1558getYieldRateInfo st = do
1559 case getMaxStreamRate st of
1561 let latency = round $ 1.0e9 / rate
1562 measured <- newIORef 0
1563 wcur <- newIORef (0,0)
1564 wcol <- newIORef (0,0)
1565 now <- getTime Monotonic
1566 wlong <- newIORef (0,now)
1567 period <- newIORef 1
1568 stopTime <- newIORef $ fromNanoSecs 0
1570 return $ Just YieldRateInfo
1571 { expectedYieldLatency = latency
1572 , workerBootstrapLatency = getStreamLatency st
1573 , workerLatencyPeriod = period
1574 , workerMeasuredLatency = measured
1575 , workerCurrentLatency = wcur
1576 , workerStopTimeStamp = stopTime
1577 , workerCollectedLatency = wcol
1578 , workerLongTermLatency = wlong
1580 Nothing -> return Nothing
1582getAheadSVar :: MonadAsync m
1584 -> ( IORef ([t m a], Int)
1585 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
1591getAheadSVar st f = do
1592 outQ <- newIORef ([], 0)
1593 outH <- newIORef (H.empty, 0)
1594 outQMv <- newEmptyMVar
1595 active <- newIORef 0
1596 wfw <- newIORef False
1597 running <- newIORef S.empty
1598 q <- newIORef ([], -1)
1599 stopMVar <- newMVar ()
1600 yl <- case getYieldLimit st of
1601 Nothing -> return Nothing
1602 Just x -> Just <$> newIORef x
1603 rateInfo <- getYieldRateInfo st
1606 maxWrk <- newIORef 0
1610 avgLat <- newIORef (0, NanoSecs 0)
1611 maxLat <- newIORef (NanoSecs 0)
1612 minLat <- newIORef (NanoSecs 0)
1613 stpTime <- newIORef Nothing
1615 let getSVar sv readOutput postProc = SVar
1616 { outputQueue = outQ
1617 , remainingYields = yl
1618 , maxBufferLimit = getMaxBuffer st
1619 , maxWorkerLimit = getMaxThreads st
1620 , yieldRateInfo = rateInfo
1621 , outputDoorBell = outQMv
1622 , readOutputQ = readOutput sv
1623 , postProcess = postProc sv
1624 , workerThreads = running
1625 , workLoop = f q outH st{streamVar = Just sv} sv
1626 , enqueue = enqueueAhead sv q
1627 , isWorkDone = isWorkDoneAhead sv q outH
1628 , needDoorBell = wfw
1629 , svarStyle = AheadVar
1630 , workerCount = active
1631 , accountThread = delThread sv
1632 , workerStopMVar = stopMVar
1635 , aheadWorkQueue = q
1638 , svarStats = SVarStats
1639 { totalDispatches = disp
1640 , maxWorkers = maxWrk
1641 , maxOutQSize = maxOq
1642 , maxHeapSize = maxHs
1643 , maxWorkQSize = maxWq
1644 , avgWorkerLatency = avgLat
1645 , minWorkerLatency = minLat
1646 , maxWorkerLatency = maxLat
1647 , svarStopTime = stpTime
1652 case getMaxStreamRate st of
1653 Nothing -> getSVar sv readOutputQBounded postProcessBounded
1654 Just _ -> getSVar sv readOutputQPaced postProcessPaced
1659 {-# INLINE isWorkDoneAhead #-}
1660 isWorkDoneAhead sv q ref = do
1662 (hp, _) <- readIORef ref
1663 return (H.size hp <= 0)
1664 queueDone <- checkEmpty q
1666 case remainingYields sv of
1670 Nothing -> return False
1671 -- XXX note that yieldsDone can only be authoritative only when there
1672 -- are no workers running. If there are active workers they can
1673 -- increment the yield count and therefore change the result.
1674 return $ yieldsDone || (queueDone && heapDone)
1677 (xs, _) <- readIORef q
1680getParallelSVar :: MonadIO m => IO (SVar t m a)
1682 outQ <- newIORef ([], 0)
1683 outQMv <- newEmptyMVar
1684 active <- newIORef 0
1685 running <- newIORef S.empty
1688 maxWrk <- newIORef 0
1692 avgLat <- newIORef (0, NanoSecs 0)
1693 maxLat <- newIORef (NanoSecs 0)
1694 minLat <- newIORef (NanoSecs 0)
1695 stpTime <- newIORef Nothing
1698 SVar { outputQueue = outQ
1699 , remainingYields = Nothing
1700 , maxBufferLimit = Unlimited
1701 , maxWorkerLimit = undefined
1702 , yieldRateInfo = Nothing
1703 , outputDoorBell = outQMv
1704 , readOutputQ = readOutputQPar sv
1705 , postProcess = allThreadsDone sv
1706 , workerThreads = running
1707 , workLoop = undefined
1708 , enqueue = undefined
1709 , isWorkDone = undefined
1710 , needDoorBell = undefined
1711 , svarStyle = ParallelVar
1712 , workerCount = active
1713 , accountThread = modifyThread sv
1714 , workerStopMVar = undefined
1717 , aheadWorkQueue = undefined
1718 , outputHeap = undefined
1720 , svarStats = SVarStats
1721 { totalDispatches = disp
1722 , maxWorkers = maxWrk
1723 , maxOutQSize = maxOq
1724 , maxHeapSize = maxHs
1725 , maxWorkQSize = maxWq
1726 , avgWorkerLatency = avgLat
1727 , minWorkerLatency = minLat
1728 , maxWorkerLatency = maxLat
1729 , svarStopTime = stpTime
1736 readOutputQPar sv = liftIO $ do
1737 withDBGMVar sv "readOutputQPar: doorbell" $ takeMVar (outputDoorBell sv)
1738 readOutputQRaw sv >>= return . fst
1740sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
1741sendFirstWorker sv m = do
1742 -- Note: We must have all the work on the queue before sending the
1743 -- pushworker, otherwise the pushworker may exit before we even get a
1745 liftIO $ enqueue sv m
1746 case yieldRateInfo sv of
1747 Nothing -> pushWorker 0 sv
1748 Just _ -> pushWorker 1 sv
1751{-# INLINABLE newAheadVar #-}
1752newAheadVar :: MonadAsync m
1755 -> ( IORef ([t m a], Int)
1756 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
1762newAheadVar st m wloop = do
1763 sv <- liftIO $ getAheadSVar st wloop
1764 sendFirstWorker sv m
1766{-# INLINABLE newParallelVar #-}
1767newParallelVar :: MonadAsync m => m (SVar t m a)
1768newParallelVar = liftIO $ getParallelSVar
1770-- XXX this errors out for Parallel/Ahead SVars
1771-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
1772-- be read back from the SVar using 'fromSVar'.
1773toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
1774toStreamVar sv m = do
1775 liftIO $ (enqueue sv) m
1776 done <- allThreadsDone sv
1777 -- XXX This is safe only when called from the consumer thread or when no
1778 -- consumer is present. There may be a race if we are not running in the
1780 -- XXX do this only if the work queue is not empty. The work may have been
1781 -- carried out by existing workers.
1783 case yieldRateInfo sv of
1784 Nothing -> pushWorker 0 sv
1785 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