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
017 package com.google.common.util.concurrent;
018
019 import com.google.common.annotations.Beta;
020 import com.google.common.annotations.VisibleForTesting;
021 import com.google.common.base.Preconditions;
022 import com.google.common.base.Ticker;
023
024 import java.util.concurrent.TimeUnit;
025
026 import 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
089 public 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 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.
225 */
226 public static RateLimiter create(double permitsPerSecond) {
227 return create(SleepingTicker.SYSTEM_TICKER, permitsPerSecond);
228 }
229
230 @VisibleForTesting
231 static RateLimiter create(SleepingTicker ticker, double permitsPerSecond) {
232 RateLimiter rateLimiter = new Bursty(ticker);
233 rateLimiter.setRate(permitsPerSecond);
234 return rateLimiter;
235 }
236
237 /**
238 * Creates a {@code RateLimiter} with the specified stable throughput, given as
239 * "permits per second" (commonly referred to as <i>QPS</i>, queries per second), and a
240 * <i>warmup period</i>, during which the {@code RateLimiter} smoothly ramps up its rate,
241 * until it reaches its maximum rate at the end of the period (as long as there are enough
242 * requests to saturate it). Similarly, if the {@code RateLimiter} is left <i>unused</i> for
243 * a duration of {@code warmupPeriod}, it will gradually return to its "cold" state,
244 * i.e. it will go through the same warming up process as when it was first created.
245 *
246 * <p>The returned {@code RateLimiter} is intended for cases where the resource that actually
247 * fulfils the requests (e.g., a remote server) needs "warmup" time, rather than
248 * being immediately accessed at the stable (maximum) rate.
249 *
250 * <p>The returned {@code RateLimiter} starts in a "cold" state (i.e. the warmup period
251 * will follow), and if it is left unused for long enough, it will return to that state.
252 *
253 * @param permitsPerSecond the rate of the returned {@code RateLimiter}, measured in
254 * how many permits become available per second
255 * @param warmupPeriod the duration of the period where the {@code RateLimiter} ramps up its
256 * rate, before reaching its stable (maximum) rate
257 * @param unit the time unit of the warmupPeriod argument
258 */
259 // TODO(user): add a burst size of 1-second-worth of permits, as in the metronome?
260 public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
261 return create(SleepingTicker.SYSTEM_TICKER, permitsPerSecond, warmupPeriod, unit);
262 }
263
264 @VisibleForTesting
265 static RateLimiter create(
266 SleepingTicker ticker, double permitsPerSecond, long warmupPeriod, TimeUnit timeUnit) {
267 RateLimiter rateLimiter = new WarmingUp(ticker, warmupPeriod, timeUnit);
268 rateLimiter.setRate(permitsPerSecond);
269 return rateLimiter;
270 }
271
272 @VisibleForTesting
273 static RateLimiter createBursty(
274 SleepingTicker ticker, double permitsPerSecond, int maxBurstSize) {
275 Bursty rateLimiter = new Bursty(ticker);
276 rateLimiter.setRate(permitsPerSecond);
277 rateLimiter.maxPermits = maxBurstSize;
278 return rateLimiter;
279 }
280
281 /**
282 * The underlying timer; used both to measure elapsed time and sleep as necessary. A separate
283 * object to facilitate testing.
284 */
285 private final SleepingTicker ticker;
286
287 /**
288 * The timestamp when the RateLimiter was created; used to avoid possible overflow/time-wrapping
289 * errors.
290 */
291 private final long offsetNanos;
292
293 /**
294 * The currently stored permits.
295 */
296 double storedPermits;
297
298 /**
299 * The maximum number of stored permits.
300 */
301 double maxPermits;
302
303 /**
304 * The interval between two unit requests, at our stable rate. E.g., a stable rate of 5 permits
305 * per second has a stable interval of 200ms.
306 */
307 double stableIntervalMicros;
308
309 /**
310 * The time when the next request (no matter its size) will be granted. After granting a request,
311 * this is pushed further in the future. Large requests push this further than small requests.
312 */
313 private long nextFreeTicketMicros = 0L; // could be either in the past or future
314
315 private RateLimiter(SleepingTicker ticker) {
316 this.ticker = ticker;
317 this.offsetNanos = ticker.read();
318 }
319
320 /**
321 * Updates the stable rate of this {@code RateLimiter}, that is, the
322 * {@code permitsPerSecond} argument provided in the factory method that
323 * constructed the {@code RateLimiter}. Currently throttled threads will <b>not</b>
324 * be awakened as a result of this invocation, thus they do not observe the new rate;
325 * only subsequent requests will.
326 *
327 * <p>Note though that, since each request repays (by waiting, if necessary) the cost
328 * of the <i>previous</i> request, this means that the very next request
329 * after an invocation to {@code setRate} will not be affected by the new rate;
330 * it will pay the cost of the previous request, which is in terms of the previous rate.
331 *
332 * <p>The behavior of the {@code RateLimiter} is not modified in any other way,
333 * e.g. if the {@code RateLimiter} was configured with a warmup period of 20 seconds,
334 * it still has a warmup period of 20 seconds after this method invocation.
335 *
336 * @param permitsPerSecond the new stable rate of this {@code RateLimiter}.
337 */
338 public final synchronized void setRate(double permitsPerSecond) {
339 Preconditions.checkArgument(permitsPerSecond > 0.0
340 && !Double.isNaN(permitsPerSecond), "rate must be positive");
341 resync(readSafeMicros());
342 double stableIntervalMicros = TimeUnit.SECONDS.toMicros(1L) / permitsPerSecond;
343 this.stableIntervalMicros = stableIntervalMicros;
344 doSetRate(permitsPerSecond, stableIntervalMicros);
345 }
346
347 abstract void doSetRate(double permitsPerSecond, double stableIntervalMicros);
348
349 /**
350 * Returns the stable rate (as {@code permits per seconds}) with which this
351 * {@code RateLimiter} is configured with. The initial value of this is the same as
352 * the {@code permitsPerSecond} argument passed in the factory method that produced
353 * this {@code RateLimiter}, and it is only updated after invocations
354 * to {@linkplain #setRate}.
355 */
356 public final synchronized double getRate() {
357 return TimeUnit.SECONDS.toMicros(1L) / stableIntervalMicros;
358 }
359
360 /**
361 * Acquires a permit from this {@code RateLimiter}, blocking until the request can be granted.
362 *
363 * <p>This method is equivalent to {@code acquire(1)}.
364 */
365 public void acquire() {
366 acquire(1);
367 }
368
369 /**
370 * Acquires the given number of permits from this {@code RateLimiter}, blocking until the
371 * request be granted.
372 *
373 * @param permits the number of permits to acquire
374 */
375 public void acquire(int permits) {
376 checkPermits(permits);
377 long microsToWait;
378 synchronized (this) {
379 microsToWait = reserveNextTicket(permits, readSafeMicros());
380 }
381 ticker.sleepMicrosUninterruptibly(microsToWait);
382 }
383
384 /**
385 * Acquires a permit from this {@code RateLimiter} if it can be obtained
386 * without exceeding the specified {@code timeout}, or returns {@code false}
387 * immediately (without waiting) if the permit would not have been granted
388 * before the timeout expired.
389 *
390 * <p>This method is equivalent to {@code tryAcquire(1, timeout, unit)}.
391 *
392 * @param timeout the maximum time to wait for the permit
393 * @param unit the time unit of the timeout argument
394 * @return {@code true} if the permit was acquired, {@code false} otherwise
395 */
396 public boolean tryAcquire(long timeout, TimeUnit unit) {
397 return tryAcquire(1, timeout, unit);
398 }
399
400 /**
401 * Acquires the given number of permits from this {@code RateLimiter} if it can be obtained
402 * without exceeding the specified {@code timeout}, or returns {@code false}
403 * immediately (without waiting) if the permits would not have been granted
404 * before the timeout expired.
405 *
406 * @param permits the number of permits to acquire
407 * @param timeout the maximum time to wait for the permits
408 * @param unit the time unit of the timeout argument
409 * @return {@code true} if the permits were acquired, {@code false} otherwise
410 */
411 public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {
412 checkPermits(permits);
413 long timeoutMicros = unit.toMicros(timeout);
414 long microsToWait;
415 synchronized (this) {
416 long nowMicros = readSafeMicros();
417 if (nextFreeTicketMicros > nowMicros + timeoutMicros) {
418 return false;
419 } else {
420 microsToWait = reserveNextTicket(permits, nowMicros);
421 }
422 }
423 ticker.sleepMicrosUninterruptibly(microsToWait);
424 return true;
425 }
426
427 private static void checkPermits(int permits) {
428 Preconditions.checkArgument(permits > 0, "Requested permits must be positive");
429 }
430
431 /**
432 * Reserves next ticket and returns the wait time that the caller must wait for.
433 */
434 private long reserveNextTicket(double requiredPermits, long nowMicros) {
435 resync(nowMicros);
436 long microsToNextFreeTicket = nextFreeTicketMicros - nowMicros;
437 double storedPermitsToSpend = Math.min(requiredPermits, this.storedPermits);
438 double freshPermits = requiredPermits - storedPermitsToSpend;
439
440 long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
441 + (long) (freshPermits * stableIntervalMicros);
442
443 this.nextFreeTicketMicros = nextFreeTicketMicros + waitMicros;
444 this.storedPermits -= storedPermitsToSpend;
445 return microsToNextFreeTicket;
446 }
447
448 /**
449 * Translates a specified portion of our currently stored permits which we want to
450 * spend/acquire, into a throttling time. Conceptually, this evaluates the integral
451 * of the underlying function we use, for the range of
452 * [(storedPermits - permitsToTake), storedPermits].
453 *
454 * This always holds: {@code 0 <= permitsToTake <= storedPermits}
455 */
456 abstract long storedPermitsToWaitTime(double storedPermits, double permitsToTake);
457
458 private void resync(long nowMicros) {
459 // if nextFreeTicket is in the past, resync to now
460 if (nowMicros > nextFreeTicketMicros) {
461 storedPermits = Math.min(maxPermits,
462 storedPermits + (nowMicros - nextFreeTicketMicros) / stableIntervalMicros);
463 nextFreeTicketMicros = nowMicros;
464 }
465 }
466
467 private long readSafeMicros() {
468 return TimeUnit.NANOSECONDS.toMicros(ticker.read() - offsetNanos);
469 }
470
471 @Override
472 public String toString() {
473 return String.format("RateLimiter[stableRate=%3.1fqps]", 1000000.0 / stableIntervalMicros);
474 }
475
476 /**
477 * This implements the following function:
478 *
479 * ^ throttling
480 * |
481 * 3*stable + /
482 * interval | /.
483 * (cold) | / .
484 * | / . <-- "warmup period" is the area of the trapezoid between
485 * 2*stable + / . halfPermits and maxPermits
486 * interval | / .
487 * | / .
488 * | / .
489 * stable +----------/ WARM . }
490 * interval | . UP . } <-- this rectangle (from 0 to maxPermits, and
491 * | . PERIOD. } height == stableInterval) defines the cooldown period,
492 * | . . } and we want cooldownPeriod == warmupPeriod
493 * |---------------------------------> storedPermits
494 * (halfPermits) (maxPermits)
495 *
496 * Before going into the details of this particular function, let's keep in mind the basics:
497 * 1) The state of the RateLimiter (storedPermits) is a vertical line in this figure.
498 * 2) When the RateLimiter is not used, this goes right (up to maxPermits)
499 * 3) When the RateLimiter is used, this goes left (down to zero), since if we have storedPermits,
500 * we serve from those first
501 * 4) When _unused_, we go right at the same speed (rate)! I.e., if our rate is
502 * 2 permits per second, and 3 unused seconds pass, we will always save 6 permits
503 * (no matter what our initial position was), up to maxPermits.
504 * If we invert the rate, we get the "stableInterval" (interval between two requests
505 * in a perfectly spaced out sequence of requests of the given rate). Thus, if you
506 * want to see "how much time it will take to go from X storedPermits to X+K storedPermits?",
507 * the answer is always stableInterval * K. In the same example, for 2 permits per second,
508 * stableInterval is 500ms. Thus to go from X storedPermits to X+6 storedPermits, we
509 * require 6 * 500ms = 3 seconds.
510 *
511 * In short, the time it takes to move to the right (save K permits) is equal to the
512 * rectangle of width == K and height == stableInterval.
513 * 4) When _used_, the time it takes, as explained in the introductory class note, is
514 * equal to the integral of our function, between X permits and X-K permits, assuming
515 * we want to spend K saved permits.
516 *
517 * In summary, the time it takes to move to the left (spend K permits), is equal to the
518 * area of the function of width == K.
519 *
520 * Let's dive into this function now:
521 *
522 * When we have storedPermits <= halfPermits (the left portion of the function), then
523 * we spend them at the exact same rate that
524 * fresh permits would be generated anyway (that rate is 1/stableInterval). We size
525 * this area to be equal to _half_ the specified warmup period. Why we need this?
526 * And why half? We'll explain shortly below (after explaining the second part).
527 *
528 * Stored permits that are beyond halfPermits, are mapped to an ascending line, that goes
529 * from stableInterval to 3 * stableInterval. The average height for that part is
530 * 2 * stableInterval, and is sized appropriately to have an area _equal_ to the
531 * specified warmup period. Thus, by point (4) above, it takes "warmupPeriod" amount of time
532 * to go from maxPermits to halfPermits.
533 *
534 * BUT, by point (3) above, it only takes "warmupPeriod / 2" amount of time to return back
535 * to maxPermits, from halfPermits! (Because the trapezoid has double the area of the rectangle
536 * of height stableInterval and equivalent width). We decided that the "cooldown period"
537 * time should be equivalent to "warmup period", thus a fully saturated RateLimiter
538 * (with zero stored permits, serving only fresh ones) can go to a fully unsaturated
539 * (with storedPermits == maxPermits) in the same amount of time it takes for a fully
540 * unsaturated RateLimiter to return to the stableInterval -- which happens in halfPermits,
541 * since beyond that point, we use a horizontal line of "stableInterval" height, simulating
542 * the regular rate.
543 *
544 * Thus, we have figured all dimensions of this shape, to give all the desired
545 * properties:
546 * - the width is warmupPeriod / stableInterval, to make cooldownPeriod == warmupPeriod
547 * - the slope starts at the middle, and goes from stableInterval to 3*stableInterval so
548 * to have halfPermits being spend in double the usual time (half the rate), while their
549 * respective rate is steadily ramping up
550 */
551 private static class WarmingUp extends RateLimiter {
552
553 final long warmupPeriodMicros;
554 /**
555 * The slope of the line from the stable interval (when permits == 0), to the cold interval
556 * (when permits == maxPermits)
557 */
558 private double slope;
559 private double halfPermits;
560
561 WarmingUp(SleepingTicker ticker, long warmupPeriod, TimeUnit timeUnit) {
562 super(ticker);
563 this.warmupPeriodMicros = timeUnit.toMicros(warmupPeriod);
564 }
565
566 @Override
567 void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
568 double oldMaxPermits = maxPermits;
569 maxPermits = warmupPeriodMicros / stableIntervalMicros;
570 halfPermits = maxPermits / 2.0;
571 // Stable interval is x, cold is 3x, so on average it's 2x. Double the time -> halve the rate
572 double coldIntervalMicros = stableIntervalMicros * 3.0;
573 slope = (coldIntervalMicros - stableIntervalMicros) / halfPermits;
574 if (oldMaxPermits == Double.POSITIVE_INFINITY) {
575 // if we don't special-case this, we would get storedPermits == NaN, below
576 storedPermits = 0.0;
577 } else {
578 storedPermits = (oldMaxPermits == 0.0)
579 ? maxPermits // initial state is cold
580 : storedPermits * maxPermits / oldMaxPermits;
581 }
582 }
583
584 @Override
585 long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
586 double availablePermitsAboveHalf = storedPermits - halfPermits;
587 long micros = 0;
588 // measuring the integral on the right part of the function (the climbing line)
589 if (availablePermitsAboveHalf > 0.0) {
590 double permitsAboveHalfToTake = Math.min(availablePermitsAboveHalf, permitsToTake);
591 micros = (long) (permitsAboveHalfToTake * (permitsToTime(availablePermitsAboveHalf)
592 + permitsToTime(availablePermitsAboveHalf - permitsAboveHalfToTake)) / 2.0);
593 permitsToTake -= permitsAboveHalfToTake;
594 }
595 // measuring the integral on the left part of the function (the horizontal line)
596 micros += (stableIntervalMicros * permitsToTake);
597 return micros;
598 }
599
600 private double permitsToTime(double permits) {
601 return stableIntervalMicros + permits * slope;
602 }
603 }
604
605 /**
606 * This implements a trivial function, where storedPermits are translated to
607 * zero throttling - thus, a client gets an infinite speedup for permits acquired out
608 * of the storedPermits pool. This is also used for the special case of the "metronome",
609 * where the width of the function is also zero; maxStoredPermits is zero, thus
610 * storedPermits and permitsToTake are always zero as well. Such a RateLimiter can
611 * not save permits when unused, thus all permits it serves are fresh, using the
612 * designated rate.
613 */
614 private static class Bursty extends RateLimiter {
615 Bursty(SleepingTicker ticker) {
616 super(ticker);
617 }
618
619 @Override
620 void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
621 double oldMaxPermits = this.maxPermits;
622 /*
623 * We allow the equivalent work of up to one second to be granted with zero waiting, if the
624 * rate limiter has been unused for as much. This is to avoid potentially producing tiny
625 * wait interval between subsequent requests for sufficiently large rates, which would
626 * unnecessarily overconstrain the thread scheduler.
627 */
628 maxPermits = permitsPerSecond; // one second worth of permits
629 storedPermits = (oldMaxPermits == 0.0)
630 ? 0.0 // initial state
631 : storedPermits * maxPermits / oldMaxPermits;
632 }
633
634 @Override
635 long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
636 return 0L;
637 }
638 }
639
640 @VisibleForTesting
641 static abstract class SleepingTicker extends Ticker {
642 abstract void sleepMicrosUninterruptibly(long micros);
643
644 static final SleepingTicker SYSTEM_TICKER = new SleepingTicker() {
645 @Override
646 public long read() {
647 return systemTicker().read();
648 }
649
650 @Override
651 public void sleepMicrosUninterruptibly(long micros) {
652 if (micros > 0) {
653 Uninterruptibles.sleepUninterruptibly(micros, TimeUnit.MICROSECONDS);
654 }
655 }
656 };
657 }
658 }