@@ -585,7 +585,8 @@ drainLatency _ss yinfo = do
585585 when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
586586#endif
587587 -- To avoid minor fluctuations update in batches
588− useCollectedBatch yinfo new col measured
588+ writeIORef col (0, 0)
589+ writeIORef measured new
589590#ifdef DIAGNOSTICS
590591 modifyIORef (avgWorkerLatency _ss) $
591592 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
@@ -1376,15 +1377,15 @@ estimateWorkers workerLimit svarYields svarElapsed wLatency targetLat range =
13761377 fromIntegral rateRecoveryTime
13771378 yieldsFreq = 1.0 / fromIntegral targetLat
13781379 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1379− adjustedLat = NanoSecs $ round $ 1.0 / totalYieldsFreq
1380− requiredLat = min (max adjustedLat (minLatency range))
1380+ requiredLat = NanoSecs $ round $ 1.0 / totalYieldsFreq
1381+ adjustedLat = min (max requiredLat (minLatency range))
13811382 (maxLatency range)
1382− in assert (requiredLat > 0) $
1383− if wLatency <= requiredLat
1383+ in assert (adjustedLat > 0) $
1384+ if wLatency <= adjustedLat
13841385 then PartialWorker deltaYields
13851386 else ManyWorkers ( fromIntegral
13861387 $ withLimit
1387− $ wLatency `div` requiredLat) deltaYields
1388+ $ wLatency `div` adjustedLat) deltaYields
13881389 else
13891390 let expectedDuration = (fromIntegral svarYields) * targetLat
13901391 sleepTime = expectedDuration - svarElapsed
@@ -1402,7 +1403,7 @@ estimateWorkers workerLimit svarYields svarElapsed wLatency targetLat range =
14021403
14031404-- | Get the worker latency without resetting workerPendingLatency
14041405-- Returns (total yield count, base time, measured latency)
1405−-- CAUTION! keep it in sync with collectLatency and useCollectedBatch
1406+-- CAUTION! keep it in sync with collectLatency
14061407getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
14071408getWorkerLatency yinfo = do
14081409 let cur = workerPendingLatency yinfo
@@ -1458,18 +1459,6 @@ updateWorkerPollingInterval yinfo latency = do
14581459
14591460 writeIORef periodRef (fromIntegral period)
14601461
1461−useCollectedBatch :: YieldRateInfo
1462− -> NanoSecs
1463− -> IORef (Count, NanoSecs)
1464− -> IORef NanoSecs
1465− -> IO ()
1466−useCollectedBatch yinfo latency colRef measRef = do
1467− updateWorkerPollingInterval yinfo latency
1468− writeIORef colRef (0, 0)
1469− -- XXX we should use some weight for the old latency as well in arriving at
1470− -- the new latency, we have the counts, we can use that for some weighting.
1471− writeIORef measRef latency
1472−
14731462-- Returns a triple, (1) yield count since last collection, (2) the base time
14741463-- when we started counting, (3) average latency in the last measurement
14751464-- period. The former two are used for accurate measurement of the going rate
@@ -1514,7 +1503,9 @@ collectLatency _ss yinfo = do
15141503 in prev > 0 && (r > 2 || r < 0.5))
15151504 || (prev == 0)
15161505 then do
1517− useCollectedBatch yinfo new col measured
1506+ updateWorkerPollingInterval yinfo (max new prev)
1507+ writeIORef col (0, 0)
1508+ writeIORef measured ((prev + new) `div` 2)
15181509#ifdef DIAGNOSTICS
15191510 modifyIORef (avgWorkerLatency _ss) $
15201511 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
@@ -1526,12 +1517,14 @@ collectLatency _ss yinfo = do
15261517 return $ tripleWith prev
15271518 else return $ tripleWith prev
15281519
1520+-- This takes away the time when the producer is idle because the consumer
1521+-- not consuming or consuming slowly.
15291522adjustLatencies :: YieldRateInfo -> IO ()
15301523adjustLatencies yinfo = do
15311524 let longTerm = svarAllTimeLatency yinfo
15321525 t0 <- readIORef (workerStopTimeStamp yinfo)
15331526 t1 <- getTime Monotonic
1534− modifyIORef longTerm $ \(c, t) -> (c, t + t1 - t0)
1527+ modifyIORef longTerm $ \(cnt, time) -> (cnt, time + t1 - t0)
15351528 return ()
15361529
15371530-- XXX in case of ahead style stream we need to take the heap size into account
@@ -1835,27 +1828,29 @@ postProcessPaced sv = do
18351828
18361829getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
18371830getYieldRateInfo st = do
1838− let toLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
1839− toMaxLatency r = if r <= 0 then maxBound else r
1831+ let -- convert rate in Hertz to latency in Nanoseconds
1832+ rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
1833+ -- handle overflowed latency value
1834+ toMaxLatency l = if l <= 0 then maxBound else l
18401835 case getStreamRate st of
18411836 Just (AvgRate rate) ->
1842− let l = toLatency rate
1837+ let l = rateToLatency rate
18431838 mx = toMaxLatency (l * 2)
18441839 in mkYieldRateInfo l (LatencyRange (l `div` 2) mx)
18451840 Just (MinRate rate) ->
1846− let l = toLatency rate
1841+ let l = rateToLatency rate
18471842 in mkYieldRateInfo l (LatencyRange (l `div` 2) l)
18481843 Just (MaxRate rate) ->
1849− let l = toLatency rate
1844+ let l = rateToLatency rate
18501845 mx = toMaxLatency (l * 2)
18511846 in mkYieldRateInfo l (LatencyRange l mx)
18521847 Just (ConstRate rate) ->
1853− let l = toLatency rate
1848+ let l = rateToLatency rate
18541849 in mkYieldRateInfo l (LatencyRange l l)
18551850 Just (Rate rate minr maxr) ->
1856− let l = toLatency rate
1857− minl = toLatency minr
1858− maxl = toLatency maxr
1851+ let l = rateToLatency rate
1852+ minl = rateToLatency maxr
1853+ maxl = rateToLatency minr
18591854 in mkYieldRateInfo l (LatencyRange minl maxl)
18601855 Nothing -> return Nothing
18611856