View Javadoc
1   /*
2    * Copyright (C) 2012 The Guava Authors
3    *
4    * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
5    * in compliance with the License. You may obtain a copy of the License at
6    *
7    * http://www.apache.org/licenses/LICENSE-2.0
8    *
9    * Unless required by applicable law or agreed to in writing, software distributed under the License
10   * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
11   * or implied. See the License for the specific language governing permissions and limitations under
12   * the License.
13   */
14  
15  package com.google.common.util.concurrent;
16  
17  import static com.google.common.base.Preconditions.checkArgument;
18  import static com.google.common.base.Preconditions.checkNotNull;
19  import static java.lang.Math.max;
20  import static java.util.concurrent.TimeUnit.MICROSECONDS;
21  import static java.util.concurrent.TimeUnit.SECONDS;
22  
23  import com.google.common.annotations.Beta;
24  import com.google.common.annotations.GwtIncompatible;
25  import com.google.common.annotations.VisibleForTesting;
26  import com.google.common.base.Stopwatch;
27  import com.google.common.util.concurrent.SmoothRateLimiter.SmoothBursty;
28  import com.google.common.util.concurrent.SmoothRateLimiter.SmoothWarmingUp;
29  import com.google.errorprone.annotations.CanIgnoreReturnValue;
30  import java.util.Locale;
31  import java.util.concurrent.TimeUnit;
32  import javax.annotation.concurrent.ThreadSafe;
33  
34  /**
35   * A rate limiter. Conceptually, a rate limiter distributes permits at a configurable rate. Each
36   * {@link #acquire()} blocks if necessary until a permit is available, and then takes it. Once
37   * acquired, permits need not be released.
38   *
39   * <p>Rate limiters are often used to restrict the rate at which some physical or logical resource
40   * is accessed. This is in contrast to {@link java.util.concurrent.Semaphore} which restricts the
41   * number of concurrent accesses instead of the rate (note though that concurrency and rate are
42   * closely related, e.g. see <a href="http://en.wikipedia.org/wiki/Little%27s_law">Little's
43   * Law</a>).
44   *
45   * <p>A {@code RateLimiter} is defined primarily by the rate at which permits are issued. Absent
46   * additional configuration, permits will be distributed at a fixed rate, defined in terms of
47   * permits per second. Permits will be distributed smoothly, with the delay between individual
48   * permits being adjusted to ensure that the configured rate is maintained.
49   *
50   * <p>It is possible to configure a {@code RateLimiter} to have a warmup period during which time
51   * the permits issued each second steadily increases until it hits the stable rate.
52   *
53   * <p>As an example, imagine that we have a list of tasks to execute, but we don't want to submit
54   * more than 2 per second: <pre>   {@code
55   *  final RateLimiter rateLimiter = RateLimiter.create(2.0); // rate is "2 permits per second"
56   *  void submitTasks(List<Runnable> tasks, Executor executor) {
57   *    for (Runnable task : tasks) {
58   *      rateLimiter.acquire(); // may wait
59   *      executor.execute(task);
60   *    }
61   *  }}</pre>
62   *
63   * <p>As another example, imagine that we produce a stream of data, and we want to cap it at 5kb per
64   * second. This could be accomplished by requiring a permit per byte, and specifying a rate of 5000
65   * permits per second: <pre>   {@code
66   *  final RateLimiter rateLimiter = RateLimiter.create(5000.0); // rate = 5000 permits per second
67   *  void submitPacket(byte[] packet) {
68   *    rateLimiter.acquire(packet.length);
69   *    networkService.send(packet);
70   *  }}</pre>
71   *
72   * <p>It is important to note that the number of permits requested <i>never</i> affects the
73   * throttling of the request itself (an invocation to {@code acquire(1)} and an invocation to
74   * {@code acquire(1000)} will result in exactly the same throttling, if any), but it affects the
75   * throttling of the <i>next</i> request. I.e., if an expensive task arrives at an idle RateLimiter,
76   * it will be granted immediately, but it is the <i>next</i> request that will experience extra
77   * throttling, thus paying for the cost of the expensive task.
78   *
79   * <p>Note: {@code RateLimiter} does not provide fairness guarantees.
80   *
81   * @author Dimitris Andreou
82   * @since 13.0
83   */
84  // TODO(user): switch to nano precision. A natural unit of cost is "bytes", and a micro precision
85  // would mean a maximum rate of "1MB/s", which might be small in some cases.
86  @ThreadSafe
87  @Beta
88  @GwtIncompatible
89  public abstract class RateLimiter {
90    /**
91     * Creates a {@code RateLimiter} with the specified stable throughput, given as
92     * "permits per second" (commonly referred to as <i>QPS</i>, queries per second).
93     *
94     * <p>The returned {@code RateLimiter} ensures that on average no more than {@code
95     * permitsPerSecond} are issued during any given second, with sustained requests being smoothly
96     * spread over each second. When the incoming request rate exceeds {@code permitsPerSecond} the
97     * rate limiter will release one permit every {@code
98     * (1.0 / permitsPerSecond)} seconds. When the rate limiter is unused, bursts of up to
99     * {@code permitsPerSecond} permits will be allowed, with subsequent requests being smoothly
100    * limited at the stable rate of {@code permitsPerSecond}.
101    *
102    * @param permitsPerSecond the rate of the returned {@code RateLimiter}, measured in how many
103    *     permits become available per second
104    * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero
105    */
106   // TODO(user): "This is equivalent to
107   // {@code createWithCapacity(permitsPerSecond, 1, TimeUnit.SECONDS)}".
108   public static RateLimiter create(double permitsPerSecond) {
109     /*
110      * The default RateLimiter configuration can save the unused permits of up to one second. This
111      * is to avoid unnecessary stalls in situations like this: A RateLimiter of 1qps, and 4 threads,
112      * all calling acquire() at these moments:
113      *
114      * T0 at 0 seconds
115      * T1 at 1.05 seconds
116      * T2 at 2 seconds
117      * T3 at 3 seconds
118      *
119      * Due to the slight delay of T1, T2 would have to sleep till 2.05 seconds, and T3 would also
120      * have to sleep till 3.05 seconds.
121      */
122     return create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer());
123   }
124 
125   @VisibleForTesting
126   static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {
127     RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
128     rateLimiter.setRate(permitsPerSecond);
129     return rateLimiter;
130   }
131 
132   /**
133    * Creates a {@code RateLimiter} with the specified stable throughput, given as
134    * "permits per second" (commonly referred to as <i>QPS</i>, queries per second), and a <i>warmup
135    * period</i>, during which the {@code RateLimiter} smoothly ramps up its rate, until it reaches
136    * its maximum rate at the end of the period (as long as there are enough requests to saturate
137    * it). Similarly, if the {@code RateLimiter} is left <i>unused</i> for a duration of
138    * {@code warmupPeriod}, it will gradually return to its "cold" state, i.e. it will go through the
139    * same warming up process as when it was first created.
140    *
141    * <p>The returned {@code RateLimiter} is intended for cases where the resource that actually
142    * fulfills the requests (e.g., a remote server) needs "warmup" time, rather than being
143    * immediately accessed at the stable (maximum) rate.
144    *
145    * <p>The returned {@code RateLimiter} starts in a "cold" state (i.e. the warmup period will
146    * follow), and if it is left unused for long enough, it will return to that state.
147    *
148    * @param permitsPerSecond the rate of the returned {@code RateLimiter}, measured in how many
149    *     permits become available per second
150    * @param warmupPeriod the duration of the period where the {@code RateLimiter} ramps up its rate,
151    *     before reaching its stable (maximum) rate
152    * @param unit the time unit of the warmupPeriod argument
153    * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero or
154    *     {@code warmupPeriod} is negative
155    */
156   public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
157     checkArgument(warmupPeriod >= 0, "warmupPeriod must not be negative: %s", warmupPeriod);
158     return create(
159         permitsPerSecond, warmupPeriod, unit, 3.0, SleepingStopwatch.createFromSystemTimer());
160   }
161 
162   @VisibleForTesting
163   static RateLimiter create(
164       double permitsPerSecond,
165       long warmupPeriod,
166       TimeUnit unit,
167       double coldFactor,
168       SleepingStopwatch stopwatch) {
169     RateLimiter rateLimiter = new SmoothWarmingUp(stopwatch, warmupPeriod, unit, coldFactor);
170     rateLimiter.setRate(permitsPerSecond);
171     return rateLimiter;
172   }
173 
174   /**
175    * The underlying timer; used both to measure elapsed time and sleep as necessary. A separate
176    * object to facilitate testing.
177    */
178   private final SleepingStopwatch stopwatch;
179 
180   // Can't be initialized in the constructor because mocks don't call the constructor.
181   private volatile Object mutexDoNotUseDirectly;
182 
183   private Object mutex() {
184     Object mutex = mutexDoNotUseDirectly;
185     if (mutex == null) {
186       synchronized (this) {
187         mutex = mutexDoNotUseDirectly;
188         if (mutex == null) {
189           mutexDoNotUseDirectly = mutex = new Object();
190         }
191       }
192     }
193     return mutex;
194   }
195 
196   RateLimiter(SleepingStopwatch stopwatch) {
197     this.stopwatch = checkNotNull(stopwatch);
198   }
199 
200   /**
201    * Updates the stable rate of this {@code RateLimiter}, that is, the {@code permitsPerSecond}
202    * argument provided in the factory method that constructed the {@code RateLimiter}. Currently
203    * throttled threads will <b>not</b> be awakened as a result of this invocation, thus they do not
204    * observe the new rate; only subsequent requests will.
205    *
206    * <p>Note though that, since each request repays (by waiting, if necessary) the cost of the
207    * <i>previous</i> request, this means that the very next request after an invocation to
208    * {@code setRate} will not be affected by the new rate; it will pay the cost of the previous
209    * request, which is in terms of the previous rate.
210    *
211    * <p>The behavior of the {@code RateLimiter} is not modified in any other way, e.g. if the
212    * {@code RateLimiter} was configured with a warmup period of 20 seconds, it still has a warmup
213    * period of 20 seconds after this method invocation.
214    *
215    * @param permitsPerSecond the new stable rate of this {@code RateLimiter}
216    * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero
217    */
218   public final void setRate(double permitsPerSecond) {
219     checkArgument(
220         permitsPerSecond > 0.0 && !Double.isNaN(permitsPerSecond), "rate must be positive");
221     synchronized (mutex()) {
222       doSetRate(permitsPerSecond, stopwatch.readMicros());
223     }
224   }
225 
226   abstract void doSetRate(double permitsPerSecond, long nowMicros);
227 
228   /**
229    * Returns the stable rate (as {@code permits per seconds}) with which this {@code RateLimiter} is
230    * configured with. The initial value of this is the same as the {@code permitsPerSecond} argument
231    * passed in the factory method that produced this {@code RateLimiter}, and it is only updated
232    * after invocations to {@linkplain #setRate}.
233    */
234   public final double getRate() {
235     synchronized (mutex()) {
236       return doGetRate();
237     }
238   }
239 
240   abstract double doGetRate();
241 
242   /**
243    * Acquires a single permit from this {@code RateLimiter}, blocking until the request can be
244    * granted. Tells the amount of time slept, if any.
245    *
246    * <p>This method is equivalent to {@code acquire(1)}.
247    *
248    * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited
249    * @since 16.0 (present in 13.0 with {@code void} return type})
250    */
251   @CanIgnoreReturnValue
252   public double acquire() {
253     return acquire(1);
254   }
255 
256   /**
257    * Acquires the given number of permits from this {@code RateLimiter}, blocking until the request
258    * can be granted. Tells the amount of time slept, if any.
259    *
260    * @param permits the number of permits to acquire
261    * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited
262    * @throws IllegalArgumentException if the requested number of permits is negative or zero
263    * @since 16.0 (present in 13.0 with {@code void} return type})
264    */
265   @CanIgnoreReturnValue
266   public double acquire(int permits) {
267     long microsToWait = reserve(permits);
268     stopwatch.sleepMicrosUninterruptibly(microsToWait);
269     return 1.0 * microsToWait / SECONDS.toMicros(1L);
270   }
271 
272   /**
273    * Reserves the given number of permits from this {@code RateLimiter} for future use, returning
274    * the number of microseconds until the reservation can be consumed.
275    *
276    * @return time in microseconds to wait until the resource can be acquired, never negative
277    */
278   final long reserve(int permits) {
279     checkPermits(permits);
280     synchronized (mutex()) {
281       return reserveAndGetWaitLength(permits, stopwatch.readMicros());
282     }
283   }
284 
285   /**
286    * Acquires a permit from this {@code RateLimiter} if it can be obtained without exceeding the
287    * specified {@code timeout}, or returns {@code false} immediately (without waiting) if the permit
288    * would not have been granted before the timeout expired.
289    *
290    * <p>This method is equivalent to {@code tryAcquire(1, timeout, unit)}.
291    *
292    * @param timeout the maximum time to wait for the permit. Negative values are treated as zero.
293    * @param unit the time unit of the timeout argument
294    * @return {@code true} if the permit was acquired, {@code false} otherwise
295    * @throws IllegalArgumentException if the requested number of permits is negative or zero
296    */
297   public boolean tryAcquire(long timeout, TimeUnit unit) {
298     return tryAcquire(1, timeout, unit);
299   }
300 
301   /**
302    * Acquires permits from this {@link RateLimiter} if it can be acquired immediately without delay.
303    *
304    * <p>This method is equivalent to {@code tryAcquire(permits, 0, anyUnit)}.
305    *
306    * @param permits the number of permits to acquire
307    * @return {@code true} if the permits were acquired, {@code false} otherwise
308    * @throws IllegalArgumentException if the requested number of permits is negative or zero
309    * @since 14.0
310    */
311   public boolean tryAcquire(int permits) {
312     return tryAcquire(permits, 0, MICROSECONDS);
313   }
314 
315   /**
316    * Acquires a permit from this {@link RateLimiter} if it can be acquired immediately without
317    * delay.
318    *
319    * <p>This method is equivalent to {@code tryAcquire(1)}.
320    *
321    * @return {@code true} if the permit was acquired, {@code false} otherwise
322    * @since 14.0
323    */
324   public boolean tryAcquire() {
325     return tryAcquire(1, 0, MICROSECONDS);
326   }
327 
328   /**
329    * Acquires the given number of permits from this {@code RateLimiter} if it can be obtained
330    * without exceeding the specified {@code timeout}, or returns {@code false} immediately (without
331    * waiting) if the permits would not have been granted before the timeout expired.
332    *
333    * @param permits the number of permits to acquire
334    * @param timeout the maximum time to wait for the permits. Negative values are treated as zero.
335    * @param unit the time unit of the timeout argument
336    * @return {@code true} if the permits were acquired, {@code false} otherwise
337    * @throws IllegalArgumentException if the requested number of permits is negative or zero
338    */
339   public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {
340     long timeoutMicros = max(unit.toMicros(timeout), 0);
341     checkPermits(permits);
342     long microsToWait;
343     synchronized (mutex()) {
344       long nowMicros = stopwatch.readMicros();
345       if (!canAcquire(nowMicros, timeoutMicros)) {
346         return false;
347       } else {
348         microsToWait = reserveAndGetWaitLength(permits, nowMicros);
349       }
350     }
351     stopwatch.sleepMicrosUninterruptibly(microsToWait);
352     return true;
353   }
354 
355   private boolean canAcquire(long nowMicros, long timeoutMicros) {
356     return queryEarliestAvailable(nowMicros) - timeoutMicros <= nowMicros;
357   }
358 
359   /**
360    * Reserves next ticket and returns the wait time that the caller must wait for.
361    *
362    * @return the required wait time, never negative
363    */
364   final long reserveAndGetWaitLength(int permits, long nowMicros) {
365     long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
366     return max(momentAvailable - nowMicros, 0);
367   }
368 
369   /**
370    * Returns the earliest time that permits are available (with one caveat).
371    *
372    * @return the time that permits are available, or, if permits are available immediately, an
373    *     arbitrary past or present time
374    */
375   abstract long queryEarliestAvailable(long nowMicros);
376 
377   /**
378    * Reserves the requested number of permits and returns the time that those permits can be used
379    * (with one caveat).
380    *
381    * @return the time that the permits may be used, or, if the permits may be used immediately, an
382    *     arbitrary past or present time
383    */
384   abstract long reserveEarliestAvailable(int permits, long nowMicros);
385 
386   @Override
387   public String toString() {
388     return String.format(Locale.ROOT, "RateLimiter[stableRate=%3.1fqps]", getRate());
389   }
390 
391   abstract static class SleepingStopwatch {
392     /** Constructor for use by subclasses. */
393     protected SleepingStopwatch() {}
394 
395     /*
396      * We always hold the mutex when calling this. TODO(cpovirk): Is that important? Perhaps we need
397      * to guarantee that each call to reserveEarliestAvailable, etc. sees a value >= the previous?
398      * Also, is it OK that we don't hold the mutex when sleeping?
399      */
400     protected abstract long readMicros();
401 
402     protected abstract void sleepMicrosUninterruptibly(long micros);
403 
404     public static final SleepingStopwatch createFromSystemTimer() {
405       return new SleepingStopwatch() {
406         final Stopwatch stopwatch = Stopwatch.createStarted();
407 
408         @Override
409         protected long readMicros() {
410           return stopwatch.elapsed(MICROSECONDS);
411         }
412 
413         @Override
414         protected void sleepMicrosUninterruptibly(long micros) {
415           if (micros > 0) {
416             Uninterruptibles.sleepUninterruptibly(micros, MICROSECONDS);
417           }
418         }
419       };
420     }
421   }
422 
423   private static void checkPermits(int permits) {
424     checkArgument(permits > 0, "Requested permits (%s) must be positive", permits);
425   }
426 }