/* * Copyright (C) 2012 The Guava Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */package com.google.common.util.concurrent;import static com.google.common.base.Preconditions.checkArgument;import static com.google.common.base.Preconditions.checkNotNull;import static java.lang.Math.max;import static java.util.concurrent.TimeUnit.MICROSECONDS;import static java.util.concurrent.TimeUnit.SECONDS;import com.google.common.annotations.Beta;import com.google.common.annotations.VisibleForTesting;import com.google.common.base.Stopwatch;import com.google.common.util.concurrent.SmoothRateLimiter.SmoothBursty;import com.google.common.util.concurrent.SmoothRateLimiter.SmoothWarmingUp;import java.util.concurrent.TimeUnit;import javax.annotation.concurrent.ThreadSafe;/** * A rate limiter. Conceptually, a rate limiter distributes permits at a * configurable rate. Each {@link #acquire()} blocks if necessary until a permit is * available, and then takes it. Once acquired, permits need not be released. * * <p>Rate limiters are often used to restrict the rate at which some * physical or logical resource is accessed. This is in contrast to {@link * java.util.concurrent.Semaphore} which restricts the number of concurrent * accesses instead of the rate (note though that concurrency and rate are closely related, * e.g. see <a href="http://en.wikipedia.org/wiki/Little's_law">Little's Law</a>). * * <p>A {@code RateLimiter} is defined primarily by the rate at which permits * are issued. Absent additional configuration, permits will be distributed at a * fixed rate, defined in terms of permits per second. Permits will be distributed * smoothly, with the delay between individual permits being adjusted to ensure * that the configured rate is maintained. * * <p>It is possible to configure a {@code RateLimiter} to have a warmup * period during which time the permits issued each second steadily increases until * it hits the stable rate. * * <p>As an example, imagine that we have a list of tasks to execute, but we don't want to * submit more than 2 per second: *<pre> {@code * final RateLimiter rateLimiter = RateLimiter.create(2.0); // rate is "2 permits per second" * void submitTasks(List<Runnable> tasks, Executor executor) { * for (Runnable task : tasks) { * rateLimiter.acquire(); // may wait * executor.execute(task); * } * } *}</pre> * * <p>As another example, imagine that we produce a stream of data, and we want to cap it * at 5kb per second. This could be accomplished by requiring a permit per byte, and specifying * a rate of 5000 permits per second: *<pre> {@code * final RateLimiter rateLimiter = RateLimiter.create(5000.0); // rate = 5000 permits per second * void submitPacket(byte[] packet) { * rateLimiter.acquire(packet.length); * networkService.send(packet); * } *}</pre> * * <p>It is important to note that the number of permits requested <i>never</i> * affect the throttling of the request itself (an invocation to {@code acquire(1)} * and an invocation to {@code acquire(1000)} will result in exactly the same throttling, if any), * but it affects the throttling of the <i>next</i> request. I.e., if an expensive task * arrives at an idle RateLimiter, it will be granted immediately, but it is the <i>next</i> * request that will experience extra throttling, thus paying for the cost of the expensive * task. * * <p>Note: {@code RateLimiter} does not provide fairness guarantees. * * @author Dimitris Andreou * @since 13.0 */// TODO(user): switch to nano precision. A natural unit of cost is "bytes", and a micro precision// would mean a maximum rate of "1MB/s", which might be small in some cases.@ThreadSafe@Betapublic abstract class RateLimiter { /** * Creates a {@code RateLimiter} with the specified stable throughput, given as * "permits per second" (commonly referred to as <i>QPS</i>, queries per second). * * <p>The returned {@code RateLimiter} ensures that on average no more than {@code * permitsPerSecond} are issued during any given second, with sustained requests * being smoothly spread over each second. When the incoming request rate exceeds * {@code permitsPerSecond} the rate limiter will release one permit every {@code * (1.0 / permitsPerSecond)} seconds. When the rate limiter is unused, * bursts of up to {@code permitsPerSecond} permits will be allowed, with subsequent * requests being smoothly limited at the stable rate of {@code permitsPerSecond}. * * @param permitsPerSecond the rate of the returned {@code RateLimiter}, measured in * how many permits become available per second * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero */ // TODO(user): "This is equivalent to // {@code createWithCapacity(permitsPerSecond, 1, TimeUnit.SECONDS)}". public static RateLimiter create(double permitsPerSecond) { /* * The default RateLimiter configuration can save the unused permits of up to one second. * This is to avoid unnecessary stalls in situations like this: A RateLimiter of 1qps, * and 4 threads, all calling acquire() at these moments: * * T0 at 0 seconds * T1 at 1.05 seconds * T2 at 2 seconds * T3 at 3 seconds * * Due to the slight delay of T1, T2 would have to sleep till 2.05 seconds, * and T3 would also have to sleep till 3.05 seconds. */ return create(SleepingStopwatch.createFromSystemTimer(), permitsPerSecond); } /* * TODO(cpovirk): make SleepingStopwatch the last parameter throughout the class so that the * overloads follow the usual convention: Foo(int), Foo(int, SleepingStopwatch) */ @VisibleForTesting static RateLimiter create(SleepingStopwatch stopwatch, double permitsPerSecond) { RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */); rateLimiter.setRate(permitsPerSecond); return rateLimiter; } /** * Creates a {@code RateLimiter} with the specified stable throughput, given as * "permits per second" (commonly referred to as <i>QPS</i>, queries per second), and a * <i>warmup period</i>, during which the {@code RateLimiter} smoothly ramps up its rate, * until it reaches its maximum rate at the end of the period (as long as there are enough * requests to saturate it). Similarly, if the {@code RateLimiter} is left <i>unused</i> for * a duration of {@code warmupPeriod}, it will gradually return to its "cold" state, * i.e. it will go through the same warming up process as when it was first created. * * <p>The returned {@code RateLimiter} is intended for cases where the resource that actually * fulfills the requests (e.g., a remote server) needs "warmup" time, rather than * being immediately accessed at the stable (maximum) rate. * * <p>The returned {@code RateLimiter} starts in a "cold" state (i.e. the warmup period * will follow), and if it is left unused for long enough, it will return to that state. * * @param permitsPerSecond the rate of the returned {@code RateLimiter}, measured in * how many permits become available per second * @param warmupPeriod the duration of the period where the {@code RateLimiter} ramps up its * rate, before reaching its stable (maximum) rate * @param unit the time unit of the warmupPeriod argument * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero or * {@code warmupPeriod} is negative */ public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit) { checkArgument(warmupPeriod >= 0, "warmupPeriod must not be negative: %s", warmupPeriod); return create(SleepingStopwatch.createFromSystemTimer(), permitsPerSecond, warmupPeriod, unit); } @VisibleForTesting static RateLimiter create( SleepingStopwatch stopwatch, double permitsPerSecond, long warmupPeriod, TimeUnit unit) { RateLimiter rateLimiter = new SmoothWarmingUp(stopwatch, warmupPeriod, unit); rateLimiter.setRate(permitsPerSecond); return rateLimiter; } /** * The underlying timer; used both to measure elapsed time and sleep as necessary. A separate * object to facilitate testing. */ private final SleepingStopwatch stopwatch; // Can't be initialized in the constructor because mocks don't call the constructor. private volatile Object mutexDoNotUseDirectly; private Object mutex() { Object mutex = mutexDoNotUseDirectly; if (mutex == null) { synchronized (this) { mutex = mutexDoNotUseDirectly; if (mutex == null) { mutexDoNotUseDirectly = mutex = new Object(); } } } return mutex; } RateLimiter(SleepingStopwatch stopwatch) { this.stopwatch = checkNotNull(stopwatch); } /** * Updates the stable rate of this {@code RateLimiter}, that is, the * {@code permitsPerSecond} argument provided in the factory method that * constructed the {@code RateLimiter}. Currently throttled threads will <b>not</b> * be awakened as a result of this invocation, thus they do not observe the new rate; * only subsequent requests will. * * <p>Note though that, since each request repays (by waiting, if necessary) the cost * of the <i>previous</i> request, this means that the very next request * after an invocation to {@code setRate} will not be affected by the new rate; * it will pay the cost of the previous request, which is in terms of the previous rate. * * <p>The behavior of the {@code RateLimiter} is not modified in any other way, * e.g. if the {@code RateLimiter} was configured with a warmup period of 20 seconds, * it still has a warmup period of 20 seconds after this method invocation. * * @param permitsPerSecond the new stable rate of this {@code RateLimiter} * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero */ public final void setRate(double permitsPerSecond) { checkArgument( permitsPerSecond > 0.0 && !Double.isNaN(permitsPerSecond), "rate must be positive"); synchronized (mutex()) { doSetRate(permitsPerSecond, stopwatch.readMicros()); } } abstract void doSetRate(double permitsPerSecond, long nowMicros); /** * Returns the stable rate (as {@code permits per seconds}) with which this * {@code RateLimiter} is configured with. The initial value of this is the same as * the {@code permitsPerSecond} argument passed in the factory method that produced * this {@code RateLimiter}, and it is only updated after invocations * to {@linkplain #setRate}. */ public final double getRate() { synchronized (mutex()) { return doGetRate(); } } abstract double doGetRate(); /** * Acquires a single permit from this {@code RateLimiter}, blocking until the * request can be granted. Tells the amount of time slept, if any. * * <p>This method is equivalent to {@code acquire(1)}. * * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited * @since 16.0 (present in 13.0 with {@code void} return type}) */ public double acquire() { return acquire(1); } /** * Acquires the given number of permits from this {@code RateLimiter}, blocking until the * request can be granted. Tells the amount of time slept, if any. * * @param permits the number of permits to acquire * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited * @throws IllegalArgumentException if the requested number of permits is negative or zero * @since 16.0 (present in 13.0 with {@code void} return type}) */ public double acquire(int permits) { long microsToWait = reserve(permits); stopwatch.sleepMicrosUninterruptibly(microsToWait); return 1.0 * microsToWait / SECONDS.toMicros(1L); } /** * Reserves the given number of permits from this {@code RateLimiter} for future use, returning * the number of microseconds until the reservation can be consumed. * * @return time in microseconds to wait until the resource can be acquired, never negative */ final long reserve(int permits) { checkPermits(permits); synchronized (mutex()) { return reserveAndGetWaitLength(permits, stopwatch.readMicros()); } } /** * Acquires a permit from this {@code RateLimiter} if it can be obtained * without exceeding the specified {@code timeout}, or returns {@code false} * immediately (without waiting) if the permit would not have been granted * before the timeout expired. * * <p>This method is equivalent to {@code tryAcquire(1, timeout, unit)}. * * @param timeout the maximum time to wait for the permit. Negative values are treated as zero. * @param unit the time unit of the timeout argument * @return {@code true} if the permit was acquired, {@code false} otherwise * @throws IllegalArgumentException if the requested number of permits is negative or zero */ public boolean tryAcquire(long timeout, TimeUnit unit) { return tryAcquire(1, timeout, unit); } /** * Acquires permits from this {@link RateLimiter} if it can be acquired immediately without delay. * * <p> * This method is equivalent to {@code tryAcquire(permits, 0, anyUnit)}. * * @param permits the number of permits to acquire * @return {@code true} if the permits were acquired, {@code false} otherwise * @throws IllegalArgumentException if the requested number of permits is negative or zero * @since 14.0 */ public boolean tryAcquire(int permits) { return tryAcquire(permits, 0, MICROSECONDS); } /** * Acquires a permit from this {@link RateLimiter} if it can be acquired immediately without * delay. * * <p> * This method is equivalent to {@code tryAcquire(1)}. * * @return {@code true} if the permit was acquired, {@code false} otherwise * @since 14.0 */ public boolean tryAcquire() { return tryAcquire(1, 0, MICROSECONDS); } /** * Acquires the given number of permits from this {@code RateLimiter} if it can be obtained * without exceeding the specified {@code timeout}, or returns {@code false} * immediately (without waiting) if the permits would not have been granted * before the timeout expired. * * @param permits the number of permits to acquire * @param timeout the maximum time to wait for the permits. Negative values are treated as zero. * @param unit the time unit of the timeout argument * @return {@code true} if the permits were acquired, {@code false} otherwise * @throws IllegalArgumentException if the requested number of permits is negative or zero */ public boolean tryAcquire(int permits, long timeout, TimeUnit unit) { long timeoutMicros = max(unit.toMicros(timeout), 0); checkPermits(permits); long microsToWait; synchronized (mutex()) { long nowMicros = stopwatch.readMicros(); if (!canAcquire(nowMicros, timeoutMicros)) { return false; } else { microsToWait = reserveAndGetWaitLength(permits, nowMicros); } } stopwatch.sleepMicrosUninterruptibly(microsToWait); return true; } private boolean canAcquire(long nowMicros, long timeoutMicros) { return queryEarliestAvailable(nowMicros) - timeoutMicros <= nowMicros; } /** * Reserves next ticket and returns the wait time that the caller must wait for. * * @return the required wait time, never negative */ final long reserveAndGetWaitLength(int permits, long nowMicros) { long momentAvailable = reserveEarliestAvailable(permits, nowMicros); return max(momentAvailable - nowMicros, 0); } /** * Returns the earliest time that permits are available (with one caveat). * * @return the time that permits are available, or, if permits are available immediately, an * arbitrary past or present time */ abstract long queryEarliestAvailable(long nowMicros); /** * Reserves the requested number of permits and returns the time that those permits can be used * (with one caveat). * * @return the time that the permits may be used, or, if the permits may be used immediately, an * arbitrary past or present time */ abstract long reserveEarliestAvailable(int permits, long nowMicros); @Override public String toString() { return String.format("RateLimiter[stableRate=%3.1fqps]", getRate()); } @VisibleForTesting abstract static class SleepingStopwatch { /* * We always hold the mutex when calling this. TODO(cpovirk): Is that important? Perhaps we need * to guarantee that each call to reserveEarliestAvailable, etc. sees a value >= the previous? * Also, is it OK that we don't hold the mutex when sleeping? */ abstract long readMicros(); abstract void sleepMicrosUninterruptibly(long micros); static final SleepingStopwatch createFromSystemTimer() { return new SleepingStopwatch() { final Stopwatch stopwatch = Stopwatch.createStarted(); @Override long readMicros() { return stopwatch.elapsed(MICROSECONDS); } @Override void sleepMicrosUninterruptibly(long micros) { if (micros > 0) { Uninterruptibles.sleepUninterruptibly(micros, MICROSECONDS); } } }; } } private static int checkPermits(int permits) { checkArgument(permits > 0, "Requested permits (%s) must be positive", permits); return permits; }}