@@ -236,7 +236,15 @@ data WorkerInfo = WorkerInfo
237-- Rate control.
237-- Rate control.
238data YieldRateInfo = YieldRateInfo
238data YieldRateInfo = YieldRateInfo
239− { expectedYieldLatency :: NanoSecs
239+ { svarExpectedLatency :: NanoSecs
241+ -- Actual latency/througput as seen from the consumer side, we count the
242+ -- yields and the time it took to generates those yields. This is used to
243+ -- increase or decrease the number of workers needed to achieve the desired
244+ -- rate. The idle time of workers is adjusted in this, so that we only
245+ -- account for the rate when the consumer actually demands data.
246+ -- XXX interval latency is enough, we can move this under diagnostics build
247+ , svarAllTimeLatency :: IORef (Count, TimeSpec)
241 -- XXX Worker latency specified by the user to be used before the first
249 -- XXX Worker latency specified by the user to be used before the first
242 -- actual measurement arrives. Not yet implemented
250 -- actual measurement arrives. Not yet implemented
@@ -248,16 +256,16 @@ data YieldRateInfo = YieldRateInfo
248 -- long time, in such cases the consumer can change it.
256 -- long time, in such cases the consumer can change it.
249 -- 0 means no latency computation
257 -- 0 means no latency computation
250 -- XXX this is derivable from workerMeasuredLatency, can be removed.
258 -- XXX this is derivable from workerMeasuredLatency, can be removed.
251− , workerLatencyPeriod :: IORef Count
259+ , workerLatencyUpdateInterval :: IORef Count
253 -- This is in progress latency stats maintained by the workers which we
261 -- This is in progress latency stats maintained by the workers which we
254 -- empty into workerCollectedLatency stats at certain intervals - whenever
262 -- empty into workerCollectedLatency stats at certain intervals - whenever
255 -- we process the stream elements yielded in this period.
263 -- we process the stream elements yielded in this period.
256 -- (yieldCount, timeTaken)
264 -- (yieldCount, timeTaken)
257− , workerCurrentLatency :: IORef (Count, NanoSecs)
265+ , workerPendingLatency :: IORef (Count, NanoSecs)
259 -- This is the second level stat which is an accmulation from
267 -- This is the second level stat which is an accmulation from
260− -- workerCurrentLatency stats. We keep accumulating latencies in this
268+ -- workerPendingLatency stats. We keep accumulating latencies in this
261 -- bucket until we have stats for a sufficient period and then we reset it
269 -- bucket until we have stats for a sufficient period and then we reset it
262 -- to start collecting for the next period and retain the computed average
270 -- to start collecting for the next period and retain the computed average
263 -- latency for the last period in workerMeasuredLatency.
271 -- latency for the last period in workerMeasuredLatency.
@@ -267,15 +275,8 @@ data YieldRateInfo = YieldRateInfo
267 -- Latency as measured by workers, aggregated for the last period.
275 -- Latency as measured by workers, aggregated for the last period.
268 , workerMeasuredLatency :: IORef NanoSecs
276 , workerMeasuredLatency :: IORef NanoSecs
270− -- Actual latency/througput as seen from the consumer side, we count the
271− -- yields and the time it took to generates those yields. This is used to
272− -- increase or decrease the number of workers needed to achieve the desired
273− -- rate. The idle time of workers is adjusted in this, so that we only
274− -- account for the rate when the consumer actually demands data.
275− , workerLongTermLatency :: IORef (Count, TimeSpec)
277 -- When the last worker exits it updates this timestamp. We use this to
278 -- When the last worker exits it updates this timestamp. We use this to
278− -- adjust workerLongTermLatency so that we do not account the idle time in
279+ -- adjust svarAllTimeLatency so that we do not account the idle time in
279 -- that. XXX Note that not all workers stop at the same time therefore
280 -- that. XXX Note that not all workers stop at the same time therefore
280 -- there is a certain inaccuracy in this mechanism.
281 -- there is a certain inaccuracy in this mechanism.
281 , workerStopTimeStamp :: IORef TimeSpec
282 , workerStopTimeStamp :: IORef TimeSpec
@@ -353,6 +354,7 @@ data State t m a = State
353 -- an explicit setting via a combinator.
354 -- an explicit setting via a combinator.
354 , _threadsHigh :: Limit
355 , _threadsHigh :: Limit
355 , _bufferHigh :: Limit
356 , _bufferHigh :: Limit
357+ -- XXX these two can be collapsed into a single type
356 , _streamLatency :: Maybe NanoSecs
358 , _streamLatency :: Maybe NanoSecs
357 , _maxStreamRate :: Maybe Double -- yields per second
359 , _maxStreamRate :: Maybe Double -- yields per second
@@ -518,9 +520,9 @@ secs k
518-- XXX Code duplicated from collectLatency
520-- XXX Code duplicated from collectLatency
519drainLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
521drainLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
520drainLatency _ss yinfo = do
522drainLatency _ss yinfo = do
521− let cur = workerCurrentLatency yinfo
523+ let cur = workerPendingLatency yinfo
522 col = workerCollectedLatency yinfo
524 col = workerCollectedLatency yinfo
523− longTerm = workerLongTermLatency yinfo
525+ longTerm = svarAllTimeLatency yinfo
524 measured = workerMeasuredLatency yinfo
526 measured = workerMeasuredLatency yinfo
526 (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
528 (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
@@ -573,7 +575,7 @@ dumpSVarStats sv ss style = do
573 (svarCnt, svarLat) <- case yieldRateInfo sv of
575 (svarCnt, svarLat) <- case yieldRateInfo sv of
574 Nothing -> return (0, 0)
576 Nothing -> return (0, 0)
575 Just yinfo -> do
577 Just yinfo -> do
576− (cnt, startTime) <- readIORef $ workerLongTermLatency yinfo
578+ (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
577 if cnt > 0
579 if cnt > 0
579 t <- readIORef (svarStopTime ss)
581 t <- readIORef (svarStopTime ss)
@@ -824,7 +826,7 @@ workerUpdateLatency yinfo winfo = do
824 t1 <- getTime Monotonic
826 t1 <- getTime Monotonic
825 writeIORef (workerLatencyStart winfo) (cnt1, t1)
827 writeIORef (workerLatencyStart winfo) (cnt1, t1)
826 let period = fromInteger $ toNanoSecs (t1 - t0)
828 let period = fromInteger $ toNanoSecs (t1 - t0)
827− let ref = workerCurrentLatency yinfo
829+ let ref = workerPendingLatency yinfo
828 atomicModifyIORefCAS ref $ \(ycnt, ytime) ->
830 atomicModifyIORefCAS ref $ \(ycnt, ytime) ->
829 ((ycnt + cnt1 - cnt0, ytime + period), ())
831 ((ycnt + cnt1 - cnt0, ytime + period), ())
@@ -851,7 +853,7 @@ checkRatePeriodic :: SVar t m a
852 -> IO Bool
854 -> IO Bool
853checkRatePeriodic sv yinfo winfo ycnt = do
855checkRatePeriodic sv yinfo winfo ycnt = do
854− i <- readIORef (workerLatencyPeriod yinfo)
856+ i <- readIORef (workerLatencyUpdateInterval yinfo)
855 if (i /= 0 && (ycnt `mod` i) == 0)
857 if (i /= 0 && (ycnt `mod` i) == 0)
857 workerUpdateLatency yinfo winfo
859 workerUpdateLatency yinfo winfo
@@ -884,7 +886,7 @@ sendYield sv winfo msg = do
884{-# INLINE workerStopUpdate #-}
886{-# INLINE workerStopUpdate #-}
885workerStopUpdate :: WorkerInfo -> YieldRateInfo -> Int -> IO ()
887workerStopUpdate :: WorkerInfo -> YieldRateInfo -> Int -> IO ()
886workerStopUpdate winfo info n = do
888workerStopUpdate winfo info n = do
887− i <- readIORef (workerLatencyPeriod info)
889+ i <- readIORef (workerLatencyUpdateInterval info)
888 when (i /= 0) $ workerUpdateLatency info winfo
890 when (i /= 0) $ workerUpdateLatency info winfo
889 when (n == 1) $ do
891 when (n == 1) $ do
890 t <- getTime Monotonic
892 t <- getTime Monotonic
@@ -1264,10 +1266,10 @@ minThreadDelay = 10^(6 :: Int)
1264-- number of yields that we are lagging by then we cannot start one worker for
1266-- number of yields that we are lagging by then we cannot start one worker for
1265-- each yield because that may be a very big number and if the latency of the
1267-- each yield because that may be a very big number and if the latency of the
1266-- workers is low these number of yields could be very high. We use this number
1268-- workers is low these number of yields could be very high. We use this number
1267−-- as the duration in which we want cover any lag. In other words we accumulate
1269+-- as the duration in which we want to cover any lag. In other words we
1268−-- the lag (lagging number of yields) such that it covers this much duration
1270+-- accumulate the lag (lagging number of yields) such that it covers this much
1269−-- for a single worker to produce those yields, before we start an extra
1271+-- duration for a single worker to produce those yields, before we start an
1270−-- worker.
1272+-- extra worker.
1271coverUpLatency :: NanoSecs
1273coverUpLatency :: NanoSecs
1272coverUpLatency = 1000000
1274coverUpLatency = 1000000
@@ -1277,7 +1279,7 @@ nanoToMicroSecs s = (fromIntegral s) `div` 1000
1277-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1279-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1278countDispatchWorkers
1280countDispatchWorkers
1279 :: Limit -> Count -> NanoSecs -> NanoSecs -> NanoSecs -> Int
1281 :: Limit -> Count -> NanoSecs -> NanoSecs -> NanoSecs -> Int
1280−countDispatchWorkers workerLimit count duration wLatency expectedLat =
1282+countDispatchWorkers workerLimit svarYields svarElapsed wLatency expectedLat =
1281 -- Calculate how many workers do we need to maintain the rate.
1283 -- Calculate how many workers do we need to maintain the rate.
1283 -- XXX Increasing the workers may reduce the overall latency per
1285 -- XXX Increasing the workers may reduce the overall latency per
@@ -1298,24 +1300,25 @@ countDispatchWorkers workerLimit count duration wLatency expectedLat =
1298 -- Calculate how many yields are we ahead or behind to match the exact
1300 -- Calculate how many yields are we ahead or behind to match the exact
1299 -- required rate. Based on that we increase or decrease the effective
1301 -- required rate. Based on that we increase or decrease the effective
1300 -- workers.
1302 -- workers.
1301− d = fromIntegral duration
1303+ d = fromIntegral svarElapsed
1302 e = fromIntegral expectedLat
1304 e = fromIntegral expectedLat
1303 n = max 1 (fromIntegral coverUpLatency / fromIntegral wLatency)
1305 n = max 1 (fromIntegral coverUpLatency / fromIntegral wLatency)
1304 :: Double
1306 :: Double
1305− extraYields = (d / e) - (fromIntegral count)
1307+ extraYields = (d / e) - (fromIntegral svarYields)
1306 extraWorkers = extraYields / n
1308 extraWorkers = extraYields / n
1307 requiredWorkers = round $ nWorkers + extraWorkers
1309 requiredWorkers = round $ nWorkers + extraWorkers
1308 in case workerLimit of
1310 in case workerLimit of
1309 Unlimited -> requiredWorkers
1311 Unlimited -> requiredWorkers
1310 Limited x -> min requiredWorkers (fromIntegral x)
1312 Limited x -> min requiredWorkers (fromIntegral x)
1312−-- | Get the worker latency without resetting workerCurrentLatency
1314+-- | Get the worker latency without resetting workerPendingLatency
1313-- Returns (total yield count, base time, measured latency)
1315-- Returns (total yield count, base time, measured latency)
1316+-- CAUTION! keep it in sync with collectLatency and useCollectedBatch
1314getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1317getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1315getWorkerLatency yinfo = do
1318getWorkerLatency yinfo = do
1316− let cur = workerCurrentLatency yinfo
1319+ let cur = workerPendingLatency yinfo
1317 col = workerCollectedLatency yinfo
1320 col = workerCollectedLatency yinfo
1318− longTerm = workerLongTermLatency yinfo
1321+ longTerm = svarAllTimeLatency yinfo
1319 measured = workerMeasuredLatency yinfo
1322 measured = workerMeasuredLatency yinfo
1321 (count, time) <- readIORef cur
1324 (count, time) <- readIORef cur
@@ -1325,11 +1328,11 @@ getWorkerLatency yinfo = do
1326 let pendingCount = colCount + count
1329 let pendingCount = colCount + count
1327 pendingTime = colTime + time
1330 pendingTime = colTime + time
1328− -- XXX Take the average of previous and the latest
1330 if pendingCount > 0
1332 if pendingCount > 0
1331− then ((pendingTime `div` (fromIntegral pendingCount)) + prev)
1333+ then let lat = pendingTime `div` (fromIntegral pendingCount)
1334+ -- XXX Give more weight to new?
1332− `div` 2
1335+ in (lat + prev) `div` 2
1333 else prev
1336 else prev
1334 return (lcount + pendingCount, ltime, new)
1337 return (lcount + pendingCount, ltime, new)
@@ -1338,7 +1341,7 @@ isBeyondMaxRate sv yinfo = do
1338 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1341 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1339 now <- getTime Monotonic
1342 now <- getTime Monotonic
1340 let duration = fromInteger $ toNanoSecs $ now - tstamp
1343 let duration = fromInteger $ toNanoSecs $ now - tstamp
1341− let expectedLat = expectedYieldLatency yinfo
1344+ let expectedLat = svarExpectedLatency yinfo
1342 let netWorkers = countDispatchWorkers (maxWorkerLimit sv) count duration
1345 let netWorkers = countDispatchWorkers (maxWorkerLimit sv) count duration
1343 wLatency expectedLat
1346 wLatency expectedLat
1344 cnt <- readIORef $ workerCount sv
1347 cnt <- readIORef $ workerCount sv
@@ -1358,7 +1361,7 @@ useCollectedBatch :: YieldRateInfo
1358 -> IORef NanoSecs
1361 -> IORef NanoSecs
1359 -> IO ()
1362 -> IO ()
1360useCollectedBatch yinfo latency colRef measRef = do
1363useCollectedBatch yinfo latency colRef measRef = do
1361− let periodRef = workerLatencyPeriod yinfo
1364+ let periodRef = workerLatencyUpdateInterval yinfo
1362 cnt = max 1 $ minThreadDelay `div` latency
1365 cnt = max 1 $ minThreadDelay `div` latency
1363 period = min cnt (fromIntegral magicMaxBuffer)
1366 period = min cnt (fromIntegral magicMaxBuffer)
@@ -1373,11 +1376,12 @@ useCollectedBatch yinfo latency colRef measRef = do
1373-- period. The former two are used for accurate measurement of the going rate
1376-- period. The former two are used for accurate measurement of the going rate
1374-- whereas the average is used for future estimates e.g. how many workers
1377-- whereas the average is used for future estimates e.g. how many workers
1375-- should be maintained to maintain the rate.
1378-- should be maintained to maintain the rate.
1379+-- CAUTION! keep it in sync with getWorkerLatency
1376collectLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1380collectLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1377collectLatency _ss yinfo = do
1381collectLatency _ss yinfo = do
1378− let cur = workerCurrentLatency yinfo
1382+ let cur = workerPendingLatency yinfo
1379 col = workerCollectedLatency yinfo
1383 col = workerCollectedLatency yinfo
1380− longTerm = workerLongTermLatency yinfo
1384+ longTerm = svarAllTimeLatency yinfo
1381 measured = workerMeasuredLatency yinfo
1385 measured = workerMeasuredLatency yinfo
1383 (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
1387 (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
@@ -1389,7 +1393,7 @@ collectLatency _ss yinfo = do
1389 pendingTime = colTime + time
1393 pendingTime = colTime + time
1391 lcount' = lcount + pendingCount
1395 lcount' = lcount + pendingCount
1392− notUpdated = (lcount', ltime, prev)
1396+ tripleWith lat = (lcount', ltime, lat)
1394 if (pendingCount > 0)
1398 if (pendingCount > 0)
@@ -1402,7 +1406,9 @@ collectLatency _ss yinfo = do
1402 maxLat <- readIORef (maxWorkerLatency _ss)
1406 maxLat <- readIORef (maxWorkerLatency _ss)
1403 when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
1407 when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
1405− -- To avoid minor fluctuations update in batches
1409+ -- When we have collected a significant sized batch we compute the new
1410+ -- latency using that batch and return the new latency, otherwise we
1411+ -- return the previous latency derived from the previous batch.
1406 if (pendingCount > fromIntegral magicMaxBuffer)
1412 if (pendingCount > fromIntegral magicMaxBuffer)
1407 || (pendingTime > minThreadDelay)
1413 || (pendingTime > minThreadDelay)
1408 || (let r = (fromIntegral new) / (fromIntegral prev) :: Double
1414 || (let r = (fromIntegral new) / (fromIntegral prev) :: Double
@@ -1415,15 +1421,15 @@ collectLatency _ss yinfo = do
1415 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
1421 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
1417 modifyIORef longTerm $ \(_, t) -> (lcount', t)
1423 modifyIORef longTerm $ \(_, t) -> (lcount', t)
1418− return (lcount', ltime, new)
1424+ return $ tripleWith new
1420 writeIORef col (pendingCount, pendingTime)
1426 writeIORef col (pendingCount, pendingTime)
1421− return notUpdated
1427+ return $ tripleWith prev
1422− else return notUpdated
1428+ else return $ tripleWith prev
1424adjustLatencies :: YieldRateInfo -> IO ()
1430adjustLatencies :: YieldRateInfo -> IO ()
1425adjustLatencies yinfo = do
1431adjustLatencies yinfo = do
1426− let longTerm = workerLongTermLatency yinfo
1432+ let longTerm = svarAllTimeLatency yinfo
1427 t0 <- readIORef (workerStopTimeStamp yinfo)
1433 t0 <- readIORef (workerStopTimeStamp yinfo)
1428 t1 <- getTime Monotonic
1434 t1 <- getTime Monotonic
1429 modifyIORef longTerm $ \(c, t) -> (c, t + t1 - t0)
1435 modifyIORef longTerm $ \(c, t) -> (c, t + t1 - t0)
@@ -1439,20 +1445,32 @@ adjustLatencies yinfo = do
1439-- False: full, no more dispatches
1445-- False: full, no more dispatches
1440dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1446dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1441dispatchWorkerPaced sv = do
1447dispatchWorkerPaced sv = do
1442− let workerLimit = maxWorkerLimit sv
1443 let yinfo = fromJust $ yieldRateInfo sv
1448 let yinfo = fromJust $ yieldRateInfo sv
1449+ (svarYields, svarElapsed, wLatency) <- do
1450+ now <- liftIO $ getTime Monotonic
1451+ (yieldCount, baseTime, lat) <-
1444− (count, tstamp, wLatency) <- liftIO $ collectLatency (svarStats sv) yinfo
1452+ liftIO $ collectLatency (svarStats sv) yinfo
1453+ let elapsed = fromInteger $ toNanoSecs $ now - baseTime
1445− -- XXX use workerBootstrapLatency
1457+ case workerBootstrapLatency yinfo of
1462+ return (yieldCount, elapsed, latency)
1446 if wLatency == 0
1464 if wLatency == 0
1447 -- Need to measure the latency with a single worker before we can perform
1465 -- Need to measure the latency with a single worker before we can perform
1448 -- any computation.
1466 -- any computation.
1449 then return False
1467 then return False
1451− now <- liftIO $ getTime Monotonic
1469+ let workerLimit = maxWorkerLimit sv
1452− let duration = fromInteger $ toNanoSecs $ now - tstamp
1453− let expectedLat = expectedYieldLatency yinfo
1470+ let expectedLat = svarExpectedLatency yinfo
1454− let netWorkers = countDispatchWorkers workerLimit count duration
1471+ let netWorkers = countDispatchWorkers
1472+ workerLimit svarYields svarElapsed
1455− wLatency expectedLat
1473+ wLatency expectedLat
1457 -- XXX we need to take yieldLimit into account here. If we are at the
1475 -- XXX we need to take yieldLimit into account here. If we are at the
1458 -- end of the limit as well as the time, we should not be sleeping.
1476 -- end of the limit as well as the time, we should not be sleeping.
@@ -1460,7 +1478,8 @@ dispatchWorkerPaced sv = do
1460 -- to take that in account.
1478 -- to take that in account.
1461 if netWorkers <= 0
1479 if netWorkers <= 0
1463− let sleepTime = ((fromIntegral count) * expectedLat) - duration
1481+ let expectedDuration = (fromIntegral svarYields) * expectedLat
1482+ let sleepTime = expectedDuration - svarElapsed
1465 when (sleepTime >= minThreadDelay) $ do
1484 when (sleepTime >= minThreadDelay) $ do
1466 liftIO $ threadDelay $ nanoToMicroSecs sleepTime
1485 liftIO $ threadDelay $ nanoToMicroSecs sleepTime
@@ -1697,14 +1716,14 @@ getYieldRateInfo st = do
1697 stopTime <- newIORef $ fromNanoSecs 0
1716 stopTime <- newIORef $ fromNanoSecs 0
1699 return $ Just YieldRateInfo
1718 return $ Just YieldRateInfo
1700− { expectedYieldLatency = latency
1719+ { svarExpectedLatency = latency
1701 , workerBootstrapLatency = getStreamLatency st
1720 , workerBootstrapLatency = getStreamLatency st
1702− , workerLatencyPeriod = period
1721+ , workerLatencyUpdateInterval = period
1703 , workerMeasuredLatency = measured
1722 , workerMeasuredLatency = measured
1704− , workerCurrentLatency = wcur
1723+ , workerPendingLatency = wcur
1705 , workerStopTimeStamp = stopTime
1724 , workerStopTimeStamp = stopTime
1706 , workerCollectedLatency = wcol
1725 , workerCollectedLatency = wcol
1707− , workerLongTermLatency = wlong
1726+ , svarAllTimeLatency = wlong
1709 Nothing -> return Nothing
1728 Nothing -> return Nothing