001/*
002 * Copyright (C) 2012 The Guava Authors
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
005 * in compliance with the License. You may obtain a copy of the License at
006 *
007 * http://www.apache.org/licenses/LICENSE-2.0
008 *
009 * Unless required by applicable law or agreed to in writing, software distributed under the License
010 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
011 * or implied. See the License for the specific language governing permissions and limitations under
012 * the License.
013 */
014
015package com.google.common.util.concurrent;
016
017import static com.google.common.base.Preconditions.checkArgument;
018import static com.google.common.base.Preconditions.checkNotNull;
019import static com.google.common.util.concurrent.Internal.saturatedToNanos;
020import static java.lang.Math.max;
021import static java.util.concurrent.TimeUnit.MICROSECONDS;
022import static java.util.concurrent.TimeUnit.SECONDS;
023
024import com.google.common.annotations.Beta;
025import com.google.common.annotations.GwtIncompatible;
026import com.google.common.annotations.VisibleForTesting;
027import com.google.common.base.Stopwatch;
028import com.google.common.util.concurrent.SmoothRateLimiter.SmoothBursty;
029import com.google.common.util.concurrent.SmoothRateLimiter.SmoothWarmingUp;
030import com.google.errorprone.annotations.CanIgnoreReturnValue;
031import java.time.Duration;
032import java.util.Locale;
033import java.util.concurrent.TimeUnit;
034import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
035
036/**
037 * A rate limiter. Conceptually, a rate limiter distributes permits at a configurable rate. Each
038 * {@link #acquire()} blocks if necessary until a permit is available, and then takes it. Once
039 * acquired, permits need not be released.
040 *
041 * <p>{@code RateLimiter} is safe for concurrent use: It will restrict the total rate of calls from
042 * all threads. Note, however, that it does not guarantee fairness.
043 *
044 * <p>Rate limiters are often used to restrict the rate at which some physical or logical resource
045 * is accessed. This is in contrast to {@link java.util.concurrent.Semaphore} which restricts the
046 * number of concurrent accesses instead of the rate (note though that concurrency and rate are
047 * closely related, e.g. see <a href="http://en.wikipedia.org/wiki/Little%27s_law">Little's
048 * Law</a>).
049 *
050 * <p>A {@code RateLimiter} is defined primarily by the rate at which permits are issued. Absent
051 * additional configuration, permits will be distributed at a fixed rate, defined in terms of
052 * permits per second. Permits will be distributed smoothly, with the delay between individual
053 * permits being adjusted to ensure that the configured rate is maintained.
054 *
055 * <p>It is possible to configure a {@code RateLimiter} to have a warmup period during which time
056 * the permits issued each second steadily increases until it hits the stable rate.
057 *
058 * <p>As an example, imagine that we have a list of tasks to execute, but we don't want to submit
059 * more than 2 per second:
060 *
061 * <pre>{@code
062 * final RateLimiter rateLimiter = RateLimiter.create(2.0); // rate is "2 permits per second"
063 * void submitTasks(List<Runnable> tasks, Executor executor) {
064 *   for (Runnable task : tasks) {
065 *     rateLimiter.acquire(); // may wait
066 *     executor.execute(task);
067 *   }
068 * }
069 * }</pre>
070 *
071 * <p>As another example, imagine that we produce a stream of data, and we want to cap it at 5kb per
072 * second. This could be accomplished by requiring a permit per byte, and specifying a rate of 5000
073 * permits per second:
074 *
075 * <pre>{@code
076 * final RateLimiter rateLimiter = RateLimiter.create(5000.0); // rate = 5000 permits per second
077 * void submitPacket(byte[] packet) {
078 *   rateLimiter.acquire(packet.length);
079 *   networkService.send(packet);
080 * }
081 * }</pre>
082 *
083 * <p>It is important to note that the number of permits requested <i>never</i> affects the
084 * throttling of the request itself (an invocation to {@code acquire(1)} and an invocation to {@code
085 * acquire(1000)} will result in exactly the same throttling, if any), but it affects the throttling
086 * of the <i>next</i> request. I.e., if an expensive task arrives at an idle RateLimiter, it will be
087 * granted immediately, but it is the <i>next</i> request that will experience extra throttling,
088 * thus paying for the cost of the expensive task.
089 *
090 * @author Dimitris Andreou
091 * @since 13.0
092 */
093// TODO(user): switch to nano precision. A natural unit of cost is "bytes", and a micro precision
094// would mean a maximum rate of "1MB/s", which might be small in some cases.
095@Beta
096@GwtIncompatible
097public abstract class RateLimiter {
098  /**
099   * Creates a {@code RateLimiter} with the specified stable throughput, given as "permits per
100   * second" (commonly referred to as <i>QPS</i>, queries per second).
101   *
102   * <p>The returned {@code RateLimiter} ensures that on average no more than {@code
103   * permitsPerSecond} are issued during any given second, with sustained requests being smoothly
104   * spread over each second. When the incoming request rate exceeds {@code permitsPerSecond} the
105   * rate limiter will release one permit every {@code (1.0 / permitsPerSecond)} seconds. When the
106   * rate limiter is unused, bursts of up to {@code permitsPerSecond} permits will be allowed, with
107   * subsequent requests being smoothly limited at the stable rate of {@code permitsPerSecond}.
108   *
109   * @param permitsPerSecond the rate of the returned {@code RateLimiter}, measured in how many
110   *     permits become available per second
111   * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero
112   */
113  // TODO(user): "This is equivalent to
114  // {@code createWithCapacity(permitsPerSecond, 1, TimeUnit.SECONDS)}".
115  public static RateLimiter create(double permitsPerSecond) {
116    /*
117     * The default RateLimiter configuration can save the unused permits of up to one second. This
118     * is to avoid unnecessary stalls in situations like this: A RateLimiter of 1qps, and 4 threads,
119     * all calling acquire() at these moments:
120     *
121     * T0 at 0 seconds
122     * T1 at 1.05 seconds
123     * T2 at 2 seconds
124     * T3 at 3 seconds
125     *
126     * Due to the slight delay of T1, T2 would have to sleep till 2.05 seconds, and T3 would also
127     * have to sleep till 3.05 seconds.
128     */
129    return create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer());
130  }
131
132  @VisibleForTesting
133  static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {
134    RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
135    rateLimiter.setRate(permitsPerSecond);
136    return rateLimiter;
137  }
138
139  /**
140   * Creates a {@code RateLimiter} with the specified stable throughput, given as "permits per
141   * second" (commonly referred to as <i>QPS</i>, queries per second), and a <i>warmup period</i>,
142   * during which the {@code RateLimiter} smoothly ramps up its rate, until it reaches its maximum
143   * rate at the end of the period (as long as there are enough requests to saturate it). Similarly,
144   * if the {@code RateLimiter} is left <i>unused</i> for a duration of {@code warmupPeriod}, it
145   * will gradually return to its "cold" state, i.e. it will go through the same warming up process
146   * as when it was first created.
147   *
148   * <p>The returned {@code RateLimiter} is intended for cases where the resource that actually
149   * fulfills the requests (e.g., a remote server) needs "warmup" time, rather than being
150   * immediately accessed at the stable (maximum) rate.
151   *
152   * <p>The returned {@code RateLimiter} starts in a "cold" state (i.e. the warmup period will
153   * follow), and if it is left unused for long enough, it will return to that state.
154   *
155   * @param permitsPerSecond the rate of the returned {@code RateLimiter}, measured in how many
156   *     permits become available per second
157   * @param warmupPeriod the duration of the period where the {@code RateLimiter} ramps up its rate,
158   *     before reaching its stable (maximum) rate
159   * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero or {@code
160   *     warmupPeriod} is negative
161   * @since 28.0
162   */
163  public static RateLimiter create(double permitsPerSecond, Duration warmupPeriod) {
164    return create(permitsPerSecond, saturatedToNanos(warmupPeriod), TimeUnit.NANOSECONDS);
165  }
166
167  /**
168   * Creates a {@code RateLimiter} with the specified stable throughput, given as "permits per
169   * second" (commonly referred to as <i>QPS</i>, queries per second), and a <i>warmup period</i>,
170   * during which the {@code RateLimiter} smoothly ramps up its rate, until it reaches its maximum
171   * rate at the end of the period (as long as there are enough requests to saturate it). Similarly,
172   * if the {@code RateLimiter} is left <i>unused</i> for a duration of {@code warmupPeriod}, it
173   * will gradually return to its "cold" state, i.e. it will go through the same warming up process
174   * as when it was first created.
175   *
176   * <p>The returned {@code RateLimiter} is intended for cases where the resource that actually
177   * fulfills the requests (e.g., a remote server) needs "warmup" time, rather than being
178   * immediately accessed at the stable (maximum) rate.
179   *
180   * <p>The returned {@code RateLimiter} starts in a "cold" state (i.e. the warmup period will
181   * follow), and if it is left unused for long enough, it will return to that state.
182   *
183   * @param permitsPerSecond the rate of the returned {@code RateLimiter}, measured in how many
184   *     permits become available per second
185   * @param warmupPeriod the duration of the period where the {@code RateLimiter} ramps up its rate,
186   *     before reaching its stable (maximum) rate
187   * @param unit the time unit of the warmupPeriod argument
188   * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero or {@code
189   *     warmupPeriod} is negative
190   */
191  @SuppressWarnings("GoodTime") // should accept a java.time.Duration
192  public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
193    checkArgument(warmupPeriod >= 0, "warmupPeriod must not be negative: %s", warmupPeriod);
194    return create(
195        permitsPerSecond, warmupPeriod, unit, 3.0, SleepingStopwatch.createFromSystemTimer());
196  }
197
198  @VisibleForTesting
199  static RateLimiter create(
200      double permitsPerSecond,
201      long warmupPeriod,
202      TimeUnit unit,
203      double coldFactor,
204      SleepingStopwatch stopwatch) {
205    RateLimiter rateLimiter = new SmoothWarmingUp(stopwatch, warmupPeriod, unit, coldFactor);
206    rateLimiter.setRate(permitsPerSecond);
207    return rateLimiter;
208  }
209
210  /**
211   * The underlying timer; used both to measure elapsed time and sleep as necessary. A separate
212   * object to facilitate testing.
213   */
214  private final SleepingStopwatch stopwatch;
215
216  // Can't be initialized in the constructor because mocks don't call the constructor.
217  @MonotonicNonNull private volatile Object mutexDoNotUseDirectly;
218
219  private Object mutex() {
220    Object mutex = mutexDoNotUseDirectly;
221    if (mutex == null) {
222      synchronized (this) {
223        mutex = mutexDoNotUseDirectly;
224        if (mutex == null) {
225          mutexDoNotUseDirectly = mutex = new Object();
226        }
227      }
228    }
229    return mutex;
230  }
231
232  RateLimiter(SleepingStopwatch stopwatch) {
233    this.stopwatch = checkNotNull(stopwatch);
234  }
235
236  /**
237   * Updates the stable rate of this {@code RateLimiter}, that is, the {@code permitsPerSecond}
238   * argument provided in the factory method that constructed the {@code RateLimiter}. Currently
239   * throttled threads will <b>not</b> be awakened as a result of this invocation, thus they do not
240   * observe the new rate; only subsequent requests will.
241   *
242   * <p>Note though that, since each request repays (by waiting, if necessary) the cost of the
243   * <i>previous</i> request, this means that the very next request after an invocation to {@code
244   * setRate} will not be affected by the new rate; it will pay the cost of the previous request,
245   * which is in terms of the previous rate.
246   *
247   * <p>The behavior of the {@code RateLimiter} is not modified in any other way, e.g. if the {@code
248   * RateLimiter} was configured with a warmup period of 20 seconds, it still has a warmup period of
249   * 20 seconds after this method invocation.
250   *
251   * @param permitsPerSecond the new stable rate of this {@code RateLimiter}
252   * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero
253   */
254  public final void setRate(double permitsPerSecond) {
255    checkArgument(
256        permitsPerSecond > 0.0 && !Double.isNaN(permitsPerSecond), "rate must be positive");
257    synchronized (mutex()) {
258      doSetRate(permitsPerSecond, stopwatch.readMicros());
259    }
260  }
261
262  abstract void doSetRate(double permitsPerSecond, long nowMicros);
263
264  /**
265   * Returns the stable rate (as {@code permits per seconds}) with which this {@code RateLimiter} is
266   * configured with. The initial value of this is the same as the {@code permitsPerSecond} argument
267   * passed in the factory method that produced this {@code RateLimiter}, and it is only updated
268   * after invocations to {@linkplain #setRate}.
269   */
270  public final double getRate() {
271    synchronized (mutex()) {
272      return doGetRate();
273    }
274  }
275
276  abstract double doGetRate();
277
278  /**
279   * Acquires a single permit from this {@code RateLimiter}, blocking until the request can be
280   * granted. Tells the amount of time slept, if any.
281   *
282   * <p>This method is equivalent to {@code acquire(1)}.
283   *
284   * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited
285   * @since 16.0 (present in 13.0 with {@code void} return type})
286   */
287  @CanIgnoreReturnValue
288  public double acquire() {
289    return acquire(1);
290  }
291
292  /**
293   * Acquires the given number of permits from this {@code RateLimiter}, blocking until the request
294   * can be granted. Tells the amount of time slept, if any.
295   *
296   * @param permits the number of permits to acquire
297   * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited
298   * @throws IllegalArgumentException if the requested number of permits is negative or zero
299   * @since 16.0 (present in 13.0 with {@code void} return type})
300   */
301  @CanIgnoreReturnValue
302  public double acquire(int permits) {
303    long microsToWait = reserve(permits);
304    stopwatch.sleepMicrosUninterruptibly(microsToWait);
305    return 1.0 * microsToWait / SECONDS.toMicros(1L);
306  }
307
308  /**
309   * Reserves the given number of permits from this {@code RateLimiter} for future use, returning
310   * the number of microseconds until the reservation can be consumed.
311   *
312   * @return time in microseconds to wait until the resource can be acquired, never negative
313   */
314  final long reserve(int permits) {
315    checkPermits(permits);
316    synchronized (mutex()) {
317      return reserveAndGetWaitLength(permits, stopwatch.readMicros());
318    }
319  }
320
321  /**
322   * Acquires a permit from this {@code RateLimiter} if it can be obtained without exceeding the
323   * specified {@code timeout}, or returns {@code false} immediately (without waiting) if the permit
324   * would not have been granted before the timeout expired.
325   *
326   * <p>This method is equivalent to {@code tryAcquire(1, timeout)}.
327   *
328   * @param timeout the maximum time to wait for the permit. Negative values are treated as zero.
329   * @return {@code true} if the permit was acquired, {@code false} otherwise
330   * @throws IllegalArgumentException if the requested number of permits is negative or zero
331   * @since 28.0
332   */
333  public boolean tryAcquire(Duration timeout) {
334    return tryAcquire(1, saturatedToNanos(timeout), TimeUnit.NANOSECONDS);
335  }
336
337  /**
338   * Acquires a permit from this {@code RateLimiter} if it can be obtained without exceeding the
339   * specified {@code timeout}, or returns {@code false} immediately (without waiting) if the permit
340   * would not have been granted before the timeout expired.
341   *
342   * <p>This method is equivalent to {@code tryAcquire(1, timeout, unit)}.
343   *
344   * @param timeout the maximum time to wait for the permit. Negative values are treated as zero.
345   * @param unit the time unit of the timeout argument
346   * @return {@code true} if the permit was acquired, {@code false} otherwise
347   * @throws IllegalArgumentException if the requested number of permits is negative or zero
348   */
349  @SuppressWarnings("GoodTime") // should accept a java.time.Duration
350  public boolean tryAcquire(long timeout, TimeUnit unit) {
351    return tryAcquire(1, timeout, unit);
352  }
353
354  /**
355   * Acquires permits from this {@link RateLimiter} if it can be acquired immediately without delay.
356   *
357   * <p>This method is equivalent to {@code tryAcquire(permits, 0, anyUnit)}.
358   *
359   * @param permits the number of permits to acquire
360   * @return {@code true} if the permits were acquired, {@code false} otherwise
361   * @throws IllegalArgumentException if the requested number of permits is negative or zero
362   * @since 14.0
363   */
364  public boolean tryAcquire(int permits) {
365    return tryAcquire(permits, 0, MICROSECONDS);
366  }
367
368  /**
369   * Acquires a permit from this {@link RateLimiter} if it can be acquired immediately without
370   * delay.
371   *
372   * <p>This method is equivalent to {@code tryAcquire(1)}.
373   *
374   * @return {@code true} if the permit was acquired, {@code false} otherwise
375   * @since 14.0
376   */
377  public boolean tryAcquire() {
378    return tryAcquire(1, 0, MICROSECONDS);
379  }
380
381  /**
382   * Acquires the given number of permits from this {@code RateLimiter} if it can be obtained
383   * without exceeding the specified {@code timeout}, or returns {@code false} immediately (without
384   * waiting) if the permits would not have been granted before the timeout expired.
385   *
386   * @param permits the number of permits to acquire
387   * @param timeout the maximum time to wait for the permits. Negative values are treated as zero.
388   * @return {@code true} if the permits were acquired, {@code false} otherwise
389   * @throws IllegalArgumentException if the requested number of permits is negative or zero
390   * @since 28.0
391   */
392  public boolean tryAcquire(int permits, Duration timeout) {
393    return tryAcquire(permits, saturatedToNanos(timeout), TimeUnit.NANOSECONDS);
394  }
395
396  /**
397   * Acquires the given number of permits from this {@code RateLimiter} if it can be obtained
398   * without exceeding the specified {@code timeout}, or returns {@code false} immediately (without
399   * waiting) if the permits would not have been granted before the timeout expired.
400   *
401   * @param permits the number of permits to acquire
402   * @param timeout the maximum time to wait for the permits. Negative values are treated as zero.
403   * @param unit the time unit of the timeout argument
404   * @return {@code true} if the permits were acquired, {@code false} otherwise
405   * @throws IllegalArgumentException if the requested number of permits is negative or zero
406   */
407  @SuppressWarnings("GoodTime") // should accept a java.time.Duration
408  public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {
409    long timeoutMicros = max(unit.toMicros(timeout), 0);
410    checkPermits(permits);
411    long microsToWait;
412    synchronized (mutex()) {
413      long nowMicros = stopwatch.readMicros();
414      if (!canAcquire(nowMicros, timeoutMicros)) {
415        return false;
416      } else {
417        microsToWait = reserveAndGetWaitLength(permits, nowMicros);
418      }
419    }
420    stopwatch.sleepMicrosUninterruptibly(microsToWait);
421    return true;
422  }
423
424  private boolean canAcquire(long nowMicros, long timeoutMicros) {
425    return queryEarliestAvailable(nowMicros) - timeoutMicros <= nowMicros;
426  }
427
428  /**
429   * Reserves next ticket and returns the wait time that the caller must wait for.
430   *
431   * @return the required wait time, never negative
432   */
433  final long reserveAndGetWaitLength(int permits, long nowMicros) {
434    long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
435    return max(momentAvailable - nowMicros, 0);
436  }
437
438  /**
439   * Returns the earliest time that permits are available (with one caveat).
440   *
441   * @return the time that permits are available, or, if permits are available immediately, an
442   *     arbitrary past or present time
443   */
444  abstract long queryEarliestAvailable(long nowMicros);
445
446  /**
447   * Reserves the requested number of permits and returns the time that those permits can be used
448   * (with one caveat).
449   *
450   * @return the time that the permits may be used, or, if the permits may be used immediately, an
451   *     arbitrary past or present time
452   */
453  abstract long reserveEarliestAvailable(int permits, long nowMicros);
454
455  @Override
456  public String toString() {
457    return String.format(Locale.ROOT, "RateLimiter[stableRate=%3.1fqps]", getRate());
458  }
459
460  abstract static class SleepingStopwatch {
461    /** Constructor for use by subclasses. */
462    protected SleepingStopwatch() {}
463
464    /*
465     * We always hold the mutex when calling this. TODO(cpovirk): Is that important? Perhaps we need
466     * to guarantee that each call to reserveEarliestAvailable, etc. sees a value >= the previous?
467     * Also, is it OK that we don't hold the mutex when sleeping?
468     */
469    protected abstract long readMicros();
470
471    protected abstract void sleepMicrosUninterruptibly(long micros);
472
473    public static SleepingStopwatch createFromSystemTimer() {
474      return new SleepingStopwatch() {
475        final Stopwatch stopwatch = Stopwatch.createStarted();
476
477        @Override
478        protected long readMicros() {
479          return stopwatch.elapsed(MICROSECONDS);
480        }
481
482        @Override
483        protected void sleepMicrosUninterruptibly(long micros) {
484          if (micros > 0) {
485            Uninterruptibles.sleepUninterruptibly(micros, MICROSECONDS);
486          }
487        }
488      };
489    }
490  }
491
492  private static void checkPermits(int permits) {
493    checkArgument(permits > 0, "Requested permits (%s) must be positive", permits);
494  }
495}