001/* 002 * Copyright (C) 2012 The Guava Authors 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 */ 016 017package com.google.common.util.concurrent; 018 019import com.google.common.annotations.Beta; 020import com.google.common.annotations.VisibleForTesting; 021import com.google.common.base.Preconditions; 022import com.google.common.base.Ticker; 023 024import java.util.concurrent.TimeUnit; 025 026import javax.annotation.concurrent.ThreadSafe; 027 028/** 029 * A rate limiter. Conceptually, a rate limiter distributes permits at a 030 * configurable rate. Each {@link #acquire()} blocks if necessary until a permit is 031 * available, and then takes it. Once acquired, permits need not be released. 032 * 033 * <p>Rate limiters are often used to restrict the rate at which some 034 * physical or logical resource is accessed. This is in contrast to {@link 035 * java.util.concurrent.Semaphore} which restricts the number of concurrent 036 * accesses instead of the rate (note though that concurrency and rate are closely related, 037 * e.g. see <a href="http://en.wikipedia.org/wiki/Little's_law">Little's Law</a>). 038 * 039 * <p>A {@code RateLimiter} is defined primarily by the rate at which permits 040 * are issued. Absent additional configuration, permits will be distributed at a 041 * fixed rate, defined in terms of permits per second. Permits will be distributed 042 * smoothly, with the delay between individual permits being adjusted to ensure 043 * that the configured rate is maintained. 044 * 045 * <p>It is possible to configure a {@code RateLimiter} to have a warmup 046 * period during which time the permits issued each second steadily increases until 047 * it hits the stable rate. 048 * 049 * <p>As an example, imagine that we have a list of tasks to execute, but we don't want to 050 * submit more than 2 per second: 051 *<pre> {@code 052 * final RateLimiter rateLimiter = RateLimiter.create(2.0); // rate is "2 permits per second" 053 * void submitTasks(List<Runnable> tasks, Executor executor) { 054 * for (Runnable task : tasks) { 055 * rateLimiter.acquire(); // may wait 056 * executor.execute(task); 057 * } 058 * } 059 *}</pre> 060 * 061 * <p>As another example, imagine that we produce a stream of data, and we want to cap it 062 * at 5kb per second. This could be accomplished by requiring a permit per byte, and specifying 063 * a rate of 5000 permits per second: 064 *<pre> {@code 065 * final RateLimiter rateLimiter = RateLimiter.create(5000.0); // rate = 5000 permits per second 066 * void submitPacket(byte[] packet) { 067 * rateLimiter.acquire(packet.length); 068 * networkService.send(packet); 069 * } 070 *}</pre> 071 * 072 * <p>It is important to note that the number of permits requested <i>never</i> 073 * affect the throttling of the request itself (an invocation to {@code acquire(1)} 074 * and an invocation to {@code acquire(1000)} will result in exactly the same throttling, if any), 075 * but it affects the throttling of the <i>next</i> request. I.e., if an expensive task 076 * arrives at an idle RateLimiter, it will be granted immediately, but it is the <i>next</i> 077 * request that will experience extra throttling, thus paying for the cost of the expensive 078 * task. 079 * 080 * <p>Note: {@code RateLimiter} does not provide fairness guarantees. 081 * 082 * @author Dimitris Andreou 083 * @since 13.0 084 */ 085// TODO(user): switch to nano precision. A natural unit of cost is "bytes", and a micro precision 086// would mean a maximum rate of "1MB/s", which might be small in some cases. 087@ThreadSafe 088@Beta 089public abstract class RateLimiter { 090 /* 091 * How is the RateLimiter designed, and why? 092 * 093 * The primary feature of a RateLimiter is its "stable rate", the maximum rate that 094 * is should allow at normal conditions. This is enforced by "throttling" incoming 095 * requests as needed, i.e. compute, for an incoming request, the appropriate throttle time, 096 * and make the calling thread wait as much. 097 * 098 * The simplest way to maintain a rate of QPS is to keep the timestamp of the last 099 * granted request, and ensure that (1/QPS) seconds have elapsed since then. For example, 100 * for a rate of QPS=5 (5 tokens per second), if we ensure that a request isn't granted 101 * earlier than 200ms after the last one, then we achieve the intended rate. 102 * If a request comes and the last request was granted only 100ms ago, then we wait for 103 * another 100ms. At this rate, serving 15 fresh permits (i.e. for an acquire(15) request) 104 * naturally takes 3 seconds. 105 * 106 * It is important to realize that such a RateLimiter has a very superficial memory 107 * of the past: it only remembers the last request. What if the RateLimiter was unused for 108 * a long period of time, then a request arrived and was immediately granted? 109 * This RateLimiter would immediately forget about that past underutilization. This may 110 * result in either underutilization or overflow, depending on the real world consequences 111 * of not using the expected rate. 112 * 113 * Past underutilization could mean that excess resources are available. Then, the RateLimiter 114 * should speed up for a while, to take advantage of these resources. This is important 115 * when the rate is applied to networking (limiting bandwidth), where past underutilization 116 * typically translates to "almost empty buffers", which can be filled immediately. 117 * 118 * On the other hand, past underutilization could mean that "the server responsible for 119 * handling the request has become less ready for future requests", i.e. its caches become 120 * stale, and requests become more likely to trigger expensive operations (a more extreme 121 * case of this example is when a server has just booted, and it is mostly busy with getting 122 * itself up to speed). 123 * 124 * To deal with such scenarios, we add an extra dimension, that of "past underutilization", 125 * modeled by "storedPermits" variable. This variable is zero when there is no 126 * underutilization, and it can grow up to maxStoredPermits, for sufficiently large 127 * underutilization. So, the requested permits, by an invocation acquire(permits), 128 * are served from: 129 * - stored permits (if available) 130 * - fresh permits (for any remaining permits) 131 * 132 * How this works is best explained with an example: 133 * 134 * For a RateLimiter that produces 1 token per second, every second 135 * that goes by with the RateLimiter being unused, we increase storedPermits by 1. 136 * Say we leave the RateLimiter unused for 10 seconds (i.e., we expected a request at time 137 * X, but we are at time X + 10 seconds before a request actually arrives; this is 138 * also related to the point made in the last paragraph), thus storedPermits 139 * becomes 10.0 (assuming maxStoredPermits >= 10.0). At that point, a request of acquire(3) 140 * arrives. We serve this request out of storedPermits, and reduce that to 7.0 (how this is 141 * translated to throttling time is discussed later). Immediately after, assume that an 142 * acquire(10) request arriving. We serve the request partly from storedPermits, 143 * using all the remaining 7.0 permits, and the remaining 3.0, we serve them by fresh permits 144 * produced by the rate limiter. 145 * 146 * We already know how much time it takes to serve 3 fresh permits: if the rate is 147 * "1 token per second", then this will take 3 seconds. But what does it mean to serve 7 148 * stored permits? As explained above, there is no unique answer. If we are primarily 149 * interested to deal with underutilization, then we want stored permits to be given out 150 * /faster/ than fresh ones, because underutilization = free resources for the taking. 151 * If we are primarily interested to deal with overflow, then stored permits could 152 * be given out /slower/ than fresh ones. Thus, we require a (different in each case) 153 * function that translates storedPermits to throtting time. 154 * 155 * This role is played by storedPermitsToWaitTime(double storedPermits, double permitsToTake). 156 * The underlying model is a continuous function mapping storedPermits 157 * (from 0.0 to maxStoredPermits) onto the 1/rate (i.e. intervals) that is effective at the given 158 * storedPermits. "storedPermits" essentially measure unused time; we spend unused time 159 * buying/storing permits. Rate is "permits / time", thus "1 / rate = time / permits". 160 * Thus, "1/rate" (time / permits) times "permits" gives time, i.e., integrals on this 161 * function (which is what storedPermitsToWaitTime() computes) correspond to minimum intervals 162 * between subsequent requests, for the specified number of requested permits. 163 * 164 * Here is an example of storedPermitsToWaitTime: 165 * If storedPermits == 10.0, and we want 3 permits, we take them from storedPermits, 166 * reducing them to 7.0, and compute the throttling for these as a call to 167 * storedPermitsToWaitTime(storedPermits = 10.0, permitsToTake = 3.0), which will 168 * evaluate the integral of the function from 7.0 to 10.0. 169 * 170 * Using integrals guarantees that the effect of a single acquire(3) is equivalent 171 * to { acquire(1); acquire(1); acquire(1); }, or { acquire(2); acquire(1); }, etc, 172 * since the integral of the function in [7.0, 10.0] is equivalent to the sum of the 173 * integrals of [7.0, 8.0], [8.0, 9.0], [9.0, 10.0] (and so on), no matter 174 * what the function is. This guarantees that we handle correctly requests of varying weight 175 * (permits), /no matter/ what the actual function is - so we can tweak the latter freely. 176 * (The only requirement, obviously, is that we can compute its integrals). 177 * 178 * Note well that if, for this function, we chose a horizontal line, at height of exactly 179 * (1/QPS), then the effect of the function is non-existent: we serve storedPermits at 180 * exactly the same cost as fresh ones (1/QPS is the cost for each). We use this trick later. 181 * 182 * If we pick a function that goes /below/ that horizontal line, it means that we reduce 183 * the area of the function, thus time. Thus, the RateLimiter becomes /faster/ after a 184 * period of underutilization. If, on the other hand, we pick a function that 185 * goes /above/ that horizontal line, then it means that the area (time) is increased, 186 * thus storedPermits are more costly than fresh permits, thus the RateLimiter becomes 187 * /slower/ after a period of underutilization. 188 * 189 * Last, but not least: consider a RateLimiter with rate of 1 permit per second, currently 190 * completely unused, and an expensive acquire(100) request comes. It would be nonsensical 191 * to just wait for 100 seconds, and /then/ start the actual task. Why wait without doing 192 * anything? A much better approach is to /allow/ the request right away (as if it was an 193 * acquire(1) request instead), and postpone /subsequent/ requests as needed. In this version, 194 * we allow starting the task immediately, and postpone by 100 seconds future requests, 195 * thus we allow for work to get done in the meantime instead of waiting idly. 196 * 197 * This has important consequences: it means that the RateLimiter doesn't remember the time 198 * of the _last_ request, but it remembers the (expected) time of the _next_ request. This 199 * also enables us to tell immediately (see tryAcquire(timeout)) whether a particular 200 * timeout is enough to get us to the point of the next scheduling time, since we always 201 * maintain that. And what we mean by "an unused RateLimiter" is also defined by that 202 * notion: when we observe that the "expected arrival time of the next request" is actually 203 * in the past, then the difference (now - past) is the amount of time that the RateLimiter 204 * was formally unused, and it is that amount of time which we translate to storedPermits. 205 * (We increase storedPermits with the amount of permits that would have been produced 206 * in that idle time). So, if rate == 1 permit per second, and arrivals come exactly 207 * one second after the previous, then storedPermits is _never_ increased -- we would only 208 * increase it for arrivals _later_ than the expected one second. 209 */ 210 211 /** 212 * Creates a {@code RateLimiter} with the specified stable throughput, given as 213 * "permits per second" (commonly referred to as <i>QPS</i>, queries per second). 214 * 215 * <p>The returned {@code RateLimiter} ensures that on average no more than {@code 216 * permitsPerSecond} are issued during any given second, with sustained requests 217 * being smoothly spread over each second. When the incoming request rate exceeds 218 * {@code permitsPerSecond} the rate limiter will release one permit every {@code 219 * (1.0 / permitsPerSecond)} seconds. When the rate limiter is unused, 220 * bursts of up to {@code permitsPerSecond} permits will be allowed, with subsequent 221 * requests being smoothly limited at the stable rate of {@code permitsPerSecond}. 222 * 223 * @param permitsPerSecond the rate of the returned {@code RateLimiter}, measured in 224 * how many permits become available per second. Must be positive 225 */ 226 // TODO(user): "This is equivalent to 227 // {@code createWithCapacity(permitsPerSecond, 1, TimeUnit.SECONDS)}". 228 public static RateLimiter create(double permitsPerSecond) { 229 /* 230 * The default RateLimiter configuration can save the unused permits of up to one second. 231 * This is to avoid unnecessary stalls in situations like this: A RateLimiter of 1qps, 232 * and 4 threads, all calling acquire() at these moments: 233 * 234 * T0 at 0 seconds 235 * T1 at 1.05 seconds 236 * T2 at 2 seconds 237 * T3 at 3 seconds 238 * 239 * Due to the slight delay of T1, T2 would have to sleep till 2.05 seconds, 240 * and T3 would also have to sleep till 3.05 seconds. 241 */ 242 return create(SleepingTicker.SYSTEM_TICKER, permitsPerSecond); 243 } 244 245 @VisibleForTesting 246 static RateLimiter create(SleepingTicker ticker, double permitsPerSecond) { 247 RateLimiter rateLimiter = new Bursty(ticker, 1.0 /* maxBurstSeconds */); 248 rateLimiter.setRate(permitsPerSecond); 249 return rateLimiter; 250 } 251 252 /** 253 * Creates a {@code RateLimiter} with the specified stable throughput, given as 254 * "permits per second" (commonly referred to as <i>QPS</i>, queries per second), and a 255 * <i>warmup period</i>, during which the {@code RateLimiter} smoothly ramps up its rate, 256 * until it reaches its maximum rate at the end of the period (as long as there are enough 257 * requests to saturate it). Similarly, if the {@code RateLimiter} is left <i>unused</i> for 258 * a duration of {@code warmupPeriod}, it will gradually return to its "cold" state, 259 * i.e. it will go through the same warming up process as when it was first created. 260 * 261 * <p>The returned {@code RateLimiter} is intended for cases where the resource that actually 262 * fulfills the requests (e.g., a remote server) needs "warmup" time, rather than 263 * being immediately accessed at the stable (maximum) rate. 264 * 265 * <p>The returned {@code RateLimiter} starts in a "cold" state (i.e. the warmup period 266 * will follow), and if it is left unused for long enough, it will return to that state. 267 * 268 * @param permitsPerSecond the rate of the returned {@code RateLimiter}, measured in 269 * how many permits become available per second. Must be positive 270 * @param warmupPeriod the duration of the period where the {@code RateLimiter} ramps up its 271 * rate, before reaching its stable (maximum) rate 272 * @param unit the time unit of the warmupPeriod argument 273 */ 274 public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit) { 275 return create(SleepingTicker.SYSTEM_TICKER, permitsPerSecond, warmupPeriod, unit); 276 } 277 278 @VisibleForTesting 279 static RateLimiter create( 280 SleepingTicker ticker, double permitsPerSecond, long warmupPeriod, TimeUnit unit) { 281 RateLimiter rateLimiter = new WarmingUp(ticker, warmupPeriod, unit); 282 rateLimiter.setRate(permitsPerSecond); 283 return rateLimiter; 284 } 285 286 @VisibleForTesting 287 static RateLimiter createWithCapacity( 288 SleepingTicker ticker, double permitsPerSecond, long maxBurstBuildup, TimeUnit unit) { 289 double maxBurstSeconds = unit.toNanos(maxBurstBuildup) / 1E+9; 290 Bursty rateLimiter = new Bursty(ticker, maxBurstSeconds); 291 rateLimiter.setRate(permitsPerSecond); 292 return rateLimiter; 293 } 294 295 /** 296 * The underlying timer; used both to measure elapsed time and sleep as necessary. A separate 297 * object to facilitate testing. 298 */ 299 private final SleepingTicker ticker; 300 301 /** 302 * The timestamp when the RateLimiter was created; used to avoid possible overflow/time-wrapping 303 * errors. 304 */ 305 private final long offsetNanos; 306 307 /** 308 * The currently stored permits. 309 */ 310 double storedPermits; 311 312 /** 313 * The maximum number of stored permits. 314 */ 315 double maxPermits; 316 317 /** 318 * The interval between two unit requests, at our stable rate. E.g., a stable rate of 5 permits 319 * per second has a stable interval of 200ms. 320 */ 321 volatile double stableIntervalMicros; 322 323 private final Object mutex = new Object(); 324 325 /** 326 * The time when the next request (no matter its size) will be granted. After granting a request, 327 * this is pushed further in the future. Large requests push this further than small requests. 328 */ 329 private long nextFreeTicketMicros = 0L; // could be either in the past or future 330 331 private RateLimiter(SleepingTicker ticker) { 332 this.ticker = ticker; 333 this.offsetNanos = ticker.read(); 334 } 335 336 /** 337 * Updates the stable rate of this {@code RateLimiter}, that is, the 338 * {@code permitsPerSecond} argument provided in the factory method that 339 * constructed the {@code RateLimiter}. Currently throttled threads will <b>not</b> 340 * be awakened as a result of this invocation, thus they do not observe the new rate; 341 * only subsequent requests will. 342 * 343 * <p>Note though that, since each request repays (by waiting, if necessary) the cost 344 * of the <i>previous</i> request, this means that the very next request 345 * after an invocation to {@code setRate} will not be affected by the new rate; 346 * it will pay the cost of the previous request, which is in terms of the previous rate. 347 * 348 * <p>The behavior of the {@code RateLimiter} is not modified in any other way, 349 * e.g. if the {@code RateLimiter} was configured with a warmup period of 20 seconds, 350 * it still has a warmup period of 20 seconds after this method invocation. 351 * 352 * @param permitsPerSecond the new stable rate of this {@code RateLimiter}. Must be positive 353 */ 354 public final void setRate(double permitsPerSecond) { 355 Preconditions.checkArgument(permitsPerSecond > 0.0 356 && !Double.isNaN(permitsPerSecond), "rate must be positive"); 357 synchronized (mutex) { 358 resync(readSafeMicros()); 359 double stableIntervalMicros = TimeUnit.SECONDS.toMicros(1L) / permitsPerSecond; 360 this.stableIntervalMicros = stableIntervalMicros; 361 doSetRate(permitsPerSecond, stableIntervalMicros); 362 } 363 } 364 365 abstract void doSetRate(double permitsPerSecond, double stableIntervalMicros); 366 367 /** 368 * Returns the stable rate (as {@code permits per seconds}) with which this 369 * {@code RateLimiter} is configured with. The initial value of this is the same as 370 * the {@code permitsPerSecond} argument passed in the factory method that produced 371 * this {@code RateLimiter}, and it is only updated after invocations 372 * to {@linkplain #setRate}. 373 */ 374 public final double getRate() { 375 return TimeUnit.SECONDS.toMicros(1L) / stableIntervalMicros; 376 } 377 378 /** 379 * Acquires a single permit from this {@code RateLimiter}, blocking until the 380 * request can be granted. Tells the amount of time slept, if any. 381 * 382 * <p>This method is equivalent to {@code acquire(1)}. 383 * 384 * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited 385 * @since 16.0 (present in 13.0 with {@code void} return type}) 386 */ 387 public double acquire() { 388 return acquire(1); 389 } 390 391 /** 392 * Acquires the given number of permits from this {@code RateLimiter}, blocking until the 393 * request can be granted. Tells the amount of time slept, if any. 394 * 395 * @param permits the number of permits to acquire 396 * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited 397 * @since 16.0 (present in 13.0 with {@code void} return type}) 398 */ 399 public double acquire(int permits) { 400 long microsToWait = reserve(permits); 401 ticker.sleepMicrosUninterruptibly(microsToWait); 402 return 1.0 * microsToWait / TimeUnit.SECONDS.toMicros(1L); 403 } 404 405 /** 406 * Reserves a single permit from this {@code RateLimiter} for future use, returning the number of 407 * microseconds until the reservation. 408 * 409 * <p>This method is equivalent to {@code reserve(1)}. 410 * 411 * @return time in microseconds to wait until the resource can be acquired. 412 */ 413 long reserve() { 414 return reserve(1); 415 } 416 417 /** 418 * Reserves the given number of permits from this {@code RateLimiter} for future use, returning 419 * the number of microseconds until the reservation can be consumed. 420 * 421 * @return time in microseconds to wait until the resource can be acquired. 422 */ 423 long reserve(int permits) { 424 checkPermits(permits); 425 synchronized (mutex) { 426 return reserveNextTicket(permits, readSafeMicros()); 427 } 428 } 429 430 /** 431 * Acquires a permit from this {@code RateLimiter} if it can be obtained 432 * without exceeding the specified {@code timeout}, or returns {@code false} 433 * immediately (without waiting) if the permit would not have been granted 434 * before the timeout expired. 435 * 436 * <p>This method is equivalent to {@code tryAcquire(1, timeout, unit)}. 437 * 438 * @param timeout the maximum time to wait for the permit 439 * @param unit the time unit of the timeout argument 440 * @return {@code true} if the permit was acquired, {@code false} otherwise 441 */ 442 public boolean tryAcquire(long timeout, TimeUnit unit) { 443 return tryAcquire(1, timeout, unit); 444 } 445 446 /** 447 * Acquires permits from this {@link RateLimiter} if it can be acquired immediately without delay. 448 * 449 * <p> 450 * This method is equivalent to {@code tryAcquire(permits, 0, anyUnit)}. 451 * 452 * @param permits the number of permits to acquire 453 * @return {@code true} if the permits were acquired, {@code false} otherwise 454 * @since 14.0 455 */ 456 public boolean tryAcquire(int permits) { 457 return tryAcquire(permits, 0, TimeUnit.MICROSECONDS); 458 } 459 460 /** 461 * Acquires a permit from this {@link RateLimiter} if it can be acquired immediately without 462 * delay. 463 * 464 * <p> 465 * This method is equivalent to {@code tryAcquire(1)}. 466 * 467 * @return {@code true} if the permit was acquired, {@code false} otherwise 468 * @since 14.0 469 */ 470 public boolean tryAcquire() { 471 return tryAcquire(1, 0, TimeUnit.MICROSECONDS); 472 } 473 474 /** 475 * Acquires the given number of permits from this {@code RateLimiter} if it can be obtained 476 * without exceeding the specified {@code timeout}, or returns {@code false} 477 * immediately (without waiting) if the permits would not have been granted 478 * before the timeout expired. 479 * 480 * @param permits the number of permits to acquire 481 * @param timeout the maximum time to wait for the permits 482 * @param unit the time unit of the timeout argument 483 * @return {@code true} if the permits were acquired, {@code false} otherwise 484 */ 485 public boolean tryAcquire(int permits, long timeout, TimeUnit unit) { 486 long timeoutMicros = unit.toMicros(timeout); 487 checkPermits(permits); 488 long microsToWait; 489 synchronized (mutex) { 490 long nowMicros = readSafeMicros(); 491 if (nextFreeTicketMicros > nowMicros + timeoutMicros) { 492 return false; 493 } else { 494 microsToWait = reserveNextTicket(permits, nowMicros); 495 } 496 } 497 ticker.sleepMicrosUninterruptibly(microsToWait); 498 return true; 499 } 500 501 private static void checkPermits(int permits) { 502 Preconditions.checkArgument(permits > 0, "Requested permits must be positive"); 503 } 504 505 /** 506 * Reserves next ticket and returns the wait time that the caller must wait for. 507 * 508 * <p>The return value is guaranteed to be non-negative. 509 */ 510 private long reserveNextTicket(double requiredPermits, long nowMicros) { 511 resync(nowMicros); 512 long microsToNextFreeTicket = Math.max(0, nextFreeTicketMicros - nowMicros); 513 double storedPermitsToSpend = Math.min(requiredPermits, this.storedPermits); 514 double freshPermits = requiredPermits - storedPermitsToSpend; 515 516 long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend) 517 + (long) (freshPermits * stableIntervalMicros); 518 519 this.nextFreeTicketMicros = nextFreeTicketMicros + waitMicros; 520 this.storedPermits -= storedPermitsToSpend; 521 return microsToNextFreeTicket; 522 } 523 524 /** 525 * Translates a specified portion of our currently stored permits which we want to 526 * spend/acquire, into a throttling time. Conceptually, this evaluates the integral 527 * of the underlying function we use, for the range of 528 * [(storedPermits - permitsToTake), storedPermits]. 529 * 530 * This always holds: {@code 0 <= permitsToTake <= storedPermits} 531 */ 532 abstract long storedPermitsToWaitTime(double storedPermits, double permitsToTake); 533 534 private void resync(long nowMicros) { 535 // if nextFreeTicket is in the past, resync to now 536 if (nowMicros > nextFreeTicketMicros) { 537 storedPermits = Math.min(maxPermits, 538 storedPermits + (nowMicros - nextFreeTicketMicros) / stableIntervalMicros); 539 nextFreeTicketMicros = nowMicros; 540 } 541 } 542 543 private long readSafeMicros() { 544 return TimeUnit.NANOSECONDS.toMicros(ticker.read() - offsetNanos); 545 } 546 547 @Override 548 public String toString() { 549 return String.format("RateLimiter[stableRate=%3.1fqps]", 1000000.0 / stableIntervalMicros); 550 } 551 552 /** 553 * This implements the following function: 554 * 555 * ^ throttling 556 * | 557 * 3*stable + / 558 * interval | /. 559 * (cold) | / . 560 * | / . <-- "warmup period" is the area of the trapezoid between 561 * 2*stable + / . halfPermits and maxPermits 562 * interval | / . 563 * | / . 564 * | / . 565 * stable +----------/ WARM . } 566 * interval | . UP . } <-- this rectangle (from 0 to maxPermits, and 567 * | . PERIOD. } height == stableInterval) defines the cooldown period, 568 * | . . } and we want cooldownPeriod == warmupPeriod 569 * |---------------------------------> storedPermits 570 * (halfPermits) (maxPermits) 571 * 572 * Before going into the details of this particular function, let's keep in mind the basics: 573 * 1) The state of the RateLimiter (storedPermits) is a vertical line in this figure. 574 * 2) When the RateLimiter is not used, this goes right (up to maxPermits) 575 * 3) When the RateLimiter is used, this goes left (down to zero), since if we have storedPermits, 576 * we serve from those first 577 * 4) When _unused_, we go right at the same speed (rate)! I.e., if our rate is 578 * 2 permits per second, and 3 unused seconds pass, we will always save 6 permits 579 * (no matter what our initial position was), up to maxPermits. 580 * If we invert the rate, we get the "stableInterval" (interval between two requests 581 * in a perfectly spaced out sequence of requests of the given rate). Thus, if you 582 * want to see "how much time it will take to go from X storedPermits to X+K storedPermits?", 583 * the answer is always stableInterval * K. In the same example, for 2 permits per second, 584 * stableInterval is 500ms. Thus to go from X storedPermits to X+6 storedPermits, we 585 * require 6 * 500ms = 3 seconds. 586 * 587 * In short, the time it takes to move to the right (save K permits) is equal to the 588 * rectangle of width == K and height == stableInterval. 589 * 4) When _used_, the time it takes, as explained in the introductory class note, is 590 * equal to the integral of our function, between X permits and X-K permits, assuming 591 * we want to spend K saved permits. 592 * 593 * In summary, the time it takes to move to the left (spend K permits), is equal to the 594 * area of the function of width == K. 595 * 596 * Let's dive into this function now: 597 * 598 * When we have storedPermits <= halfPermits (the left portion of the function), then 599 * we spend them at the exact same rate that 600 * fresh permits would be generated anyway (that rate is 1/stableInterval). We size 601 * this area to be equal to _half_ the specified warmup period. Why we need this? 602 * And why half? We'll explain shortly below (after explaining the second part). 603 * 604 * Stored permits that are beyond halfPermits, are mapped to an ascending line, that goes 605 * from stableInterval to 3 * stableInterval. The average height for that part is 606 * 2 * stableInterval, and is sized appropriately to have an area _equal_ to the 607 * specified warmup period. Thus, by point (4) above, it takes "warmupPeriod" amount of time 608 * to go from maxPermits to halfPermits. 609 * 610 * BUT, by point (3) above, it only takes "warmupPeriod / 2" amount of time to return back 611 * to maxPermits, from halfPermits! (Because the trapezoid has double the area of the rectangle 612 * of height stableInterval and equivalent width). We decided that the "cooldown period" 613 * time should be equivalent to "warmup period", thus a fully saturated RateLimiter 614 * (with zero stored permits, serving only fresh ones) can go to a fully unsaturated 615 * (with storedPermits == maxPermits) in the same amount of time it takes for a fully 616 * unsaturated RateLimiter to return to the stableInterval -- which happens in halfPermits, 617 * since beyond that point, we use a horizontal line of "stableInterval" height, simulating 618 * the regular rate. 619 * 620 * Thus, we have figured all dimensions of this shape, to give all the desired 621 * properties: 622 * - the width is warmupPeriod / stableInterval, to make cooldownPeriod == warmupPeriod 623 * - the slope starts at the middle, and goes from stableInterval to 3*stableInterval so 624 * to have halfPermits being spend in double the usual time (half the rate), while their 625 * respective rate is steadily ramping up 626 */ 627 private static class WarmingUp extends RateLimiter { 628 629 final long warmupPeriodMicros; 630 /** 631 * The slope of the line from the stable interval (when permits == 0), to the cold interval 632 * (when permits == maxPermits) 633 */ 634 private double slope; 635 private double halfPermits; 636 637 WarmingUp(SleepingTicker ticker, long warmupPeriod, TimeUnit timeUnit) { 638 super(ticker); 639 this.warmupPeriodMicros = timeUnit.toMicros(warmupPeriod); 640 } 641 642 @Override 643 void doSetRate(double permitsPerSecond, double stableIntervalMicros) { 644 double oldMaxPermits = maxPermits; 645 maxPermits = warmupPeriodMicros / stableIntervalMicros; 646 halfPermits = maxPermits / 2.0; 647 // Stable interval is x, cold is 3x, so on average it's 2x. Double the time -> halve the rate 648 double coldIntervalMicros = stableIntervalMicros * 3.0; 649 slope = (coldIntervalMicros - stableIntervalMicros) / halfPermits; 650 if (oldMaxPermits == Double.POSITIVE_INFINITY) { 651 // if we don't special-case this, we would get storedPermits == NaN, below 652 storedPermits = 0.0; 653 } else { 654 storedPermits = (oldMaxPermits == 0.0) 655 ? maxPermits // initial state is cold 656 : storedPermits * maxPermits / oldMaxPermits; 657 } 658 } 659 660 @Override 661 long storedPermitsToWaitTime(double storedPermits, double permitsToTake) { 662 double availablePermitsAboveHalf = storedPermits - halfPermits; 663 long micros = 0; 664 // measuring the integral on the right part of the function (the climbing line) 665 if (availablePermitsAboveHalf > 0.0) { 666 double permitsAboveHalfToTake = Math.min(availablePermitsAboveHalf, permitsToTake); 667 micros = (long) (permitsAboveHalfToTake * (permitsToTime(availablePermitsAboveHalf) 668 + permitsToTime(availablePermitsAboveHalf - permitsAboveHalfToTake)) / 2.0); 669 permitsToTake -= permitsAboveHalfToTake; 670 } 671 // measuring the integral on the left part of the function (the horizontal line) 672 micros += (stableIntervalMicros * permitsToTake); 673 return micros; 674 } 675 676 private double permitsToTime(double permits) { 677 return stableIntervalMicros + permits * slope; 678 } 679 } 680 681 /** 682 * This implements a "bursty" RateLimiter, where storedPermits are translated to 683 * zero throttling. The maximum number of permits that can be saved (when the RateLimiter is 684 * unused) is defined in terms of time, in this sense: if a RateLimiter is 2qps, and this 685 * time is specified as 10 seconds, we can save up to 2 * 10 = 20 permits. 686 */ 687 private static class Bursty extends RateLimiter { 688 /** The work (permits) of how many seconds can be saved up if this RateLimiter is unused? */ 689 final double maxBurstSeconds; 690 691 Bursty(SleepingTicker ticker, double maxBurstSeconds) { 692 super(ticker); 693 this.maxBurstSeconds = maxBurstSeconds; 694 } 695 696 @Override 697 void doSetRate(double permitsPerSecond, double stableIntervalMicros) { 698 double oldMaxPermits = this.maxPermits; 699 maxPermits = maxBurstSeconds * permitsPerSecond; 700 storedPermits = (oldMaxPermits == 0.0) 701 ? 0.0 // initial state 702 : storedPermits * maxPermits / oldMaxPermits; 703 } 704 705 @Override 706 long storedPermitsToWaitTime(double storedPermits, double permitsToTake) { 707 return 0L; 708 } 709 } 710 711 @VisibleForTesting 712 static abstract class SleepingTicker extends Ticker { 713 abstract void sleepMicrosUninterruptibly(long micros); 714 715 static final SleepingTicker SYSTEM_TICKER = new SleepingTicker() { 716 @Override 717 public long read() { 718 return systemTicker().read(); 719 } 720 721 @Override 722 public void sleepMicrosUninterruptibly(long micros) { 723 if (micros > 0) { 724 Uninterruptibles.sleepUninterruptibly(micros, TimeUnit.MICROSECONDS); 725 } 726 } 727 }; 728 } 729}