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.toNanosSaturated;
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 javax.annotation.CheckForNull;
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
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   * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero or {@code
161   *     warmupPeriod} is negative
162   * @since 28.0
163   */
164  public static RateLimiter create(double permitsPerSecond, Duration warmupPeriod) {
165    return create(permitsPerSecond, toNanosSaturated(warmupPeriod), TimeUnit.NANOSECONDS);
166  }
167
168  /**
169   * Creates a {@code RateLimiter} with the specified stable throughput, given as "permits per
170   * second" (commonly referred to as <i>QPS</i>, queries per second), and a <i>warmup period</i>,
171   * during which the {@code RateLimiter} smoothly ramps up its rate, until it reaches its maximum
172   * rate at the end of the period (as long as there are enough requests to saturate it). Similarly,
173   * if the {@code RateLimiter} is left <i>unused</i> for a duration of {@code warmupPeriod}, it
174   * will gradually return to its "cold" state, i.e. it will go through the same warming up process
175   * as when it was first created.
176   *
177   * <p>The returned {@code RateLimiter} is intended for cases where the resource that actually
178   * fulfills the requests (e.g., a remote server) needs "warmup" time, rather than being
179   * immediately accessed at the stable (maximum) rate.
180   *
181   * <p>The returned {@code RateLimiter} starts in a "cold" state (i.e. the warmup period will
182   * follow), and if it is left unused for long enough, it will return to that state.
183   *
184   * @param permitsPerSecond the rate of the returned {@code RateLimiter}, measured in how many
185   *     permits become available per second
186   * @param warmupPeriod the duration of the period where the {@code RateLimiter} ramps up its rate,
187   *     before reaching its stable (maximum) rate
188   * @param unit the time unit of the warmupPeriod argument
189   * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero or {@code
190   *     warmupPeriod} is negative
191   */
192  @SuppressWarnings("GoodTime") // should accept a java.time.Duration
193  public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
194    checkArgument(warmupPeriod >= 0, "warmupPeriod must not be negative: %s", warmupPeriod);
195    return create(
196        permitsPerSecond, warmupPeriod, unit, 3.0, SleepingStopwatch.createFromSystemTimer());
197  }
198
199  @VisibleForTesting
200  static RateLimiter create(
201      double permitsPerSecond,
202      long warmupPeriod,
203      TimeUnit unit,
204      double coldFactor,
205      SleepingStopwatch stopwatch) {
206    RateLimiter rateLimiter = new SmoothWarmingUp(stopwatch, warmupPeriod, unit, coldFactor);
207    rateLimiter.setRate(permitsPerSecond);
208    return rateLimiter;
209  }
210
211  /**
212   * The underlying timer; used both to measure elapsed time and sleep as necessary. A separate
213   * object to facilitate testing.
214   */
215  private final SleepingStopwatch stopwatch;
216
217  // Can't be initialized in the constructor because mocks don't call the constructor.
218  @CheckForNull private volatile Object mutexDoNotUseDirectly;
219
220  private Object mutex() {
221    Object mutex = mutexDoNotUseDirectly;
222    if (mutex == null) {
223      synchronized (this) {
224        mutex = mutexDoNotUseDirectly;
225        if (mutex == null) {
226          mutexDoNotUseDirectly = mutex = new Object();
227        }
228      }
229    }
230    return mutex;
231  }
232
233  RateLimiter(SleepingStopwatch stopwatch) {
234    this.stopwatch = checkNotNull(stopwatch);
235  }
236
237  /**
238   * Updates the stable rate of this {@code RateLimiter}, that is, the {@code permitsPerSecond}
239   * argument provided in the factory method that constructed the {@code RateLimiter}. Currently
240   * throttled threads will <b>not</b> be awakened as a result of this invocation, thus they do not
241   * observe the new rate; only subsequent requests will.
242   *
243   * <p>Note though that, since each request repays (by waiting, if necessary) the cost of the
244   * <i>previous</i> request, this means that the very next request after an invocation to {@code
245   * setRate} will not be affected by the new rate; it will pay the cost of the previous request,
246   * which is in terms of the previous rate.
247   *
248   * <p>The behavior of the {@code RateLimiter} is not modified in any other way, e.g. if the {@code
249   * RateLimiter} was configured with a warmup period of 20 seconds, it still has a warmup period of
250   * 20 seconds after this method invocation.
251   *
252   * @param permitsPerSecond the new stable rate of this {@code RateLimiter}
253   * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero
254   */
255  public final void setRate(double permitsPerSecond) {
256    checkArgument(
257        permitsPerSecond > 0.0 && !Double.isNaN(permitsPerSecond), "rate must be positive");
258    synchronized (mutex()) {
259      doSetRate(permitsPerSecond, stopwatch.readMicros());
260    }
261  }
262
263  abstract void doSetRate(double permitsPerSecond, long nowMicros);
264
265  /**
266   * Returns the stable rate (as {@code permits per seconds}) with which this {@code RateLimiter} is
267   * configured with. The initial value of this is the same as the {@code permitsPerSecond} argument
268   * passed in the factory method that produced this {@code RateLimiter}, and it is only updated
269   * after invocations to {@linkplain #setRate}.
270   */
271  public final double getRate() {
272    synchronized (mutex()) {
273      return doGetRate();
274    }
275  }
276
277  abstract double doGetRate();
278
279  /**
280   * Acquires a single permit from this {@code RateLimiter}, blocking until the request can be
281   * granted. Tells the amount of time slept, if any.
282   *
283   * <p>This method is equivalent to {@code acquire(1)}.
284   *
285   * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited
286   * @since 16.0 (present in 13.0 with {@code void} return type})
287   */
288  @CanIgnoreReturnValue
289  public double acquire() {
290    return acquire(1);
291  }
292
293  /**
294   * Acquires the given number of permits from this {@code RateLimiter}, blocking until the request
295   * can be granted. Tells the amount of time slept, if any.
296   *
297   * @param permits the number of permits to acquire
298   * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited
299   * @throws IllegalArgumentException if the requested number of permits is negative or zero
300   * @since 16.0 (present in 13.0 with {@code void} return type})
301   */
302  @CanIgnoreReturnValue
303  public double acquire(int permits) {
304    long microsToWait = reserve(permits);
305    stopwatch.sleepMicrosUninterruptibly(microsToWait);
306    return 1.0 * microsToWait / SECONDS.toMicros(1L);
307  }
308
309  /**
310   * Reserves the given number of permits from this {@code RateLimiter} for future use, returning
311   * the number of microseconds until the reservation can be consumed.
312   *
313   * @return time in microseconds to wait until the resource can be acquired, never negative
314   */
315  final long reserve(int permits) {
316    checkPermits(permits);
317    synchronized (mutex()) {
318      return reserveAndGetWaitLength(permits, stopwatch.readMicros());
319    }
320  }
321
322  /**
323   * Acquires a permit from this {@code RateLimiter} if it can be obtained without exceeding the
324   * specified {@code timeout}, or returns {@code false} immediately (without waiting) if the permit
325   * would not have been granted before the timeout expired.
326   *
327   * <p>This method is equivalent to {@code tryAcquire(1, timeout)}.
328   *
329   * @param timeout the maximum time to wait for the permit. Negative values are treated as zero.
330   * @return {@code true} if the permit was acquired, {@code false} otherwise
331   * @throws IllegalArgumentException if the requested number of permits is negative or zero
332   * @since 28.0
333   */
334  public boolean tryAcquire(Duration timeout) {
335    return tryAcquire(1, toNanosSaturated(timeout), TimeUnit.NANOSECONDS);
336  }
337
338  /**
339   * Acquires a permit from this {@code RateLimiter} if it can be obtained without exceeding the
340   * specified {@code timeout}, or returns {@code false} immediately (without waiting) if the permit
341   * would not have been granted before the timeout expired.
342   *
343   * <p>This method is equivalent to {@code tryAcquire(1, timeout, unit)}.
344   *
345   * @param timeout the maximum time to wait for the permit. Negative values are treated as zero.
346   * @param unit the time unit of the timeout argument
347   * @return {@code true} if the permit was acquired, {@code false} otherwise
348   * @throws IllegalArgumentException if the requested number of permits is negative or zero
349   */
350  @SuppressWarnings("GoodTime") // should accept a java.time.Duration
351  public boolean tryAcquire(long timeout, TimeUnit unit) {
352    return tryAcquire(1, timeout, unit);
353  }
354
355  /**
356   * Acquires permits from this {@link RateLimiter} if it can be acquired immediately without delay.
357   *
358   * <p>This method is equivalent to {@code tryAcquire(permits, 0, anyUnit)}.
359   *
360   * @param permits the number of permits to acquire
361   * @return {@code true} if the permits were acquired, {@code false} otherwise
362   * @throws IllegalArgumentException if the requested number of permits is negative or zero
363   * @since 14.0
364   */
365  public boolean tryAcquire(int permits) {
366    return tryAcquire(permits, 0, MICROSECONDS);
367  }
368
369  /**
370   * Acquires a permit from this {@link RateLimiter} if it can be acquired immediately without
371   * delay.
372   *
373   * <p>This method is equivalent to {@code tryAcquire(1)}.
374   *
375   * @return {@code true} if the permit was acquired, {@code false} otherwise
376   * @since 14.0
377   */
378  public boolean tryAcquire() {
379    return tryAcquire(1, 0, MICROSECONDS);
380  }
381
382  /**
383   * Acquires the given number of permits from this {@code RateLimiter} if it can be obtained
384   * without exceeding the specified {@code timeout}, or returns {@code false} immediately (without
385   * waiting) if the permits would not have been granted before the timeout expired.
386   *
387   * @param permits the number of permits to acquire
388   * @param timeout the maximum time to wait for the permits. Negative values are treated as zero.
389   * @return {@code true} if the permits were acquired, {@code false} otherwise
390   * @throws IllegalArgumentException if the requested number of permits is negative or zero
391   * @since 28.0
392   */
393  public boolean tryAcquire(int permits, Duration timeout) {
394    return tryAcquire(permits, toNanosSaturated(timeout), TimeUnit.NANOSECONDS);
395  }
396
397  /**
398   * Acquires the given number of permits from this {@code RateLimiter} if it can be obtained
399   * without exceeding the specified {@code timeout}, or returns {@code false} immediately (without
400   * waiting) if the permits would not have been granted before the timeout expired.
401   *
402   * @param permits the number of permits to acquire
403   * @param timeout the maximum time to wait for the permits. Negative values are treated as zero.
404   * @param unit the time unit of the timeout argument
405   * @return {@code true} if the permits were acquired, {@code false} otherwise
406   * @throws IllegalArgumentException if the requested number of permits is negative or zero
407   */
408  @SuppressWarnings("GoodTime") // should accept a java.time.Duration
409  public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {
410    long timeoutMicros = max(unit.toMicros(timeout), 0);
411    checkPermits(permits);
412    long microsToWait;
413    synchronized (mutex()) {
414      long nowMicros = stopwatch.readMicros();
415      if (!canAcquire(nowMicros, timeoutMicros)) {
416        return false;
417      } else {
418        microsToWait = reserveAndGetWaitLength(permits, nowMicros);
419      }
420    }
421    stopwatch.sleepMicrosUninterruptibly(microsToWait);
422    return true;
423  }
424
425  private boolean canAcquire(long nowMicros, long timeoutMicros) {
426    return queryEarliestAvailable(nowMicros) - timeoutMicros <= nowMicros;
427  }
428
429  /**
430   * Reserves next ticket and returns the wait time that the caller must wait for.
431   *
432   * @return the required wait time, never negative
433   */
434  final long reserveAndGetWaitLength(int permits, long nowMicros) {
435    long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
436    return max(momentAvailable - nowMicros, 0);
437  }
438
439  /**
440   * Returns the earliest time that permits are available (with one caveat).
441   *
442   * @return the time that permits are available, or, if permits are available immediately, an
443   *     arbitrary past or present time
444   */
445  abstract long queryEarliestAvailable(long nowMicros);
446
447  /**
448   * Reserves the requested number of permits and returns the time that those permits can be used
449   * (with one caveat).
450   *
451   * @return the time that the permits may be used, or, if the permits may be used immediately, an
452   *     arbitrary past or present time
453   */
454  abstract long reserveEarliestAvailable(int permits, long nowMicros);
455
456  @Override
457  public String toString() {
458    return String.format(Locale.ROOT, "RateLimiter[stableRate=%3.1fqps]", getRate());
459  }
460
461  abstract static class SleepingStopwatch {
462    /** Constructor for use by subclasses. */
463    protected SleepingStopwatch() {}
464
465    /*
466     * We always hold the mutex when calling this. TODO(cpovirk): Is that important? Perhaps we need
467     * to guarantee that each call to reserveEarliestAvailable, etc. sees a value >= the previous?
468     * Also, is it OK that we don't hold the mutex when sleeping?
469     */
470    protected abstract long readMicros();
471
472    protected abstract void sleepMicrosUninterruptibly(long micros);
473
474    public static SleepingStopwatch createFromSystemTimer() {
475      return new SleepingStopwatch() {
476        final Stopwatch stopwatch = Stopwatch.createStarted();
477
478        @Override
479        protected long readMicros() {
480          return stopwatch.elapsed(MICROSECONDS);
481        }
482
483        @Override
484        protected void sleepMicrosUninterruptibly(long micros) {
485          if (micros > 0) {
486            Uninterruptibles.sleepUninterruptibly(micros, MICROSECONDS);
487          }
488        }
489      };
490    }
491  }
492
493  private static void checkPermits(int permits) {
494    checkArgument(permits > 0, "Requested permits (%s) must be positive", permits);
495  }
496}