@@ -235,16 +235,32 @@ data WorkerInfo = WorkerInfo
235235 , workerLatencyStart :: IORef (Count, TimeSpec)
236236 }
237237
238−-- | Specify the stream yield rate in yields per second (@Hertz@). We try to
239−-- keep the average rate at 'rateGoal', if the actual average rate goes above
240−-- or below the goal we try to recover it by increasing or decreasing the
241−-- instantaneous rate but keeping it within 'rateLow' and 'rateHigh' limits.
238+
239+-- | Specifies the stream yield rate in yields per second (@Hertz@).
240+-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
241+-- allow only as many yields as we have accumulated as per 'rateGoal' since the
242+-- start of time. If the consumer or the producer is slower or faster, the
243+-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
244+-- between the two by increasing or decreasing the pull rate from the producer.
245+-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
246+-- much as 'rateBuffer'.
247+--
248+-- 'rateLow' puts a bound on how low the instantaneous rate can go when
249+-- recovering the rate gap. In other words, it determines the maximum yield
250+-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
251+-- rate can go when recovering the rate gap. In other words, it determines the
252+-- minimum yield latency. We reduce the latency by increasing concurrency,
253+-- therefore we can say that it puts an upper bound on concurrency.
254+--
255+-- If the 'rateGoal' is 0 or negative the stream never yields a value.
256+-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
242257--
243258-- @since 0.5.0
244259data Rate = Rate
245− { rateLow :: Double -- ^ The lower rate limit
246− , rateGoal :: Double -- ^ The target rate we want to achieve
247− , rateHigh :: Double -- ^ The upper rate limit
260+ { rateLow :: Double -- ^ The lower rate limit
261+ , rateGoal :: Double -- ^ The target rate we want to achieve
262+ , rateHigh :: Double -- ^ The upper rate limit
263+ , rateBuffer :: Int -- ^ Maximum slack from the goal
248264 }
249265
250266data LatencyRange = LatencyRange
@@ -254,8 +270,10 @@ data LatencyRange = LatencyRange
254270
255271-- Rate control.
256272data YieldRateInfo = YieldRateInfo
257− { svarLatencyTarget :: NanoSecs
258− , svarLatencyRange :: LatencyRange
273+ { svarLatencyTarget :: NanoSecs
274+ , svarLatencyRange :: LatencyRange
275+ , svarRateBuffer :: Int
276+ , svarGainedLostYields :: IORef Count
259277
260278 -- Actual latency/througput as seen from the consumer side, we count the
261279 -- yields and the time it took to generates those yields. This is used to
@@ -585,22 +603,23 @@ dumpSVarStats sv ss style = do
585603 minLat <- readIORef $ minWorkerLatency ss
586604 maxLat <- readIORef $ maxWorkerLatency ss
587605 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
588− (svarCnt, svarLat) <- case yieldRateInfo sv of
589− Nothing -> return (0, 0)
606+ (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
607+ Nothing -> return (0, 0, 0)
590608 Just yinfo -> do
591609 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
592610 if cnt > 0
593611 then do
594612 t <- readIORef (svarStopTime ss)
613+ gl <- readIORef (svarGainedLostYields yinfo)
595614 case t of
596615 Nothing -> do
597616 now <- getTime Monotonic
598617 let interval = toNanoSecs (now - startTime)
599− return $ (cnt, interval `div` fromIntegral cnt)
618+ return $ (cnt, gl, interval `div` fromIntegral cnt)
600619 Just stopTime -> do
601620 let interval = toNanoSecs (stopTime - startTime)
602− return $ (cnt, interval `div` fromIntegral cnt)
603− else return (0, 0)
621+ return $ (cnt, gl, interval `div` fromIntegral cnt)
622+ else return (0, 0, 0)
604623
605624 return $ unlines
606625 [ "total dispatches = " ++ show dispatches
@@ -629,6 +648,9 @@ dumpSVarStats sv ss style = do
629648 ++ (if svarCnt > 0
630649 then "\nSVar yield count = " ++ show svarCnt
631650 else "")
651+ ++ (if svarGainLossCnt > 0
652+ then "\nSVar gain/loss yield count = " ++ show svarGainLossCnt
653+ else "")
632654 ]
633655
634656{-# NOINLINE dumpSVar #-}
@@ -1301,12 +1323,14 @@ data Work
13011323estimateWorkers
13021324 :: Limit
13031325 -> Count
1326+ -> Count
13041327 -> NanoSecs
13051328 -> NanoSecs
13061329 -> NanoSecs
13071330 -> LatencyRange
13081331 -> Work
1309−estimateWorkers workerLimit svarYields svarElapsed wLatency targetLat range =
1332+estimateWorkers workerLimit svarYields gainLossYields
1333+ svarElapsed wLatency targetLat range =
13101334 -- XXX we can have a maxEfficiency combinator as well which runs the
13111335 -- producer at the maximal efficiency i.e. the number of workers are chosen
13121336 -- such that the latency is minimum or within a range. Or we can call it
@@ -1339,7 +1363,8 @@ estimateWorkers workerLimit svarYields svarElapsed wLatency targetLat range =
13391363 -- When the worker latency is lower than required latency we begin with
13401364 -- a yield and then wait rather than first waiting and then yielding.
13411365 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1342− deltaYields = fromIntegral targetYields - svarYields
1366+ effectiveYields = svarYields + gainLossYields
1367+ deltaYields = fromIntegral targetYields - effectiveYields
13431368
13441369 -- We recover the deficit by running at a higher/lower rate for a
13451370 -- certain amount of time. To keep the effective rate in reasonable
@@ -1362,7 +1387,7 @@ estimateWorkers workerLimit svarYields svarElapsed wLatency targetLat range =
13621387 $ withLimit
13631388 $ wLatency `div` adjustedLat) deltaYields
13641389 else
1365− let expectedDuration = (fromIntegral svarYields) * targetLat
1390+ let expectedDuration = fromIntegral effectiveYields * targetLat
13661391 sleepTime = expectedDuration - svarElapsed
13671392 maxSleepTime = maxLatency range - wLatency
13681393 s = min sleepTime maxSleepTime
@@ -1407,8 +1432,9 @@ isBeyondMaxRate sv yinfo = do
14071432 now <- getTime Monotonic
14081433 let duration = fromInteger $ toNanoSecs $ now - tstamp
14091434 let targetLat = svarLatencyTarget yinfo
1410− let work = estimateWorkers (maxWorkerLimit sv) count duration wLatency
1411− targetLat (svarLatencyRange yinfo)
1435+ gainLoss <- readIORef (svarGainedLostYields yinfo)
1436+ let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1437+ wLatency targetLat (svarLatencyRange yinfo)
14121438 cnt <- readIORef $ workerCount sv
14131439 return $ case work of
14141440 -- XXX set the worker's maxYields or polling interval based on yields
@@ -1536,7 +1562,8 @@ dispatchWorkerPaced sv = do
15361562 let workerLimit = maxWorkerLimit sv
15371563 let targetLat = svarLatencyTarget yinfo
15381564 let range = svarLatencyRange yinfo
1539− let work = estimateWorkers workerLimit svarYields svarElapsed
1565+ gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1566+ let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
15401567 wLatency targetLat range
15411568
15421569 -- XXX we need to take yieldLimit into account here. If we are at the
@@ -1563,12 +1590,15 @@ dispatchWorkerPaced sv = do
15631590 return False
15641591 PartialWorker yields -> do
15651592 assert (yields > 0) (return ())
1593+ updateGainedLostYields yinfo yields
1594+
15661595 done <- allThreadsDone sv
15671596 when done $ void $ dispatchWorker yields sv
15681597 return False
15691598 ManyWorkers netWorkers yields -> do
15701599 assert (netWorkers >= 1) (return ())
15711600 assert (yields >= 0) (return ())
1601+ updateGainedLostYields yinfo yields
15721602
15731603 let periodRef = workerPollingInterval yinfo
15741604 ycnt = max 1 $ yields `div` fromIntegral netWorkers
@@ -1595,6 +1625,15 @@ dispatchWorkerPaced sv = do
15951625
15961626 where
15971627
1628+ updateGainedLostYields yinfo yields = do
1629+ let buf = fromIntegral $ svarRateBuffer yinfo
1630+ when (yields /= 0 && abs yields > buf) $ do
1631+ let delta =
1632+ if yields > 0
1633+ then yields - buf
1634+ else yields + buf
1635+ liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1636+
15981637 dispatchN n = do
15991638 if n == 0
16001639 then return True
@@ -1806,16 +1845,16 @@ getYieldRateInfo st = do
18061845 -- convert rate in Hertz to latency in Nanoseconds
18071846 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
18081847 case getStreamRate st of
1809− Just (Rate low goal high) ->
1848+ Just (Rate low goal high buf) ->
18101849 let l = rateToLatency goal
18111850 minl = rateToLatency high
18121851 maxl = rateToLatency low
1813− in mkYieldRateInfo l (LatencyRange minl maxl)
1852+ in mkYieldRateInfo l (LatencyRange minl maxl) buf
18141853 Nothing -> return Nothing
18151854
18161855 where
18171856
1818− mkYieldRateInfo latency latRange = do
1857+ mkYieldRateInfo latency latRange buf = do
18191858 measured <- newIORef 0
18201859 wcur <- newIORef (0,0)
18211860 wcol <- newIORef (0,0)
@@ -1823,10 +1862,13 @@ getYieldRateInfo st = do
18231862 wlong <- newIORef (0,now)
18241863 period <- newIORef 1
18251864 stopTime <- newIORef $ fromNanoSecs 0
1865+ gainLoss <- newIORef (Count 0)
18261866
18271867 return $ Just YieldRateInfo
18281868 { svarLatencyTarget = latency
18291869 , svarLatencyRange = latRange
1870+ , svarRateBuffer = buf
1871+ , svarGainedLostYields = gainLoss
18301872 , workerBootstrapLatency = getStreamLatency st
18311873 , workerPollingInterval = period
18321874 , workerMeasuredLatency = measured