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