@@ -1,12 +1,13 @@
1−{-# LANGUAGE CPP #-}
1+{-# LANGUAGE CPP #-}
2−{-# LANGUAGE KindSignatures #-}
2+{-# LANGUAGE KindSignatures #-}
3−{-# LANGUAGE ConstraintKinds #-}
3+{-# LANGUAGE ConstraintKinds #-}
4−{-# LANGUAGE FlexibleContexts #-}
4+{-# LANGUAGE FlexibleContexts #-}
5−{-# LANGUAGE FlexibleInstances #-}
5+{-# LANGUAGE FlexibleInstances #-}
6+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
6−{-# LANGUAGE LambdaCase #-}
7+{-# LANGUAGE LambdaCase #-}
7−{-# LANGUAGE MagicHash #-}
8+{-# LANGUAGE MagicHash #-}
8−{-# LANGUAGE MultiParamTypeClasses #-}
9+{-# LANGUAGE MultiParamTypeClasses #-}
9−{-# LANGUAGE UnboxedTuples #-}
10+{-# LANGUAGE UnboxedTuples #-}
12-- Module : Streamly.SVar
13-- Module : Streamly.SVar
@@ -21,27 +22,37 @@
21module Streamly.SVar
22module Streamly.SVar
23 MonadAsync
24 MonadAsync
25 , SVarStyle (..)
25 , SVarStyle (..)
26− , defaultMaxBuffer
26+ , SVar (..)
27− , defaultMaxThreads
27+
28− , State (..)
28+ -- State threaded around the stream
29 , defState
31 , defState
30 , rstState
32 , rstState
32 , newAheadVar
44 , newAheadVar
33 , newParallelVar
45 , newParallelVar
37 , atomicModifyIORefCAS
47 , atomicModifyIORefCAS
38 , ChildEvent (..)
50 , ChildEvent (..)
39 , AheadHeapEntry (..)
51 , AheadHeapEntry (..)
40 , sendYield
53 , sendYield
41 , sendStop
54 , sendStop
42 , enqueueLIFO
55 , enqueueLIFO
45 , enqueueFIFO
56 , enqueueFIFO
46 , enqueueAhead
57 , enqueueAhead
47 , pushWorkerPar
58 , pushWorkerPar
@@ -50,18 +61,30 @@ module Streamly.SVar
50 , dequeueAhead
61 , dequeueAhead
51 , dequeueFromHeap
62 , dequeueFromHeap
66+ , workerUpdateLatency
69+ , decrementYieldLimit
70+ , incrementYieldLimit
53 , postProcessBounded
71 , postProcessBounded
54 , readOutputQBounded
73 , readOutputQBounded
55− , sendWorker
74+ , readOutputQPaced
75+ , dispatchWorkerPaced
56 , delThread
77 , delThread
60import Control.Concurrent
83import Control.Concurrent
61 (ThreadId, myThreadId, threadDelay, getNumCapabilities)
84 (ThreadId, myThreadId, threadDelay, getNumCapabilities)
62import Control.Concurrent.MVar
85import Control.Concurrent.MVar
63− (MVar, newEmptyMVar, tryPutMVar, takeMVar)
86+ (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar)
64−import Control.Exception (SomeException(..), catch, mask)
87+import Control.Exception (SomeException(..), catch, mask, assert)
65import Control.Monad (when)
88import Control.Monad (when)
66import Control.Monad.Catch (MonadThrow)
89import Control.Monad.Catch (MonadThrow)
67import Control.Monad.IO.Class (MonadIO(..))
90import Control.Monad.IO.Class (MonadIO(..))
@@ -69,17 +92,20 @@ import Control.Monad.Trans.Control (MonadBaseControl, control)
69import Data.Atomics
92import Data.Atomics
70 (casIORef, readForCAS, peekTicket, atomicModifyIORefCAS_,
93 (casIORef, readForCAS, peekTicket, atomicModifyIORefCAS_,
71 writeBarrier, storeLoadBarrier)
94 writeBarrier, storeLoadBarrier)
72−import Data.Concurrent.Queue.MichaelScott
95+import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
73− (LinkedQueue, pushL, tryPopR)
74import Data.Functor (void)
96import Data.Functor (void)
75import Data.Heap (Heap, Entry(..))
97import Data.Heap (Heap, Entry(..))
98+import Data.Int (Int64)
76import Data.IORef
99import Data.IORef
77− (IORef, modifyIORef, newIORef, readIORef, atomicModifyIORef)
100+ (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
78import Data.Maybe (fromJust)
101import Data.Maybe (fromJust)
79import Data.Set (Set)
102import Data.Set (Set)
103+import Data.Word (Word64)
80import GHC.Conc (ThreadId(..))
104import GHC.Conc (ThreadId(..))
81import GHC.Exts
105import GHC.Exts
82import GHC.IO (IO(..))
106import GHC.IO (IO(..))
108+ (TimeSpec, Clock(Monotonic), getTime, toNanoSecs, fromNanoSecs)
84import qualified Data.Heap as H
110import qualified Data.Heap as H
85import qualified Data.Set as S
111import qualified Data.Set as S
@@ -94,10 +120,33 @@ import Control.Concurrent.MVar (tryTakeMVar)
94import Control.Exception
120import Control.Exception
95 (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
121 (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
96 BlockedIndefinitelyOnSTM(..))
122 BlockedIndefinitelyOnSTM(..))
97−import Data.IORef (writeIORef)
98import System.IO (hPutStrLn, stderr)
123import System.IO (hPutStrLn, stderr)
126+newtype NanoSecs = NanoSecs Word64
138+newtype Count = Count Word64
101------------------------------------------------------------------------------
150------------------------------------------------------------------------------
102-- Parent child thread communication type
151-- Parent child thread communication type
103------------------------------------------------------------------------------
152------------------------------------------------------------------------------
@@ -148,61 +197,155 @@ data SVarStyle =
148-- New work is enqueued either at the time of creation of the SVar or as a
197-- New work is enqueued either at the time of creation of the SVar or as a
149-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
198-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
150-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
199-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
152−-- XXX can we use forall t m.
201+-- We measure the individual worker latencies to estimate the number of workers
202+-- needed or the amount of time we have to sleep between dispatches to achieve
203+-- a particular rate when controlled pace mode it used.
204+data WorkerInfo = WorkerInfo
205+ { workerYieldMax :: Count -- 0 means unlimited
206+ -- total number of yields by the worker till now
207+ , workerYieldCount :: IORef Count
208+ -- yieldCount at start, timestamp
209+ , workerLatencyStart :: IORef (Count, TimeSpec)
213+data YieldRateInfo = YieldRateInfo
214+ { expectedYieldLatency :: NanoSecs
216+ -- XXX Worker latency specified by the user to be used before the first
217+ -- actual measurement arrives. Not yet implemented
218+ , workerBootstrapLatency :: Maybe NanoSecs
220+ -- After how many yields the worker should update the latency information.
221+ -- If the latency is high, this count is kept lower and vice-versa. XXX If
222+ -- the latency suddenly becomes too high this count may remain too high for
223+ -- long time, in such cases the consumer can change it.
224+ -- 0 means no latency computation
225+ -- XXX this is derivable from workerMeasuredLatency, can be removed.
226+ , workerLatencyPeriod :: IORef Count
228+ -- This is in progress latency stats maintained by the workers which we
229+ -- empty into workerCollectedLatency stats at certain intervals - whenever
230+ -- we process the stream elements yielded in this period.
231+ -- (yieldCount, timeTaken)
232+ , workerCurrentLatency :: IORef (Count, NanoSecs)
234+ -- This is the second level stat which is an accmulation from
235+ -- workerCurrentLatency stats. We keep accumulating latencies in this
236+ -- bucket until we have stats for a sufficient period and then we reset it
237+ -- to start collecting for the next period and retain the computed average
238+ -- latency for the last period in workerMeasuredLatency.
239+ -- (yieldCount, timeTaken)
240+ , workerCollectedLatency :: IORef (Count, NanoSecs)
242+ -- Latency as measured by workers, aggregated for the last period.
243+ , workerMeasuredLatency :: IORef NanoSecs
245+ -- Actual latency/througput as seen from the consumer side, we count the
246+ -- yields and the time it took to generates those yields. This is used to
247+ -- increase or decrease the number of workers needed to achieve the desired
248+ -- rate. The idle time of workers is adjusted in this, so that we only
249+ -- account for the rate when the consumer actually demands data.
250+ , workerLongTermLatency :: IORef (Count, TimeSpec)
252+ -- When the last worker exits it updates this timestamp. We use this to
253+ -- adjust workerLongTermLatency so that we do not account the idle time in
254+ -- that. XXX Note that not all workers stop at the same time therefore
255+ -- there is a certain inaccuracy in this mechanism.
256+ , workerStopTimeStamp :: IORef TimeSpec
257+ , workerStopMVar :: MVar ()
260+data Limit = Unlimited | Limited Word deriving Show
153−data SVar t m a =
262+data SVar t m a = SVar
155− -- Read only state
264+ -- Read only state
156− svarStyle :: SVarStyle
265+ svarStyle :: SVarStyle
158− -- Shared output queue (events, length)
267+ -- Shared output queue (events, length)
159− , outputQueue :: IORef ([ChildEvent a], Int)
268+ , outputQueue :: IORef ([ChildEvent a], Int)
160− , maxYieldLimit :: Maybe (IORef Int)
161− , outputDoorBell :: MVar () -- signal the consumer about output
269+ , outputDoorBell :: MVar () -- signal the consumer about output
162− , readOutputQ :: m [ChildEvent a]
270+ , readOutputQ :: m [ChildEvent a]
163− , postProcess :: m Bool
271+ , postProcess :: m Bool
273+ -- Combined/aggregate parameters
274+ , maxWorkerLimit :: Limit
275+ , maxBufferLimit :: Limit
276+ , remainingYields :: Maybe (IORef Count)
277+ , yieldRateInfo :: Maybe YieldRateInfo
165− -- Used only by bounded SVar types
279+ -- Used only by bounded SVar types
166− , enqueue :: t m a -> IO ()
280+ , enqueue :: t m a -> IO ()
167− , isWorkDone :: IO Bool
281+ , isWorkDone :: IO Bool
168− , needDoorBell :: IORef Bool
282+ , needDoorBell :: IORef Bool
169− , workLoop :: m ()
283+ , workLoop :: WorkerInfo -> m ()
171− -- Shared, thread tracking
285+ -- Shared, thread tracking
172− , workerThreads :: IORef (Set ThreadId)
286+ , workerThreads :: IORef (Set ThreadId)
173− , workerCount :: IORef Int
287+ , workerCount :: IORef Int
174− , accountThread :: ThreadId -> m ()
288+ , accountThread :: ThreadId -> m ()
175#ifdef DIAGNOSTICS
290#ifdef DIAGNOSTICS
176− , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a))
291+ , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)
179− -- Shared work queue (stream, seqNo)
292+ -- Shared work queue (stream, seqNo)
180− , aheadWorkQueue :: IORef ([t m a], Int)
293+ , aheadWorkQueue :: IORef ([t m a], Int)
181− , totalDispatches :: IORef Int
294+ , totalDispatches :: IORef Int
182− , maxWorkers :: IORef Int
295+ , maxWorkers :: IORef Int
183− , maxOutQSize :: IORef Int
296+ , maxOutQSize :: IORef Int
184− , maxHeapSize :: IORef Int
297+ , maxHeapSize :: IORef Int
185− , maxWorkQSize :: IORef Int
298+ , maxWorkQSize :: IORef Int
302+-------------------------------------------------------------------------------
303+-- State for concurrency control
304+-------------------------------------------------------------------------------
306+-- XXX we can put the resettable fields in a oneShotConfig field and others in
307+-- a persistentConfig field. That way reset would be fast and scalable
308+-- irrespective of the number of fields.
310+-- XXX make all these Limited types and use phantom types to distinguish them
189data State t m a = State
311data State t m a = State
312+ { -- one shot configuration, automatically reset for each API call
190− { streamVar :: Maybe (SVar t m a)
313+ streamVar :: Maybe (SVar t m a)
191− , yieldLimit :: Maybe Int
314+ , _yieldLimit :: Maybe Count
192− , threadsHigh :: Int
315+
193− , bufferHigh :: Int
316+ -- persistent configuration, state that remains valid until changed by
317+ -- an explicit setting via a combinator.
318+ , _threadsHigh :: Limit
319+ , _bufferHigh :: Limit
320+ , _streamLatency :: Maybe NanoSecs
321+ , _maxStreamRate :: Maybe Double -- yields per second
196−defaultMaxThreads, defaultMaxBuffer :: Int
324+-------------------------------------------------------------------------------
197−defaultMaxThreads = 1500
325+-- State defaults and reset
198−defaultMaxBuffer = 1500
326+-------------------------------------------------------------------------------
328+-- A magical value for the buffer size arrived at by running the smallest
329+-- possible task and measuring the optimal value of the buffer for that. This
330+-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
331+-- core-i7 processor.
332+magicMaxBuffer :: Word
333+magicMaxBuffer = 1500
335+defaultMaxThreads, defaultMaxBuffer :: Limit
336+defaultMaxThreads = Limited magicMaxBuffer
337+defaultMaxBuffer = Limited magicMaxBuffer
339+-- The fields prefixed by an _ are not to be accessed or updated directly but
340+-- via smart accessor APIs.
200defState :: State t m a
341defState :: State t m a
201defState = State
342defState = State
202 { streamVar = Nothing
343 { streamVar = Nothing
203− , yieldLimit = Nothing
344+ , _yieldLimit = Nothing
204− , threadsHigh = defaultMaxThreads
345+ , _threadsHigh = defaultMaxThreads
205− , bufferHigh = defaultMaxBuffer
346+ , _bufferHigh = defaultMaxBuffer
347+ , _maxStreamRate = Nothing
348+ , _streamLatency = Nothing
208-- XXX if perf gets affected we can have all the Nothing params in a single
351-- XXX if perf gets affected we can have all the Nothing params in a single
@@ -215,9 +358,84 @@ defState = State
215rstState :: State t m a -> State t m b
358rstState :: State t m a -> State t m b
216rstState st = st
359rstState st = st
217 { streamVar = Nothing
360 { streamVar = Nothing
218− , yieldLimit = Nothing
361+ , _yieldLimit = Nothing
364+-------------------------------------------------------------------------------
365+-- Smart get/set routines for State
366+-------------------------------------------------------------------------------
368+-- Use get/set routines instead of directly accessing the State fields
369+setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
370+setYieldLimit lim st =
377+ else Just (fromIntegral n)
380+getYieldLimit :: State t m a -> Maybe Count
381+getYieldLimit = _yieldLimit
383+setMaxThreads :: Int -> State t m a -> State t m a
389+ then defaultMaxThreads
390+ else Limited (fromIntegral n)
393+getMaxThreads :: State t m a -> Limit
394+getMaxThreads = _threadsHigh
396+setMaxBuffer :: Int -> State t m a -> State t m a
402+ then defaultMaxBuffer
403+ else Limited (fromIntegral n)
406+getMaxBuffer :: State t m a -> Limit
407+getMaxBuffer = _bufferHigh
409+setMaxStreamRate :: Double -> State t m a -> State t m a
410+setMaxStreamRate n st =
411+ st { _maxStreamRate =
419+getMaxStreamRate :: State t m a -> Maybe Double
420+getMaxStreamRate = _maxStreamRate
422+setStreamLatency :: Int -> State t m a -> State t m a
423+setStreamLatency n st =
424+ st { _streamLatency =
429+ else Just (fromIntegral n)
432+getStreamLatency :: State t m a -> Maybe NanoSecs
433+getStreamLatency = _streamLatency
435+-------------------------------------------------------------------------------
436+-- Dumping the SVar for debug/diag
437+-------------------------------------------------------------------------------
221#ifdef DIAGNOSTICS
439#ifdef DIAGNOSTICS
222{-# NOINLINE dumpSVar #-}
440{-# NOINLINE dumpSVar #-}
223dumpSVar :: SVar t m a -> IO String
441dumpSVar :: SVar t m a -> IO String
@@ -287,6 +505,10 @@ withDBGMVar :: SVar t m a -> String -> IO () -> IO ()
287withDBGMVar _ _ action = action
505withDBGMVar _ _ action = action
508+-------------------------------------------------------------------------------
510+-------------------------------------------------------------------------------
290-- Slightly faster version of CAS. Gained some improvement by avoiding the use
512-- Slightly faster version of CAS. Gained some improvement by avoiding the use
291-- of "evaluate" because we know we do not have exceptions in fn.
513-- of "evaluate" because we know we do not have exceptions in fn.
292{-# INLINE atomicModifyIORefCAS #-}
514{-# INLINE atomicModifyIORefCAS #-}
@@ -346,8 +568,10 @@ doFork action exHandler =
347-- | This function is used by the producer threads to queue output for the
569-- | This function is used by the producer threads to queue output for the
348-- consumer thread to consume. Returns whether the queue has more space.
570-- consumer thread to consume. Returns whether the queue has more space.
349−send :: Int -> SVar t m a -> ChildEvent a -> IO Bool
571+send :: SVar t m a -> ChildEvent a -> IO Bool
350−send maxOutputQLen sv msg = do
572+send sv msg = do
573+ -- XXX can the access to outputQueue and maxBufferLimit be made faster
351 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
575 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
352 ((msg : es, n + 1), n)
576 ((msg : es, n + 1), n)
353 when (len <= 0) $ do
577 when (len <= 0) $ do
@@ -361,22 +585,116 @@ send maxOutputQLen sv msg = do
361 -- The important point is that the consumer is guaranteed to receive a
585 -- The important point is that the consumer is guaranteed to receive a
362 -- doorbell if something was added to the queue after it empties it.
586 -- doorbell if something was added to the queue after it empties it.
363 void $ tryPutMVar (outputDoorBell sv) ()
587 void $ tryPutMVar (outputDoorBell sv) ()
364− return (len < maxOutputQLen || maxOutputQLen < 0)
366−{-# NOINLINE sendYield #-}
589+ -- XXX we should reserve the buffer when we pick up the work from the
367−sendYield :: Int -> SVar t m a -> ChildEvent a -> IO Bool
590+ -- queue, instead of checking it here when it is too late.
368−sendYield maxOutputQLen sv msg = do
591+ let limit = maxBufferLimit sv
369− ylimit <- case maxYieldLimit sv of
592+ case limit of
370− Nothing -> return True
593+ Unlimited -> return True
371− Just ref -> atomicModifyIORefCAS ref $ \x -> (x - 1, x > 1)
594+ Limited lim -> do
372− r <- send maxOutputQLen sv msg
595+ active <- readIORef (workerCount sv)
373− return $ r && ylimit
596+ return $ len < ((fromIntegral lim) - active)
598+-- XXX We assume that a worker always yields a value. If we can have
599+-- workers that return without yielding anything our computations to
600+-- determine the number of workers may be off.
601+workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
602+workerUpdateLatency yinfo winfo = do
603+ cnt1 <- readIORef (workerYieldCount winfo)
604+ (cnt0, t0) <- readIORef (workerLatencyStart winfo)
605+ t1 <- getTime Monotonic
606+ writeIORef (workerLatencyStart winfo) (cnt1, t1)
607+ let period = fromInteger $ toNanoSecs (t1 - t0)
608+ let ref = workerCurrentLatency yinfo
609+ atomicModifyIORefCAS ref $ \(ycnt, ytime) ->
610+ ((ycnt + cnt1 - cnt0, ytime + period), ())
612+updateYieldCount :: WorkerInfo -> IO Count
613+updateYieldCount winfo = do
614+ cnt <- readIORef (workerYieldCount winfo)
616+ writeIORef (workerYieldCount winfo) cnt1
619+isBeyondMaxYield :: Count -> WorkerInfo -> Bool
620+isBeyondMaxYield cnt winfo =
621+ let ymax = workerYieldMax winfo
622+ in ymax /= 0 && cnt >= ymax
624+-- XXX we should do rate control periodically based on the total yields rather
625+-- than based on the worker local yields as other workers may have yielded more
626+-- and we should stop based on the aggregate yields. However, latency update
627+-- period can be based on individual worker yields.
628+{-# NOINLINE checkRatePeriodic #-}
629+checkRatePeriodic :: SVar t m a
634+checkRatePeriodic sv yinfo winfo ycnt = do
635+ i <- readIORef (workerLatencyPeriod yinfo)
636+ if (i /= 0 && (ycnt `mod` i) == 0)
638+ workerUpdateLatency yinfo winfo
639+ isBeyondMaxRate sv yinfo
375−{-# NOINLINE sendStop #-}
642+-- CAUTION! this also updates the yield count and therefore should be called
376−sendStop :: SVar t m a -> IO ()
643+-- only when we are actually yielding an element.
377−sendStop sv = do
644+{-# NOINLINE workerRateControl #-}
645+workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
646+workerRateControl sv yinfo winfo = do
647+ cnt <- updateYieldCount winfo
648+ beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
649+ return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
651+-- XXX Can we make access to remainingYields and yieldRateInfo fields in sv
652+-- faster, along with the fields in sv required by send?
653+-- XXX make it noinline
654+decrementYieldLimit :: SVar t m a -> IO Bool
655+decrementYieldLimit sv =
656+ case remainingYields sv of
657+ Nothing -> return True
659+ r <- atomicModifyIORefCAS ref $ \x ->
660+ (if x >= 1 then x - 1 else 0, x)
663+-- XXX increment is not safe when the yield limit was already 0
664+-- and therefore was not decremented. We need yield limit to be
665+-- int and negative if we want this to work.
666+incrementYieldLimit :: SVar t m a -> IO ()
667+incrementYieldLimit sv =
668+ case remainingYields sv of
669+ Nothing -> return ()
378− liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
670+ Just ref -> atomicModifyIORefCAS_ ref (+ 1)
672+-- XXX we should do rate control here but not latency update in case of ahead
673+-- streams. latency update must be done when we yield directly to outputQueue
674+-- or when we yield to heap.
675+{-# INLINE sendYield #-}
676+sendYield :: SVar t m a -> WorkerInfo -> ChildEvent a -> IO Bool
677+sendYield sv winfo msg = do
680+ case yieldRateInfo sv of
681+ Nothing -> return True
682+ Just yinfo -> workerRateControl sv yinfo winfo
683+ return $ r && rateLimitOk
685+{-# INLINABLE sendStop #-}
686+sendStop :: SVar t m a -> WorkerInfo -> IO ()
687+sendStop sv winfo = do
688+ n <- liftIO $ atomicModifyIORefCAS (workerCount sv) $ \n -> (n - 1, n)
689+ case yieldRateInfo sv of
690+ Nothing -> return ()
692+ i <- readIORef (workerLatencyPeriod info)
693+ when (i /= 0) $ workerUpdateLatency info winfo
695+ t <- getTime Monotonic
696+ writeIORef (workerStopTimeStamp info) t
379− myThreadId >>= \tid -> void $ send (-1) sv (ChildStop tid Nothing)
697+ myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
381-------------------------------------------------------------------------------
699-------------------------------------------------------------------------------
@@ -401,25 +719,6 @@ enqueueLIFO sv q m = do
401 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
719 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
402 void $ tryPutMVar (outputDoorBell sv) ()
720 void $ tryPutMVar (outputDoorBell sv) ()
404−{-# INLINE workLoopLIFO #-}
405−workLoopLIFO :: MonadIO m
406− => (State t m a -> IORef [t m a] -> t m a -> m () -> m ())
407− -> State t m a -> IORef [t m a] -> m ()
408−workLoopLIFO f st q = run
412− sv = fromJust $ streamVar st
416− Nothing -> liftIO $ sendStop sv
417− Just m -> f st q m run
419− dequeue = liftIO $ atomicModifyIORefCAS q $ \case
421− x : xs -> (xs, Just x)
423-------------------------------------------------------------------------------
722-------------------------------------------------------------------------------
425-------------------------------------------------------------------------------
724-------------------------------------------------------------------------------
@@ -443,21 +742,6 @@ enqueueFIFO sv q m = do
443 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
742 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
444 void $ tryPutMVar (outputDoorBell sv) ()
743 void $ tryPutMVar (outputDoorBell sv) ()
446−{-# INLINE workLoopFIFO #-}
447−workLoopFIFO :: MonadIO m
448− => (State t m a -> LinkedQueue (t m a) -> t m a -> m () -> m ())
449− -> State t m a -> LinkedQueue (t m a) -> m ()
450−workLoopFIFO f st q = run
454− sv = fromJust $ streamVar st
456− work <- liftIO $ tryPopR q
458− Nothing -> liftIO $ sendStop sv
459− Just m -> f st q m run
461-------------------------------------------------------------------------------
745-------------------------------------------------------------------------------
463-------------------------------------------------------------------------------
747-------------------------------------------------------------------------------
@@ -510,10 +794,18 @@ workLoopFIFO f st q = run
510-- that they do not hog the resources and hinder the progress of the threads in
794-- that they do not hog the resources and hinder the progress of the threads in
511-- front of them.
795-- front of them.
513−-- Left associated ahead expressions are expensive. We start a new SVar for
797+-- XXX Left associated ahead expressions are expensive. We start a new SVar for
514-- each left associative expression. The queue is used only for right
798-- each left associative expression. The queue is used only for right
515-- associated expression, we queue the right expression and execute the left.
799-- associated expression, we queue the right expression and execute the left.
516−-- Thererefore the queue never has more than on item in it.
800+-- Thererefore the queue never has more than one item in it.
802+-- XXX we can fix this. When we queue more than one item on the queue we can
803+-- mark the previously queued item as not-runnable. The not-runnable item is
804+-- not dequeued until the already running one has finished and at that time we
805+-- would also know the exact sequence number of the already queued item.
807+-- we can even run the already queued items but they will have to be sorted in
808+-- layers in the heap. We can use a list of heaps for that.
517{-# INLINE enqueueAhead #-}
809{-# INLINE enqueueAhead #-}
518enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
810enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
519enqueueAhead sv q m = do
811enqueueAhead sv q m = do
@@ -571,7 +863,7 @@ dequeueFromHeap
571 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
863 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
572 -> IO (Maybe (Entry Int (AheadHeapEntry t m a)))
864 -> IO (Maybe (Entry Int (AheadHeapEntry t m a)))
573dequeueFromHeap hpRef = do
865dequeueFromHeap hpRef = do
574− atomicModifyIORefCAS hpRef $ \hp@(h, snum) -> do
866+ atomicModifyIORef hpRef $ \hp@(h, snum) -> do
575 let r = H.uncons h
867 let r = H.uncons h
576 case r of
868 case r of
577 Nothing -> (hp, Nothing)
869 Nothing -> (hp, Nothing)
@@ -637,7 +929,7 @@ allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
637handleChildException :: SVar t m a -> SomeException -> IO ()
929handleChildException :: SVar t m a -> SomeException -> IO ()
638handleChildException sv e = do
930handleChildException sv e = do
639 tid <- myThreadId
931 tid <- myThreadId
640− void $ send (-1) sv (ChildStop tid (Just e))
932+ void $ send sv (ChildStop tid (Just e))
642#ifdef DIAGNOSTICS
934#ifdef DIAGNOSTICS
643recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
935recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
@@ -649,13 +941,24 @@ recordMaxWorkers sv = liftIO $ do
651{-# NOINLINE pushWorker #-}
943{-# NOINLINE pushWorker #-}
652−pushWorker :: MonadAsync m => SVar t m a -> m ()
944+pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
653−pushWorker sv = do
945+pushWorker yieldMax sv = do
654 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
946 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
655#ifdef DIAGNOSTICS
947#ifdef DIAGNOSTICS
656 recordMaxWorkers sv
948 recordMaxWorkers sv
950+ -- XXX we can make this allocation conditional, it might matter when
951+ -- significant number of workers are being sent.
953+ cntRef <- liftIO $ newIORef 0
954+ t <- liftIO $ getTime Monotonic
955+ lat <- liftIO $ newIORef (0, t)
957+ { workerYieldMax = yieldMax
958+ , workerYieldCount = cntRef
959+ , workerLatencyStart = lat
658− doFork (workLoop sv) (handleChildException sv) >>= addThread sv
961+ doFork (workLoop sv winfo) (handleChildException sv) >>= addThread sv
660-- XXX we can push the workerCount modification in accountThread and use the
963-- XXX we can push the workerCount modification in accountThread and use the
661-- same pushWorker for Parallel case as well.
964-- same pushWorker for Parallel case as well.
@@ -676,49 +979,313 @@ pushWorkerPar sv wloop = do
677 doFork wloop (handleChildException sv) >>= modifyThread sv
980 doFork wloop (handleChildException sv) >>= modifyThread sv
983+-- True: can dispatch more
984+-- False: cannot dispatch any more
679−dispatchWorker :: MonadAsync m => Int -> SVar t m a -> m ()
985+dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
986+dispatchWorker yieldCount sv = do
680−dispatchWorker maxWorkerLimit sv = do
987+ let workerLimit = maxWorkerLimit sv
988+ -- XXX in case of Ahead streams we should not send more than one worker
989+ -- when the work queue is done but heap is not done.
681 done <- liftIO $ isWorkDone sv
990 done <- liftIO $ isWorkDone sv
682− when (not done) $ do
991+ if (not done)
683 -- Note that the worker count is only decremented during event
993 -- Note that the worker count is only decremented during event
684 -- processing in fromStreamVar and therefore it is safe to read and
994 -- processing in fromStreamVar and therefore it is safe to read and
685 -- use it without a lock.
995 -- use it without a lock.
686− cnt <- liftIO $ readIORef $ workerCount sv
996+ active <- liftIO $ readIORef $ workerCount sv
687 -- Note that we may deadlock if the previous workers (tasks in the
997 -- Note that we may deadlock if the previous workers (tasks in the
688 -- stream) wait/depend on the future workers (tasks in the stream)
998 -- stream) wait/depend on the future workers (tasks in the stream)
689 -- executing. In that case we should either configure the maxWorker
999 -- executing. In that case we should either configure the maxWorker
690 -- count to higher or use parallel style instead of ahead or async
1000 -- count to higher or use parallel style instead of ahead or async
691 -- style.
1001 -- style.
692− limit <- case maxYieldLimit sv of
1002+ limit <- case remainingYields sv of
693− Nothing -> return maxWorkerLimit
1003+ Nothing -> return workerLimit
694 Just x -> do
1004 Just x -> do
695− lim <- liftIO $ readIORef x
1005+ n <- liftIO $ readIORef x
696 return $
1006 return $
697− if maxWorkerLimit > 0
1007+ case workerLimit of
698− then min maxWorkerLimit lim
1008+ Unlimited -> Limited (fromIntegral n)
699− else lim
1009+ Limited lim -> Limited $ min lim (fromIntegral n)
700− when (cnt < limit || limit < 0) $ pushWorker sv
1010+
1011+ -- XXX for ahead streams shall we take the heap yields into account for
1012+ -- controlling the dispatch? We should not dispatch if the heap has
1013+ -- already got the limit covered.
1014+ let dispatch = pushWorker yieldCount sv >> return True
1016+ Unlimited -> dispatch
1017+ -- Note that the use of remainingYields and workerCount is not
1018+ -- atomic and the counts may even have changed between reading and
1019+ -- using them here, so this is just approximate logic and we cannot
1020+ -- rely on it for correctness. We may actually dispatch more
1021+ -- workers than required.
1022+ Limited lim | active < (fromIntegral lim) -> dispatch
702−{-# NOINLINE sendWorkerWait #-}
1026+-- | This is a magic number and it is overloaded, and used at several places to
1027+-- achieve batching:
1029+-- 1. If we have to sleep to slowdown this is the minimum period that we
1030+-- accumulate before we sleep. Also, workers do not stop until this much
1031+-- sleep time is accumulated.
1032+-- 3. Collected latencies are computed and transferred to measured latency
1033+-- after a minimum of this period.
1034+minThreadDelay :: NanoSecs
1035+minThreadDelay = 10^(6 :: Int)
1037+-- | Another magic number! When we have to start more workers to cover up a
1038+-- number of yields that we are lagging by then we cannot start one worker for
1039+-- each yield because that may be a very big number and if the latency of the
1040+-- workers is low these number of yields could be very high. We use this number
1041+-- as the duration in which we want cover any lag. In other words we accumulate
1042+-- the lag (lagging number of yields) such that it covers this much duration
1043+-- for a single worker to produce those yields, before we start an extra
1045+coverUpLatency :: NanoSecs
1046+coverUpLatency = 1000000
1048+nanoToMicroSecs :: NanoSecs -> Int
1049+nanoToMicroSecs s = (fromIntegral s) `div` 1000
1051+-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1052+countDispatchWorkers
1053+ :: Limit -> Count -> NanoSecs -> NanoSecs -> NanoSecs -> Int
1054+countDispatchWorkers workerLimit count duration wLatency expectedLat =
1055+ -- Calculate how many workers do we need to maintain the rate.
1057+ -- XXX Increasing the workers may reduce the overall latency per
1058+ -- yield of the workers, sometimes quite a bit. We can keep a
1059+ -- latency histogram for 1,2,4..2^20 workers so that we can decide
1060+ -- which bucket is optimal and decide the number of workers based
1063+ -- XXX put a cap on extraworkers? Should not be more than nWorkers.
1064+ -- Bridge the gap slowly rather than in a burst in one go.
1066+ -- XXX we can have a maxEfficiency combinator as well which runs the
1067+ -- producer at the maximal efficiency i.e. the number of workers are chosen
1068+ -- such that the latency is minimum or within a range. Or we can call it
1071+ let nWorkers = (fromIntegral wLatency) / (fromIntegral expectedLat)
1072+ -- Calculate how many yields are we ahead or behind to match the exact
1073+ -- required rate. Based on that we increase or decrease the effective
1075+ d = fromIntegral duration
1076+ e = fromIntegral expectedLat
1077+ n = max 1 (fromIntegral coverUpLatency / fromIntegral wLatency)
1079+ extraYields = (d / e) - (fromIntegral count)
1080+ extraWorkers = extraYields / n
1081+ requiredWorkers = round $ nWorkers + extraWorkers
1082+ in case workerLimit of
1083+ Unlimited -> requiredWorkers
1084+ Limited x -> min requiredWorkers (fromIntegral x)
1086+-- | Get the worker latency without resetting workerCurrentLatency
1087+-- Returns (total yield count, base time, measured latency)
1088+getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1089+getWorkerLatency yinfo = do
1090+ let cur = workerCurrentLatency yinfo
1091+ col = workerCollectedLatency yinfo
1092+ longTerm = workerLongTermLatency yinfo
1093+ measured = workerMeasuredLatency yinfo
1095+ (count, time) <- readIORef cur
1096+ (colCount, colTime) <- readIORef col
1097+ (lcount, ltime) <- readIORef longTerm
1098+ prev <- readIORef measured
1100+ let pendingCount = colCount + count
1101+ pendingTime = colTime + time
1102+ -- Take the average of previous and the latest
1104+ if pendingCount > 0
1105+ then ((pendingTime `div` (fromIntegral pendingCount)) + prev)
1108+ return (lcount + pendingCount, ltime, new)
1110+isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1111+isBeyondMaxRate sv yinfo = do
1112+ (count, tstamp, wLatency) <- getWorkerLatency yinfo
1113+ now <- getTime Monotonic
1114+ let duration = fromInteger $ toNanoSecs $ now - tstamp
1115+ let expectedLat = expectedYieldLatency yinfo
1116+ let netWorkers = countDispatchWorkers (maxWorkerLimit sv) count duration
1117+ wLatency expectedLat
1118+ cnt <- readIORef $ workerCount sv
1121+ let sleepTime = ((fromIntegral count) * expectedLat) - duration
1122+ if sleepTime >= minThreadDelay
1125+ else if cnt > netWorkers
1129+useCollectedBatch :: YieldRateInfo
1131+ -> IORef (Count, NanoSecs)
1134+useCollectedBatch yinfo latency colRef measRef = do
1135+ let periodRef = workerLatencyPeriod yinfo
1136+ cnt = max 1 $ minThreadDelay `div` latency
1137+ period = min cnt (fromIntegral magicMaxBuffer)
1139+ -- XXX we should use some weight for the old latency as well in arriving at
1140+ -- the new latency, we have the counts, we can use that for some weighting.
1141+ writeIORef periodRef (fromIntegral period)
1142+ writeIORef colRef (0, 0)
1143+ writeIORef measRef latency
1145+-- Returns a triple, (1) yield count since last collection, (2) the base time
1146+-- when we started counting, (3) average latency in the last measurement
1147+-- period. The former two are used for accurate measurement of the going rate
1148+-- whereas the average is used for future estimates e.g. how many workers
1149+-- should be maintained to maintain the rate.
1150+collectLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1151+collectLatency yinfo = do
1152+ let cur = workerCurrentLatency yinfo
1153+ col = workerCollectedLatency yinfo
1154+ longTerm = workerLongTermLatency yinfo
1155+ measured = workerMeasuredLatency yinfo
1157+ (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
1158+ (colCount, colTime) <- readIORef col
1159+ (lcount, ltime) <- readIORef longTerm
1160+ prev <- readIORef measured
1162+ let pendingCount = colCount + count
1163+ pendingTime = colTime + time
1165+ lcount' = lcount + pendingCount
1166+ notUpdated = (lcount', ltime, prev)
1168+ if (pendingCount > 0)
1170+ let new = pendingTime `div` (fromIntegral pendingCount)
1171+ -- To avoid minor fluctuations update in batches
1172+ if (pendingCount > fromIntegral magicMaxBuffer)
1173+ || (pendingTime > minThreadDelay)
1174+ || (let r = (fromIntegral new) / (fromIntegral prev) :: Double
1175+ in prev > 0 && (r > 2 || r < 0.5))
1178+ useCollectedBatch yinfo new col measured
1179+ modifyIORef longTerm $ \(_, t) -> (lcount', t)
1180+ return (lcount', ltime, new)
1182+ writeIORef col (pendingCount, pendingTime)
1184+ else return notUpdated
1186+adjustLatencies :: YieldRateInfo -> IO ()
1187+adjustLatencies yinfo = do
1188+ let longTerm = workerLongTermLatency yinfo
1189+ t0 <- readIORef (workerStopTimeStamp yinfo)
1190+ t1 <- getTime Monotonic
1191+ modifyIORef longTerm $ \(c, t) -> (c, t + t1 - t0)
1194+-- XXX in case of ahead style stream we need to take the heap size into account
1195+-- because we return the workers on the basis of that which causes a condition
1196+-- where we keep dispatching and they keep returning. So we must have exactly
1197+-- the same logic for not dispatching and for returning.
1200+-- True: can dispatch more
1201+-- False: full, no more dispatches
703−sendWorkerWait :: MonadAsync m => Int -> SVar t m a -> m ()
1202+dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1203+dispatchWorkerPaced sv = do
704−sendWorkerWait maxWorkerLimit sv = do
1204+ let workerLimit = maxWorkerLimit sv
705− -- Note that we are guaranteed to have at least one outstanding worker when
1205+ let yinfo = fromJust $ yieldRateInfo sv
706− -- we enter this function. So if we sleep we are guaranteed to be woken up
1206+ (count, tstamp, wLatency) <- liftIO $ collectLatency yinfo
707− -- by a outputDoorBell, when the worker exits.
1207+ -- XXX use workerBootstrapLatency
1209+ -- Need to measure the latency with a single worker before we can perform
1210+ -- any computation.
1213+ now <- liftIO $ getTime Monotonic
1214+ let duration = fromInteger $ toNanoSecs $ now - tstamp
1215+ let expectedLat = expectedYieldLatency yinfo
1216+ let netWorkers = countDispatchWorkers workerLimit count duration
1217+ wLatency expectedLat
1219+ -- XXX we need to take yieldLimit into account here. If we are at the
1220+ -- end of the limit as well as the time, we should not be sleeping.
1221+ -- If we are not actually planning to dispatch any more workers we need
1222+ -- to take that in account.
1225+ let sleepTime = ((fromIntegral count) * expectedLat) - duration
1227+ when (sleepTime >= minThreadDelay) $ do
1228+ liftIO $ threadDelay $ nanoToMicroSecs sleepTime
1230+ cnt <- liftIO $ readIORef $ workerCount sv
1231+ when (cnt < 1) $ void $ do
1232+ let n = countDispatchYields (expectedLat - wLatency)
1233+ in dispatchWorker (fromIntegral (max n 1)) sv
1236+ -- XXX stagger the workers over a period
1237+ cnt <- liftIO $ readIORef $ workerCount sv
1238+ if (cnt < netWorkers)
1239+ then dispatchWorker 0 sv
1244+ countDispatchYields :: NanoSecs -> Int
1245+ countDispatchYields idleTime =
1246+ assert (idleTime >= 0) $
1247+ -- we do not want to sleep for less than minThreadDelay
1248+ let maxYieldsPerWorker = fromIntegral magicMaxBuffer
1249+ in if idleTime < minThreadDelay
1252+ then maxYieldsPerWorker
1254+ let n = (minThreadDelay + idleTime) `div` idleTime
1255+ in min (fromIntegral n) maxYieldsPerWorker
1258+sendWorkerDelayPaced :: SVar t m a -> IO ()
1259+sendWorkerDelayPaced _ = return ()
1261+sendWorkerDelay :: SVar t m a -> IO ()
1262+sendWorkerDelay sv = do
709 -- XXX we need a better way to handle this than hardcoded delays. The
1263 -- XXX we need a better way to handle this than hardcoded delays. The
710 -- delays may be different for different systems.
1264 -- delays may be different for different systems.
711− ncpu <- liftIO $ getNumCapabilities
1265+ ncpu <- getNumCapabilities
712 if ncpu <= 1
1266 if ncpu <= 1
714 if (svarStyle sv == AheadVar)
1268 if (svarStyle sv == AheadVar)
715− then liftIO $ threadDelay 100
1269+ then threadDelay 100
716− else liftIO $ threadDelay 25
1270+ else threadDelay 25
718 if (svarStyle sv == AheadVar)
1272 if (svarStyle sv == AheadVar)
719− then liftIO $ threadDelay 100
1273+ then threadDelay 100
720− else liftIO $ threadDelay 10
1274+ else threadDelay 10
1276+{-# NOINLINE sendWorkerWait #-}
1279+ => (SVar t m a -> IO ())
1280+ -> (SVar t m a -> m Bool)
1283+sendWorkerWait delay dispatch sv = do
1284+ -- Note that we are guaranteed to have at least one outstanding worker when
1285+ -- we enter this function. So if we sleep we are guaranteed to be woken up
1286+ -- by a outputDoorBell, when the worker exits.
722 (_, n) <- liftIO $ readIORef (outputQueue sv)
1289 (_, n) <- liftIO $ readIORef (outputQueue sv)
723 when (n <= 0) $ do
1290 when (n <= 0) $ do
724 -- The queue may be empty temporarily if the worker has dequeued the
1291 -- The queue may be empty temporarily if the worker has dequeued the
@@ -753,7 +1320,7 @@ sendWorkerWait maxWorkerLimit sv = do
754 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
1321 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
755 liftIO $ storeLoadBarrier
1322 liftIO $ storeLoadBarrier
756− dispatchWorker maxWorkerLimit sv
1323+ canDoMore <- dispatch sv
758 -- XXX test for the case when we miss sending a worker when the worker
1325 -- XXX test for the case when we miss sending a worker when the worker
759 -- count is more than 1500.
1326 -- count is more than 1500.
@@ -762,14 +1329,13 @@ sendWorkerWait maxWorkerLimit sv = do
762 -- least one outstanding worker. Otherwise we could be sleeping
1329 -- least one outstanding worker. Otherwise we could be sleeping
763 -- forever.
1330 -- forever.
765− done <- liftIO $ isWorkDone sv
1332+ if canDoMore
766− if done
1333+ then sendWorkerWait delay dispatch sv
767− then do
1334+ else do
768 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
1335 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
769 $ takeMVar (outputDoorBell sv)
1336 $ takeMVar (outputDoorBell sv)
770 (_, len) <- liftIO $ readIORef (outputQueue sv)
1337 (_, len) <- liftIO $ readIORef (outputQueue sv)
771− when (len <= 0) $ sendWorkerWait maxWorkerLimit sv
1338+ when (len <= 0) $ sendWorkerWait delay dispatch sv
772− else sendWorkerWait maxWorkerLimit sv
774{-# INLINE readOutputQRaw #-}
1340{-# INLINE readOutputQRaw #-}
775readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
1341readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
@@ -781,8 +1347,8 @@ readOutputQRaw sv = do
782 return (list, len)
1348 return (list, len)
784−readOutputQBounded :: MonadAsync m => Int -> SVar t m a -> m [ChildEvent a]
1350+readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
785−readOutputQBounded n sv = do
1351+readOutputQBounded sv = do
786 (list, len) <- liftIO $ readOutputQRaw sv
1352 (list, len) <- liftIO $ readOutputQRaw sv
787 -- When there is no output seen we dispatch more workers to help
1353 -- When there is no output seen we dispatch more workers to help
788 -- out if there is work pending in the work queue.
1354 -- out if there is work pending in the work queue.
@@ -802,32 +1368,96 @@ readOutputQBounded n sv = do
802 cnt <- liftIO $ readIORef $ workerCount sv
1368 cnt <- liftIO $ readIORef $ workerCount sv
803 when (cnt <= 0) $ do
1369 when (cnt <= 0) $ do
804 done <- liftIO $ isWorkDone sv
1370 done <- liftIO $ isWorkDone sv
805− when (not done) $ pushWorker sv
1371+ when (not done) $ pushWorker 0 sv
807 {-# INLINE blockingRead #-}
1373 {-# INLINE blockingRead #-}
808 blockingRead = do
1374 blockingRead = do
809− sendWorkerWait n sv
1375+ sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
1376+ liftIO $ (readOutputQRaw sv >>= return . fst)
1378+readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1379+readOutputQPaced sv = do
1380+ (list, len) <- liftIO $ readOutputQRaw sv
1384+ -- XXX send a worker proactively, if needed, even before we start
1385+ -- processing the output.
1386+ void $ dispatchWorkerPaced sv
1391+ {-# INLINE blockingRead #-}
1393+ sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
810 liftIO $ (readOutputQRaw sv >>= return . fst)
1394 liftIO $ (readOutputQRaw sv >>= return . fst)
812postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
1396postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
813postProcessBounded sv = do
1397postProcessBounded sv = do
814 workersDone <- allThreadsDone sv
1398 workersDone <- allThreadsDone sv
815− -- There may still be work pending even if there are no workers
1399+ -- There may still be work pending even if there are no workers pending
816− -- pending because all the workers may return if the
1400+ -- because all the workers may return if the outputQueue becomes full. In
817− -- outputQueue becomes full. In that case send off a worker to
818− -- kickstart the work again.
1401+ -- that case send off a worker to kickstart the work again.
1404+ r <- liftIO $ isWorkDone sv
1405+ -- XXX do we need to dispatch many here?
1406+ when (not r) $ pushWorker 0 sv
1407+ -- void $ dispatchWorker sv
1411+postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
1412+postProcessPaced sv = do
1413+ workersDone <- allThreadsDone sv
1414+ -- XXX If during consumption we figure out we are getting delayed then we
1415+ -- should trigger dispatch there as well. We should try to check on the
1416+ -- workers after consuming every n item from the buffer?
819 if workersDone
1417 if workersDone
821 r <- liftIO $ isWorkDone sv
1419 r <- liftIO $ isWorkDone sv
822− when (not r) $ pushWorker sv
1420+ when (not r) $ do
1421+ let yinfo = fromJust $ yieldRateInfo sv
1422+ liftIO $ adjustLatencies yinfo
1423+ void $ dispatchWorkerPaced sv
823 return r
1424 return r
824 else return False
1425 else return False
1427+getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
1428+getYieldRateInfo st = do
1429+ case getMaxStreamRate st of
1431+ let latency = round $ 1.0e9 / rate
1432+ measured <- newIORef 0
1433+ wcur <- newIORef (0,0)
1434+ wcol <- newIORef (0,0)
1435+ now <- getTime Monotonic
1436+ wlong <- newIORef (0,now)
1437+ period <- newIORef 1
1438+ stopTime <- newIORef $ fromNanoSecs 0
1439+ stopMVar <- newMVar ()
1441+ return $ Just YieldRateInfo
1442+ { expectedYieldLatency = latency
1443+ , workerBootstrapLatency = getStreamLatency st
1444+ , workerLatencyPeriod = period
1445+ , workerMeasuredLatency = measured
1446+ , workerCurrentLatency = wcur
1447+ , workerStopTimeStamp = stopTime
1448+ , workerCollectedLatency = wcol
1449+ , workerLongTermLatency = wlong
1450+ , workerStopMVar = stopMVar
1452+ Nothing -> return Nothing
826getAheadSVar :: MonadAsync m
1454getAheadSVar :: MonadAsync m
827 => State t m a
1455 => State t m a
829− -> IORef ([t m a], Int)
1456+ -> ( IORef ([t m a], Int)
830 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
1457 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
831 -> m ())
1461 -> m ())
832 -> IO (SVar t m a)
1462 -> IO (SVar t m a)
833getAheadSVar st f = do
1463getAheadSVar st f = do
@@ -838,53 +1468,71 @@ getAheadSVar st f = do
838 wfw <- newIORef False
1468 wfw <- newIORef False
839 running <- newIORef S.empty
1469 running <- newIORef S.empty
840 q <- newIORef ([], -1)
1470 q <- newIORef ([], -1)
841− yl <- case yieldLimit st of
1471+ yl <- case getYieldLimit st of
842 Nothing -> return Nothing
1472 Nothing -> return Nothing
843 Just x -> Just <$> newIORef x
1473 Just x -> Just <$> newIORef x
1474+ rateInfo <- getYieldRateInfo st
845#ifdef DIAGNOSTICS
1476#ifdef DIAGNOSTICS
846− disp <- newIORef 0
1477+ disp <- newIORef 0
847 maxWrk <- newIORef 0
1478 maxWrk <- newIORef 0
848 maxOq <- newIORef 0
1479 maxOq <- newIORef 0
849 maxHs <- newIORef 0
1480 maxHs <- newIORef 0
850 maxWq <- newIORef 0
1481 maxWq <- newIORef 0
1484+ let getSVar sv readOutput postProc = SVar
853− SVar { outputQueue = outQ
1485+ { outputQueue = outQ
854− , maxYieldLimit = yl
1486+ , remainingYields = yl
1487+ , maxBufferLimit = getMaxBuffer st
1488+ , maxWorkerLimit = getMaxThreads st
1489+ , yieldRateInfo = rateInfo
855− , outputDoorBell = outQMv
1490+ , outputDoorBell = outQMv
856− , readOutputQ = readOutputQBounded (threadsHigh st) sv
1491+ , readOutputQ = readOutput sv
857− , postProcess = postProcessBounded sv
1492+ , postProcess = postProc sv
858− , workerThreads = running
1493+ , workerThreads = running
859− -- , workLoop = workLoopAhead sv q outH
860− , workLoop = f st{streamVar = Just sv} q outH
1494+ , workLoop = f q outH st{streamVar = Just sv} sv
861− , enqueue = enqueueAhead sv q
1495+ , enqueue = enqueueAhead sv q
862− , isWorkDone = isWorkDoneAhead q outH
1496+ , isWorkDone = isWorkDoneAhead sv q outH
863− , needDoorBell = wfw
1497+ , needDoorBell = wfw
864− , svarStyle = AheadVar
1498+ , svarStyle = AheadVar
865− , workerCount = active
1499+ , workerCount = active
866− , accountThread = delThread sv
1500+ , accountThread = delThread sv
867#ifdef DIAGNOSTICS
1501#ifdef DIAGNOSTICS
868− , aheadWorkQueue = q
1502+ , aheadWorkQueue = q
869− , outputHeap = outH
1503+ , outputHeap = outH
870− , totalDispatches = disp
1504+ , totalDispatches = disp
871− , maxWorkers = maxWrk
1505+ , maxWorkers = maxWrk
872− , maxOutQSize = maxOq
1506+ , maxOutQSize = maxOq
873− , maxHeapSize = maxHs
1507+ , maxHeapSize = maxHs
874− , maxWorkQSize = maxWq
1508+ , maxWorkQSize = maxWq
1513+ case getMaxStreamRate st of
1514+ Nothing -> getSVar sv readOutputQBounded postProcessBounded
1515+ Just _ -> getSVar sv readOutputQPaced postProcessPaced
877 in return sv
1516 in return sv
881 {-# INLINE isWorkDoneAhead #-}
1520 {-# INLINE isWorkDoneAhead #-}
882− isWorkDoneAhead q ref = do
1521+ isWorkDoneAhead sv q ref = do
883 heapDone <- do
1522 heapDone <- do
884 (hp, _) <- readIORef ref
1523 (hp, _) <- readIORef ref
885 return (H.size hp <= 0)
1524 return (H.size hp <= 0)
886 queueDone <- checkEmpty q
1525 queueDone <- checkEmpty q
1527+ case remainingYields sv of
1529+ n <- readIORef yref
1531+ Nothing -> return False
1532+ -- XXX note that yieldsDone can only be authoritative only when there
1533+ -- are no workers running. If there are active workers they can
1534+ -- increment the yield count and therefore change the result.
887− return $ queueDone && heapDone
1535+ return $ yieldsDone || (queueDone && heapDone)
889 checkEmpty q = do
1537 checkEmpty q = do
890 (xs, _) <- readIORef q
1538 (xs, _) <- readIORef q
@@ -905,7 +1553,10 @@ getParallelSVar = do
906 let sv =
1554 let sv =
907 SVar { outputQueue = outQ
1555 SVar { outputQueue = outQ
908− , maxYieldLimit = Nothing
1556+ , remainingYields = Nothing
1557+ , maxBufferLimit = Unlimited
1558+ , maxWorkerLimit = undefined
1559+ , yieldRateInfo = Nothing
909 , outputDoorBell = outQMv
1560 , outputDoorBell = outQMv
910 , readOutputQ = readOutputQPar sv
1561 , readOutputQ = readOutputQPar sv
911 , postProcess = allThreadsDone sv
1562 , postProcess = allThreadsDone sv
@@ -935,27 +1586,31 @@ getParallelSVar = do
935 withDBGMVar sv "readOutputQPar: doorbell" $ takeMVar (outputDoorBell sv)
1586 withDBGMVar sv "readOutputQPar: doorbell" $ takeMVar (outputDoorBell sv)
936 readOutputQRaw sv >>= return . fst
1587 readOutputQRaw sv >>= return . fst
938−sendWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
1589+sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
939−sendWorker sv m = do
1590+sendFirstWorker sv m = do
940 -- Note: We must have all the work on the queue before sending the
1591 -- Note: We must have all the work on the queue before sending the
941 -- pushworker, otherwise the pushworker may exit before we even get a
1592 -- pushworker, otherwise the pushworker may exit before we even get a
942 -- chance to push.
1593 -- chance to push.
943 liftIO $ enqueue sv m
1594 liftIO $ enqueue sv m
1595+ case yieldRateInfo sv of
1596+ Nothing -> pushWorker 0 sv
944− pushWorker sv
1597+ Just _ -> pushWorker 1 sv
945 return sv
1598 return sv
947{-# INLINABLE newAheadVar #-}
1600{-# INLINABLE newAheadVar #-}
948newAheadVar :: MonadAsync m
1601newAheadVar :: MonadAsync m
949 => State t m a
1602 => State t m a
950 -> t m a
1603 -> t m a
952− -> IORef ([t m a], Int)
1604+ -> ( IORef ([t m a], Int)
953 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
1605 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
954 -> m ())
1609 -> m ())
955 -> m (SVar t m a)
1610 -> m (SVar t m a)
956newAheadVar st m wloop = do
1611newAheadVar st m wloop = do
957 sv <- liftIO $ getAheadSVar st wloop
1612 sv <- liftIO $ getAheadSVar st wloop
958− sendWorker sv m
1613+ sendFirstWorker sv m
960{-# INLINABLE newParallelVar #-}
1615{-# INLINABLE newParallelVar #-}
961newParallelVar :: MonadAsync m => m (SVar t m a)
1616newParallelVar :: MonadAsync m => m (SVar t m a)
@@ -971,4 +1626,9 @@ toStreamVar sv m = do
971 -- XXX This is safe only when called from the consumer thread or when no
1626 -- XXX This is safe only when called from the consumer thread or when no
972 -- consumer is present. There may be a race if we are not running in the
1627 -- consumer is present. There may be a race if we are not running in the
973 -- consumer thread.
1628 -- consumer thread.
1629+ -- XXX do this only if the work queue is not empty. The work may have been
1630+ -- carried out by existing workers.
974− when done $ pushWorker sv
1631+ when done $
1632+ case yieldRateInfo sv of
1633+ Nothing -> pushWorker 0 sv
1634+ Just _ -> pushWorker 1 sv