Repository: eclipse-vertx/vert.x

Problem statement:
HTTP client fixed event loop pool
HTTP client uses event loops provided by actual requests to assign connection event-loops. In some case it is desirable to have the client use its own event loop as a pool, like when it is shared between verticles or statically.
Shared HttpClient support
A few Vert.x clients (e.g. JDBC, Cassandra, Mongo) have `createShared` methods. The purpose of these creation methods is that, when you deploy multiple verticle instances, a single resource (JDBC datasource, Mongo client, Cassandra session) is created and reused.

While this requirement was obvious for Vert.x clients wrapping existing libraries, most built-in clients don't have them. So if a user has to open at most `C` connections to an HTTP backend, and deploys `V` verticle instances, it is advised to create an `HttpClient` with `C/V` as max pool size.

On the upside, the user will get maximum performance because the HttpClient will be operated on the verticle context. On the downside, the configuration process is not straightforward.

For Vert.x 4, HttpClient (and some other built-in clients too) have been made safe to use on different contexts. So it is possible to add `createShared` methods now.

The benefit is simplified configuration. The drawback is that this will require an extra hop between the context that actually created the `HttpClient` and the context using it. This should be acceptable by most apps though (the user pays at least this price when sending messages over the eventbus).


Hints:

keep in mind this is not in the 4.0 priority list and I want us to focus on 4.0 priority tasks.

Patch:
diff --git a/src/main/generated/io/vertx/core/http/HttpClientOptionsConverter.java b/src/main/generated/io/vertx/core/http/HttpClientOptionsConverter.java
--- a/src/main/generated/io/vertx/core/http/HttpClientOptionsConverter.java
+++ b/src/main/generated/io/vertx/core/http/HttpClientOptionsConverter.java
@@ -150,6 +150,11 @@ static void fromJson(Iterable<java.util.Map.Entry<String, Object>> json, HttpCli
             obj.setPoolCleanerPeriod(((Number)member.getValue()).intValue());
           }
           break;
+        case "poolEventLoopSize":
+          if (member.getValue() instanceof Number) {
+            obj.setPoolEventLoopSize(((Number)member.getValue()).intValue());
+          }
+          break;
         case "protocolVersion":
           if (member.getValue() instanceof String) {
             obj.setProtocolVersion(io.vertx.core.http.HttpVersion.valueOf((String)member.getValue()));
@@ -249,6 +254,7 @@ static void toJson(HttpClientOptions obj, java.util.Map<String, Object> json) {
     json.put("pipelining", obj.isPipelining());
     json.put("pipeliningLimit", obj.getPipeliningLimit());
     json.put("poolCleanerPeriod", obj.getPoolCleanerPeriod());
+    json.put("poolEventLoopSize", obj.getPoolEventLoopSize());
     if (obj.getProtocolVersion() != null) {
       json.put("protocolVersion", obj.getProtocolVersion().name());
     }
diff --git a/src/main/java/examples/HTTPExamples.java b/src/main/java/examples/HTTPExamples.java
--- a/src/main/java/examples/HTTPExamples.java
+++ b/src/main/java/examples/HTTPExamples.java
@@ -11,7 +11,9 @@
 
 package examples;
 
+import io.vertx.core.AbstractVerticle;
 import io.vertx.core.AsyncResult;
+import io.vertx.core.DeploymentOptions;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.MultiMap;
@@ -1163,4 +1165,38 @@ public static void setIdentityContentEncodingHeader(HttpServerRequest request) {
       .putHeader(HttpHeaders.CONTENT_ENCODING, HttpHeaders.IDENTITY)
       .sendFile("/path/to/image.jpg");
   }
+
+  public static void httpClientSharing1(Vertx vertx) {
+    HttpClient client = vertx.createSharedHttpClient();
+    vertx.deployVerticle(() -> new AbstractVerticle() {
+      @Override
+      public void start() throws Exception {
+        // Use the client
+      }
+    }, new DeploymentOptions().setInstances(4));
+  }
+
+  public static void httpClientSharing2(Vertx vertx) {
+    vertx.deployVerticle(() -> new AbstractVerticle() {
+      HttpClient client;
+      @Override
+      public void start() {
+        // Get or create a shared client
+        // this actually creates a lease to the client
+        // when the verticle is undeployed, the lease will be released automaticaly
+        client = vertx.createSharedHttpClient("my-client");
+      }
+    }, new DeploymentOptions().setInstances(4));
+  }
+
+  public static void httpClientSharing3(Vertx vertx) {
+    vertx.deployVerticle(() -> new AbstractVerticle() {
+      HttpClient client;
+      @Override
+      public void start() {
+        // The client creates and use two event-loops for 4 instances
+        client = vertx.createSharedHttpClient("my-client", new HttpClientOptions().setPoolEventLoopSize(2));
+      }
+    }, new DeploymentOptions().setInstances(4));
+  }
 }
diff --git a/src/main/java/io/vertx/core/Vertx.java b/src/main/java/io/vertx/core/Vertx.java
--- a/src/main/java/io/vertx/core/Vertx.java
+++ b/src/main/java/io/vertx/core/Vertx.java
@@ -176,6 +176,30 @@ static Future<Vertx> clusteredVertx(VertxOptions options) {
    */
   HttpClient createHttpClient(HttpClientOptions options);
 
+  /**
+   * Like {@link #createSharedHttpClient(HttpClientOptions)}, using default options.
+   */
+  HttpClient createSharedHttpClient();
+
+  /**
+   * Like {@link #createSharedHttpClient(String, HttpClientOptions)}, using the default shared client name.
+   */
+  HttpClient createSharedHttpClient(HttpClientOptions options);
+
+  /**
+   * Like {@link #createSharedHttpClient(String, HttpClientOptions)}, using default options.
+   */
+  HttpClient createSharedHttpClient(String name);
+
+  /**
+   * Create a HTTP/HTTPS client using the specified name and options.
+   *
+   * @param name    the shared client name
+   * @param options the options to use
+   * @return the client
+   */
+  HttpClient createSharedHttpClient(String name, HttpClientOptions options);
+
   /**
    * Create a HTTP/HTTPS client using default options
    *
diff --git a/src/main/java/io/vertx/core/http/HttpClientOptions.java b/src/main/java/io/vertx/core/http/HttpClientOptions.java
--- a/src/main/java/io/vertx/core/http/HttpClientOptions.java
+++ b/src/main/java/io/vertx/core/http/HttpClientOptions.java
@@ -198,6 +198,11 @@ public class HttpClientOptions extends ClientOptionsBase {
    */
   public static final int DEFAULT_POOL_CLEANER_PERIOD = 1000;
 
+  /**
+   * Default pool event loop size = 0 (reuse current event-loop)
+   */
+  public static final int DEFAULT_POOL_EVENT_LOOP_SIZE = 0;
+
   /**
    * Default WebSocket closing timeout = 10 second
    */
@@ -219,6 +224,7 @@ public class HttpClientOptions extends ClientOptionsBase {
   private int http2ConnectionWindowSize;
   private int http2KeepAliveTimeout;
   private int poolCleanerPeriod;
+  private int poolEventLoopSize;
 
   private boolean tryUseCompression;
   private int maxWebSocketFrameSize;
@@ -292,6 +298,7 @@ public HttpClientOptions(HttpClientOptions other) {
     this.forceSni = other.forceSni;
     this.decoderInitialBufferSize = other.getDecoderInitialBufferSize();
     this.poolCleanerPeriod = other.getPoolCleanerPeriod();
+    this.poolEventLoopSize = other.getPoolEventLoopSize();
     this.tryUsePerFrameWebSocketCompression = other.tryUsePerFrameWebSocketCompression;
     this.tryUsePerMessageWebSocketCompression = other.tryUsePerMessageWebSocketCompression;
     this.webSocketAllowClientNoContext = other.webSocketAllowClientNoContext;
@@ -359,6 +366,7 @@ private void init() {
     webSocketRequestServerNoContext = DEFAULT_WEBSOCKET_REQUEST_SERVER_NO_CONTEXT;
     webSocketClosingTimeout = DEFAULT_WEBSOCKET_CLOSING_TIMEOUT;
     poolCleanerPeriod = DEFAULT_POOL_CLEANER_PERIOD;
+    poolEventLoopSize = DEFAULT_POOL_EVENT_LOOP_SIZE;
     tracingPolicy = DEFAULT_TRACING_POLICY;
   }
 
@@ -1325,6 +1333,33 @@ public HttpClientOptions setPoolCleanerPeriod(int poolCleanerPeriod) {
     return this;
   }
 
+  /**
+   * @return the max number of event-loop a pool will use, the default value is {@code 0} which implies
+   * to reuse the current event-loop
+   */
+  public int getPoolEventLoopSize() {
+    return poolEventLoopSize;
+  }
+
+  /**
+   * Set the number of event-loop the pool use.
+   *
+   * <ul>
+   *   <li>when the size is {@code 0}, the client pool will use the current event-loop</li>
+   *   <li>otherwise the client will create and use its own event loop</li>
+   * </ul>
+   *
+   * The default size is {@code 0}.
+   *
+   * @param poolEventLoopSize  the new size
+   * @return a reference to this, so the API can be used fluently
+   */
+  public HttpClientOptions setPoolEventLoopSize(int poolEventLoopSize) {
+    Arguments.require(poolEventLoopSize >= 0, "poolEventLoopSize must be >= 0");
+    this.poolEventLoopSize = poolEventLoopSize;
+    return this;
+  }
+
   /**
    * @return the tracing policy
    */
diff --git a/src/main/java/io/vertx/core/http/impl/HttpClientHolder.java b/src/main/java/io/vertx/core/http/impl/HttpClientHolder.java
new file mode 100644
--- /dev/null
+++ b/src/main/java/io/vertx/core/http/impl/HttpClientHolder.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright (c) 2011-2021 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+
+package io.vertx.core.http.impl;
+
+import io.vertx.core.CompositeFuture;
+import io.vertx.core.Future;
+import io.vertx.core.http.HttpClient;
+import io.vertx.core.http.HttpClientOptions;
+import io.vertx.core.impl.CloseFuture;
+import io.vertx.core.impl.VertxInternal;
+import io.vertx.core.shareddata.Shareable;
+
+public class HttpClientHolder implements Shareable {
+
+  private final int count;
+  private final HttpClient client;
+  private final CloseFuture closeFuture;
+
+  public HttpClientHolder() {
+    count = 1;
+    client = null;
+    closeFuture = null;
+  }
+
+  private HttpClientHolder(int count, HttpClient client, CloseFuture closeFuture) {
+    this.count = count;
+    this.client = client;
+    this.closeFuture = closeFuture;
+  }
+
+  public HttpClient get() {
+    return client;
+  }
+
+  public HttpClientHolder increment() {
+    return client == null ? null : new HttpClientHolder(count + 1, client, closeFuture);
+  }
+
+  public HttpClientHolder init(VertxInternal vertx, HttpClientOptions options) {
+    CloseFuture closeFuture = new CloseFuture();
+    HttpClient client = new HttpClientImpl(vertx, options, closeFuture);
+    return new HttpClientHolder(count, client, closeFuture);
+  }
+
+  public HttpClientHolder decrement() {
+    return count == 1 ? null : new HttpClientHolder(count - 1, client, closeFuture);
+  }
+
+  public Future<Void> close() {
+    return CompositeFuture.all(client.close(), closeFuture.close()).mapEmpty();
+  }
+
+  @Override
+  public String toString() {
+    return "HttpClientHolder{" +
+      "count=" + count +
+      '}';
+  }
+}
diff --git a/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java b/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java
--- a/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java
+++ b/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java
@@ -29,6 +29,7 @@
 import io.vertx.core.net.ProxyOptions;
 import io.vertx.core.net.ProxyType;
 import io.vertx.core.net.SocketAddress;
+import io.vertx.core.net.impl.pool.ConnectionPool;
 import io.vertx.core.net.impl.pool.Endpoint;
 import io.vertx.core.net.impl.pool.Lease;
 import io.vertx.core.spi.metrics.ClientMetrics;
@@ -44,6 +45,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -132,6 +134,7 @@ public class HttpClientImpl implements HttpClient, MetricsProvider, Closeable {
   private Predicate<SocketAddress> proxyFilter;
   private volatile Handler<HttpConnection> connectionHandler;
   private volatile Function<HttpClientResponse, Future<RequestOptions>> redirectHandler = DEFAULT_HANDLER;
+  private final Function<ContextInternal, EventLoopContext> contextProvider;
 
   public HttpClientImpl(VertxInternal vertx, HttpClientOptions options, CloseFuture closeFuture) {
     this.vertx = vertx;
@@ -169,7 +172,22 @@ public HttpClientImpl(VertxInternal vertx, HttpClientOptions options, CloseFutur
     httpCM = httpConnectionManager();
     if (options.getPoolCleanerPeriod() > 0 && (options.getKeepAliveTimeout() > 0L || options.getHttp2KeepAliveTimeout() > 0L)) {
       PoolChecker checker = new PoolChecker(this);
-      timerID = vertx.setTimer(options.getPoolCleanerPeriod(), checker);
+      ContextInternal timerContext = vertx.createEventLoopContext();
+      timerID = timerContext.setTimer(options.getPoolCleanerPeriod(), checker);
+    }
+    int eventLoopSize = options.getPoolEventLoopSize();
+    if (eventLoopSize > 0) {
+      EventLoopContext[] eventLoops = new EventLoopContext[eventLoopSize];
+      for (int i = 0;i < eventLoopSize;i++) {
+        eventLoops[i] = vertx.createEventLoopContext();
+      }
+      AtomicInteger idx = new AtomicInteger();
+      contextProvider = ctx -> {
+        int i = idx.getAndIncrement();
+        return eventLoops[i % eventLoopSize];
+      };
+    } else {
+      contextProvider = ConnectionPool.EVENT_LOOP_CONTEXT_PROVIDER;
     }
 
     closeFuture.add(netClient);
@@ -230,6 +248,10 @@ private ConnectionManager<EndpointKey, HttpClientConnection> webSocketConnection
     });
   }
 
+  Function<ContextInternal, EventLoopContext> contextProvider() {
+    return contextProvider;
+  }
+
   private int getPort(RequestOptions request) {
     Integer port = request.getPort();
     if (port != null) {
diff --git a/src/main/java/io/vertx/core/http/impl/SharedClientHttpStreamEndpoint.java b/src/main/java/io/vertx/core/http/impl/SharedClientHttpStreamEndpoint.java
--- a/src/main/java/io/vertx/core/http/impl/SharedClientHttpStreamEndpoint.java
+++ b/src/main/java/io/vertx/core/http/impl/SharedClientHttpStreamEndpoint.java
@@ -68,10 +68,13 @@ public SharedClientHttpStreamEndpoint(HttpClientImpl client,
                                         HttpChannelConnector connector,
                                         Runnable dispose) {
     super(metrics, dispose);
+
+    ConnectionPool<HttpClientConnection> pool = ConnectionPool.pool(this, new int[]{http1MaxSize, http2MaxSize}, queueMaxSize)
+      .connectionSelector(LIFO_SELECTOR).contextProvider(client.contextProvider());
+
     this.client = client;
     this.connector = connector;
-    this.pool = ConnectionPool.pool(this, new int[] { http1MaxSize, http2MaxSize }, queueMaxSize)
-      .connectionSelector(LIFO_SELECTOR);
+    this.pool = pool;
   }
 
   @Override
diff --git a/src/main/java/io/vertx/core/http/impl/SharedHttpClient.java b/src/main/java/io/vertx/core/http/impl/SharedHttpClient.java
new file mode 100644
--- /dev/null
+++ b/src/main/java/io/vertx/core/http/impl/SharedHttpClient.java
@@ -0,0 +1,207 @@
+/*
+ * Copyright (c) 2011-2021 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+
+package io.vertx.core.http.impl;
+
+import io.vertx.core.*;
+import io.vertx.core.http.*;
+import io.vertx.core.impl.VertxInternal;
+import io.vertx.core.shareddata.LocalMap;
+
+import java.util.List;
+import java.util.function.Function;
+
+public class SharedHttpClient implements HttpClient, Closeable {
+
+  private static final String MAP_NAME = "__vertx.shared.httpClients";
+  public static final String DEFAULT_CLIENT_NAME = "SharedHttpClient.DEFAULT";
+
+  private final VertxInternal vertx;
+  private final String name;
+  private final HttpClient delegate;
+
+  private SharedHttpClient(VertxInternal vertx, String name, HttpClient delegate) {
+    this.vertx = vertx;
+    this.name = name;
+    this.delegate = delegate;
+  }
+
+  public static SharedHttpClient create(VertxInternal vertx, String name, HttpClientOptions options) {
+    LocalMap<String, HttpClientHolder> localMap = vertx.sharedData().getLocalMap(MAP_NAME);
+    HttpClient client;
+    HttpClientHolder current, candidate;
+    for (; ; ) {
+      current = localMap.get(name);
+      if (current != null) {
+        candidate = current.increment();
+        if (candidate != null && localMap.replaceIfPresent(name, current, candidate)) {
+          client = candidate.get();
+          break;
+        }
+      } else {
+        candidate = new HttpClientHolder();
+        if (localMap.putIfAbsent(name, candidate) == null) {
+          candidate = candidate.init(vertx, options);
+          client = candidate.get();
+          localMap.put(name, candidate);
+          break;
+        }
+      }
+    }
+    return new SharedHttpClient(vertx, name, client);
+  }
+
+  @Override
+  public void close(Handler<AsyncResult<Void>> handler) {
+    Future<Void> future = close();
+    if (handler != null) {
+      future.onComplete(handler);
+    }
+  }
+
+  @Override
+  public Future<Void> close() {
+    Promise<Void> promise = vertx.promise();
+    LocalMap<String, HttpClientHolder> localMap = vertx.sharedData().getLocalMap(MAP_NAME);
+    HttpClientHolder current, candidate;
+    for (; ; ) {
+      current = localMap.get(name);
+      candidate = current.decrement();
+      if (candidate == null) {
+        if (localMap.removeIfPresent(name, current)) {
+          current.close().onComplete(promise);
+          break;
+        }
+      } else if (localMap.replace(name, current, candidate)) {
+        promise.complete();
+        break;
+      }
+    }
+    return promise.future();
+  }
+
+  @Override
+  public void close(Promise<Void> completion) {
+    close().onComplete(completion);
+  }
+
+  @Override
+  protected void finalize() throws Throwable {
+    // Make sure the shared client count gets decreased if there are no more references to this instance
+    close();
+    super.finalize();
+  }
+
+  @Override
+  public void request(RequestOptions options, Handler<AsyncResult<HttpClientRequest>> handler) {
+    delegate.request(options, handler);
+  }
+
+  @Override
+  public Future<HttpClientRequest> request(RequestOptions options) {
+    return delegate.request(options);
+  }
+
+  @Override
+  public void request(HttpMethod method, int port, String host, String requestURI, Handler<AsyncResult<HttpClientRequest>> handler) {
+    delegate.request(method, port, host, requestURI, handler);
+  }
+
+  @Override
+  public Future<HttpClientRequest> request(HttpMethod method, int port, String host, String requestURI) {
+    return delegate.request(method, port, host, requestURI);
+  }
+
+  @Override
+  public void request(HttpMethod method, String host, String requestURI, Handler<AsyncResult<HttpClientRequest>> handler) {
+    delegate.request(method, host, requestURI, handler);
+  }
+
+  @Override
+  public Future<HttpClientRequest> request(HttpMethod method, String host, String requestURI) {
+    return delegate.request(method, host, requestURI);
+  }
+
+  @Override
+  public void request(HttpMethod method, String requestURI, Handler<AsyncResult<HttpClientRequest>> handler) {
+    delegate.request(method, requestURI, handler);
+  }
+
+  @Override
+  public Future<HttpClientRequest> request(HttpMethod method, String requestURI) {
+    return delegate.request(method, requestURI);
+  }
+
+  @Override
+  public void webSocket(int port, String host, String requestURI, Handler<AsyncResult<WebSocket>> handler) {
+    delegate.webSocket(port, host, requestURI, handler);
+  }
+
+  @Override
+  public Future<WebSocket> webSocket(int port, String host, String requestURI) {
+    return delegate.webSocket(port, host, requestURI);
+  }
+
+  @Override
+  public void webSocket(String host, String requestURI, Handler<AsyncResult<WebSocket>> handler) {
+    delegate.webSocket(host, requestURI, handler);
+  }
+
+  @Override
+  public Future<WebSocket> webSocket(String host, String requestURI) {
+    return delegate.webSocket(host, requestURI);
+  }
+
+  @Override
+  public void webSocket(String requestURI, Handler<AsyncResult<WebSocket>> handler) {
+    delegate.webSocket(requestURI, handler);
+  }
+
+  @Override
+  public Future<WebSocket> webSocket(String requestURI) {
+    return delegate.webSocket(requestURI);
+  }
+
+  @Override
+  public void webSocket(WebSocketConnectOptions options, Handler<AsyncResult<WebSocket>> handler) {
+    delegate.webSocket(options, handler);
+  }
+
+  @Override
+  public Future<WebSocket> webSocket(WebSocketConnectOptions options) {
+    return delegate.webSocket(options);
+  }
+
+  @Override
+  public void webSocketAbs(String url, MultiMap headers, WebsocketVersion version, List<String> subProtocols, Handler<AsyncResult<WebSocket>> handler) {
+    delegate.webSocketAbs(url, headers, version, subProtocols, handler);
+  }
+
+  @Override
+  public Future<WebSocket> webSocketAbs(String url, MultiMap headers, WebsocketVersion version, List<String> subProtocols) {
+    return delegate.webSocketAbs(url, headers, version, subProtocols);
+  }
+
+  @Override
+  public HttpClient connectionHandler(Handler<HttpConnection> handler) {
+    return delegate.connectionHandler(handler);
+  }
+
+  @Override
+  public HttpClient redirectHandler(Function<HttpClientResponse, Future<RequestOptions>> handler) {
+    return delegate.redirectHandler(handler);
+  }
+
+  @Override
+  public Function<HttpClientResponse, Future<RequestOptions>> redirectHandler() {
+    return delegate.redirectHandler();
+  }
+}
diff --git a/src/main/java/io/vertx/core/impl/VertxImpl.java b/src/main/java/io/vertx/core/impl/VertxImpl.java
--- a/src/main/java/io/vertx/core/impl/VertxImpl.java
+++ b/src/main/java/io/vertx/core/impl/VertxImpl.java
@@ -43,6 +43,7 @@
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.core.http.impl.HttpClientImpl;
 import io.vertx.core.http.impl.HttpServerImpl;
+import io.vertx.core.http.impl.SharedHttpClient;
 import io.vertx.core.impl.future.PromiseInternal;
 import io.vertx.core.impl.logging.Logger;
 import io.vertx.core.impl.logging.LoggerFactory;
@@ -74,7 +75,10 @@
 import java.lang.ref.WeakReference;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -361,6 +365,31 @@ public HttpClient createHttpClient(HttpClientOptions options) {
     return client;
   }
 
+  @Override
+  public HttpClient createSharedHttpClient() {
+    return createSharedHttpClient(SharedHttpClient.DEFAULT_CLIENT_NAME, new HttpClientOptions());
+  }
+
+  @Override
+  public HttpClient createSharedHttpClient(HttpClientOptions options) {
+    return createSharedHttpClient(SharedHttpClient.DEFAULT_CLIENT_NAME, options);
+  }
+
+  @Override
+  public HttpClient createSharedHttpClient(String name) {
+    return createSharedHttpClient(name, new HttpClientOptions());
+  }
+
+  @Override
+  public HttpClient createSharedHttpClient(String name, HttpClientOptions options) {
+    CloseFuture closeFuture = new CloseFuture(log);
+    SharedHttpClient client = SharedHttpClient.create(this, name, options);
+    closeFuture.add(client);
+    CloseFuture parentCloseFuture = resolveCloseFuture();
+    parentCloseFuture.add(closeFuture);
+    return client;
+  }
+
   @Override
   public HttpClient createHttpClient() {
     return createHttpClient(new HttpClientOptions());


Test patch:
diff --git a/src/test/java/io/vertx/core/http/Http1xTest.java b/src/test/java/io/vertx/core/http/Http1xTest.java
--- a/src/test/java/io/vertx/core/http/Http1xTest.java
+++ b/src/test/java/io/vertx/core/http/Http1xTest.java
@@ -13,6 +13,7 @@
 
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
+import io.netty.channel.EventLoop;
 import io.netty.handler.codec.TooLongFrameException;
 import io.vertx.core.*;
 import io.vertx.core.Future;
@@ -5181,4 +5182,34 @@ private void testNetSocketUpgradeSuccess(Buffer payload) {
     }));
     await();
   }
+
+  @Test
+  public void testClientEventLoopSize() throws Exception {
+    Assume.assumeTrue("Domain socket don't pass this test", testAddress.isInetSocket());
+    server.requestHandler(req -> {
+      req.response().end();
+    });
+    startServer();
+    int size = 4;
+    int maxPoolSize = size + 2;
+    client.close();
+    client = vertx.createHttpClient(new HttpClientOptions()
+      .setMaxPoolSize(maxPoolSize)
+      .setPoolEventLoopSize(size));
+    List<EventLoop> eventLoops = Collections.synchronizedList(new ArrayList<>());
+    client.connectionHandler(conn -> eventLoops.add(((ContextInternal)Vertx.currentContext()).nettyEventLoop()));
+    List<Future> futures = new ArrayList<>();
+    for (int i = 0;i < size * 2;i++) {
+      futures.add(client
+        .request(requestOptions)
+        .compose(HttpClientRequest::send)
+        .compose(HttpClientResponse::body));
+    }
+    CompositeFuture.all(futures).onComplete(onSuccess(v -> {
+      assertEquals(maxPoolSize, eventLoops.size());
+      assertEquals(size, new HashSet<>(eventLoops).size());
+      testComplete();
+    }));
+    await();
+  }
 }
diff --git a/src/test/java/io/vertx/core/http/SharedHttpClientTest.java b/src/test/java/io/vertx/core/http/SharedHttpClientTest.java
new file mode 100644
--- /dev/null
+++ b/src/test/java/io/vertx/core/http/SharedHttpClientTest.java
@@ -0,0 +1,228 @@
+/*
+ * Copyright (c) 2011-2021 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+
+package io.vertx.core.http;
+
+import io.vertx.core.*;
+import io.vertx.core.eventbus.Message;
+import io.vertx.core.impl.VertxInternal;
+import io.vertx.core.json.JsonObject;
+import io.vertx.test.core.VertxTestBase;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import static io.vertx.core.http.HttpMethod.GET;
+
+public class SharedHttpClientTest extends VertxTestBase {
+
+  int sharedPoolSize = 7;
+  int clientVerticleInstances = 8;
+
+  int requestsPerVerticle = 50 * sharedPoolSize;
+  int requestsTotal = clientVerticleInstances * requestsPerVerticle;
+
+  @Test
+  public void testVerticlesUseSamePool() throws Exception {
+    CountDownLatch receivedLatch = new CountDownLatch(requestsTotal);
+    ServerVerticle serverVerticle = new ServerVerticle();
+
+    vertx.deployVerticle(serverVerticle, onSuccess(serverId -> {
+
+      HttpClientOptions clientOptions = httpClientOptions(serverVerticle, sharedPoolSize);
+      DeploymentOptions deploymentOptions = deploymentOptions(clientVerticleInstances, clientOptions);
+
+      Supplier<Verticle> verticleSupplier = () -> new ClientVerticle(clientVerticle -> {
+        // Verify the reply context is the same as of the deployment
+        // We can't compare to the verticle context because the reply context is a DuplicatedContext
+        assertEquals(clientVerticle.context.deploymentID(), Vertx.currentContext().deploymentID());
+        receivedLatch.countDown();
+      });
+
+      vertx.deployVerticle(verticleSupplier, deploymentOptions, onSuccess(clientId -> {
+        vertx.eventBus().publish(ClientVerticle.TRIGGER_ADDRESS, requestsPerVerticle);
+      }));
+    }));
+
+    waitUntil(() -> serverVerticle.connections.size() == sharedPoolSize);
+    serverVerticle.replyLatch.complete();
+    awaitLatch(receivedLatch);
+    assertEquals(serverVerticle.maxConnections, sharedPoolSize);
+  }
+
+  @Test
+  public void testSharedPoolClosedAutomatically() throws Exception {
+    CountDownLatch receivedLatch = new CountDownLatch(requestsTotal);
+    ServerVerticle serverVerticle = new ServerVerticle();
+    AtomicReference<String> clientDeploymentId = new AtomicReference<>();
+
+    vertx.deployVerticle(serverVerticle, onSuccess(serverId -> {
+
+      HttpClientOptions clientOptions = httpClientOptions(serverVerticle, sharedPoolSize)
+        // Make sure connections stay alive for the duration of the test if the server is not closed
+        .setKeepAliveTimeout(3600);
+      DeploymentOptions deploymentOptions = deploymentOptions(clientVerticleInstances, clientOptions);
+
+      Supplier<Verticle> verticleSupplier = () -> new ClientVerticle(clientVerticle -> receivedLatch.countDown());
+
+      vertx.deployVerticle(verticleSupplier, deploymentOptions, onSuccess(clientId -> {
+        clientDeploymentId.set(clientId);
+        vertx.eventBus().publish(ClientVerticle.TRIGGER_ADDRESS, requestsPerVerticle);
+      }));
+    }));
+
+    waitUntil(() -> serverVerticle.connections.size() == sharedPoolSize);
+    serverVerticle.replyLatch.complete();
+    awaitLatch(receivedLatch);
+
+    CountDownLatch undeployLatch = new CountDownLatch(1);
+    vertx.undeploy(clientDeploymentId.get(), onSuccess(v -> {
+      undeployLatch.countDown();
+    }));
+
+    awaitLatch(undeployLatch);
+    assertWaitUntil(() -> serverVerticle.connections.size() == 0);
+  }
+
+  @Test
+  public void testSharedPoolRetainedByOtherDeployment() throws Exception {
+    int keepAliveTimeoutSeconds = 3;
+
+    CountDownLatch receivedLatch = new CountDownLatch(requestsTotal);
+    ServerVerticle serverVerticle = new ServerVerticle();
+    AtomicReference<String> clientDeploymentId = new AtomicReference<>();
+
+    vertx.deployVerticle(serverVerticle, onSuccess(serverId -> {
+
+      HttpClientOptions clientOptions = httpClientOptions(serverVerticle, sharedPoolSize)
+        .setKeepAliveTimeout(keepAliveTimeoutSeconds);
+      DeploymentOptions deploymentOptions = deploymentOptions(clientVerticleInstances, clientOptions);
+
+      Supplier<Verticle> verticleSupplier = () -> new ClientVerticle(clientVerticle -> receivedLatch.countDown());
+
+      vertx.deployVerticle(verticleSupplier, deploymentOptions, onSuccess(clientId -> {
+        clientDeploymentId.set(clientId);
+        vertx.eventBus().publish(ClientVerticle.TRIGGER_ADDRESS, requestsPerVerticle);
+      }));
+    }));
+
+    waitUntil(() -> serverVerticle.connections.size() == sharedPoolSize);
+
+    CountDownLatch deployLatch = new CountDownLatch(1);
+    vertx.deployVerticle(new AbstractVerticle() {
+      @Override
+      public void start() throws Exception {
+        vertx.createSharedHttpClient(ClientVerticle.SHARED_CLIENT_NAME, new HttpClientOptions());
+      }
+    }, onSuccess(v -> {
+      deployLatch.countDown();
+    }));
+    awaitLatch(deployLatch);
+
+    serverVerticle.replyLatch.complete();
+    awaitLatch(receivedLatch);
+
+    CountDownLatch undeployLatch = new CountDownLatch(1);
+    vertx.undeploy(clientDeploymentId.get(), onSuccess(v -> {
+      undeployLatch.countDown();
+    }));
+
+    awaitLatch(undeployLatch);
+
+    waitFor(2);
+    vertx.setTimer((1000 * keepAliveTimeoutSeconds) / 2, l -> {
+      assertTrue(serverVerticle.connections.size() > 0);
+      complete();
+    });
+    vertx.setTimer(2 * 1000 * keepAliveTimeoutSeconds, l -> {
+      assertTrue(serverVerticle.connections.size() == 0);
+      complete();
+    });
+    await();
+  }
+
+  private static class ClientVerticle extends AbstractVerticle implements Handler<Message<Integer>> {
+
+    static final String TRIGGER_ADDRESS = UUID.randomUUID().toString();
+    static final String SHARED_CLIENT_NAME = UUID.randomUUID().toString();
+
+    final Consumer<ClientVerticle> onResponseReceived;
+
+    volatile Context context;
+    HttpClient client;
+
+    ClientVerticle(Consumer<ClientVerticle> onResponseReceived) {
+      this.onResponseReceived = onResponseReceived;
+    }
+
+    @Override
+    public void start(Promise<Void> startPromise) throws Exception {
+      context = super.context;
+      client = vertx.createSharedHttpClient(SHARED_CLIENT_NAME, new HttpClientOptions(config().getJsonObject("httpClientOptions")));
+      vertx.eventBus().consumer(TRIGGER_ADDRESS, this).completionHandler(startPromise);
+    }
+
+    @Override
+    public void handle(Message<Integer> message) {
+      for (int i = 0; i < message.body(); i++) {
+        client.request(GET, "/").compose(HttpClientRequest::send).onComplete(ar -> onResponseReceived.accept(this));
+      }
+    }
+  }
+
+  private static class ServerVerticle extends AbstractVerticle implements Handler<HttpServerRequest> {
+
+    volatile Promise<Void> replyLatch;
+    volatile int port;
+    Set<HttpConnection> connections = Collections.synchronizedSet(new HashSet<>());
+    volatile int maxConnections;
+
+    @Override
+    public void start(Promise<Void> startPromise) throws Exception {
+      replyLatch = ((VertxInternal) vertx).promise();
+      vertx.createHttpServer()
+        .requestHandler(this)
+        .listen(0)
+        .onSuccess(server -> port = server.actualPort())
+        .<Void>mapEmpty()
+        .onComplete(startPromise);
+    }
+
+    @Override
+    public void handle(HttpServerRequest req) {
+      HttpConnection connection = req.connection();
+      connections.add(connection);
+      connection.closeHandler(v -> connections.remove(connection));
+      maxConnections = Math.max(maxConnections, connections.size());
+      replyLatch.future().onComplete(ar -> req.response().end());
+    }
+  }
+
+  private static HttpClientOptions httpClientOptions(ServerVerticle serverVerticle, int sharedPoolSize) {
+    return new HttpClientOptions()
+      .setDefaultPort(serverVerticle.port)
+      .setMaxPoolSize(sharedPoolSize);
+  }
+
+  private static DeploymentOptions deploymentOptions(int instances, HttpClientOptions options) {
+    return new DeploymentOptions()
+      .setInstances(instances)
+      .setConfig(new JsonObject()
+        .put("httpClientOptions", options.toJson()));
+  }
+}