This is an automated email from the ASF dual-hosted git repository.
rustyrazorblade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cassandra-easy-stress.git
The following commit(s) were added to refs/heads/main by this push:
new 83ee78c Fix the rate limiter optimizer's control loop
83ee78c is described below
commit 83ee78c7678b350b3c20f267616834817262ea06
Author: Jon Haddad <[email protected]>
AuthorDate: Fri Aug 28 17:18:00 2026 -0700
Fix the rate limiter optimizer's control loop
The optimizer read a p99 that lagged its own control interval by minutes, so
every adjustment compounded on stale evidence. Twelve 10% cuts in a minute
took
the rate to 28% of capacity while the measured p99 had barely moved.
- Give the operation timers a SlidingTimeWindowArrayReservoir of 10 seconds.
The Dropwizard default still weights samples from about five minutes ago.
- Wait one full latency window after any change, so each decision sees the
rate
it last set rather than the one before it.
- Reduce against achieved throughput, not the nominal limit. Under overload
the
client falls behind its limit, so cutting the limit alone changed nothing.
- Remove the guard that blocked reductions when utilization was low. It
blocked
the only corrective action in exactly the case that needed it.
- Replace the cube-root increase gain with one based on the unused fraction
of
the latency budget. The old formula divided a cube root of milliseconds by
milliseconds, so its response depended on the size of the target.
- Add a rate floor at a hundredth of the starting rate.
- Count deletions and populate operations towards the minimum-samples gate.
The
populate branch could never run before, and delete-only workloads never
optimized.
- Skip the optimizer entirely without a latency target. A plain --rate run
no
longer spends its first minute ramping towards a rate it was given.
- Make its Timer a daemon and cancel it on shutdown; guard the shared state.
The reported p99 now covers ten seconds rather than a five-minute decay. The
console column, the Prometheus export, and the server API all change with
it.
The HDR and Parquet outputs are unaffected; they record raw operation
timings.
---
.../org/apache/cassandra/easystress/Metrics.kt | 31 +++-
.../cassandra/easystress/RateLimiterOptimizer.kt | 173 ++++++++++++++-------
.../apache/cassandra/easystress/commands/Run.kt | 27 +++-
.../org/apache/cassandra/easystress/MetricsTest.kt | 51 ++++++
.../easystress/RateLimiterOptimizerTest.kt | 104 +++++++++++++
5 files changed, 318 insertions(+), 68 deletions(-)
diff --git a/src/main/kotlin/org/apache/cassandra/easystress/Metrics.kt
b/src/main/kotlin/org/apache/cassandra/easystress/Metrics.kt
index f8686a5..736996a 100644
--- a/src/main/kotlin/org/apache/cassandra/easystress/Metrics.kt
+++ b/src/main/kotlin/org/apache/cassandra/easystress/Metrics.kt
@@ -19,6 +19,8 @@ package org.apache.cassandra.easystress
import com.codahale.metrics.MetricRegistry
import com.codahale.metrics.ScheduledReporter
+import com.codahale.metrics.SlidingTimeWindowArrayReservoir
+import com.codahale.metrics.Timer
import io.prometheus.client.CollectorRegistry
import io.prometheus.client.dropwizard.DropwizardExports
import io.prometheus.client.exporter.HTTPServer
@@ -67,11 +69,11 @@ class Metrics(
}
var errors = metricRegistry.meter("errors")
- val mutations = metricRegistry.timer("mutations")
- val selects = metricRegistry.timer("selects")
- val deletions = metricRegistry.timer("deletions")
+ val mutations = shortWindowTimer("mutations")
+ val selects = shortWindowTimer("selects")
+ val deletions = shortWindowTimer("deletions")
- val populate = metricRegistry.timer("populateMutations")
+ val populate = shortWindowTimer("populateMutations")
// Throughput trackers for metrics
val selectThroughputTracker = getTracker { selects.count }.start()
@@ -79,6 +81,19 @@ class Metrics(
val deletionThroughputTracker = getTracker { deletions.count }.start()
val populateThroughputTracker = getTracker { populate.count }.start()
+ /**
+ * Creates a timer whose percentiles cover only the recent past.
+ *
+ * The Dropwizard default is an ExponentiallyDecayingReservoir, which
still weights samples
+ * from about five minutes ago. The rate limiter optimizer reads these
percentiles every few
+ * seconds, so that default makes it react to a signal it has already
acted on, and the console
+ * report shows a p99 that lags the run. A short sliding window fixes
both.
+ */
+ private fun shortWindowTimer(name: String) =
+ metricRegistry.timer(name) {
+ Timer(SlidingTimeWindowArrayReservoir(LATENCY_WINDOW_SECONDS,
TimeUnit.SECONDS))
+ }
+
/**
* We track throughput using separate structures than Dropwizard
*/
@@ -102,4 +117,12 @@ class Metrics(
fun getDeletionThroughput() =
deletionThroughputTracker.getCurrentThroughput()
fun getPopulateThroughput() =
populateThroughputTracker.getCurrentThroughput()
+
+ companion object {
+ /**
+ * The window latency percentiles are calculated over. This must stay
short enough that the
+ * rate limiter optimizer sees the effect of its last adjustment
before it makes the next one.
+ */
+ const val LATENCY_WINDOW_SECONDS = 10L
+ }
}
diff --git
a/src/main/kotlin/org/apache/cassandra/easystress/RateLimiterOptimizer.kt
b/src/main/kotlin/org/apache/cassandra/easystress/RateLimiterOptimizer.kt
index d6d30e1..e349a30 100644
--- a/src/main/kotlin/org/apache/cassandra/easystress/RateLimiterOptimizer.kt
+++ b/src/main/kotlin/org/apache/cassandra/easystress/RateLimiterOptimizer.kt
@@ -19,11 +19,8 @@ package org.apache.cassandra.easystress
import com.google.common.util.concurrent.RateLimiter
import org.apache.logging.log4j.kotlin.logger
-import java.lang.Math.cbrt
-import java.lang.Math.min
import java.util.Optional
import java.util.concurrent.TimeUnit
-import kotlin.math.sqrt
/**
* Dynamically adjusts the rate limiter based on observed latency metrics.
@@ -32,6 +29,8 @@ import kotlin.math.sqrt
* 1. Gradually increases load during an initial "step phase"
* 2. Monitors latencies to ensure they stay within target thresholds
* 3. Dynamically adjusts throughput based on current performance metrics
+ *
+ * The loop only acts once the latency metrics reflect its last change. See
[hasSettled].
*/
class RateLimiterOptimizer(
val rateLimiter: RateLimiter,
@@ -39,15 +38,51 @@ class RateLimiterOptimizer(
val maxReadLatency: Long?,
val maxWriteLatency: Long?,
var isStepPhase: Boolean = true,
+ val clock: () -> Long = System::currentTimeMillis,
) {
companion object {
val log = logger()
+
+ /** Cut the rate to this fraction of achieved throughput when latency
exceeds the target. */
+ const val REDUCTION_FACTOR = 0.90
+
+ /** The largest single increase, applied when latency is far below the
target. */
+ const val MAX_INCREASE = 0.05
+
+ /** Hold the rate steady above this fraction of the target, so the
loop does not hunt. */
+ const val DEAD_BAND = 0.90
+
+ /** Never raise a limit the client is not already using. */
+ const val MIN_UTILIZATION = 0.90
+
+ /** Wait for this many operations before trusting the latency
percentiles. */
+ const val MIN_OPERATIONS = 100L
+
+ /** The step phase reaches the target rate in this many steps. */
+ const val STEPS = 10
+
+ /** The rate never falls below the starting rate divided by this. */
+ const val RATE_FLOOR_DIVISOR = 100.0
}
val durationFactor = 1.0 / TimeUnit.MILLISECONDS.toNanos(1)
val initial: Double = rateLimiter.rate
- val stepValue = initial / 10.0
+ val stepValue = initial / STEPS
+
+ /**
+ * A floor stops a run of reductions from driving the rate arbitrarily
close to zero, which
+ * leaves the test stalled with no way back up.
+ */
+ private val minRate = initial / RATE_FLOOR_DIVISOR
+
+ /**
+ * Latency percentiles cover [Metrics.LATENCY_WINDOW_SECONDS]. Acting
again before that window
+ * has passed means acting on the previous rate, which compounds every
adjustment.
+ */
+ private val settleMs =
TimeUnit.SECONDS.toMillis(Metrics.LATENCY_WINDOW_SECONDS)
+
+ private var lastChangeAtMs = 0L
init {
println("Stepping rate limiter by $stepValue to $initial")
@@ -62,6 +97,7 @@ class RateLimiterOptimizer(
*
* @return The updated rate limit value
*/
+ @Synchronized
fun execute(): Double {
// Handle fresh start or when reset() was called
if (isStepPhase) {
@@ -69,11 +105,17 @@ class RateLimiterOptimizer(
}
// Skip optimization if we don't have enough metrics
- if (getTotalOperations() < 100) {
+ if (getTotalOperations() < MIN_OPERATIONS) {
log.info("Not enough operations performed yet to optimize")
return rateLimiter.rate
}
+ // Skip optimization until the metrics describe the rate we set last
time
+ if (!hasSettled()) {
+ log.debug("Waiting for the latency window to catch up with the
last change")
+ return rateLimiter.rate
+ }
+
// Get current latency metrics and optimize if available
return getCurrentAndMaxLatency()
.map { (currentLatency, maxLatency) ->
optimizeRateLimit(currentLatency, maxLatency) }
@@ -85,7 +127,7 @@ class RateLimiterOptimizer(
*/
private fun handleStepPhase(): Double {
log.info("Stepping rate limiter by $stepValue")
- val newValue = min(rateLimiter.rate + stepValue, initial)
+ val newValue = minOf(rateLimiter.rate + stepValue, initial)
// Check if we've reached the initial target
if (newValue >= initial) {
@@ -93,7 +135,7 @@ class RateLimiterOptimizer(
isStepPhase = false
}
- rateLimiter.rate = newValue
+ applyRate(newValue)
log.info("New rate limiter value: ${rateLimiter.rate}")
return rateLimiter.rate
}
@@ -106,7 +148,7 @@ class RateLimiterOptimizer(
maxLatency: Long,
): Double {
val currentRate = rateLimiter.rate
- val newLimit = getNextValue(currentRate, currentLatency, maxLatency)
+ val newLimit = getNextValue(currentRate, getCurrentTotalThroughput(),
currentLatency, maxLatency)
// No change needed
if (newLimit == currentRate) {
@@ -114,43 +156,39 @@ class RateLimiterOptimizer(
return currentRate
}
- // Get current throughput for decision making
- val currentThroughput = getCurrentTotalThroughput()
- val utilizationRatio = currentThroughput / currentRate
-
- // Handle rate increases - only if we're utilizing enough of current
capacity
- if (newLimit > currentRate) {
- if (utilizationRatio < 0.9) {
- log.info(
- "Not increasing rate limiter, current utilization too low
(${utilizationRatio.format(2)})" +
- " - throughput: $currentThroughput, limit:
$currentRate",
- )
- return currentRate
- }
- } else if (newLimit < currentRate && utilizationRatio < 0.9) {
- // Handle rate decreases - avoid oscillation
- log.info(
- "Not decreasing rate limiter despite high latency - throughput
($currentThroughput) " +
- "is well below current limit ($currentRate)",
- )
- return currentRate
- }
-
- // Apply the new rate limit
log.info("Updating rate limiter from $currentRate to $newLimit")
+ return applyRate(newLimit)
+ }
+
+ /**
+ * Sets the rate limiter and records when, so the next decision waits for
metrics that describe it.
+ */
+ private fun applyRate(newLimit: Double): Double {
rateLimiter.rate = newLimit
+ lastChangeAtMs = clock()
return newLimit
}
+ /**
+ * Reports whether a full latency window has passed since the last change.
+ */
+ private fun hasSettled(): Boolean = clock() - lastChangeAtMs >= settleMs
+
/**
* Format a double to specified decimal places
*/
private fun Double.format(decimals: Int): String =
"%.${decimals}f".format(this)
/**
- * Added to prevent the rate limiter from acting when queries aren't
running, generally during populate phase
+ * Counts every operation type, so a workload of any shape can be
optimized.
+ *
+ * The optimizer needs a meaningful number of samples behind its
percentiles before it acts.
*/
- fun getTotalOperations(): Long = metrics.mutations.count +
metrics.selects.count
+ fun getTotalOperations(): Long =
+ metrics.mutations.count +
+ metrics.selects.count +
+ metrics.deletions.count +
+ metrics.populate.count
fun getCurrentTotalThroughput(): Double =
metrics.getSelectThroughput() +
@@ -239,55 +277,71 @@ class RateLimiterOptimizer(
}
/**
- * Calculates the optimal rate limit value based on current performance
metrics.
+ * Calculates the rate limit to use next, based on current performance
metrics.
*
* This implements an adaptive algorithm with three cases:
- * 1. If latency exceeds target: reduce throughput quickly (by 10%)
+ * 1. If latency exceeds target: cut back to below the throughput actually
being achieved
* 2. If within 90% of target latency: maintain current throughput to
avoid oscillation
* 3. If well below target: increase throughput proportionally to
available headroom
*
* @param currentRate The current rate limit value
+ * @param currentThroughput The throughput actually being achieved
(ops/sec)
* @param currentLatency The current observed latency (in ms)
* @param maxLatency The maximum acceptable latency (in ms)
- * @return The calculated new rate limit
+ * @return The calculated new rate limit, or currentRate to leave it alone
*/
fun getNextValue(
currentRate: Double,
+ currentThroughput: Double,
currentLatency: Double,
maxLatency: Long,
): Double {
- val maxLatencyDouble = maxLatency.toDouble()
- val latencyRatio = currentLatency / maxLatencyDouble
+ val latencyRatio = currentLatency / maxLatency.toDouble()
- // Case 1: Latency is too high - reduce throughput
+ // Case 1: Latency is too high - reduce throughput.
+ //
+ // The reduction applies to the throughput actually achieved, not to
the nominal limit.
+ // Under overload the client falls behind its own limit, so a limit
that sits above the
+ // achieved rate constrains nothing and cutting it changes nothing.
if (latencyRatio > 1.0) {
- val reductionFactor = 0.90
+ // A throughput tracker that has not warmed up yet reports zero.
Falling back to the
+ // current rate stops that from slamming the limiter down to the
floor.
+ val basis = if (currentThroughput > 0.0) minOf(currentRate,
currentThroughput) else currentRate
+ val newLimit = (basis * REDUCTION_FACTOR).coerceAtLeast(minRate)
log.info(
"Latency exceeded target: ${currentLatency.format(2)}ms >
${maxLatency}ms, " +
- "reducing throughput by ${(1 - reductionFactor) * 100}%",
+ "reducing from $currentRate to ${newLimit.format(1)} " +
+ "(achieved throughput ${currentThroughput.format(1)})",
)
- return currentRate * reductionFactor
- } else if (latencyRatio > 0.90) {
- // Case 2: Within 90% of target - maintain current throughput
+ return newLimit
+ }
+
+ // Case 2: Within the dead band below target - maintain current
throughput
+ if (latencyRatio > DEAD_BAND) {
log.info("Latency approaching target (${(latencyRatio *
100).format(1)}% of max), maintaining throughput")
return currentRate
- } else {
- // Case 3: Well below target - increase proportionally to
available headroom
- // Calculate increase factor - more aggressive for low latencies,
gentler as we approach target
- // Uses cube root to provide a non-linear response curve
- // Small latency requirements (< 10ms) will use smaller
adjustments due to sensitivity
- val maxIncreaseFactor = (1.0 + sqrt(maxLatencyDouble) /
100.0).coerceAtMost(1.05)
- val headroom = maxLatencyDouble - currentLatency
- val adjustmentFactor = (1.0 + cbrt(headroom) /
maxLatencyDouble).coerceAtMost(maxIncreaseFactor)
-
- val newLimit = currentRate * adjustmentFactor
+ }
+
+ // Case 3: Well below target - increase, but only a limit we are
already using.
+ val utilizationRatio = currentThroughput / currentRate
+ if (utilizationRatio < MIN_UTILIZATION) {
log.info(
- "Latency (${currentLatency.format(2)}ms) well below target
(${maxLatency}ms): " +
- "increasing throughput by ${((adjustmentFactor - 1) *
100).format(1)}% " +
- "from $currentRate to ${newLimit.format(1)}",
+ "Not increasing rate limiter, current utilization too low
(${utilizationRatio.format(2)})" +
+ " - throughput: $currentThroughput, limit: $currentRate",
)
- return newLimit
+ return currentRate
}
+
+ // The increase scales with the fraction of the latency budget still
unused. That fraction
+ // is dimensionless, so the response does not change when the target
does.
+ val adjustmentFactor = 1.0 + MAX_INCREASE * (1.0 - latencyRatio)
+ val newLimit = currentRate * adjustmentFactor
+ log.info(
+ "Latency (${currentLatency.format(2)}ms) well below target
(${maxLatency}ms): " +
+ "increasing throughput by ${((adjustmentFactor - 1) *
100).format(1)}% " +
+ "from $currentRate to ${newLimit.format(1)}",
+ )
+ return newLimit
}
/**
@@ -309,9 +363,10 @@ class RateLimiterOptimizer(
* Resets the optimizer to its initial state, starting the step phase
again.
* This is typically called after a populate phase completes or when
workload parameters change.
*/
+ @Synchronized
fun reset() {
log.info("Resetting rate limiter optimizer to step phase, starting
rate: $stepValue")
isStepPhase = true
- rateLimiter.rate = stepValue
+ applyRate(stepValue)
}
}
diff --git a/src/main/kotlin/org/apache/cassandra/easystress/commands/Run.kt
b/src/main/kotlin/org/apache/cassandra/easystress/commands/Run.kt
index 8b69696..b856788 100644
--- a/src/main/kotlin/org/apache/cassandra/easystress/commands/Run.kt
+++ b/src/main/kotlin/org/apache/cassandra/easystress/commands/Run.kt
@@ -471,16 +471,32 @@ class Run(
val metrics = createMetrics()
currentMetrics = metrics // Store for external access
+ // The optimizer steers the rate limiter towards a latency target.
Without a target it has
+ // nothing to steer towards, so running it would only cost the user
the ramp up.
+ val hasLatencyTarget = maxReadLatency != null || maxWriteLatency !=
null
+
+ if (useOptimizer && !hasLatencyTarget) {
+ println(
+ "No latency target set, running at a fixed rate of $rate. " +
+ "Set --max-read-latency or --max-write-latency to let the
optimizer find the rate.",
+ )
+ }
+
// set up the rate limiter optimizer and put it on a schedule
+ var optimizerTimer: Timer? = null
val optimizer =
- if (useOptimizer) {
+ if (useOptimizer && hasLatencyTarget) {
val opt = RateLimiterOptimizer(rateLimiter, metrics,
maxReadLatency, maxWriteLatency)
opt.reset()
- // Schedule the optimizer to run periodically
- Timer().schedule(10000, 5000) {
- opt.execute()
- }
+ // Schedule the optimizer to run periodically. It is a daemon
so it can never hold
+ // the JVM open, and it is cancelled in the finally block
below.
+ optimizerTimer =
+ Timer("rate-limiter-optimizer", true).apply {
+ schedule(10000, 5000) {
+ opt.execute()
+ }
+ }
opt
} else {
null
@@ -538,6 +554,7 @@ class Run(
} finally {
// we need to be able to run multiple tests in the same JVM
// without this cleanup we could have the metrics runner still
running and it will cause subsequent tests to fail
+ optimizerTimer?.cancel()
metrics.shutdown()
currentMetrics = null // Clear reference when done
collector.close(context)
diff --git a/src/test/kotlin/org/apache/cassandra/easystress/MetricsTest.kt
b/src/test/kotlin/org/apache/cassandra/easystress/MetricsTest.kt
new file mode 100644
index 0000000..5719ad6
--- /dev/null
+++ b/src/test/kotlin/org/apache/cassandra/easystress/MetricsTest.kt
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.easystress
+
+import com.codahale.metrics.MetricRegistry
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.Test
+import java.util.concurrent.TimeUnit
+
+class MetricsTest {
+ /**
+ * The rate limiter optimizer reads these percentiles every few seconds.
The Dropwizard default
+ * reservoir weights samples from about five minutes ago, which makes the
optimizer act on a
+ * signal it has already responded to.
+ *
+ * The default reservoir also caps itself at 1028 samples. A sliding time
window keeps every
+ * sample recorded inside the window, so the sample count tells the two
apart.
+ */
+ @Test
+ fun latencyPercentilesCoverOnlyTheRecentPast() {
+ val metrics = Metrics(MetricRegistry(), emptyList(), 0)
+
+ try {
+ val timers = listOf(metrics.mutations, metrics.selects,
metrics.deletions, metrics.populate)
+
+ for (timer in timers) {
+ repeat(2000) {
+ timer.update(1, TimeUnit.MILLISECONDS)
+ }
+ assertThat(timer.snapshot.size()).isEqualTo(2000)
+ }
+ } finally {
+ metrics.shutdown()
+ }
+ }
+}
diff --git
a/src/test/kotlin/org/apache/cassandra/easystress/RateLimiterOptimizerTest.kt
b/src/test/kotlin/org/apache/cassandra/easystress/RateLimiterOptimizerTest.kt
index 0c800c4..069ff54 100644
---
a/src/test/kotlin/org/apache/cassandra/easystress/RateLimiterOptimizerTest.kt
+++
b/src/test/kotlin/org/apache/cassandra/easystress/RateLimiterOptimizerTest.kt
@@ -26,6 +26,7 @@ import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.extension.ExtendWith
import java.util.Optional
+import java.util.concurrent.TimeUnit
@ExtendWith(MockKExtension::class)
class RateLimiterOptimizerTest {
@@ -76,4 +77,107 @@ class RateLimiterOptimizerTest {
val newRate = optimizer.execute()
assertThat(newRate).isGreaterThan(10.0)
}
+
+ /**
+ * Under overload the client falls behind its own limit, so the limit
constrains nothing.
+ * Cutting the limit has to bring it below what the client is actually
achieving.
+ */
+ @Test
+ fun testReducesAgainstAchievedThroughput() {
+ val maxLatency = 100L
+ val optimizer = spyk(RateLimiterOptimizer(rateLimiter, metrics,
maxLatency, maxLatency, isStepPhase = false))
+ every { optimizer.getCurrentAndMaxLatency() } returns pair(200.0,
maxLatency)
+ every { optimizer.getTotalOperations() } returns 100
+
+ // the limit is 1000, but the client is only managing 400 ops/sec
+ every { optimizer.getCurrentTotalThroughput() } returns 400.0
+
+ val newRate = optimizer.execute()
+
+ // 400 * 0.9, not 1000 * 0.9
+ assertThat(newRate).isEqualTo(360.0)
+ }
+
+ /**
+ * The throughput tracker reports zero until it has warmed up. That must
not be read as a
+ * client managing no operations at all.
+ */
+ @Test
+ fun testIgnoresAColdThroughputTracker() {
+ val maxLatency = 100L
+ val optimizer = spyk(RateLimiterOptimizer(rateLimiter, metrics,
maxLatency, maxLatency, isStepPhase = false))
+ every { optimizer.getCurrentAndMaxLatency() } returns pair(200.0,
maxLatency)
+ every { optimizer.getTotalOperations() } returns 100
+ every { optimizer.getCurrentTotalThroughput() } returns 0.0
+
+ val newRate = optimizer.execute()
+
+ // one ordinary reduction against the current rate, not a drop to the
floor
+ assertThat(newRate).isEqualTo(900.0)
+ }
+
+ /**
+ * A run of reductions must not drive the rate arbitrarily close to zero.
+ */
+ @Test
+ fun testRateHasAFloor() {
+ val maxLatency = 100L
+ val optimizer = spyk(RateLimiterOptimizer(rateLimiter, metrics,
maxLatency, maxLatency, isStepPhase = false))
+ every { optimizer.getCurrentAndMaxLatency() } returns pair(500.0,
maxLatency)
+ every { optimizer.getTotalOperations() } returns 100
+ every { optimizer.getCurrentTotalThroughput() } returns 1.0
+
+ val newRate = optimizer.execute()
+
+ // the starting rate is 1000, so the floor is 10
+ assertThat(newRate).isEqualTo(10.0)
+ }
+
+ /**
+ * Acting again before the latency window has passed means acting on the
previous rate, which
+ * compounds every adjustment.
+ */
+ @Test
+ fun testWaitsForTheLatencyWindowBeforeChangingAgain() {
+ val maxLatency = 100L
+ var now = 1_000_000L
+ val optimizer =
+ spyk(
+ RateLimiterOptimizer(
+ rateLimiter,
+ metrics,
+ maxLatency,
+ maxLatency,
+ isStepPhase = false,
+ clock = { now },
+ ),
+ )
+ every { optimizer.getCurrentAndMaxLatency() } returns pair(110.0,
maxLatency)
+ every { optimizer.getTotalOperations() } returns 100
+ every { optimizer.getCurrentTotalThroughput() } returns 1000.0
+
+ val firstRate = optimizer.execute()
+ assertThat(firstRate).isEqualTo(900.0)
+
+ // no time has passed, so the metrics still describe the old rate
+ assertThat(optimizer.execute()).isEqualTo(900.0)
+
+ // one full window later the optimizer may act again
+ now += TimeUnit.SECONDS.toMillis(Metrics.LATENCY_WINDOW_SECONDS)
+ assertThat(optimizer.execute()).isLessThan(900.0)
+ }
+
+ /**
+ * The increase depends on the fraction of the latency budget still
unused, not on the size of
+ * the budget. The same fraction has to produce the same increase at any
target.
+ */
+ @Test
+ fun testIncreaseDoesNotDependOnTheSizeOfTheTarget() {
+ val optimizer = RateLimiterOptimizer(rateLimiter, metrics, 100, 100,
isStepPhase = false)
+
+ val smallTarget = optimizer.getNextValue(1000.0, 1000.0, 10.0, 100)
+ val largeTarget = optimizer.getNextValue(1000.0, 1000.0, 100.0, 1000)
+
+ assertThat(smallTarget).isEqualTo(largeTarget)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
