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(permitsPerSecond > 0.0, "rate must be positive"); 229 synchronized (mutex()) { 230 doSetRate(permitsPerSecond, stopwatch.readMicros()); 231 } 232 } 233 234 abstract void doSetRate(double permitsPerSecond, long nowMicros); 235 236 /** 237 * Returns the stable rate (as {@code permits per seconds}) with which this {@code RateLimiter} is 238 * configured with. The initial value of this is the same as the {@code permitsPerSecond} argument 239 * passed in the factory method that produced this {@code RateLimiter}, and it is only updated 240 * after invocations to {@linkplain #setRate}. 241 */ 242 public final double getRate() { 243 synchronized (mutex()) { 244 return doGetRate(); 245 } 246 } 247 248 abstract double doGetRate(); 249 250 /** 251 * Acquires a single permit from this {@code RateLimiter}, blocking until the request can be 252 * granted. Tells the amount of time slept, if any. 253 * 254 * <p>This method is equivalent to {@code acquire(1)}. 255 * 256 * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited 257 * @since 16.0 (present in 13.0 with {@code void} return type}) 258 */ 259 @CanIgnoreReturnValue 260 public double acquire() { 261 return acquire(1); 262 } 263 264 /** 265 * Acquires the given number of permits from this {@code RateLimiter}, blocking until the request 266 * can be granted. Tells the amount of time slept, if any. 267 * 268 * @param permits the number of permits to acquire 269 * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited 270 * @throws IllegalArgumentException if the requested number of permits is negative or zero 271 * @since 16.0 (present in 13.0 with {@code void} return type}) 272 */ 273 @CanIgnoreReturnValue 274 public double acquire(int permits) { 275 long microsToWait = reserve(permits); 276 stopwatch.sleepMicrosUninterruptibly(microsToWait); 277 return 1.0 * microsToWait / SECONDS.toMicros(1L); 278 } 279 280 /** 281 * Reserves the given number of permits from this {@code RateLimiter} for future use, returning 282 * the number of microseconds until the reservation can be consumed. 283 * 284 * @return time in microseconds to wait until the resource can be acquired, never negative 285 */ 286 final long reserve(int permits) { 287 checkPermits(permits); 288 synchronized (mutex()) { 289 return reserveAndGetWaitLength(permits, stopwatch.readMicros()); 290 } 291 } 292 293 /** 294 * Acquires a permit from this {@code RateLimiter} if it can be obtained without exceeding the 295 * specified {@code timeout}, or returns {@code false} immediately (without waiting) if the permit 296 * would not have been granted before the timeout expired. 297 * 298 * <p>This method is equivalent to {@code tryAcquire(1, timeout, unit)}. 299 * 300 * @param timeout the maximum time to wait for the permit. Negative values are treated as zero. 301 * @param unit the time unit of the timeout argument 302 * @return {@code true} if the permit was acquired, {@code false} otherwise 303 * @throws IllegalArgumentException if the requested number of permits is negative or zero 304 */ 305 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 306 public boolean tryAcquire(long timeout, TimeUnit unit) { 307 return tryAcquire(1, timeout, unit); 308 } 309 310 /** 311 * Acquires permits from this {@link RateLimiter} if it can be acquired immediately without delay. 312 * 313 * <p>This method is equivalent to {@code tryAcquire(permits, 0, anyUnit)}. 314 * 315 * @param permits the number of permits to acquire 316 * @return {@code true} if the permits were acquired, {@code false} otherwise 317 * @throws IllegalArgumentException if the requested number of permits is negative or zero 318 * @since 14.0 319 */ 320 public boolean tryAcquire(int permits) { 321 return tryAcquire(permits, 0, MICROSECONDS); 322 } 323 324 /** 325 * Acquires a permit from this {@link RateLimiter} if it can be acquired immediately without 326 * delay. 327 * 328 * <p>This method is equivalent to {@code tryAcquire(1)}. 329 * 330 * @return {@code true} if the permit was acquired, {@code false} otherwise 331 * @since 14.0 332 */ 333 public boolean tryAcquire() { 334 return tryAcquire(1, 0, MICROSECONDS); 335 } 336 337 /** 338 * Acquires the given number of permits from this {@code RateLimiter} if it can be obtained 339 * without exceeding the specified {@code timeout}, or returns {@code false} immediately (without 340 * waiting) if the permits would not have been granted before the timeout expired. 341 * 342 * @param permits the number of permits to acquire 343 * @param timeout the maximum time to wait for the permits. Negative values are treated as zero. 344 * @param unit the time unit of the timeout argument 345 * @return {@code true} if the permits were acquired, {@code false} otherwise 346 * @throws IllegalArgumentException if the requested number of permits is negative or zero 347 */ 348 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 349 public boolean tryAcquire(int permits, long timeout, TimeUnit unit) { 350 long timeoutMicros = max(unit.toMicros(timeout), 0); 351 checkPermits(permits); 352 long microsToWait; 353 synchronized (mutex()) { 354 long nowMicros = stopwatch.readMicros(); 355 if (!canAcquire(nowMicros, timeoutMicros)) { 356 return false; 357 } else { 358 microsToWait = reserveAndGetWaitLength(permits, nowMicros); 359 } 360 } 361 stopwatch.sleepMicrosUninterruptibly(microsToWait); 362 return true; 363 } 364 365 private boolean canAcquire(long nowMicros, long timeoutMicros) { 366 return queryEarliestAvailable(nowMicros) - timeoutMicros <= nowMicros; 367 } 368 369 /** 370 * Reserves next ticket and returns the wait time that the caller must wait for. 371 * 372 * @return the required wait time, never negative 373 */ 374 final long reserveAndGetWaitLength(int permits, long nowMicros) { 375 long momentAvailable = reserveEarliestAvailable(permits, nowMicros); 376 return max(momentAvailable - nowMicros, 0); 377 } 378 379 /** 380 * Returns the earliest time that permits are available (with one caveat). 381 * 382 * @return the time that permits are available, or, if permits are available immediately, an 383 * arbitrary past or present time 384 */ 385 abstract long queryEarliestAvailable(long nowMicros); 386 387 /** 388 * Reserves the requested number of permits and returns the time that those permits can be used 389 * (with one caveat). 390 * 391 * @return the time that the permits may be used, or, if the permits may be used immediately, an 392 * arbitrary past or present time 393 */ 394 abstract long reserveEarliestAvailable(int permits, long nowMicros); 395 396 @Override 397 public String toString() { 398 return String.format(Locale.ROOT, "RateLimiter[stableRate=%3.1fqps]", getRate()); 399 } 400 401 abstract static class SleepingStopwatch { 402 /** Constructor for use by subclasses. */ 403 protected SleepingStopwatch() {} 404 405 /* 406 * We always hold the mutex when calling this. TODO(cpovirk): Is that important? Perhaps we need 407 * to guarantee that each call to reserveEarliestAvailable, etc. sees a value >= the previous? 408 * Also, is it OK that we don't hold the mutex when sleeping? 409 */ 410 protected abstract long readMicros(); 411 412 protected abstract void sleepMicrosUninterruptibly(long micros); 413 414 public static SleepingStopwatch createFromSystemTimer() { 415 return new SleepingStopwatch() { 416 final Stopwatch stopwatch = Stopwatch.createStarted(); 417 418 @Override 419 protected long readMicros() { 420 return stopwatch.elapsed(MICROSECONDS); 421 } 422 423 @Override 424 protected void sleepMicrosUninterruptibly(long micros) { 425 if (micros > 0) { 426 Uninterruptibles.sleepUninterruptibly(micros, MICROSECONDS); 427 } 428 } 429 }; 430 } 431 } 432 433 private static void checkPermits(int permits) { 434 checkArgument(permits > 0, "Requested permits (%s) must be positive", permits); 435 } 436}