New client pool implementation (#11736)

#8100 overhauled the connection pooling, but its primary purpose was to improve the separation of concerns, and to extend connection pooling to non-exchange methods. From an efficiency standpoint, it is not very good. In particular, it uses a central "doSomeWork" method to dispatch requests onto connections, which is shared between all threads and only dispatches in a serial fashion. The connections were also shared with no regard to which event loop they were created on.

#11300 bolted on "client affinity" functionality: Requests arriving on an event loop would preferentially be dispatched to connections running on the same event loop, which can reduce context switches and improve latency. However, the actual dispatch still happens in the central `doSomeWork` method serially, now further weighed down by the client affinity algorithm.

Benchmarks connected to #11704 show that this model breaks down in scenarios with many connections serving many requests on multiple event loops. `doSomeWork` becomes a bottleneck, since only one event loop can execute it at a time, and the connection selection algorithm for client affinity makes this even heavier.

This PR introduces a new connection pool built with scalability in mind from the start. Instead of a global pool of connections, the pool is split into "local pools", one for each event loop. Requests that are created on one event loop can be dispatched purely within that loop, without coordination with other loops. Contention between loops is kept to a minimum.

If load is uneven between event loops, requests can still spill into a global request queue as a fallback, and be picked up by less busy loops. This ensures progress, but should not be a hot path. The logic for moving between event loops or into the global queue is unfortunately quite complex.

Author: yawkat

http-client-core/src/main/java/io/micronaut/http/client/HttpClientConfiguration.java

@@ -915,7 +915,7 @@ public static class ConnectionPoolConfiguration implements Toggleable {
         private int maxPendingConnections = 4;
 
         private int maxConcurrentRequestsPerHttp2Connection = Integer.MAX_VALUE;
-        private int maxConcurrentHttp1Connections = Integer.MAX_VALUE;
+        private int maxConcurrentHttp1Connections = 1000;
         private int maxConcurrentHttp2Connections = 1;
 
         private int maxPendingAcquires = Integer.MAX_VALUE;
@@ -927,6 +927,9 @@ public static class ConnectionPoolConfiguration implements Toggleable {
         @NonNull
         private HttpClientConfiguration.ConnectionPoolConfiguration.ConnectionLocality connectionLocality = ConnectionLocality.PREFERRED;
 
+        @NonNull
+        private PoolVersion version = PoolVersion.V4_9;
+
         /**
          * Whether connection pooling is enabled.
          * [available in the Netty HTTP client]
@@ -1088,6 +1091,26 @@ public void setConnectionLocality(@NonNull HttpClientConfiguration.ConnectionPoo
             this.connectionLocality = connectionLocality;
         }
 
+        /**
+         * The version of the connection pool implementation. Defaults to {@code V4_9}, can be set
+         * to {@code V4_0} for compatibility.
+         *
+         * @return The pool version
+         */
+        public @NonNull PoolVersion getVersion() {
+            return version;
+        }
+
+        /**
+         * The version of the connection pool implementation. Defaults to {@code V4_9}, can be set
+         * to {@code V4_0} for compatibility.
+         *
+         * @param version The pool version
+         */
+        public void setVersion(@NonNull PoolVersion version) {
+            this.version = version;
+        }
+
         /**
          * Options for {@link #connectionLocality}.
          *
@@ -1120,6 +1143,22 @@ public enum ConnectionLocality {
              */
             ENFORCED_ALWAYS,
         }
+
+        /**
+         * Different pool implementation versions, for compatibility.
+         *
+         * @since 4.9.0
+         */
+        public enum PoolVersion {
+            /**
+             * The connection pool introduced in micronaut-core 4.0.0.
+             */
+            V4_0,
+            /**
+             * The connection pool introduced in micronaut-core 4.9.0.
+             */
+            V4_9
+        }
     }
 
     /**

http-client-core/src/main/java/io/micronaut/http/client/loadbalance/FixedLoadBalancer.java

@@ -15,6 +15,7 @@
  */
 package io.micronaut.http.client.loadbalance;
 
+import io.micronaut.core.annotation.Internal;
 import io.micronaut.core.annotation.Nullable;
 import io.micronaut.core.async.publisher.Publishers;
 import io.micronaut.discovery.ServiceInstance;
@@ -36,6 +37,7 @@
  */
 public class FixedLoadBalancer implements LoadBalancer  {
     private final Publisher<ServiceInstance> publisher;
+    private final ServiceInstance serviceInstance;
     private final URI uri;
 
     /**
@@ -56,7 +58,8 @@ public FixedLoadBalancer(URL url) {
      */
     public FixedLoadBalancer(URI uri) {
         this.uri = uri;
-        this.publisher = Publishers.just(ServiceInstance.of(uri.getHost(), uri));
+        serviceInstance = ServiceInstance.of(uri.getHost(), uri);
+        this.publisher = Publishers.just(serviceInstance);
     }
 
     @Override
@@ -84,6 +87,16 @@ public URI getUri() {
         return uri;
     }
 
+    /**
+     * The fixed {@link ServiceInstance} returned by {@link #select}. Internal use only.
+     *
+     * @return The service instance
+     */
+    @Internal
+    public ServiceInstance getServiceInstance() {
+        return serviceInstance;
+    }
+
     @Override
     public Optional<String> getContextPath() {
         return Optional.ofNullable(getUri().getPath());

http-client/src/main/java/io/micronaut/http/client/netty/BlockHint.java

@@ -19,7 +19,7 @@
 import io.micronaut.core.annotation.NonNull;
 import io.micronaut.core.annotation.Nullable;
 import io.micronaut.http.client.exceptions.HttpClientException;
-import io.netty.channel.EventLoop;
+import io.netty.util.concurrent.EventExecutor;
 
 /**
  * Information about what threads are blocked waiting for a request to complete. This is used to
@@ -64,7 +64,7 @@ static HttpClientException createException() {
                 "https://docs.micronaut.io/latest/guide/index.html#clientConfiguration");
     }
 
-    boolean blocks(EventLoop eventLoop) {
+    boolean blocks(EventExecutor eventLoop) {
         BlockHint bh = this;
         while (bh != null) {
             if (eventLoop.inEventLoop(bh.blockedThread)) {