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