@@ -236,7 +236,15 @@ data WorkerInfo = WorkerInfo
236236
237237-- Rate control.
238238data YieldRateInfo = YieldRateInfo
239− { expectedYieldLatency :: NanoSecs
239+ { svarExpectedLatency :: NanoSecs
240+
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)
240248
241249 -- XXX Worker latency specified by the user to be used before the first
242250 -- actual measurement arrives. Not yet implemented
@@ -248,16 +256,16 @@ data YieldRateInfo = YieldRateInfo
248256 -- long time, in such cases the consumer can change it.
249257 -- 0 means no latency computation
250258 -- XXX this is derivable from workerMeasuredLatency, can be removed.
251− , workerLatencyPeriod :: IORef Count
259+ , workerLatencyUpdateInterval :: IORef Count
252260
253261 -- This is in progress latency stats maintained by the workers which we
254262 -- empty into workerCollectedLatency stats at certain intervals - whenever
255263 -- we process the stream elements yielded in this period.
256264 -- (yieldCount, timeTaken)
257− , workerCurrentLatency :: IORef (Count, NanoSecs)
265+ , workerPendingLatency :: IORef (Count, NanoSecs)
258266
259267 -- 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
261269 -- bucket until we have stats for a sufficient period and then we reset it
262270 -- to start collecting for the next period and retain the computed average
263271 -- latency for the last period in workerMeasuredLatency.
@@ -267,15 +275,8 @@ data YieldRateInfo = YieldRateInfo
267275 -- Latency as measured by workers, aggregated for the last period.
268276 , workerMeasuredLatency :: IORef NanoSecs
269277
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)
276−
277278 -- 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
279280 -- that. XXX Note that not all workers stop at the same time therefore
280281 -- there is a certain inaccuracy in this mechanism.
281282 , workerStopTimeStamp :: IORef TimeSpec
@@ -353,6 +354,7 @@ data State t m a = State
353354 -- an explicit setting via a combinator.
354355 , _threadsHigh :: Limit
355356 , _bufferHigh :: Limit
357+ -- XXX these two can be collapsed into a single type
356358 , _streamLatency :: Maybe NanoSecs
357359 , _maxStreamRate :: Maybe Double -- yields per second
358360 }
@@ -518,9 +520,9 @@ secs k
518520-- XXX Code duplicated from collectLatency
519521drainLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
520522drainLatency _ss yinfo = do
521− let cur = workerCurrentLatency yinfo
523+ let cur = workerPendingLatency yinfo
522524 col = workerCollectedLatency yinfo
523− longTerm = workerLongTermLatency yinfo
525+ longTerm = svarAllTimeLatency yinfo
524526 measured = workerMeasuredLatency yinfo
525527
526528 (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
@@ -573,7 +575,7 @@ dumpSVarStats sv ss style = do
573575 (svarCnt, svarLat) <- case yieldRateInfo sv of
574576 Nothing -> return (0, 0)
575577 Just yinfo -> do
576− (cnt, startTime) <- readIORef $ workerLongTermLatency yinfo
578+ (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
577579 if cnt > 0
578580 then do
579581 t <- readIORef (svarStopTime ss)
@@ -824,7 +826,7 @@ workerUpdateLatency yinfo winfo = do
824826 t1 <- getTime Monotonic
825827 writeIORef (workerLatencyStart winfo) (cnt1, t1)
826828 let period = fromInteger $ toNanoSecs (t1 - t0)
827− let ref = workerCurrentLatency yinfo
829+ let ref = workerPendingLatency yinfo
828830 atomicModifyIORefCAS ref $ \(ycnt, ytime) ->
829831 ((ycnt + cnt1 - cnt0, ytime + period), ())
830832
@@ -851,7 +853,7 @@ checkRatePeriodic :: SVar t m a
851853 -> Count
852854 -> IO Bool
853855checkRatePeriodic sv yinfo winfo ycnt = do
854− i <- readIORef (workerLatencyPeriod yinfo)
856+ i <- readIORef (workerLatencyUpdateInterval yinfo)
855857 if (i /= 0 && (ycnt `mod` i) == 0)
856858 then do
857859 workerUpdateLatency yinfo winfo
@@ -884,7 +886,7 @@ sendYield sv winfo msg = do
884886{-# INLINE workerStopUpdate #-}
885887workerStopUpdate :: WorkerInfo -> YieldRateInfo -> Int -> IO ()
886888workerStopUpdate winfo info n = do
887− i <- readIORef (workerLatencyPeriod info)
889+ i <- readIORef (workerLatencyUpdateInterval info)
888890 when (i /= 0) $ workerUpdateLatency info winfo
889891 when (n == 1) $ do
890892 t <- getTime Monotonic
@@ -1264,10 +1266,10 @@ minThreadDelay = 10^(6 :: Int)
12641266-- number of yields that we are lagging by then we cannot start one worker for
12651267-- each yield because that may be a very big number and if the latency of the
12661268-- 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
1268−-- the lag (lagging number of yields) such that it covers this much duration
1269−-- for a single worker to produce those yields, before we start an extra
1270−-- worker.
1269+-- as the duration in which we want to cover any lag. In other words we
1270+-- accumulate the lag (lagging number of yields) such that it covers this much
1271+-- duration for a single worker to produce those yields, before we start an
1272+-- extra worker.
12711273coverUpLatency :: NanoSecs
12721274coverUpLatency = 1000000
12731275
@@ -1277,7 +1279,7 @@ nanoToMicroSecs s = (fromIntegral s) `div` 1000
12771279-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
12781280countDispatchWorkers
12791281 :: Limit -> Count -> NanoSecs -> NanoSecs -> NanoSecs -> Int
1280−countDispatchWorkers workerLimit count duration wLatency expectedLat =
1282+countDispatchWorkers workerLimit svarYields svarElapsed wLatency expectedLat =
12811283 -- Calculate how many workers do we need to maintain the rate.
12821284
12831285 -- XXX Increasing the workers may reduce the overall latency per
@@ -1298,24 +1300,25 @@ countDispatchWorkers workerLimit count duration wLatency expectedLat =
12981300 -- Calculate how many yields are we ahead or behind to match the exact
12991301 -- required rate. Based on that we increase or decrease the effective
13001302 -- workers.
1301− d = fromIntegral duration
1303+ d = fromIntegral svarElapsed
13021304 e = fromIntegral expectedLat
13031305 n = max 1 (fromIntegral coverUpLatency / fromIntegral wLatency)
13041306 :: Double
1305− extraYields = (d / e) - (fromIntegral count)
1307+ extraYields = (d / e) - (fromIntegral svarYields)
13061308 extraWorkers = extraYields / n
13071309 requiredWorkers = round $ nWorkers + extraWorkers
13081310 in case workerLimit of
13091311 Unlimited -> requiredWorkers
13101312 Limited x -> min requiredWorkers (fromIntegral x)
13111313
1312−-- | Get the worker latency without resetting workerCurrentLatency
1314+-- | Get the worker latency without resetting workerPendingLatency
13131315-- Returns (total yield count, base time, measured latency)
1316+-- CAUTION! keep it in sync with collectLatency and useCollectedBatch
13141317getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
13151318getWorkerLatency yinfo = do
1316− let cur = workerCurrentLatency yinfo
1319+ let cur = workerPendingLatency yinfo
13171320 col = workerCollectedLatency yinfo
1318− longTerm = workerLongTermLatency yinfo
1321+ longTerm = svarAllTimeLatency yinfo
13191322 measured = workerMeasuredLatency yinfo
13201323
13211324 (count, time) <- readIORef cur
@@ -1325,11 +1328,11 @@ getWorkerLatency yinfo = do
13251328
13261329 let pendingCount = colCount + count
13271330 pendingTime = colTime + time
1328− -- XXX Take the average of previous and the latest
13291331 new =
13301332 if pendingCount > 0
1331− then ((pendingTime `div` (fromIntegral pendingCount)) + prev)
1332− `div` 2
1333+ then let lat = pendingTime `div` (fromIntegral pendingCount)
1334+ -- XXX Give more weight to new?
1335+ in (lat + prev) `div` 2
13331336 else prev
13341337 return (lcount + pendingCount, ltime, new)
13351338
@@ -1338,7 +1341,7 @@ isBeyondMaxRate sv yinfo = do
13381341 (count, tstamp, wLatency) <- getWorkerLatency yinfo
13391342 now <- getTime Monotonic
13401343 let duration = fromInteger $ toNanoSecs $ now - tstamp
1341− let expectedLat = expectedYieldLatency yinfo
1344+ let expectedLat = svarExpectedLatency yinfo
13421345 let netWorkers = countDispatchWorkers (maxWorkerLimit sv) count duration
13431346 wLatency expectedLat
13441347 cnt <- readIORef $ workerCount sv
@@ -1358,7 +1361,7 @@ useCollectedBatch :: YieldRateInfo
13581361 -> IORef NanoSecs
13591362 -> IO ()
13601363useCollectedBatch yinfo latency colRef measRef = do
1361− let periodRef = workerLatencyPeriod yinfo
1364+ let periodRef = workerLatencyUpdateInterval yinfo
13621365 cnt = max 1 $ minThreadDelay `div` latency
13631366 period = min cnt (fromIntegral magicMaxBuffer)
13641367
@@ -1373,11 +1376,12 @@ useCollectedBatch yinfo latency colRef measRef = do
13731376-- period. The former two are used for accurate measurement of the going rate
13741377-- whereas the average is used for future estimates e.g. how many workers
13751378-- should be maintained to maintain the rate.
1379+-- CAUTION! keep it in sync with getWorkerLatency
13761380collectLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
13771381collectLatency _ss yinfo = do
1378− let cur = workerCurrentLatency yinfo
1382+ let cur = workerPendingLatency yinfo
13791383 col = workerCollectedLatency yinfo
1380− longTerm = workerLongTermLatency yinfo
1384+ longTerm = svarAllTimeLatency yinfo
13811385 measured = workerMeasuredLatency yinfo
13821386
13831387 (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
@@ -1389,7 +1393,7 @@ collectLatency _ss yinfo = do
13891393 pendingTime = colTime + time
13901394
13911395 lcount' = lcount + pendingCount
1392− notUpdated = (lcount', ltime, prev)
1396+ tripleWith lat = (lcount', ltime, lat)
13931397
13941398 if (pendingCount > 0)
13951399 then do
@@ -1402,7 +1406,9 @@ collectLatency _ss yinfo = do
14021406 maxLat <- readIORef (maxWorkerLatency _ss)
14031407 when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
14041408#endif
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.
14061412 if (pendingCount > fromIntegral magicMaxBuffer)
14071413 || (pendingTime > minThreadDelay)
14081414 || (let r = (fromIntegral new) / (fromIntegral prev) :: Double
@@ -1415,15 +1421,15 @@ collectLatency _ss yinfo = do
14151421 \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
14161422#endif
14171423 modifyIORef longTerm $ \(_, t) -> (lcount', t)
1418− return (lcount', ltime, new)
1424+ return $ tripleWith new
14191425 else do
14201426 writeIORef col (pendingCount, pendingTime)
1421− return notUpdated
1422− else return notUpdated
1427+ return $ tripleWith prev
1428+ else return $ tripleWith prev
14231429
14241430adjustLatencies :: YieldRateInfo -> IO ()
14251431adjustLatencies yinfo = do
1426− let longTerm = workerLongTermLatency yinfo
1432+ let longTerm = svarAllTimeLatency yinfo
14271433 t0 <- readIORef (workerStopTimeStamp yinfo)
14281434 t1 <- getTime Monotonic
14291435 modifyIORef longTerm $ \(c, t) -> (c, t + t1 - t0)
@@ -1439,20 +1445,32 @@ adjustLatencies yinfo = do
14391445-- False: full, no more dispatches
14401446dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
14411447dispatchWorkerPaced sv = do
1442− let workerLimit = maxWorkerLimit sv
14431448 let yinfo = fromJust $ yieldRateInfo sv
1444− (count, tstamp, wLatency) <- liftIO $ collectLatency (svarStats sv) yinfo
1445− -- XXX use workerBootstrapLatency
1449+ (svarYields, svarElapsed, wLatency) <- do
1450+ now <- liftIO $ getTime Monotonic
1451+ (yieldCount, baseTime, lat) <-
1452+ liftIO $ collectLatency (svarStats sv) yinfo
1453+ let elapsed = fromInteger $ toNanoSecs $ now - baseTime
1454+ let latency =
1455+ if lat == 0
1456+ then
1457+ case workerBootstrapLatency yinfo of
1458+ Nothing -> lat
1459+ Just t -> t
1460+ else lat
1461+
1462+ return (yieldCount, elapsed, latency)
1463+
14461464 if wLatency == 0
14471465 -- Need to measure the latency with a single worker before we can perform
14481466 -- any computation.
14491467 then return False
14501468 else do
1451− now <- liftIO $ getTime Monotonic
1452− let duration = fromInteger $ toNanoSecs $ now - tstamp
1453− let expectedLat = expectedYieldLatency yinfo
1454− let netWorkers = countDispatchWorkers workerLimit count duration
1455− wLatency expectedLat
1469+ let workerLimit = maxWorkerLimit sv
1470+ let expectedLat = svarExpectedLatency yinfo
1471+ let netWorkers = countDispatchWorkers
1472+ workerLimit svarYields svarElapsed
1473+ wLatency expectedLat
14561474
14571475 -- XXX we need to take yieldLimit into account here. If we are at the
14581476 -- end of the limit as well as the time, we should not be sleeping.
@@ -1460,7 +1478,8 @@ dispatchWorkerPaced sv = do
14601478 -- to take that in account.
14611479 if netWorkers <= 0
14621480 then do
1463− let sleepTime = ((fromIntegral count) * expectedLat) - duration
1481+ let expectedDuration = (fromIntegral svarYields) * expectedLat
1482+ let sleepTime = expectedDuration - svarElapsed
14641483
14651484 when (sleepTime >= minThreadDelay) $ do
14661485 liftIO $ threadDelay $ nanoToMicroSecs sleepTime
@@ -1697,14 +1716,14 @@ getYieldRateInfo st = do
16971716 stopTime <- newIORef $ fromNanoSecs 0
16981717
16991718 return $ Just YieldRateInfo
1700− { expectedYieldLatency = latency
1719+ { svarExpectedLatency = latency
17011720 , workerBootstrapLatency = getStreamLatency st
1702− , workerLatencyPeriod = period
1721+ , workerLatencyUpdateInterval = period
17031722 , workerMeasuredLatency = measured
1704− , workerCurrentLatency = wcur
1723+ , workerPendingLatency = wcur
17051724 , workerStopTimeStamp = stopTime
17061725 , workerCollectedLatency = wcol
1707− , workerLongTermLatency = wlong
1726+ , svarAllTimeLatency = wlong
17081727 }
17091728 Nothing -> return Nothing
17101729