Better behaviour in the presence of 429s

http-clients/src/main/java/com/palantir/remoting3/clients/ClientConfigurations.java

@@ -87,7 +87,7 @@ public static ClientConfiguration of(
                 .enableGcmCipherSuites(DEFAULT_ENABLE_GCM_CIPHERS)
                 .proxy(ProxySelector.getDefault())
                 .proxyCredentials(Optional.empty())
-                .maxNumRetries(uris.size())
+                .maxNumRetries(DEFAULT_MAX_NUM_RETRIES)
                 .backoffSlotSize(DEFAULT_BACKOFF_SLOT_SIZE)
                 .nodeSelectionStrategy(DEFAULT_NODE_SELECTION_STRATEGY)
                 .failedUrlCooldown(DEFAULT_FAILED_URL_COOLDOWN)

okhttp-clients/build.gradle

@@ -7,6 +7,7 @@ dependencies {
     compile project(':http-clients')
     compile project(':tracing-okhttp3')
     compile 'com.google.guava:guava'
+    compile 'com.netflix.concurrency-limits:concurrency-limits-core'
     compile 'com.palantir.safe-logging:preconditions'
     compile 'com.palantir.tritium:tritium-registry'
     compile 'com.squareup.okhttp3:logging-interceptor'

okhttp-clients/src/main/java/com/palantir/remoting3/okhttp/ConcurrencyLimiters.java

@@ -0,0 +1,177 @@
+/*
+ * Copyright 2018 Palantir Technologies, Inc. All rights reserved.
+ *
+ * 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 com.palantir.remoting3.okhttp;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.RemovalCause;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import com.netflix.concurrency.limits.Limiter;
+import com.netflix.concurrency.limits.limit.AIMDLimit;
+import com.netflix.concurrency.limits.limit.TracingLimitDecorator;
+import com.netflix.concurrency.limits.limiter.DefaultLimiter;
+import com.netflix.concurrency.limits.strategy.SimpleStrategy;
+import com.palantir.remoting3.tracing.okhttp3.OkhttpTraceInterceptor;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import okhttp3.Request;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Remoting calls may observe 429 or 503 responses from the server, at which point they back off in order to
+ * reduce excess load. Unfortunately this state on backing off is stored per-call, so 429s or 503s in one call do not
+ * cause any request rate slowdown in subsequent calls. This class affects this by adjusting the number of requests
+ * that might be dispatched to a given endpoint.
+ * <p>
+ * This is based on Netflix's <a href="https://github.com/Netflix/concurrency-limits/">Concurrency Limits</a> library,
+ * which provides a number of primitives for this.
+ * <p>
+ * In order to use this class, one should get a Limiter for their request, which returns a future. once the Future is
+ * done, the caller can assume that the request is schedulable. After the request completes, the caller <b>must</b>
+ * call one of the methods on {@link Limiter.Listener} in order to provide feedback about the request's success.
+ * If this is not done, throughput will be negatively affected. We attempt to eventually recover to avoid a total
+ * deadlock, but this is not guaranteed.
+ */
+final class ConcurrencyLimiters {
+    private static final Logger log = LoggerFactory.getLogger(ConcurrencyLimiters.class);
+    private static final String FALLBACK = "";
+
+    // If a request is never marked as complete and is thrown away, recover on the next GC instead of deadlocking
+    private static final Map<Limiter.Listener, Runnable> activeListeners = CacheBuilder.newBuilder()
+            .weakKeys()
+            .<Limiter.Listener, Runnable>removalListener(notification -> {
+                if (notification.getCause().equals(RemovalCause.COLLECTED)) {
+                    log.warn("Concurrency limiter was leaked."
+                            + " This implies a remoting bug or classpath issue, and may cause degraded performance");
+                    notification.getValue().run();
+                }
+            })
+            .build()
+            .asMap();
+
+    private final ConcurrentMap<String, ConcurrencyLimiter> limiters = new ConcurrentHashMap<>();
+
+    private static Limiter<Void> newLimiter() {
+        return DefaultLimiter.newBuilder()
+                .limit(TracingLimitDecorator.wrap(AIMDLimit.newBuilder().initialLimit(1).build()))
+                .build(new SimpleStrategy<>());
+    }
+
+    @VisibleForTesting
+    ConcurrencyLimiter limiter(String name) {
+        return limiters.computeIfAbsent(name, key -> new ConcurrencyLimiter(activeListeners, newLimiter()));
+    }
+
+    ConcurrencyLimiter limiter(Request request) {
+        final String limiterKey;
+        String pathTemplate = request.header(OkhttpTraceInterceptor.PATH_TEMPLATE_HEADER);
+        if (pathTemplate == null) {
+            limiterKey = FALLBACK;
+        } else {
+            limiterKey = request.method() + " " + pathTemplate;
+        }
+        return limiter(limiterKey);
+    }
+
+    /**
+     * The Netflix library provides either a blocking approach or a non-blocking approach which might say
+     * you can't be scheduled at this time. All of our HTTP calls are asynchronous, so we really want to get
+     * a {@link ListenableFuture} that we can add a callback to. This class then is a translation of
+     * {@link com.netflix.concurrency.limits.limiter.BlockingLimiter} to be asynchronous, maintaining a queue
+     * of currently waiting requests.
+     * <p>
+     * Upon a request finishing, we check if there are any waiting requests, and if there are we attempt to trigger
+     * some more.
+     */
+    static final class ConcurrencyLimiter {
+        private final Map<Limiter.Listener, Runnable> activeListeners;
+        private final Queue<SettableFuture<Limiter.Listener>> waitingRequests = new LinkedBlockingQueue<>();
+        private final Limiter<Void> limiter;
+
+        public ConcurrencyLimiter(
+                Map<Limiter.Listener, Runnable> activeListeners,
+                Limiter<Void> limiter) {
+            this.activeListeners = activeListeners;
+            this.limiter = limiter;
+        }
+
+        public ListenableFuture<Limiter.Listener> acquire() {
+            Optional<Limiter.Listener> maybeListener = limiter.acquire(null);
+            if (maybeListener.isPresent()) {
+                return Futures.immediateFuture(wrap(activeListeners, maybeListener.get()));
+            }
+            SettableFuture<Limiter.Listener> future = SettableFuture.create();
+            waitingRequests.add(future);
+            return future;
+        }
+
+        private void processQueue() {
+            while (!waitingRequests.isEmpty()) {
+                Optional<Limiter.Listener> maybeAcquired = limiter.acquire(null);
+                if (!maybeAcquired.isPresent()) {
+                    return;
+                }
+                Limiter.Listener acquired = maybeAcquired.get();
+                SettableFuture<Limiter.Listener> head = waitingRequests.poll();
+                if (head == null) {
+                    acquired.onIgnore();
+                } else {
+                    head.set(acquired);
+                }
+            }
+        }
+
+        private Limiter.Listener wrap(
+                Map<Limiter.Listener, Runnable> activeListeners, Limiter.Listener listener) {
+            Limiter.Listener res = new Limiter.Listener() {
+                @Override
+                public void onSuccess() {
+                    listener.onSuccess();
+                    activeListeners.remove(this);
+                    processQueue();
+                }
+
+                @Override
+                public void onIgnore() {
+                    listener.onIgnore();
+                    activeListeners.remove(this);
+                    processQueue();
+                }
+
+                @Override
+                public void onDropped() {
+                    listener.onDropped();
+                    activeListeners.remove(this);
+                    processQueue();
+                }
+            };
+            activeListeners.put(res, () -> {
+                listener.onIgnore();
+                processQueue();
+            });
+            return res;
+        }
+
+    }
+}

okhttp-clients/src/main/java/com/palantir/remoting3/okhttp/OkHttpClients.java

@@ -31,12 +31,12 @@
 import com.palantir.tritium.metrics.registry.MetricName;
 import com.palantir.tritium.metrics.registry.TaggedMetricRegistry;
 import java.util.Collection;
-import java.util.Random;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import okhttp3.ConnectionPool;
 import okhttp3.ConnectionSpec;
@@ -221,7 +221,8 @@ private static RemotingOkHttpClient createInternal(
 
         return new RemotingOkHttpClient(
                 client.build(),
-                () -> new ExponentialBackoff(config.maxNumRetries(), config.backoffSlotSize(), new Random()),
+                () -> new ExponentialBackoff(
+                        config.maxNumRetries(), config.backoffSlotSize(), ThreadLocalRandom.current()),
                 urlSelector,
                 schedulingExecutor,
                 executionExecutor);