@@ -1,12 +1,13 @@
1−{-# LANGUAGE CPP #-}
2−{-# LANGUAGE KindSignatures #-}
3−{-# LANGUAGE ConstraintKinds #-}
4−{-# LANGUAGE FlexibleContexts #-}
5−{-# LANGUAGE FlexibleInstances #-}
6−{-# LANGUAGE LambdaCase #-}
7−{-# LANGUAGE MagicHash #-}
8−{-# LANGUAGE MultiParamTypeClasses #-}
9−{-# LANGUAGE UnboxedTuples #-}
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE KindSignatures #-}
3+{-# LANGUAGE ConstraintKinds #-}
4+{-# LANGUAGE FlexibleContexts #-}
5+{-# LANGUAGE FlexibleInstances #-}
6+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
7+{-# LANGUAGE LambdaCase #-}
8+{-# LANGUAGE MagicHash #-}
9+{-# LANGUAGE MultiParamTypeClasses #-}
10+{-# LANGUAGE UnboxedTuples #-}
1011
1112-- |
1213-- Module : Streamly.SVar
@@ -21,27 +22,37 @@
2122module Streamly.SVar
2223 (
2324 MonadAsync
24− , SVar (..)
2525 , SVarStyle (..)
26− , defaultMaxBuffer
27− , defaultMaxThreads
28− , State (..)
26+ , SVar (..)
27+
28+ -- State threaded around the stream
29+ , Limit (..)
30+ , State (streamVar)
2931 , defState
3032 , rstState
31−
33+ , getMaxThreads
34+ , setMaxThreads
35+ , getMaxBuffer
36+ , setMaxBuffer
37+ , getMaxStreamRate
38+ , setMaxStreamRate
39+ , setStreamLatency
40+ , getYieldLimit
41+ , setYieldLimit
42+
43+ -- SVar related
3244 , newAheadVar
3345 , newParallelVar
3446
35− , toStreamVar
36−
3747 , atomicModifyIORefCAS
48+ , WorkerInfo (..)
49+ , YieldRateInfo (..)
3850 , ChildEvent (..)
3951 , AheadHeapEntry (..)
52+ , send
4053 , sendYield
4154 , sendStop
4255 , enqueueLIFO
43− , workLoopLIFO
44− , workLoopFIFO
4556 , enqueueFIFO
4657 , enqueueAhead
4758 , pushWorkerPar
@@ -50,18 +61,30 @@ module Streamly.SVar
5061 , dequeueAhead
5162 , dequeueFromHeap
5263
64+ , getYieldRateInfo
65+ , collectLatency
66+ , workerUpdateLatency
67+ , workerRateControl
68+ , updateYieldCount
69+ , decrementYieldLimit
70+ , incrementYieldLimit
5371 , postProcessBounded
72+ , postProcessPaced
5473 , readOutputQBounded
55− , sendWorker
74+ , readOutputQPaced
75+ , dispatchWorkerPaced
76+ , sendFirstWorker
5677 , delThread
78+
79+ , toStreamVar
5780 )
5881where
5982
6083import Control.Concurrent
6184 (ThreadId, myThreadId, threadDelay, getNumCapabilities)
6285import Control.Concurrent.MVar
63− (MVar, newEmptyMVar, tryPutMVar, takeMVar)
64−import Control.Exception (SomeException(..), catch, mask)
86+ (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar)
87+import Control.Exception (SomeException(..), catch, mask, assert)
6588import Control.Monad (when)
6689import Control.Monad.Catch (MonadThrow)
6790import Control.Monad.IO.Class (MonadIO(..))
@@ -69,17 +92,20 @@ import Control.Monad.Trans.Control (MonadBaseControl, control)
6992import Data.Atomics
7093 (casIORef, readForCAS, peekTicket, atomicModifyIORefCAS_,
7194 writeBarrier, storeLoadBarrier)
72−import Data.Concurrent.Queue.MichaelScott
73− (LinkedQueue, pushL, tryPopR)
95+import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
7496import Data.Functor (void)
7597import Data.Heap (Heap, Entry(..))
98+import Data.Int (Int64)
7699import Data.IORef
77− (IORef, modifyIORef, newIORef, readIORef, atomicModifyIORef)
100+ (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
78101import Data.Maybe (fromJust)
79102import Data.Set (Set)
103+import Data.Word (Word64)
80104import GHC.Conc (ThreadId(..))
81105import GHC.Exts
82106import GHC.IO (IO(..))
107+import System.Clock
108+ (TimeSpec, Clock(Monotonic), getTime, toNanoSecs, fromNanoSecs)
83109
84110import qualified Data.Heap as H
85111import qualified Data.Set as S
@@ -94,10 +120,33 @@ import Control.Concurrent.MVar (tryTakeMVar)
94120import Control.Exception
95121 (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
96122 BlockedIndefinitelyOnSTM(..))
97−import Data.IORef (writeIORef)
98123import System.IO (hPutStrLn, stderr)
99124#endif
100125
126+newtype NanoSecs = NanoSecs Word64
127+ deriving ( Eq
128+ , Read
129+ , Show
130+ , Enum
131+ , Bounded
132+ , Num
133+ , Real
134+ , Integral
135+ , Ord
136+ )
137+
138+newtype Count = Count Word64
139+ deriving ( Eq
140+ , Read
141+ , Show
142+ , Enum
143+ , Bounded
144+ , Num
145+ , Real
146+ , Integral
147+ , Ord
148+ )
149+
101150------------------------------------------------------------------------------
102151-- Parent child thread communication type
103152------------------------------------------------------------------------------
@@ -148,61 +197,155 @@ data SVarStyle =
148197-- New work is enqueued either at the time of creation of the SVar or as a
149198-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
150199-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
151−--
152−-- XXX can we use forall t m.
153−data SVar t m a =
154− SVar {
155− -- Read only state
156− svarStyle :: SVarStyle
157−
158− -- Shared output queue (events, length)
159− , outputQueue :: IORef ([ChildEvent a], Int)
160− , maxYieldLimit :: Maybe (IORef Int)
161− , outputDoorBell :: MVar () -- signal the consumer about output
162− , readOutputQ :: m [ChildEvent a]
163− , postProcess :: m Bool
164−
165− -- Used only by bounded SVar types
166− , enqueue :: t m a -> IO ()
167− , isWorkDone :: IO Bool
168− , needDoorBell :: IORef Bool
169− , workLoop :: m ()
170−
171− -- Shared, thread tracking
172− , workerThreads :: IORef (Set ThreadId)
173− , workerCount :: IORef Int
174− , accountThread :: ThreadId -> m ()
200+
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)
210+ }
211+
212+-- Rate control.
213+data YieldRateInfo = YieldRateInfo
214+ { expectedYieldLatency :: NanoSecs
215+
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
219+
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
227+
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)
233+
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)
241+
242+ -- Latency as measured by workers, aggregated for the last period.
243+ , workerMeasuredLatency :: IORef NanoSecs
244+
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)
251+
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 ()
258+ }
259+
260+data Limit = Unlimited | Limited Word deriving Show
261+
262+data SVar t m a = SVar
263+ {
264+ -- Read only state
265+ svarStyle :: SVarStyle
266+
267+ -- Shared output queue (events, length)
268+ , outputQueue :: IORef ([ChildEvent a], Int)
269+ , outputDoorBell :: MVar () -- signal the consumer about output
270+ , readOutputQ :: m [ChildEvent a]
271+ , postProcess :: m Bool
272+
273+ -- Combined/aggregate parameters
274+ , maxWorkerLimit :: Limit
275+ , maxBufferLimit :: Limit
276+ , remainingYields :: Maybe (IORef Count)
277+ , yieldRateInfo :: Maybe YieldRateInfo
278+
279+ -- Used only by bounded SVar types
280+ , enqueue :: t m a -> IO ()
281+ , isWorkDone :: IO Bool
282+ , needDoorBell :: IORef Bool
283+ , workLoop :: WorkerInfo -> m ()
284+
285+ -- Shared, thread tracking
286+ , workerThreads :: IORef (Set ThreadId)
287+ , workerCount :: IORef Int
288+ , accountThread :: ThreadId -> m ()
289+
175290#ifdef DIAGNOSTICS
176− , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a))
177− , Int
178− )
179− -- Shared work queue (stream, seqNo)
180− , aheadWorkQueue :: IORef ([t m a], Int)
181− , totalDispatches :: IORef Int
182− , maxWorkers :: IORef Int
183− , maxOutQSize :: IORef Int
184− , maxHeapSize :: IORef Int
185− , maxWorkQSize :: IORef Int
291+ , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)
292+ -- Shared work queue (stream, seqNo)
293+ , aheadWorkQueue :: IORef ([t m a], Int)
294+ , totalDispatches :: IORef Int
295+ , maxWorkers :: IORef Int
296+ , maxOutQSize :: IORef Int
297+ , maxHeapSize :: IORef Int
298+ , maxWorkQSize :: IORef Int
186299#endif
187− }
300+ }
301+
302+-------------------------------------------------------------------------------
303+-- State for concurrency control
304+-------------------------------------------------------------------------------
188305
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.
309+--
310+-- XXX make all these Limited types and use phantom types to distinguish them
189311data State t m a = State
190− { streamVar :: Maybe (SVar t m a)
191− , yieldLimit :: Maybe Int
192− , threadsHigh :: Int
193− , bufferHigh :: Int
312+ { -- one shot configuration, automatically reset for each API call
313+ streamVar :: Maybe (SVar t m a)
314+ , _yieldLimit :: Maybe Count
315+
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
194322 }
195323
196−defaultMaxThreads, defaultMaxBuffer :: Int
197−defaultMaxThreads = 1500
198−defaultMaxBuffer = 1500
324+-------------------------------------------------------------------------------
325+-- State defaults and reset
326+-------------------------------------------------------------------------------
199327
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
334+
335+defaultMaxThreads, defaultMaxBuffer :: Limit
336+defaultMaxThreads = Limited magicMaxBuffer
337+defaultMaxBuffer = Limited magicMaxBuffer
338+
339+-- The fields prefixed by an _ are not to be accessed or updated directly but
340+-- via smart accessor APIs.
200341defState :: State t m a
201342defState = State
202343 { streamVar = Nothing
203− , yieldLimit = Nothing
204− , threadsHigh = defaultMaxThreads
205− , bufferHigh = defaultMaxBuffer
344+ , _yieldLimit = Nothing
345+ , _threadsHigh = defaultMaxThreads
346+ , _bufferHigh = defaultMaxBuffer
347+ , _maxStreamRate = Nothing
348+ , _streamLatency = Nothing
206349 }
207350
208351-- XXX if perf gets affected we can have all the Nothing params in a single
@@ -215,9 +358,84 @@ defState = State
215358rstState :: State t m a -> State t m b
216359rstState st = st
217360 { streamVar = Nothing
218− , yieldLimit = Nothing
361+ , _yieldLimit = Nothing
219362 }
220363
364+-------------------------------------------------------------------------------
365+-- Smart get/set routines for State
366+-------------------------------------------------------------------------------
367+
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 =
371+ st { _yieldLimit =
372+ case lim of
373+ Nothing -> Nothing
374+ Just n ->
375+ if n <= 0
376+ then Just 0
377+ else Just (fromIntegral n)
378+ }
379+
380+getYieldLimit :: State t m a -> Maybe Count
381+getYieldLimit = _yieldLimit
382+
383+setMaxThreads :: Int -> State t m a -> State t m a
384+setMaxThreads n st =
385+ st { _threadsHigh =
386+ if n < 0
387+ then Unlimited
388+ else if n == 0
389+ then defaultMaxThreads
390+ else Limited (fromIntegral n)
391+ }
392+
393+getMaxThreads :: State t m a -> Limit
394+getMaxThreads = _threadsHigh
395+
396+setMaxBuffer :: Int -> State t m a -> State t m a
397+setMaxBuffer n st =
398+ st { _bufferHigh =
399+ if n < 0
400+ then Unlimited
401+ else if n == 0
402+ then defaultMaxBuffer
403+ else Limited (fromIntegral n)
404+ }
405+
406+getMaxBuffer :: State t m a -> Limit
407+getMaxBuffer = _bufferHigh
408+
409+setMaxStreamRate :: Double -> State t m a -> State t m a
410+setMaxStreamRate n st =
411+ st { _maxStreamRate =
412+ if n < 0
413+ then Nothing
414+ else if n == 0
415+ then Nothing
416+ else Just n
417+ }
418+
419+getMaxStreamRate :: State t m a -> Maybe Double
420+getMaxStreamRate = _maxStreamRate
421+
422+setStreamLatency :: Int -> State t m a -> State t m a
423+setStreamLatency n st =
424+ st { _streamLatency =
425+ if n < 0
426+ then Nothing
427+ else if n == 0
428+ then Nothing
429+ else Just (fromIntegral n)
430+ }
431+
432+getStreamLatency :: State t m a -> Maybe NanoSecs
433+getStreamLatency = _streamLatency
434+
435+-------------------------------------------------------------------------------
436+-- Dumping the SVar for debug/diag
437+-------------------------------------------------------------------------------
438+
221439#ifdef DIAGNOSTICS
222440{-# NOINLINE dumpSVar #-}
223441dumpSVar :: SVar t m a -> IO String
@@ -287,6 +505,10 @@ withDBGMVar :: SVar t m a -> String -> IO () -> IO ()
287505withDBGMVar _ _ action = action
288506#endif
289507
508+-------------------------------------------------------------------------------
509+-- CAS
510+-------------------------------------------------------------------------------
511+
290512-- Slightly faster version of CAS. Gained some improvement by avoiding the use
291513-- of "evaluate" because we know we do not have exceptions in fn.
292514{-# INLINE atomicModifyIORefCAS #-}
@@ -346,8 +568,10 @@ doFork action exHandler =
346568
347569-- | This function is used by the producer threads to queue output for the
348570-- consumer thread to consume. Returns whether the queue has more space.
349−send :: Int -> SVar t m a -> ChildEvent a -> IO Bool
350−send maxOutputQLen sv msg = do
571+send :: SVar t m a -> ChildEvent a -> IO Bool
572+send sv msg = do
573+ -- XXX can the access to outputQueue and maxBufferLimit be made faster
574+ -- somehow?
351575 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
352576 ((msg : es, n + 1), n)
353577 when (len <= 0) $ do
@@ -361,22 +585,116 @@ send maxOutputQLen sv msg = do
361585 -- The important point is that the consumer is guaranteed to receive a
362586 -- doorbell if something was added to the queue after it empties it.
363587 void $ tryPutMVar (outputDoorBell sv) ()
364− return (len < maxOutputQLen || maxOutputQLen < 0)
365588
366−{-# NOINLINE sendYield #-}
367−sendYield :: Int -> SVar t m a -> ChildEvent a -> IO Bool
368−sendYield maxOutputQLen sv msg = do
369− ylimit <- case maxYieldLimit sv of
370− Nothing -> return True
371− Just ref -> atomicModifyIORefCAS ref $ \x -> (x - 1, x > 1)
372− r <- send maxOutputQLen sv msg
373− return $ r && ylimit
589+ -- XXX we should reserve the buffer when we pick up the work from the
590+ -- queue, instead of checking it here when it is too late.
591+ let limit = maxBufferLimit sv
592+ case limit of
593+ Unlimited -> return True
594+ Limited lim -> do
595+ active <- readIORef (workerCount sv)
596+ return $ len < ((fromIntegral lim) - active)
597+
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), ())
611+
612+updateYieldCount :: WorkerInfo -> IO Count
613+updateYieldCount winfo = do
614+ cnt <- readIORef (workerYieldCount winfo)
615+ let cnt1 = cnt + 1
616+ writeIORef (workerYieldCount winfo) cnt1
617+ return cnt1
618+
619+isBeyondMaxYield :: Count -> WorkerInfo -> Bool
620+isBeyondMaxYield cnt winfo =
621+ let ymax = workerYieldMax winfo
622+ in ymax /= 0 && cnt >= ymax
623+
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
630+ -> YieldRateInfo
631+ -> WorkerInfo
632+ -> Count
633+ -> IO Bool
634+checkRatePeriodic sv yinfo winfo ycnt = do
635+ i <- readIORef (workerLatencyPeriod yinfo)
636+ if (i /= 0 && (ycnt `mod` i) == 0)
637+ then do
638+ workerUpdateLatency yinfo winfo
639+ isBeyondMaxRate sv yinfo
640+ else return False
374641
375−{-# NOINLINE sendStop #-}
376−sendStop :: SVar t m a -> IO ()
377−sendStop sv = do
378− liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
379− myThreadId >>= \tid -> void $ send (-1) sv (ChildStop tid Nothing)
642+-- CAUTION! this also updates the yield count and therefore should be called
643+-- only when we are actually yielding an element.
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)
650+
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
658+ Just ref -> do
659+ r <- atomicModifyIORefCAS ref $ \x ->
660+ (if x >= 1 then x - 1 else 0, x)
661+ return $ r >= 1
662+
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 ()
670+ Just ref -> atomicModifyIORefCAS_ ref (+ 1)
671+
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
678+ r <- send sv msg
679+ rateLimitOk <-
680+ case yieldRateInfo sv of
681+ Nothing -> return True
682+ Just yinfo -> workerRateControl sv yinfo winfo
683+ return $ r && rateLimitOk
684+
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 ()
691+ Just info -> do
692+ i <- readIORef (workerLatencyPeriod info)
693+ when (i /= 0) $ workerUpdateLatency info winfo
694+ when (n == 1) $ do
695+ t <- getTime Monotonic
696+ writeIORef (workerStopTimeStamp info) t
697+ myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
380698
381699-------------------------------------------------------------------------------
382700-- Async
@@ -401,25 +719,6 @@ enqueueLIFO sv q m = do
401719 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
402720 void $ tryPutMVar (outputDoorBell sv) ()
403721
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
409−
410− where
411−
412− sv = fromJust $ streamVar st
413− run = do
414− work <- dequeue
415− case work of
416− Nothing -> liftIO $ sendStop sv
417− Just m -> f st q m run
418−
419− dequeue = liftIO $ atomicModifyIORefCAS q $ \case
420− [] -> ([], Nothing)
421− x : xs -> (xs, Just x)
422−
423722-------------------------------------------------------------------------------
424723-- WAsync
425724-------------------------------------------------------------------------------
@@ -443,21 +742,6 @@ enqueueFIFO sv q m = do
443742 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
444743 void $ tryPutMVar (outputDoorBell sv) ()
445744
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
451−
452− where
453−
454− sv = fromJust $ streamVar st
455− run = do
456− work <- liftIO $ tryPopR q
457− case work of
458− Nothing -> liftIO $ sendStop sv
459− Just m -> f st q m run
460−
461745-------------------------------------------------------------------------------
462746-- Ahead
463747-------------------------------------------------------------------------------
@@ -510,10 +794,18 @@ workLoopFIFO f st q = run
510794-- that they do not hog the resources and hinder the progress of the threads in
511795-- front of them.
512796
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
514798-- each left associative expression. The queue is used only for right
515799-- 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.
801+--
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.
806+--
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.
517809{-# INLINE enqueueAhead #-}
518810enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
519811enqueueAhead sv q m = do
@@ -571,7 +863,7 @@ dequeueFromHeap
571863 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
572864 -> IO (Maybe (Entry Int (AheadHeapEntry t m a)))
573865dequeueFromHeap hpRef = do
574− atomicModifyIORefCAS hpRef $ \hp@(h, snum) -> do
866+ atomicModifyIORef hpRef $ \hp@(h, snum) -> do
575867 let r = H.uncons h
576868 case r of
577869 Nothing -> (hp, Nothing)
@@ -637,7 +929,7 @@ allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
637929handleChildException :: SVar t m a -> SomeException -> IO ()
638930handleChildException sv e = do
639931 tid <- myThreadId
640− void $ send (-1) sv (ChildStop tid (Just e))
932+ void $ send sv (ChildStop tid (Just e))
641933
642934#ifdef DIAGNOSTICS
643935recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
@@ -649,13 +941,24 @@ recordMaxWorkers sv = liftIO $ do
649941#endif
650942
651943{-# NOINLINE pushWorker #-}
652−pushWorker :: MonadAsync m => SVar t m a -> m ()
653−pushWorker sv = do
944+pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
945+pushWorker yieldMax sv = do
654946 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
655947#ifdef DIAGNOSTICS
656948 recordMaxWorkers sv
657949#endif
658− doFork (workLoop sv) (handleChildException sv) >>= addThread sv
950+ -- XXX we can make this allocation conditional, it might matter when
951+ -- significant number of workers are being sent.
952+ winfo <- do
953+ cntRef <- liftIO $ newIORef 0
954+ t <- liftIO $ getTime Monotonic
955+ lat <- liftIO $ newIORef (0, t)
956+ return $ WorkerInfo
957+ { workerYieldMax = yieldMax
958+ , workerYieldCount = cntRef
959+ , workerLatencyStart = lat
960+ }
961+ doFork (workLoop sv winfo) (handleChildException sv) >>= addThread sv
659962
660963-- XXX we can push the workerCount modification in accountThread and use the
661964-- same pushWorker for Parallel case as well.
@@ -676,49 +979,313 @@ pushWorkerPar sv wloop = do
676979#endif
677980 doFork wloop (handleChildException sv) >>= modifyThread sv
678981
679−dispatchWorker :: MonadAsync m => Int -> SVar t m a -> m ()
680−dispatchWorker maxWorkerLimit sv = do
982+-- Returns:
983+-- True: can dispatch more
984+-- False: cannot dispatch any more
985+dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
986+dispatchWorker yieldCount 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.
681990 done <- liftIO $ isWorkDone sv
682− when (not done) $ do
991+ if (not done)
992+ then do
683993 -- Note that the worker count is only decremented during event
684994 -- processing in fromStreamVar and therefore it is safe to read and
685995 -- use it without a lock.
686− cnt <- liftIO $ readIORef $ workerCount sv
996+ active <- liftIO $ readIORef $ workerCount sv
687997 -- Note that we may deadlock if the previous workers (tasks in the
688998 -- stream) wait/depend on the future workers (tasks in the stream)
689999 -- executing. In that case we should either configure the maxWorker
6901000 -- count to higher or use parallel style instead of ahead or async
6911001 -- style.
692− limit <- case maxYieldLimit sv of
693− Nothing -> return maxWorkerLimit
1002+ limit <- case remainingYields sv of
1003+ Nothing -> return workerLimit
6941004 Just x -> do
695− lim <- liftIO $ readIORef x
1005+ n <- liftIO $ readIORef x
6961006 return $
697− if maxWorkerLimit > 0
698− then min maxWorkerLimit lim
699− else lim
700− when (cnt < limit || limit < 0) $ pushWorker sv
1007+ case workerLimit of
1008+ Unlimited -> Limited (fromIntegral n)
1009+ Limited lim -> Limited $ min lim (fromIntegral n)
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
1015+ in case limit of
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
1023+ _ -> return False
1024+ else return False
7011025
702−{-# NOINLINE sendWorkerWait #-}
703−sendWorkerWait :: MonadAsync m => Int -> SVar t m a -> m ()
704−sendWorkerWait maxWorkerLimit sv = do
705− -- Note that we are guaranteed to have at least one outstanding worker when
706− -- we enter this function. So if we sleep we are guaranteed to be woken up
707− -- by a outputDoorBell, when the worker exits.
1026+-- | This is a magic number and it is overloaded, and used at several places to
1027+-- achieve batching:
1028+--
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)
1036+
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
1044+-- worker.
1045+coverUpLatency :: NanoSecs
1046+coverUpLatency = 1000000
1047+
1048+nanoToMicroSecs :: NanoSecs -> Int
1049+nanoToMicroSecs s = (fromIntegral s) `div` 1000
1050+
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.
1056+
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
1061+ -- on that.
1062+ --
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.
1065+ --
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
1069+ -- maxLatency.
1070+ --
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
1074+ -- workers.
1075+ d = fromIntegral duration
1076+ e = fromIntegral expectedLat
1077+ n = max 1 (fromIntegral coverUpLatency / fromIntegral wLatency)
1078+ :: Double
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)
1085+
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
1094+
1095+ (count, time) <- readIORef cur
1096+ (colCount, colTime) <- readIORef col
1097+ (lcount, ltime) <- readIORef longTerm
1098+ prev <- readIORef measured
1099+
1100+ let pendingCount = colCount + count
1101+ pendingTime = colTime + time
1102+ -- Take the average of previous and the latest
1103+ new =
1104+ if pendingCount > 0
1105+ then ((pendingTime `div` (fromIntegral pendingCount)) + prev)
1106+ `div` 2
1107+ else prev
1108+ return (lcount + pendingCount, ltime, new)
1109+
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
1119+ if netWorkers <= 0
1120+ then do
1121+ let sleepTime = ((fromIntegral count) * expectedLat) - duration
1122+ if sleepTime >= minThreadDelay
1123+ then return True
1124+ else return False
1125+ else if cnt > netWorkers
1126+ then return True
1127+ else return False
1128+
1129+useCollectedBatch :: YieldRateInfo
1130+ -> NanoSecs
1131+ -> IORef (Count, NanoSecs)
1132+ -> IORef NanoSecs
1133+ -> IO ()
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)
1138+
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
1144+
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
1156+
1157+ (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
1158+ (colCount, colTime) <- readIORef col
1159+ (lcount, ltime) <- readIORef longTerm
1160+ prev <- readIORef measured
1161+
1162+ let pendingCount = colCount + count
1163+ pendingTime = colTime + time
1164+
1165+ lcount' = lcount + pendingCount
1166+ notUpdated = (lcount', ltime, prev)
1167+
1168+ if (pendingCount > 0)
1169+ then do
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))
1176+ || (prev == 0)
1177+ then do
1178+ useCollectedBatch yinfo new col measured
1179+ modifyIORef longTerm $ \(_, t) -> (lcount', t)
1180+ return (lcount', ltime, new)
1181+ else do
1182+ writeIORef col (pendingCount, pendingTime)
1183+ return notUpdated
1184+ else return notUpdated
1185+
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)
1192+ return ()
1193+
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.
1198+--
1199+-- Returns:
1200+-- True: can dispatch more
1201+-- False: full, no more dispatches
1202+dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1203+dispatchWorkerPaced sv = do
1204+ let workerLimit = maxWorkerLimit sv
1205+ let yinfo = fromJust $ yieldRateInfo sv
1206+ (count, tstamp, wLatency) <- liftIO $ collectLatency yinfo
1207+ -- XXX use workerBootstrapLatency
1208+ if wLatency == 0
1209+ -- Need to measure the latency with a single worker before we can perform
1210+ -- any computation.
1211+ then return False
1212+ else do
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
1218+
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.
1223+ if netWorkers <= 0
1224+ then do
1225+ let sleepTime = ((fromIntegral count) * expectedLat) - duration
1226+
1227+ when (sleepTime >= minThreadDelay) $ do
1228+ liftIO $ threadDelay $ nanoToMicroSecs sleepTime
1229+
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
1234+ return False
1235+ else do
1236+ -- XXX stagger the workers over a period
1237+ cnt <- liftIO $ readIORef $ workerCount sv
1238+ if (cnt < netWorkers)
1239+ then dispatchWorker 0 sv
1240+ else return False
1241+
1242+ where
7081243
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
1250+ then
1251+ if idleTime == 0
1252+ then maxYieldsPerWorker
1253+ else
1254+ let n = (minThreadDelay + idleTime) `div` idleTime
1255+ in min (fromIntegral n) maxYieldsPerWorker
1256+ else 1
1257+
1258+sendWorkerDelayPaced :: SVar t m a -> IO ()
1259+sendWorkerDelayPaced _ = return ()
1260+
1261+sendWorkerDelay :: SVar t m a -> IO ()
1262+sendWorkerDelay sv = do
7091263 -- XXX we need a better way to handle this than hardcoded delays. The
7101264 -- delays may be different for different systems.
711− ncpu <- liftIO $ getNumCapabilities
1265+ ncpu <- getNumCapabilities
7121266 if ncpu <= 1
7131267 then
7141268 if (svarStyle sv == AheadVar)
715− then liftIO $ threadDelay 100
716− else liftIO $ threadDelay 25
1269+ then threadDelay 100
1270+ else threadDelay 25
7171271 else
7181272 if (svarStyle sv == AheadVar)
719− then liftIO $ threadDelay 100
720− else liftIO $ threadDelay 10
1273+ then threadDelay 100
1274+ else threadDelay 10
7211275
1276+{-# NOINLINE sendWorkerWait #-}
1277+sendWorkerWait
1278+ :: MonadAsync m
1279+ => (SVar t m a -> IO ())
1280+ -> (SVar t m a -> m Bool)
1281+ -> SVar t m a
1282+ -> m ()
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.
1287+
1288+ liftIO $ delay sv
7221289 (_, n) <- liftIO $ readIORef (outputQueue sv)
7231290 when (n <= 0) $ do
7241291 -- The queue may be empty temporarily if the worker has dequeued the
@@ -753,7 +1320,7 @@ sendWorkerWait maxWorkerLimit sv = do
7531320
7541321 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
7551322 liftIO $ storeLoadBarrier
756− dispatchWorker maxWorkerLimit sv
1323+ canDoMore <- dispatch sv
7571324
7581325 -- XXX test for the case when we miss sending a worker when the worker
7591326 -- count is more than 1500.
@@ -762,14 +1329,13 @@ sendWorkerWait maxWorkerLimit sv = do
7621329 -- least one outstanding worker. Otherwise we could be sleeping
7631330 -- forever.
7641331
765− done <- liftIO $ isWorkDone sv
766− if done
767− then do
1332+ if canDoMore
1333+ then sendWorkerWait delay dispatch sv
1334+ else do
7681335 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
7691336 $ takeMVar (outputDoorBell sv)
7701337 (_, len) <- liftIO $ readIORef (outputQueue sv)
771− when (len <= 0) $ sendWorkerWait maxWorkerLimit sv
772− else sendWorkerWait maxWorkerLimit sv
1338+ when (len <= 0) $ sendWorkerWait delay dispatch sv
7731339
7741340{-# INLINE readOutputQRaw #-}
7751341readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
@@ -781,8 +1347,8 @@ readOutputQRaw sv = do
7811347#endif
7821348 return (list, len)
7831349
784−readOutputQBounded :: MonadAsync m => Int -> SVar t m a -> m [ChildEvent a]
785−readOutputQBounded n sv = do
1350+readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1351+readOutputQBounded sv = do
7861352 (list, len) <- liftIO $ readOutputQRaw sv
7871353 -- When there is no output seen we dispatch more workers to help
7881354 -- out if there is work pending in the work queue.
@@ -802,32 +1368,96 @@ readOutputQBounded n sv = do
8021368 cnt <- liftIO $ readIORef $ workerCount sv
8031369 when (cnt <= 0) $ do
8041370 done <- liftIO $ isWorkDone sv
805− when (not done) $ pushWorker sv
1371+ when (not done) $ pushWorker 0 sv
8061372
8071373 {-# INLINE blockingRead #-}
8081374 blockingRead = do
809− sendWorkerWait n sv
1375+ sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
1376+ liftIO $ (readOutputQRaw sv >>= return . fst)
1377+
1378+readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
1379+readOutputQPaced sv = do
1380+ (list, len) <- liftIO $ readOutputQRaw sv
1381+ if len <= 0
1382+ then blockingRead
1383+ else do
1384+ -- XXX send a worker proactively, if needed, even before we start
1385+ -- processing the output.
1386+ void $ dispatchWorkerPaced sv
1387+ return list
1388+
1389+ where
1390+
1391+ {-# INLINE blockingRead #-}
1392+ blockingRead = do
1393+ sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
8101394 liftIO $ (readOutputQRaw sv >>= return . fst)
8111395
8121396postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
8131397postProcessBounded sv = do
8141398 workersDone <- allThreadsDone sv
815− -- There may still be work pending even if there are no workers
816− -- pending because all the workers may return if the
817− -- outputQueue becomes full. In that case send off a worker to
818− -- kickstart the work again.
1399+ -- There may still be work pending even if there are no workers pending
1400+ -- because all the workers may return if the outputQueue becomes full. In
1401+ -- that case send off a worker to kickstart the work again.
1402+ if workersDone
1403+ then do
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
1408+ return r
1409+ else return False
1410+
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?
8191417 if workersDone
8201418 then do
8211419 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
8231424 return r
8241425 else return False
8251426
1427+getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
1428+getYieldRateInfo st = do
1429+ case getMaxStreamRate st of
1430+ Just rate -> do
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 ()
1440+
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
1451+ }
1452+ Nothing -> return Nothing
1453+
8261454getAheadSVar :: MonadAsync m
8271455 => State t m a
828− -> ( State t m a
829− -> IORef ([t m a], Int)
1456+ -> ( IORef ([t m a], Int)
8301457 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
1458+ -> State t m a
1459+ -> SVar t m a
1460+ -> WorkerInfo
8311461 -> m ())
8321462 -> IO (SVar t m a)
8331463getAheadSVar st f = do
@@ -838,53 +1468,71 @@ getAheadSVar st f = do
8381468 wfw <- newIORef False
8391469 running <- newIORef S.empty
8401470 q <- newIORef ([], -1)
841− yl <- case yieldLimit st of
1471+ yl <- case getYieldLimit st of
8421472 Nothing -> return Nothing
8431473 Just x -> Just <$> newIORef x
1474+ rateInfo <- getYieldRateInfo st
8441475
8451476#ifdef DIAGNOSTICS
846− disp <- newIORef 0
1477+ disp <- newIORef 0
8471478 maxWrk <- newIORef 0
8481479 maxOq <- newIORef 0
8491480 maxHs <- newIORef 0
8501481 maxWq <- newIORef 0
8511482#endif
852− let sv =
853− SVar { outputQueue = outQ
854− , maxYieldLimit = yl
855− , outputDoorBell = outQMv
856− , readOutputQ = readOutputQBounded (threadsHigh st) sv
857− , postProcess = postProcessBounded sv
858− , workerThreads = running
859− -- , workLoop = workLoopAhead sv q outH
860− , workLoop = f st{streamVar = Just sv} q outH
861− , enqueue = enqueueAhead sv q
862− , isWorkDone = isWorkDoneAhead q outH
863− , needDoorBell = wfw
864− , svarStyle = AheadVar
865− , workerCount = active
866− , accountThread = delThread sv
1483+
1484+ let getSVar sv readOutput postProc = SVar
1485+ { outputQueue = outQ
1486+ , remainingYields = yl
1487+ , maxBufferLimit = getMaxBuffer st
1488+ , maxWorkerLimit = getMaxThreads st
1489+ , yieldRateInfo = rateInfo
1490+ , outputDoorBell = outQMv
1491+ , readOutputQ = readOutput sv
1492+ , postProcess = postProc sv
1493+ , workerThreads = running
1494+ , workLoop = f q outH st{streamVar = Just sv} sv
1495+ , enqueue = enqueueAhead sv q
1496+ , isWorkDone = isWorkDoneAhead sv q outH
1497+ , needDoorBell = wfw
1498+ , svarStyle = AheadVar
1499+ , workerCount = active
1500+ , accountThread = delThread sv
8671501#ifdef DIAGNOSTICS
868− , aheadWorkQueue = q
869− , outputHeap = outH
870− , totalDispatches = disp
871− , maxWorkers = maxWrk
872− , maxOutQSize = maxOq
873− , maxHeapSize = maxHs
874− , maxWorkQSize = maxWq
1502+ , aheadWorkQueue = q
1503+ , outputHeap = outH
1504+ , totalDispatches = disp
1505+ , maxWorkers = maxWrk
1506+ , maxOutQSize = maxOq
1507+ , maxHeapSize = maxHs
1508+ , maxWorkQSize = maxWq
8751509#endif
876− }
1510+ }
1511+
1512+ let sv =
1513+ case getMaxStreamRate st of
1514+ Nothing -> getSVar sv readOutputQBounded postProcessBounded
1515+ Just _ -> getSVar sv readOutputQPaced postProcessPaced
8771516 in return sv
8781517
8791518 where
8801519
8811520 {-# INLINE isWorkDoneAhead #-}
882− isWorkDoneAhead q ref = do
1521+ isWorkDoneAhead sv q ref = do
8831522 heapDone <- do
8841523 (hp, _) <- readIORef ref
8851524 return (H.size hp <= 0)
8861525 queueDone <- checkEmpty q
887− return $ queueDone && heapDone
1526+ yieldsDone <-
1527+ case remainingYields sv of
1528+ Just yref -> do
1529+ n <- readIORef yref
1530+ return (n <= 0)
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.
1535+ return $ yieldsDone || (queueDone && heapDone)
8881536
8891537 checkEmpty q = do
8901538 (xs, _) <- readIORef q
@@ -905,7 +1553,10 @@ getParallelSVar = do
9051553#endif
9061554 let sv =
9071555 SVar { outputQueue = outQ
908− , maxYieldLimit = Nothing
1556+ , remainingYields = Nothing
1557+ , maxBufferLimit = Unlimited
1558+ , maxWorkerLimit = undefined
1559+ , yieldRateInfo = Nothing
9091560 , outputDoorBell = outQMv
9101561 , readOutputQ = readOutputQPar sv
9111562 , postProcess = allThreadsDone sv
@@ -935,27 +1586,31 @@ getParallelSVar = do
9351586 withDBGMVar sv "readOutputQPar: doorbell" $ takeMVar (outputDoorBell sv)
9361587 readOutputQRaw sv >>= return . fst
9371588
938−sendWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
939−sendWorker sv m = do
1589+sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
1590+sendFirstWorker sv m = do
9401591 -- Note: We must have all the work on the queue before sending the
9411592 -- pushworker, otherwise the pushworker may exit before we even get a
9421593 -- chance to push.
9431594 liftIO $ enqueue sv m
944− pushWorker sv
1595+ case yieldRateInfo sv of
1596+ Nothing -> pushWorker 0 sv
1597+ Just _ -> pushWorker 1 sv
9451598 return sv
9461599
9471600{-# INLINABLE newAheadVar #-}
9481601newAheadVar :: MonadAsync m
9491602 => State t m a
9501603 -> t m a
951− -> ( State t m a
952− -> IORef ([t m a], Int)
1604+ -> ( IORef ([t m a], Int)
9531605 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)
1606+ -> State t m a
1607+ -> SVar t m a
1608+ -> WorkerInfo
9541609 -> m ())
9551610 -> m (SVar t m a)
9561611newAheadVar st m wloop = do
9571612 sv <- liftIO $ getAheadSVar st wloop
958− sendWorker sv m
1613+ sendFirstWorker sv m
9591614
9601615{-# INLINABLE newParallelVar #-}
9611616newParallelVar :: MonadAsync m => m (SVar t m a)
@@ -971,4 +1626,9 @@ toStreamVar sv m = do
9711626 -- XXX This is safe only when called from the consumer thread or when no
9721627 -- consumer is present. There may be a race if we are not running in the
9731628 -- consumer thread.
974− when done $ pushWorker sv
1629+ -- XXX do this only if the work queue is not empty. The work may have been
1630+ -- carried out by existing workers.
1631+ when done $
1632+ case yieldRateInfo sv of
1633+ Nothing -> pushWorker 0 sv
1634+ Just _ -> pushWorker 1 sv