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