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