diff --git a/extractor/src/test/java/org/schabi/newpipe/downloader/DownloaderTestImpl.java b/extractor/src/test/java/org/schabi/newpipe/downloader/DownloaderTestImpl.java index 1ba47e086..0bcd313e9 100644 --- a/extractor/src/test/java/org/schabi/newpipe/downloader/DownloaderTestImpl.java +++ b/extractor/src/test/java/org/schabi/newpipe/downloader/DownloaderTestImpl.java @@ -1,5 +1,6 @@ package org.schabi.newpipe.downloader; +import org.schabi.newpipe.downloader.ratelimiting.RateLimitedClientWrapper; import org.schabi.newpipe.extractor.downloader.Downloader; import org.schabi.newpipe.extractor.downloader.Request; import org.schabi.newpipe.extractor.downloader.Response; @@ -24,10 +25,11 @@ public final class DownloaderTestImpl extends Downloader { private static final String USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:128.0) Gecko/20100101 Firefox/128.0"; private static DownloaderTestImpl instance; - private final OkHttpClient client; + private final RateLimitedClientWrapper clientWrapper; private DownloaderTestImpl(final OkHttpClient.Builder builder) { - this.client = builder.readTimeout(30, TimeUnit.SECONDS).build(); + this.clientWrapper = new RateLimitedClientWrapper( + builder.readTimeout(30, TimeUnit.SECONDS).build()); } /** @@ -66,22 +68,22 @@ public final class DownloaderTestImpl extends Downloader { .method(httpMethod, requestBody).url(url) .addHeader("User-Agent", USER_AGENT); - for (Map.Entry> pair : headers.entrySet()) { + for (final Map.Entry> pair : headers.entrySet()) { final String headerName = pair.getKey(); final List headerValueList = pair.getValue(); if (headerValueList.size() > 1) { requestBuilder.removeHeader(headerName); - for (String headerValue : headerValueList) { + for (final String headerValue : headerValueList) { requestBuilder.addHeader(headerName, headerValue); } } else if (headerValueList.size() == 1) { requestBuilder.header(headerName, headerValueList.get(0)); } - } - final okhttp3.Response response = client.newCall(requestBuilder.build()).execute(); + final okhttp3.Response response = + clientWrapper.executeRequestWithLimit(requestBuilder.build()); if (response.code() == 429) { response.close(); diff --git a/extractor/src/test/java/org/schabi/newpipe/downloader/ratelimiting/RateLimitedClientWrapper.java b/extractor/src/test/java/org/schabi/newpipe/downloader/ratelimiting/RateLimitedClientWrapper.java new file mode 100644 index 000000000..ab9661871 --- /dev/null +++ b/extractor/src/test/java/org/schabi/newpipe/downloader/ratelimiting/RateLimitedClientWrapper.java @@ -0,0 +1,76 @@ +package org.schabi.newpipe.downloader.ratelimiting; + +import java.io.IOException; +import java.net.ProtocolException; +import java.time.Duration; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.function.Predicate; + +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; + +public class RateLimitedClientWrapper { + + private static final int REQUEST_RATE_LIMITED_WAIT_MS = 5_000; + private static final Map, RateLimiter> FORCED_RATE_LIMITERS = Map.ofEntries( + Map.entry(host -> host.endsWith("youtube.com"), + RateLimiter.create(1.6, Duration.ofSeconds(1), 3.0)) + ); + + private final OkHttpClient client; + private final Map hostRateLimiters = new LinkedHashMap<>(); + + public RateLimitedClientWrapper(final OkHttpClient client) { + this.client = client; + } + + protected RateLimiter getRateLimiterFor(final Request request) { + return hostRateLimiters.computeIfAbsent(request.url().host(), host -> + FORCED_RATE_LIMITERS.entrySet() + .stream() + .filter(e -> e.getKey().test(host)) + .findFirst() + .map(Map.Entry::getValue) + .orElseGet(() -> + // Default rate limiter per domain + RateLimiter.create(5, Duration.ofSeconds(5), 3.0))); + } + + public Response executeRequestWithLimit(final Request request) throws IOException { + Exception cause = null; + for (int tries = 1; tries <= 3; tries++) { + try { + final double rateLimitedSec = getRateLimiterFor(request).acquire(); + System.out.println( + "[RATE-LIMIT] Waited " + rateLimitedSec + "s for " + request.url()); + + final Response response = client.newCall(request).execute(); + if(response.code() != 429) { // 429 = Too many requests + return response; + } + cause = new IllegalStateException("HTTP 429 - Too many requests"); + } catch (final ProtocolException pre) { + if (!pre.getMessage().startsWith("Too many follow-up")) { // -> Too many requests + throw pre; + } + cause = pre; + } + + final int waitMs = REQUEST_RATE_LIMITED_WAIT_MS * tries; + System.out.println( + "[TOO-MANY-REQUESTS] Waiting " + waitMs + "ms for " + request.url()); + try { + Thread.sleep(waitMs); + } catch (final InterruptedException iex) { + Thread.currentThread().interrupt(); + } + } + throw new IllegalStateException("Retrying/Rate-limiting for " + request.url() + "failed", cause); + } + + public OkHttpClient getClient() { + return client; + } +} diff --git a/extractor/src/test/java/org/schabi/newpipe/downloader/ratelimiting/RateLimiter.java b/extractor/src/test/java/org/schabi/newpipe/downloader/ratelimiting/RateLimiter.java new file mode 100644 index 000000000..1b066e877 --- /dev/null +++ b/extractor/src/test/java/org/schabi/newpipe/downloader/ratelimiting/RateLimiter.java @@ -0,0 +1,464 @@ +/* + * 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 org.schabi.newpipe.downloader.ratelimiting; + +import static java.lang.Math.max; +import static java.util.concurrent.TimeUnit.MICROSECONDS; +import static java.util.concurrent.TimeUnit.NANOSECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; + +import com.google.errorprone.annotations.CanIgnoreReturnValue; + +import org.schabi.newpipe.downloader.ratelimiting.SmoothRateLimiter.SmoothWarmingUp; + +import java.time.Duration; +import java.util.Locale; +import java.util.Objects; +import java.util.concurrent.TimeUnit; + +/** + * 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. + * + *

{@code RateLimiter} is safe for concurrent use: It will restrict the total rate of calls from + * all threads. Note, however, that it does not guarantee fairness. + * + *

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 Little's + * Law). + * + *

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. + * + *

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. + * + *

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: + * + * 

{@code
+ * final RateLimiter rateLimiter = RateLimiter.create(2.0); // rate is "2 permits per second"
+ * void submitTasks(List tasks, Executor executor) {
+ *   for (Runnable task : tasks) {
+ *     rateLimiter.acquire(); // may wait
+ *     executor.execute(task);
+ *   }
+ * }
+ * }
+ * + *

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: + * + * 

{@code
+ * final RateLimiter rateLimiter = RateLimiter.create(5000.0); // rate = 5000 permits per second
+ * void submitPacket(byte[] packet) {
+ *   rateLimiter.acquire(packet.length);
+ *   networkService.send(packet);
+ * }
+ * }
+ * + *

It is important to note that the number of permits requested never affects 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 next request. I.e., if an expensive task arrives at an idle RateLimiter, it will be + * granted immediately, but it is the next request that will experience extra throttling, + * thus paying for the cost of the expensive task. + * + * @author Dimitris Andreou + * @since 13.0 + */ +public abstract class RateLimiter { + /** + * Creates a {@code RateLimiter} with the specified stable throughput, given as "permits per + * second" (commonly referred to as QPS, queries per second), and a warmup period, + * 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 unused 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. + * + *

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. + * + *

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 coldFactor see {@link SmoothWarmingUp} + * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero or {@code + * warmupPeriod} is negative + * @since 28.0 (but only since 33.4.0 in the Android flavor) + */ + public static RateLimiter create( + final double permitsPerSecond, + final Duration warmupPeriod, + final double coldFactor + ) { + return create(permitsPerSecond, warmupPeriod.toNanos(), TimeUnit.NANOSECONDS, coldFactor); + } + + /** + * Creates a {@code RateLimiter} with the specified stable throughput, given as "permits per + * second" (commonly referred to as QPS, queries per second), and a warmup period, + * 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 unused 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. + * + *

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. + * + *

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 + * @param coldFactor see {@link SmoothWarmingUp} + * @throws IllegalArgumentException if {@code permitsPerSecond} is negative or zero or {@code + * warmupPeriod} is negative + */ + @SuppressWarnings("GoodTime") // should accept a java.time.Duration + public static RateLimiter create( + final double permitsPerSecond, + final long warmupPeriod, + final TimeUnit unit, + final double coldFactor + ) { + if(warmupPeriod < 0) { + throw new IllegalArgumentException( + "warmupPeriod must not be negative: " + warmupPeriod); + } + return create( + permitsPerSecond, warmupPeriod, unit, coldFactor, SleepingStopwatch.createFromSystemTimer()); + } + + static RateLimiter create( + final double permitsPerSecond, + final long warmupPeriod, + final TimeUnit unit, + final double coldFactor, + final SleepingStopwatch stopwatch) { + final RateLimiter rateLimiter = new SmoothWarmingUp(stopwatch, warmupPeriod, unit, coldFactor); + 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(final SleepingStopwatch stopwatch) { + this.stopwatch = Objects.requireNonNull(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 not be awakened as a result of this invocation, thus they do not + * observe the new rate; only subsequent requests will. + * + *

Note though that, since each request repays (by waiting, if necessary) the cost of the + * previous 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. + * + *

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(final double permitsPerSecond) { + if(permitsPerSecond <= 0.0) { + throw new IllegalArgumentException("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. + * + *

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}) + */ + @CanIgnoreReturnValue + 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}) + */ + @CanIgnoreReturnValue + public double acquire(final int permits) { + final 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(final 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. + * + *

This method is equivalent to {@code tryAcquire(1, timeout)}. + * + * @param timeout the maximum time to wait for the permit. Negative values are treated as zero. + * @return {@code true} if the permit was acquired, {@code false} otherwise + * @throws IllegalArgumentException if the requested number of permits is negative or zero + * @since 28.0 (but only since 33.4.0 in the Android flavor) + */ + public boolean tryAcquire(final Duration timeout) { + return tryAcquire(1, timeout.toNanos(), TimeUnit.NANOSECONDS); + } + + /** + * 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. + * + *

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 + */ + @SuppressWarnings("GoodTime") // should accept a java.time.Duration + public boolean tryAcquire(final long timeout, final TimeUnit unit) { + return tryAcquire(1, timeout, unit); + } + + /** + * Acquires permits from this {@link RateLimiter} if it can be acquired immediately without delay. + * + *

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(final int permits) { + return tryAcquire(permits, 0, MICROSECONDS); + } + + /** + * Acquires a permit from this {@link RateLimiter} if it can be acquired immediately without + * delay. + * + *

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. + * @return {@code true} if the permits were acquired, {@code false} otherwise + * @throws IllegalArgumentException if the requested number of permits is negative or zero + * @since 28.0 (but only since 33.4.0 in the Android flavor) + */ + public boolean tryAcquire(final int permits, final Duration timeout) { + return tryAcquire(permits, timeout.toNanos(), TimeUnit.NANOSECONDS); + } + + /** + * 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 + */ + @SuppressWarnings("GoodTime") // should accept a java.time.Duration + public boolean tryAcquire(final int permits, final long timeout, final TimeUnit unit) { + final long timeoutMicros = max(unit.toMicros(timeout), 0); + checkPermits(permits); + final long microsToWait; + synchronized (mutex()) { + final long nowMicros = stopwatch.readMicros(); + if (!canAcquire(nowMicros, timeoutMicros)) { + return false; + } else { + microsToWait = reserveAndGetWaitLength(permits, nowMicros); + } + } + stopwatch.sleepMicrosUninterruptibly(microsToWait); + return true; + } + + private boolean canAcquire(final long nowMicros, final 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(final int permits, final long nowMicros) { + final 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(Locale.ROOT, "RateLimiter[stableRate=%3.1fqps]", getRate()); + } + + abstract static class SleepingStopwatch { + /** Constructor for use by subclasses. */ + protected SleepingStopwatch() {} + + protected abstract long readMicros(); + + protected abstract void sleepMicrosUninterruptibly(long micros); + + public static SleepingStopwatch createFromSystemTimer() { + return new SleepingStopwatch() { + final long startTimeNano = System.nanoTime(); + + @Override + protected long readMicros() { + return MICROSECONDS.convert(System.nanoTime() - startTimeNano, NANOSECONDS); + } + + @Override + protected void sleepMicrosUninterruptibly(final long micros) { + if (micros > 0) { + final long sleepMs = MICROSECONDS.toMillis(micros); + if (sleepMs > 0) { + try { + Thread.sleep(sleepMs); + } catch (final InterruptedException iex) { + Thread.currentThread().interrupt(); + } + } + } + } + }; + } + } + + private static void checkPermits(final int permits) { + if(permits <= 0.0) { + throw new IllegalArgumentException( + "Requested permits (" + permits + ") must be positive"); + } + } +} \ No newline at end of file diff --git a/extractor/src/test/java/org/schabi/newpipe/downloader/ratelimiting/SmoothRateLimiter.java b/extractor/src/test/java/org/schabi/newpipe/downloader/ratelimiting/SmoothRateLimiter.java new file mode 100644 index 000000000..f72b127c2 --- /dev/null +++ b/extractor/src/test/java/org/schabi/newpipe/downloader/ratelimiting/SmoothRateLimiter.java @@ -0,0 +1,368 @@ +/* + * 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 org.schabi.newpipe.downloader.ratelimiting; + +import static java.lang.Math.min; +import static java.util.concurrent.TimeUnit.SECONDS; + +import java.util.concurrent.TimeUnit; + +abstract class SmoothRateLimiter extends RateLimiter { + /* + * How is the RateLimiter designed, and why? + * + * The primary feature of a RateLimiter is its "stable rate", the maximum rate that it should + * allow in normal conditions. This is enforced by "throttling" incoming requests as needed. For + * example, we could compute the appropriate throttle time for an incoming request, and make the + * calling thread wait for that time. + * + * The simplest way to maintain a rate of QPS is to keep the timestamp of the last granted + * request, and ensure that (1/QPS) seconds have elapsed since then. For example, for a rate of + * QPS=5 (5 tokens per second), if we ensure that a request isn't granted earlier than 200ms after + * the last one, then we achieve the intended rate. If a request comes and the last request was + * granted only 100ms ago, then we wait for another 100ms. At this rate, serving 15 fresh permits + * (i.e. for an acquire(15) request) naturally takes 3 seconds. + * + * It is important to realize that such a RateLimiter has a very superficial memory of the past: + * it only remembers the last request. What if the RateLimiter was unused for a long period of + * time, then a request arrived and was immediately granted? This RateLimiter would immediately + * forget about that past underutilization. This may result in either underutilization or + * overflow, depending on the real world consequences of not using the expected rate. + * + * Past underutilization could mean that excess resources are available. Then, the RateLimiter + * should speed up for a while, to take advantage of these resources. This is important when the + * rate is applied to networking (limiting bandwidth), where past underutilization typically + * translates to "almost empty buffers", which can be filled immediately. + * + * On the other hand, past underutilization could mean that "the server responsible for handling + * the request has become less ready for future requests", i.e. its caches become stale, and + * requests become more likely to trigger expensive operations (a more extreme case of this + * example is when a server has just booted, and it is mostly busy with getting itself up to + * speed). + * + * To deal with such scenarios, we add an extra dimension, that of "past underutilization", + * modeled by "storedPermits" variable. This variable is zero when there is no underutilization, + * and it can grow up to maxStoredPermits, for sufficiently large underutilization. So, the + * requested permits, by an invocation acquire(permits), are served from: + * + * - stored permits (if available) + * + * - fresh permits (for any remaining permits) + * + * How this works is best explained with an example: + * + * For a RateLimiter that produces 1 token per second, every second that goes by with the + * RateLimiter being unused, we increase storedPermits by 1. Say we leave the RateLimiter unused + * for 10 seconds (i.e., we expected a request at time X, but we are at time X + 10 seconds before + * a request actually arrives; this is also related to the point made in the last paragraph), thus + * storedPermits becomes 10.0 (assuming maxStoredPermits >= 10.0). At that point, a request of + * acquire(3) arrives. We serve this request out of storedPermits, and reduce that to 7.0 (how + * this is translated to throttling time is discussed later). Immediately after, assume that an + * acquire(10) request arriving. We serve the request partly from storedPermits, using all the + * remaining 7.0 permits, and the remaining 3.0, we serve them by fresh permits produced by the + * rate limiter. + * + * We already know how much time it takes to serve 3 fresh permits: if the rate is + * "1 token per second", then this will take 3 seconds. But what does it mean to serve 7 stored + * permits? As explained above, there is no unique answer. If we are primarily interested to deal + * with underutilization, then we want stored permits to be given out /faster/ than fresh ones, + * because underutilization = free resources for the taking. If we are primarily interested to + * deal with overflow, then stored permits could be given out /slower/ than fresh ones. Thus, we + * require a (different in each case) function that translates storedPermits to throttling time. + * + * This role is played by storedPermitsToWaitTime(double storedPermits, double permitsToTake). The + * underlying model is a continuous function mapping storedPermits (from 0.0 to maxStoredPermits) + * onto the 1/rate (i.e. intervals) that is effective at the given storedPermits. "storedPermits" + * essentially measure unused time; we spend unused time buying/storing permits. Rate is + * "permits / time", thus "1 / rate = time / permits". Thus, "1/rate" (time / permits) times + * "permits" gives time, i.e., integrals on this function (which is what storedPermitsToWaitTime() + * computes) correspond to minimum intervals between subsequent requests, for the specified number + * of requested permits. + * + * Here is an example of storedPermitsToWaitTime: If storedPermits == 10.0, and we want 3 permits, + * we take them from storedPermits, reducing them to 7.0, and compute the throttling for these as + * a call to storedPermitsToWaitTime(storedPermits = 10.0, permitsToTake = 3.0), which will + * evaluate the integral of the function from 7.0 to 10.0. + * + * Using integrals guarantees that the effect of a single acquire(3) is equivalent to { + * acquire(1); acquire(1); acquire(1); }, or { acquire(2); acquire(1); }, etc, since the integral + * of the function in [7.0, 10.0] is equivalent to the sum of the integrals of [7.0, 8.0], [8.0, + * 9.0], [9.0, 10.0] (and so on), no matter what the function is. This guarantees that we handle + * correctly requests of varying weight (permits), /no matter/ what the actual function is - so we + * can tweak the latter freely. (The only requirement, obviously, is that we can compute its + * integrals). + * + * Note well that if, for this function, we chose a horizontal line, at height of exactly (1/QPS), + * then the effect of the function is non-existent: we serve storedPermits at exactly the same + * cost as fresh ones (1/QPS is the cost for each). We use this trick later. + * + * If we pick a function that goes /below/ that horizontal line, it means that we reduce the area + * of the function, thus time. Thus, the RateLimiter becomes /faster/ after a period of + * underutilization. If, on the other hand, we pick a function that goes /above/ that horizontal + * line, then it means that the area (time) is increased, thus storedPermits are more costly than + * fresh permits, thus the RateLimiter becomes /slower/ after a period of underutilization. + * + * Last, but not least: consider a RateLimiter with rate of 1 permit per second, currently + * completely unused, and an expensive acquire(100) request comes. It would be nonsensical to just + * wait for 100 seconds, and /then/ start the actual task. Why wait without doing anything? A much + * better approach is to /allow/ the request right away (as if it was an acquire(1) request + * instead), and postpone /subsequent/ requests as needed. In this version, we allow starting the + * task immediately, and postpone by 100 seconds future requests, thus we allow for work to get + * done in the meantime instead of waiting idly. + * + * This has important consequences: it means that the RateLimiter doesn't remember the time of the + * _last_ request, but it remembers the (expected) time of the _next_ request. This also enables + * us to tell immediately (see tryAcquire(timeout)) whether a particular timeout is enough to get + * us to the point of the next scheduling time, since we always maintain that. And what we mean by + * "an unused RateLimiter" is also defined by that notion: when we observe that the + * "expected arrival time of the next request" is actually in the past, then the difference (now - + * past) is the amount of time that the RateLimiter was formally unused, and it is that amount of + * time which we translate to storedPermits. (We increase storedPermits with the amount of permits + * that would have been produced in that idle time). So, if rate == 1 permit per second, and + * arrivals come exactly one second after the previous, then storedPermits is _never_ increased -- + * we would only increase it for arrivals _later_ than the expected one second. + */ + + /** + * This implements the following function where coldInterval = coldFactor * stableInterval. + * + * 

+     *          ^ throttling
+     *          |
+     *    cold  +                  /
+     * interval |                 /.
+     *          |                / .
+     *          |               /  .   ← "warmup period" is the area of the trapezoid between
+     *          |              /   .     thresholdPermits and maxPermits
+     *          |             /    .
+     *          |            /     .
+     *          |           /      .
+     *   stable +----------/  WARM .
+     * interval |          .   UP  .
+     *          |          . PERIOD.
+     *          |          .       .
+     *        0 +----------+-------+--------------→ storedPermits
+     *          0 thresholdPermits maxPermits
+     *
+ * + * Before going into the details of this particular function, let's keep in mind the basics: + * + *
    + *
  1. The state of the RateLimiter (storedPermits) is a vertical line in this figure. + *
  2. When the RateLimiter is not used, this goes right (up to maxPermits) + *
  3. When the RateLimiter is used, this goes left (down to zero), since if we have + * storedPermits, we serve from those first + *
  4. When _unused_, we go right at a constant rate! The rate at which we move to the right is + * chosen as maxPermits / warmupPeriod. This ensures that the time it takes to go from 0 to + * maxPermits is equal to warmupPeriod. + *
  5. When _used_, the time it takes, as explained in the introductory class note, is equal to + * the integral of our function, between X permits and X-K permits, assuming we want to + * spend K saved permits. + *
+ * + *

In summary, the time it takes to move to the left (spend K permits), is equal to the area of + * the function of width == K. + * + *

Assuming we have saturated demand, the time to go from maxPermits to thresholdPermits is + * equal to warmupPeriod. And the time to go from thresholdPermits to 0 is warmupPeriod/2. (The + * reason that this is warmupPeriod/2 is to maintain the behavior of the original implementation + * where coldFactor was hard coded as 3.) + * + *

It remains to calculate thresholdsPermits and maxPermits. + * + *

    + *
  • The time to go from thresholdPermits to 0 is equal to the integral of the function + * between 0 and thresholdPermits. This is thresholdPermits * stableIntervals. By (5) it is + * also equal to warmupPeriod/2. Therefore + *
    + * thresholdPermits = 0.5 * warmupPeriod / stableInterval + *
    + *
  • The time to go from maxPermits to thresholdPermits is equal to the integral of the + * function between thresholdPermits and maxPermits. This is the area of the pictured + * trapezoid, and it is equal to 0.5 * (stableInterval + coldInterval) * (maxPermits - + * thresholdPermits). It is also equal to warmupPeriod, so + *
    + * maxPermits = thresholdPermits + 2 * warmupPeriod / (stableInterval + coldInterval) + *
    + *
+ */ + static final class SmoothWarmingUp extends SmoothRateLimiter { + private final long warmupPeriodMicros; + + /** + * The slope of the line from the stable interval (when permits == 0), to the cold interval + * (when permits == maxPermits) + */ + private double slope; + + private double thresholdPermits; + private final double coldFactor; + + SmoothWarmingUp( + final SleepingStopwatch stopwatch, + final long warmupPeriod, + final TimeUnit timeUnit, + final double coldFactor) { + super(stopwatch); + this.warmupPeriodMicros = timeUnit.toMicros(warmupPeriod); + this.coldFactor = coldFactor; + } + + @Override + void doSetRate(final double permitsPerSecond, final double stableIntervalMicros) { + final double oldMaxPermits = maxPermits; + final double coldIntervalMicros = stableIntervalMicros * coldFactor; + thresholdPermits = 0.5 * warmupPeriodMicros / stableIntervalMicros; + maxPermits = + thresholdPermits + 2.0 * warmupPeriodMicros / (stableIntervalMicros + coldIntervalMicros); + slope = (coldIntervalMicros - stableIntervalMicros) / (maxPermits - thresholdPermits); + if (oldMaxPermits == Double.POSITIVE_INFINITY) { + // if we don't special-case this, we would get storedPermits == NaN, below + storedPermits = 0.0; + } else { + storedPermits = + (oldMaxPermits == 0.0) + ? maxPermits // initial state is cold + : storedPermits * maxPermits / oldMaxPermits; + } + } + + @Override + long storedPermitsToWaitTime(final double storedPermits, double permitsToTake) { + final double availablePermitsAboveThreshold = storedPermits - thresholdPermits; + long micros = 0; + // measuring the integral on the right part of the function (the climbing line) + if (availablePermitsAboveThreshold > 0.0) { + final double permitsAboveThresholdToTake = min(availablePermitsAboveThreshold, permitsToTake); + final double length = + permitsToTime(availablePermitsAboveThreshold) + + permitsToTime(availablePermitsAboveThreshold - permitsAboveThresholdToTake); + micros = (long) (permitsAboveThresholdToTake * length / 2.0); + permitsToTake -= permitsAboveThresholdToTake; + } + // measuring the integral on the left part of the function (the horizontal line) + micros += (long) (stableIntervalMicros * permitsToTake); + return micros; + } + + private double permitsToTime(final double permits) { + return stableIntervalMicros + permits * slope; + } + + @Override + double coolDownIntervalMicros() { + return warmupPeriodMicros / maxPermits; + } + } + + /** The currently stored permits. */ + double storedPermits; + + /** The maximum number of stored permits. */ + double maxPermits; + + /** + * The interval between two unit requests, at our stable rate. E.g., a stable rate of 5 permits + * per second has a stable interval of 200ms. + */ + double stableIntervalMicros; + + /** + * The time when the next request (no matter its size) will be granted. After granting a request, + * this is pushed further in the future. Large requests push this further than small requests. + */ + private long nextFreeTicketMicros = 0L; // could be either in the past or future + + private SmoothRateLimiter(final SleepingStopwatch stopwatch) { + super(stopwatch); + } + + @Override + final void doSetRate(final double permitsPerSecond, final long nowMicros) { + resync(nowMicros); + final double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond; + this.stableIntervalMicros = stableIntervalMicros; + doSetRate(permitsPerSecond, stableIntervalMicros); + } + + abstract void doSetRate(double permitsPerSecond, double stableIntervalMicros); + + @Override + final double doGetRate() { + return SECONDS.toMicros(1L) / stableIntervalMicros; + } + + @Override + final long queryEarliestAvailable(final long nowMicros) { + return nextFreeTicketMicros; + } + + @Override + final long reserveEarliestAvailable(final int requiredPermits, final long nowMicros) { + resync(nowMicros); + final long returnValue = nextFreeTicketMicros; + final double storedPermitsToSpend = min(requiredPermits, this.storedPermits); + final double freshPermits = requiredPermits - storedPermitsToSpend; + final long waitMicros = + storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend) + + (long) (freshPermits * stableIntervalMicros); + + this.nextFreeTicketMicros = saturatedAdd(nextFreeTicketMicros, waitMicros); + this.storedPermits -= storedPermitsToSpend; + return returnValue; + } + + /** + * Returns the sum of {@code a} and {@code b} unless it would overflow or underflow in which case + * {@code Long.MAX_VALUE} or {@code Long.MIN_VALUE} is returned, respectively. + * + * @since 20.0 + */ + @SuppressWarnings("ShortCircuitBoolean") + static long saturatedAdd(final long a, final long b) { + final long naiveSum = a + b; + if ((a ^ b) < 0 || (a ^ naiveSum) >= 0) { + // If a and b have different signs or a has the same sign as the result then there was no + // overflow, return. + return naiveSum; + } + // we did over/under flow, if the sign is negative we should return MAX otherwise MIN + return Long.MAX_VALUE + ((naiveSum >>> (Long.SIZE - 1)) ^ 1); + } + + /** + * Translates a specified portion of our currently stored permits which we want to spend/acquire, + * into a throttling time. Conceptually, this evaluates the integral of the underlying function we + * use, for the range of [(storedPermits - permitsToTake), storedPermits]. + * + *

This always holds: {@code 0 <= permitsToTake <= storedPermits} + */ + abstract long storedPermitsToWaitTime(double storedPermits, double permitsToTake); + + /** + * Returns the number of microseconds during cool down that we have to wait to get a new permit. + */ + abstract double coolDownIntervalMicros(); + + /** Updates {@code storedPermits} and {@code nextFreeTicketMicros} based on the current time. */ + void resync(final long nowMicros) { + // if nextFreeTicket is in the past, resync to now + if (nowMicros > nextFreeTicketMicros) { + final double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros(); + storedPermits = min(maxPermits, storedPermits + newPermits); + nextFreeTicketMicros = nowMicros; + } + } +} \ No newline at end of file diff --git a/extractor/src/test/java/org/schabi/newpipe/downloader/ratelimiting/package-info.java b/extractor/src/test/java/org/schabi/newpipe/downloader/ratelimiting/package-info.java new file mode 100644 index 000000000..41a6c319e --- /dev/null +++ b/extractor/src/test/java/org/schabi/newpipe/downloader/ratelimiting/package-info.java @@ -0,0 +1,10 @@ +/** + * Contains classes for Rate limiting. + *
+ * This code is based on + * google/guava + * (Apache 2.0 license) but was modified/refactored for our use. + * + * @author litetex + */ +package org.schabi.newpipe.downloader.ratelimiting;