1 /*
2 * Copyright (C) 2012 The Guava Authors
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
5 * in compliance with the License. You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software distributed under the License
10 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
11 * or implied. See the License for the specific language governing permissions and limitations under
12 * the License.
13 */
14
15 package com.google.common.util.concurrent;
16
17 import static com.google.common.base.Preconditions.checkArgument;
18 import static com.google.common.base.Preconditions.checkNotNull;
19 import static com.google.common.util.concurrent.Internal.toNanosSaturated;
20 import static java.lang.Math.max;
21 import static java.util.concurrent.TimeUnit.MICROSECONDS;
22 import static java.util.concurrent.TimeUnit.SECONDS;
23
24 import com.google.common.annotations.Beta;
25 import com.google.common.annotations.GwtIncompatible;
26 import com.google.common.annotations.VisibleForTesting;
27 import com.google.common.base.Stopwatch;
28 import com.google.common.util.concurrent.SmoothRateLimiter.SmoothBursty;
29 import com.google.common.util.concurrent.SmoothRateLimiter.SmoothWarmingUp;
30 import com.google.errorprone.annotations.CanIgnoreReturnValue;
31 import java.time.Duration;
32 import java.util.Locale;
33 import java.util.concurrent.TimeUnit;
34 import javax.annotation.CheckForNull;
35
36 /**
37 * A rate limiter. Conceptually, a rate limiter distributes permits at a configurable rate. Each
38 * {@link #acquire()} blocks if necessary until a permit is available, and then takes it. Once
39 * acquired, permits need not be released.
40 *
41 * <p>{@code RateLimiter} is safe for concurrent use: It will restrict the total rate of calls from
42 * all threads. Note, however, that it does not guarantee fairness.
43 *
44 * <p>Rate limiters are often used to restrict the rate at which some physical or logical resource
45 * is accessed. This is in contrast to {@link java.util.concurrent.Semaphore} which restricts the
46 * number of concurrent accesses instead of the rate (note though that concurrency and rate are
47 * closely related, e.g. see <a href="http://en.wikipedia.org/wiki/Little%27s_law">Little's
48 * Law</a>).
49 *
50 * <p>A {@code RateLimiter} is defined primarily by the rate at which permits are issued. Absent
51 * additional configuration, permits will be distributed at a fixed rate, defined in terms of
52 * permits per second. Permits will be distributed smoothly, with the delay between individual
53 * permits being adjusted to ensure that the configured rate is maintained.
54 *
55 * <p>It is possible to configure a {@code RateLimiter} to have a warmup period during which time
56 * the permits issued each second steadily increases until it hits the stable rate.
57 *
58 * <p>As an example, imagine that we have a list of tasks to execute, but we don't want to submit
59 * more than 2 per second:
60 *
61 * <pre>{@code
62 * final RateLimiter rateLimiter = RateLimiter.create(2.0); // rate is "2 permits per second"
63 * void submitTasks(List<Runnable> tasks, Executor executor) {
64 * for (Runnable task : tasks) {
65 * rateLimiter.acquire(); // may wait
66 * executor.execute(task);
67 * }
68 * }
69 * }</pre>
70 *
71 * <p>As another example, imagine that we produce a stream of data, and we want to cap it at 5kb per
72 * second. This could be accomplished by requiring a permit per byte, and specifying a rate of 5000
73 * permits per second:
74 *
75 * <pre>{@code
76 * final RateLimiter rateLimiter = RateLimiter.create(5000.0); // rate = 5000 permits per second
77 * void submitPacket(byte[] packet) {
78 * rateLimiter.acquire(packet.length);
79 * networkService.send(packet);
80 * }
81 * }</pre>
82 *
83 * <p>It is important to note that the number of permits requested <i>never</i> affects the
84 * throttling of the request itself (an invocation to {@code acquire(1)} and an invocation to {@code
85 * acquire(1000)} will result in exactly the same throttling, if any), but it affects the throttling
86 * of the <i>next</i> request. I.e., if an expensive task arrives at an idle RateLimiter, it will be
87 * granted immediately, but it is the <i>next</i> request that will experience extra throttling,
88 * thus paying for the cost of the expensive task.
89 *
90 * @author Dimitris Andreou
91 * @since 13.0
92 */
93 // TODO(user): switch to nano precision. A natural unit of cost is "bytes", and a micro precision
94 // would mean a maximum rate of "1MB/s", which might be small in some cases.
95 @Beta
96 @GwtIncompatible
97 @ElementTypesAreNonnullByDefault
98 public abstract class RateLimiter {
99 /**
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 }