@@ -585,7 +585,8 @@ drainLatency _ss yinfo = do
585 when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
585 when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
587 -- To avoid minor fluctuations update in batches
587 -- To avoid minor fluctuations update in batches
588− useCollectedBatch yinfo new col measured
588+ writeIORef col (0, 0)
589+ writeIORef measured new
589#ifdef DIAGNOSTICS
590#ifdef DIAGNOSTICS
590 modifyIORef (avgWorkerLatency _ss) $
591 modifyIORef (avgWorkerLatency _ss) $
591 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
592 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
@@ -1376,15 +1377,15 @@ estimateWorkers workerLimit svarYields svarElapsed wLatency targetLat range =
1376 fromIntegral rateRecoveryTime
1377 fromIntegral rateRecoveryTime
1377 yieldsFreq = 1.0 / fromIntegral targetLat
1378 yieldsFreq = 1.0 / fromIntegral targetLat
1378 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1379 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1379− adjustedLat = NanoSecs $ round $ 1.0 / totalYieldsFreq
1380+ requiredLat = NanoSecs $ round $ 1.0 / totalYieldsFreq
1380− requiredLat = min (max adjustedLat (minLatency range))
1381+ adjustedLat = min (max requiredLat (minLatency range))
1381 (maxLatency range)
1382 (maxLatency range)
1382− in assert (requiredLat > 0) $
1383+ in assert (adjustedLat > 0) $
1383− if wLatency <= requiredLat
1384+ if wLatency <= adjustedLat
1384 then PartialWorker deltaYields
1385 then PartialWorker deltaYields
1385 else ManyWorkers ( fromIntegral
1386 else ManyWorkers ( fromIntegral
1386 $ withLimit
1387 $ withLimit
1387− $ wLatency `div` requiredLat) deltaYields
1388+ $ wLatency `div` adjustedLat) deltaYields
1389 let expectedDuration = (fromIntegral svarYields) * targetLat
1390 let expectedDuration = (fromIntegral svarYields) * targetLat
1390 sleepTime = expectedDuration - svarElapsed
1391 sleepTime = expectedDuration - svarElapsed
@@ -1402,7 +1403,7 @@ estimateWorkers workerLimit svarYields svarElapsed wLatency targetLat range =
1403-- | Get the worker latency without resetting workerPendingLatency
1404-- | Get the worker latency without resetting workerPendingLatency
1404-- Returns (total yield count, base time, measured latency)
1405-- 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
1406getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1407getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1407getWorkerLatency yinfo = do
1408getWorkerLatency yinfo = do
1408 let cur = workerPendingLatency yinfo
1409 let cur = workerPendingLatency yinfo
@@ -1458,18 +1459,6 @@ updateWorkerPollingInterval yinfo latency = do
1459 writeIORef periodRef (fromIntegral period)
1460 writeIORef periodRef (fromIntegral period)
1461−useCollectedBatch :: YieldRateInfo
1463− -> IORef (Count, NanoSecs)
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
1473-- Returns a triple, (1) yield count since last collection, (2) the base time
1462-- Returns a triple, (1) yield count since last collection, (2) the base time
1474-- when we started counting, (3) average latency in the last measurement
1463-- when we started counting, (3) average latency in the last measurement
1475-- period. The former two are used for accurate measurement of the going rate
1464-- period. The former two are used for accurate measurement of the going rate
@@ -1514,7 +1503,9 @@ collectLatency _ss yinfo = do
1514 in prev > 0 && (r > 2 || r < 0.5))
1503 in prev > 0 && (r > 2 || r < 0.5))
1515 || (prev == 0)
1504 || (prev == 0)
1517− useCollectedBatch yinfo new col measured
1506+ updateWorkerPollingInterval yinfo (max new prev)
1507+ writeIORef col (0, 0)
1508+ writeIORef measured ((prev + new) `div` 2)
1518#ifdef DIAGNOSTICS
1509#ifdef DIAGNOSTICS
1519 modifyIORef (avgWorkerLatency _ss) $
1510 modifyIORef (avgWorkerLatency _ss) $
1520 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
1511 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
@@ -1526,12 +1517,14 @@ collectLatency _ss yinfo = do
1526 return $ tripleWith prev
1517 return $ tripleWith prev
1527 else return $ tripleWith prev
1518 else return $ tripleWith prev
1520+-- This takes away the time when the producer is idle because the consumer
1521+-- not consuming or consuming slowly.
1529adjustLatencies :: YieldRateInfo -> IO ()
1522adjustLatencies :: YieldRateInfo -> IO ()
1530adjustLatencies yinfo = do
1523adjustLatencies yinfo = do
1531 let longTerm = svarAllTimeLatency yinfo
1524 let longTerm = svarAllTimeLatency yinfo
1532 t0 <- readIORef (workerStopTimeStamp yinfo)
1525 t0 <- readIORef (workerStopTimeStamp yinfo)
1533 t1 <- getTime Monotonic
1526 t1 <- getTime Monotonic
1534− modifyIORef longTerm $ \(c, t) -> (c, t + t1 - t0)
1527+ modifyIORef longTerm $ \(cnt, time) -> (cnt, time + t1 - t0)
1535 return ()
1528 return ()
1537-- XXX in case of ahead style stream we need to take the heap size into account
1530-- XXX in case of ahead style stream we need to take the heap size into account
@@ -1835,27 +1828,29 @@ postProcessPaced sv = do
1836getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
1829getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
1837getYieldRateInfo st = do
1830getYieldRateInfo st = do
1831+ let -- convert rate in Hertz to latency in Nanoseconds
1838− let toLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
1832+ rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
1833+ -- handle overflowed latency value
1839− toMaxLatency r = if r <= 0 then maxBound else r
1834+ toMaxLatency l = if l <= 0 then maxBound else l
1840 case getStreamRate st of
1835 case getStreamRate st of
1841 Just (AvgRate rate) ->
1836 Just (AvgRate rate) ->
1842− let l = toLatency rate
1837+ let l = rateToLatency rate
1843 mx = toMaxLatency (l * 2)
1838 mx = toMaxLatency (l * 2)
1844 in mkYieldRateInfo l (LatencyRange (l `div` 2) mx)
1839 in mkYieldRateInfo l (LatencyRange (l `div` 2) mx)
1845 Just (MinRate rate) ->
1840 Just (MinRate rate) ->
1846− let l = toLatency rate
1841+ let l = rateToLatency rate
1847 in mkYieldRateInfo l (LatencyRange (l `div` 2) l)
1842 in mkYieldRateInfo l (LatencyRange (l `div` 2) l)
1848 Just (MaxRate rate) ->
1843 Just (MaxRate rate) ->
1849− let l = toLatency rate
1844+ let l = rateToLatency rate
1850 mx = toMaxLatency (l * 2)
1845 mx = toMaxLatency (l * 2)
1851 in mkYieldRateInfo l (LatencyRange l mx)
1846 in mkYieldRateInfo l (LatencyRange l mx)
1852 Just (ConstRate rate) ->
1847 Just (ConstRate rate) ->
1853− let l = toLatency rate
1848+ let l = rateToLatency rate
1854 in mkYieldRateInfo l (LatencyRange l l)
1849 in mkYieldRateInfo l (LatencyRange l l)
1855 Just (Rate rate minr maxr) ->
1850 Just (Rate rate minr maxr) ->
1856− let l = toLatency rate
1851+ let l = rateToLatency rate
1857− minl = toLatency minr
1852+ minl = rateToLatency maxr
1858− maxl = toLatency maxr
1853+ maxl = rateToLatency minr
1859 in mkYieldRateInfo l (LatencyRange minl maxl)
1854 in mkYieldRateInfo l (LatencyRange minl maxl)
1860 Nothing -> return Nothing
1855 Nothing -> return Nothing