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