View Javadoc
1   /*
2    * Copyright (C) 2012 The Guava Authors
3    *
4    * Licensed under the Apache License, Version 2.0 (the "License");
5    * you may not use this file except in compliance with the License.
6    * You may obtain a copy of the License at
7    *
8    * http://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS,
12   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13   * See the License for the specific language governing permissions and
14   * limitations under the License.
15   */
16  
17  package com.google.common.util.concurrent;
18  
19  import com.google.common.annotations.Beta;
20  import com.google.common.annotations.VisibleForTesting;
21  import com.google.common.base.Preconditions;
22  import com.google.common.base.Ticker;
23  
24  import java.util.concurrent.TimeUnit;
25  
26  import javax.annotation.concurrent.ThreadSafe;
27  
28  /**
29   * A rate limiter. Conceptually, a rate limiter distributes permits at a
30   * configurable rate. Each {@link #acquire()} blocks if necessary until a permit is
31   * available, and then takes it. Once acquired, permits need not be released.
32   *
33   * <p>Rate limiters are often used to restrict the rate at which some
34   * physical or logical resource is accessed. This is in contrast to {@link
35   * java.util.concurrent.Semaphore} which restricts the number of concurrent
36   * accesses instead of the rate (note though that concurrency and rate are closely related,
37   * e.g. see <a href="http://en.wikipedia.org/wiki/Little's_law">Little's Law</a>).
38   *
39   * <p>A {@code RateLimiter} is defined primarily by the rate at which permits
40   * are issued. Absent additional configuration, permits will be distributed at a
41   * fixed rate, defined in terms of permits per second. Permits will be distributed
42   * smoothly, with the delay between individual permits being adjusted to ensure
43   * that the configured rate is maintained.
44   *
45   * <p>It is possible to configure a {@code RateLimiter} to have a warmup
46   * period during which time the permits issued each second steadily increases until
47   * it hits the stable rate.
48   *
49   * <p>As an example, imagine that we have a list of tasks to execute, but we don't want to
50   * submit more than 2 per second:
51   *<pre>  {@code
52   *  final RateLimiter rateLimiter = RateLimiter.create(2.0); // rate is "2 permits per second"
53   *  void submitTasks(List<Runnable> tasks, Executor executor) {
54   *    for (Runnable task : tasks) {
55   *      rateLimiter.acquire(); // may wait
56   *      executor.execute(task);
57   *    }
58   *  }
59   *}</pre>
60   *
61   * <p>As another example, imagine that we produce a stream of data, and we want to cap it
62   * at 5kb per second. This could be accomplished by requiring a permit per byte, and specifying
63   * a rate of 5000 permits per second:
64   *<pre>  {@code
65   *  final RateLimiter rateLimiter = RateLimiter.create(5000.0); // rate = 5000 permits per second
66   *  void submitPacket(byte[] packet) {
67   *    rateLimiter.acquire(packet.length);
68   *    networkService.send(packet);
69   *  }
70   *}</pre>
71   *
72   * <p>It is important to note that the number of permits requested <i>never</i>
73   * affect the throttling of the request itself (an invocation to {@code acquire(1)}
74   * and an invocation to {@code acquire(1000)} will result in exactly the same throttling, if any),
75   * but it affects the throttling of the <i>next</i> request. I.e., if an expensive task
76   * arrives at an idle RateLimiter, it will be granted immediately, but it is the <i>next</i>
77   * request that will experience extra throttling, thus paying for the cost of the expensive
78   * task.
79   *
80   * <p>Note: {@code RateLimiter} does not provide fairness guarantees.
81   *
82   * @author Dimitris Andreou
83   * @since 13.0
84   */
85  // TODO(user): switch to nano precision. A natural unit of cost is "bytes", and a micro precision
86  //     would mean a maximum rate of "1MB/s", which might be small in some cases.
87  @ThreadSafe
88  @Beta
89  public abstract class RateLimiter {
90    /*
91     * How is the RateLimiter designed, and why?
92     *
93     * The primary feature of a RateLimiter is its "stable rate", the maximum rate that
94     * is should allow at normal conditions. This is enforced by "throttling" incoming
95     * requests as needed, i.e. compute, for an incoming request, the appropriate throttle time,
96     * and make the calling thread wait as much.
97     *
98     * The simplest way to maintain a rate of QPS is to keep the timestamp of the last
99     * 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 }