Repository: eclipse-vertx/vert.x

Problem statement:
Resource lifecycle improvements
The core client lifecycle (`HttpClient` and `NetClient`) is correctly not bound to the Vert.x instance, i.e some clients might be not closed after a Vert.x instance is closed. This can raise an issue when a client is closed afterward and the related resources (event-loop) are not available anymore resulting in tasks rejections (exceptions).

Such exceptions can also be triggered by client finalisation that attempt to close a client when it is not referenced anymore.

Changes:

- any client will be closed when the creating scope is closed which can be
  - the verticle that created this client
  - the vertx instance
  - the event-bus (that creates net client)
- the referenced client should not prevent GC gc-ing them when they are not reachable anymore which imply that the close hooks should use weak references
- a few useless finalisers are removed because there is no need to clean anything
- non deployment context do not need to create a close hooks because they will not be closed and thus we don't need to create useless objects
- TCP servers don't need to to be closed anymore in the vertx close process as they can be added as shutdown hooks of the vertx instance (that is a simplification)



Hints:

Patch:
diff --git a/src/main/java/io/vertx/core/Context.java b/src/main/java/io/vertx/core/Context.java
--- a/src/main/java/io/vertx/core/Context.java
+++ b/src/main/java/io/vertx/core/Context.java
@@ -276,6 +276,6 @@ static boolean isOnVertxThread() {
   void addCloseHook(Closeable hook);
 
   @GenIgnore(GenIgnore.PERMITTED_TYPE)
-  boolean removeCloseHook(Closeable hook);
+  void removeCloseHook(Closeable hook);
 
 }
diff --git a/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java b/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
--- a/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
+++ b/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
@@ -477,15 +477,5 @@ public void close(Promise<Void> completion) {
     }
 
   }
-
-  @Override
-  protected void finalize() throws Throwable {
-    // Make sure this gets cleaned up if there are no more references to it
-    // so as not to leave connections and resources dangling until the system is shutdown
-    // which could make the JVM run out of file handles.
-    close(Promise.promise());
-    super.finalize();
-  }
-
 }
 
diff --git a/src/main/java/io/vertx/core/eventbus/impl/MessageProducerImpl.java b/src/main/java/io/vertx/core/eventbus/impl/MessageProducerImpl.java
--- a/src/main/java/io/vertx/core/eventbus/impl/MessageProducerImpl.java
+++ b/src/main/java/io/vertx/core/eventbus/impl/MessageProducerImpl.java
@@ -83,11 +83,4 @@ public void close(Handler<AsyncResult<Void>> handler) {
       fut.onComplete(handler);
     }
   }
-
-  // Just in case user forget to call close()
-  @Override
-  protected void finalize() throws Throwable {
-    close();
-    super.finalize();
-  }
 }
diff --git a/src/main/java/io/vertx/core/eventbus/impl/clustered/ConnectionHolder.java b/src/main/java/io/vertx/core/eventbus/impl/clustered/ConnectionHolder.java
--- a/src/main/java/io/vertx/core/eventbus/impl/clustered/ConnectionHolder.java
+++ b/src/main/java/io/vertx/core/eventbus/impl/clustered/ConnectionHolder.java
@@ -59,7 +59,7 @@ class ConnectionHolder {
     NetClientOptions clientOptions = new NetClientOptions(options.toJson());
     ClusteredEventBus.setCertOptions(clientOptions, options.getKeyCertOptions());
     ClusteredEventBus.setTrustOptions(clientOptions, options.getTrustOptions());
-    client = new NetClientImpl(eventBus.vertx(), clientOptions, false);
+    client = new NetClientImpl(eventBus.vertx(), clientOptions, null);
   }
 
   void connect() {
diff --git a/src/main/java/io/vertx/core/http/HttpClient.java b/src/main/java/io/vertx/core/http/HttpClient.java
--- a/src/main/java/io/vertx/core/http/HttpClient.java
+++ b/src/main/java/io/vertx/core/http/HttpClient.java
@@ -1387,4 +1387,9 @@ public interface HttpClient extends Measured {
    */
   Future<Void> close();
 
+  /**
+   * @return a future closed when the client is closed
+   */
+  Future<Void> closeFuture();
+
 }
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
@@ -14,7 +14,6 @@
 import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.group.ChannelGroupFuture;
 import io.netty.channel.group.DefaultChannelGroup;
-import io.netty.util.concurrent.GlobalEventExecutor;
 import io.vertx.core.Closeable;
 import io.vertx.core.Context;
 import io.vertx.core.Future;
@@ -23,6 +22,7 @@
 import io.vertx.core.*;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.*;
+import io.vertx.core.impl.CloseHooks;
 import io.vertx.core.net.impl.clientconnection.ConnectionManager;
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.impl.PromiseInternal;
@@ -40,6 +40,7 @@
 import io.vertx.core.spi.metrics.MetricsProvider;
 import io.vertx.core.streams.ReadStream;
 
+import java.lang.ref.WeakReference;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.Arrays;
@@ -57,7 +58,7 @@
  *
  * @author <a href="http://tfox.org">Tim Fox</a>
  */
-public class HttpClientImpl implements HttpClient, MetricsProvider {
+public class HttpClientImpl implements HttpClient, MetricsProvider, Closeable {
 
   // Pattern to check we are not dealing with an absoluate URI
   private static final Pattern ABS_URI_START_PATTERN = Pattern.compile("^\\p{Alpha}[\\p{Alpha}\\p{Digit}+.\\-]*:");
@@ -111,25 +112,26 @@ public class HttpClientImpl implements HttpClient, MetricsProvider {
   private final VertxInternal vertx;
   private final ChannelGroup channelGroup;
   private final HttpClientOptions options;
-  private final ContextInternal context;
   private final ConnectionManager<EndpointKey, HttpClientConnection> webSocketCM;
   private final ConnectionManager<EndpointKey, HttpClientConnection> httpCM;
-  private final Closeable closeHook;
+  private final CloseHooks closeHooks;
   private final ProxyType proxyType;
   private final SSLHelper sslHelper;
   private final HttpClientMetrics metrics;
   private final boolean keepAlive;
   private final boolean pipelining;
+  private final PromiseInternal<Void> closePromise;
+  private final Future<Void> closeFuture;
   private long timerID;
   private volatile boolean closed;
   private volatile Handler<HttpConnection> connectionHandler;
   private volatile Function<HttpClientResponse, Future<HttpClientRequest>> redirectHandler = DEFAULT_HANDLER;
 
-  public HttpClientImpl(VertxInternal vertx, HttpClientOptions options) {
+  public HttpClientImpl(VertxInternal vertx, CloseHooks hooks, HttpClientOptions options) {
     this.vertx = vertx;
     this.metrics = vertx.metricsSPI() != null ? vertx.metricsSPI().createHttpClientMetrics(options) : null;
     this.options = new HttpClientOptions(options);
-    this.channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
+    this.channelGroup = new DefaultChannelGroup(vertx.getAcceptorEventLoopGroup().next());
     List<HttpVersion> alpnVersions = options.getAlpnVersions();
     if (alpnVersions == null || alpnVersions.isEmpty()) {
       switch (options.getProtocolVersion()) {
@@ -146,33 +148,56 @@ public HttpClientImpl(VertxInternal vertx, HttpClientOptions options) {
     this.sslHelper = new SSLHelper(options, options.getKeyCertOptions(), options.getTrustOptions()).
         setApplicationProtocols(alpnVersions);
     sslHelper.validate(vertx);
-    context = vertx.getOrCreateContext();
-    closeHook = completionHandler -> {
-      HttpClientImpl.this.close();
-      completionHandler.handle(Future.succeededFuture());
-    };
     if(options.getProtocolVersion() == HttpVersion.HTTP_2 && Context.isOnWorkerThread()) {
       throw new IllegalStateException("Cannot use HttpClient with HTTP_2 in a worker");
     }
-    if (context.deploymentID() != null) {
-      context.addCloseHook(closeHook);
-    }
     if (!keepAlive && pipelining) {
       throw new IllegalStateException("Cannot have pipelining with no keep alive");
     }
+    closePromise = (PromiseInternal) Promise.promise();
+    if (metrics != null) {
+      closeFuture = closePromise.future().compose(v -> {
+        metrics.close();
+        return Future.succeededFuture();
+      });
+    } else {
+      closeFuture = closePromise.future();
+    }
+    closeHooks = hooks;
     webSocketCM = webSocketConnectionManager();
     httpCM = httpConnectionManager();
     proxyType = options.getProxyOptions() != null ? options.getProxyOptions().getType() : null;
     if (options.getPoolCleanerPeriod() > 0 && (options.getKeepAliveTimeout() > 0L || options.getHttp2KeepAliveTimeout() > 0L)) {
-      timerID = vertx.setTimer(options.getPoolCleanerPeriod(), id -> checkExpired());
+      PoolChecker checker = new PoolChecker(this);
+      timerID = vertx.setTimer(options.getPoolCleanerPeriod(), checker);
     }
   }
 
-  private void checkExpired() {
+  /**
+   * A weak ref to the client so it can be finalized.
+   */
+  private static class PoolChecker implements Handler<Long> {
+
+    final WeakReference<HttpClientImpl> ref;
+
+    private PoolChecker(HttpClientImpl client) {
+      ref = new WeakReference<>(client);
+    }
+
+    @Override
+    public void handle(Long event) {
+      HttpClientImpl client = ref.get();
+      if (client != null) {
+        client.checkExpired(this);
+      }
+    }
+  }
+
+  private void checkExpired(Handler<Long> checker) {
     httpCM.forEach(EXPIRED_CHECKER);
     synchronized (this) {
       if (!closed) {
-        timerID = vertx.setTimer(options.getPoolCleanerPeriod(), id -> checkExpired());
+        timerID = vertx.setTimer(options.getPoolCleanerPeriod(), checker);
       }
     }
   }
@@ -1189,43 +1214,50 @@ public Future<HttpClientResponse> delete(String requestURI) {
     return send(HttpMethod.DELETE, requestURI);
   }
 
+  @Override
+  public void close(Promise<Void> completion) {
+    boolean close;
+    synchronized (this) {
+      close = !closed;
+      if (close) {
+        closed = true;
+        if (timerID >= 0) {
+          vertx.cancelTimer(timerID);
+          timerID = -1;
+        }
+      }
+    }
+    if (close) {
+      webSocketCM.close();
+      httpCM.close();
+      ChannelGroupFuture fut = channelGroup.close();
+      fut.addListener(closePromise);
+    }
+    if (completion != null) {
+      closePromise.future().onComplete(completion);
+    }
+  }
+
   @Override
   public void close(Handler<AsyncResult<Void>> handler) {
+    if (closeHooks != null) {
+      closeHooks.remove(this);
+    }
     ContextInternal closingCtx = vertx.getOrCreateContext();
-    close(closingCtx.promise(handler));
+    close(handler != null ? closingCtx.promise(handler) : null);
   }
 
   @Override
   public Future<Void> close() {
+    if (closeHooks != null) {
+      closeHooks.remove(this);
+    }
     ContextInternal closingCtx = vertx.getOrCreateContext();
     Promise<Void> promise = closingCtx.promise();
     close(promise);
     return promise.future();
   }
 
-  private void close(PromiseInternal<Void> promise) {
-    synchronized (this) {
-      checkClosed();
-      closed = true;
-      if (timerID >= 0) {
-        vertx.cancelTimer(timerID);
-        timerID = -1;
-      }
-    }
-    if (context.deploymentID() != null) {
-      context.removeCloseHook(closeHook);
-    }
-    webSocketCM.close();
-    httpCM.close();
-    ChannelGroupFuture fut = channelGroup.close();
-    fut.addListener(promise);
-    promise.future().onComplete(ar -> {
-      if (metrics != null) {
-        metrics.close();
-      }
-    });
-  }
-
   @Override
   public boolean isMetricsEnabled() {
     return getMetrics() != null;
@@ -1344,12 +1376,17 @@ private synchronized void checkClosed() {
     }
   }
 
+  @Override
+  public Future<Void> closeFuture() {
+    return closePromise.future();
+  }
+
   @Override
   protected void finalize() throws Throwable {
     // Make sure this gets cleaned up if there are no more references to it
     // so as not to leave connections and resources dangling until the system is shutdown
     // which could make the JVM run out of file handles.
-    close((PromiseInternal<Void>) Promise.<Void>promise());
+    close((Handler<AsyncResult<Void>>) null);
     super.finalize();
   }
 }
diff --git a/src/main/java/io/vertx/core/http/impl/HttpServerImpl.java b/src/main/java/io/vertx/core/http/impl/HttpServerImpl.java
--- a/src/main/java/io/vertx/core/http/impl/HttpServerImpl.java
+++ b/src/main/java/io/vertx/core/http/impl/HttpServerImpl.java
@@ -265,15 +265,6 @@ public SSLHelper getSslHelper() {
     return sslHelper;
   }
 
-  @Override
-  protected void finalize() throws Throwable {
-    // Make sure this gets cleaned up if there are no more references to it
-    // so as not to leave connections and resources dangling until the system is shutdown
-    // which could make the JVM run out of file handles.
-    close();
-    super.finalize();
-  }
-
   boolean requestAccept() {
     return requestStream.accept();
   }
diff --git a/src/main/java/io/vertx/core/impl/CloseHooks.java b/src/main/java/io/vertx/core/impl/CloseHooks.java
--- a/src/main/java/io/vertx/core/impl/CloseHooks.java
+++ b/src/main/java/io/vertx/core/impl/CloseHooks.java
@@ -18,22 +18,24 @@
 import io.vertx.core.Promise;
 import io.vertx.core.impl.logging.Logger;
 
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.WeakHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
-class CloseHooks {
+public class CloseHooks {
 
   private final Logger log;
   private boolean closeHooksRun;
-  private Set<Closeable> closeHooks;
+  private Map<Closeable, CloseHooks> closeHooks;
 
   CloseHooks(Logger log) {
     this.log = log;
+    this.closeHooks = new WeakHashMap<>();
   }
 
   /**
@@ -41,25 +43,21 @@ class CloseHooks {
    *
    * @param hook the hook to add
    */
-  synchronized void add(Closeable hook) {
+  public synchronized void add(Closeable hook) {
     if (closeHooks == null) {
-      // Has to be concurrent as can be removed from non context thread
-      closeHooks = new HashSet<>();
+      throw new IllegalStateException();
     }
-    closeHooks.add(hook);
+    closeHooks.put(hook, this);
   }
 
   /**
    * Remove an existing hook.
    *
    * @param hook the hook to remove
-   * @return {@code} true if the hook was removed
    */
-  synchronized boolean remove(Closeable hook) {
+  public synchronized void remove(Closeable hook) {
     if (closeHooks != null) {
-      return closeHooks.remove(hook);
-    } else {
-      return false;
+      closeHooks.remove(hook);
     }
   }
 
@@ -69,51 +67,38 @@ synchronized boolean remove(Closeable hook) {
    * @param completionHandler called when all hooks have beene executed
    */
   void run(Handler<AsyncResult<Void>> completionHandler) {
-    Set<Closeable> copy = null;
+    Map<Closeable, CloseHooks> copy;
     synchronized (this) {
       if (closeHooksRun) {
         // Sanity check
         throw new IllegalStateException("Close hooks already run");
       }
       closeHooksRun = true;
-      if (closeHooks != null && !closeHooks.isEmpty()) {
-        // Must copy before looping as can be removed during loop otherwise
-        copy = new HashSet<>(closeHooks);
-      }
+      copy = closeHooks;
+      closeHooks = null;
     }
-    if (copy != null && !copy.isEmpty()) {
-      int num = copy.size();
-      if (num != 0) {
-        AtomicInteger count = new AtomicInteger();
-        AtomicBoolean failed = new AtomicBoolean();
-        for (Closeable hook : copy) {
-          Promise<Void> promise = Promise.promise();
-          promise.future().onComplete(ar -> {
-            if (ar.failed()) {
-              if (failed.compareAndSet(false, true)) {
-                // Only report one failure
-                completionHandler.handle(Future.failedFuture(ar.cause()));
-              }
-            } else {
-              if (count.incrementAndGet() == num) {
-                // closeHooksRun = true;
-                completionHandler.handle(Future.succeededFuture());
-              }
-            }
-          });
-          try {
-            hook.close(promise);
-          } catch (Throwable t) {
-            log.warn("Failed to run close hooks", t);
-            promise.tryFail(t);
+    // We want an immutable version of the list holding strong references to avoid racing against finalization
+    List<Closeable> list = new ArrayList<>(copy.size());
+    copy.keySet().forEach(list::add);
+    int num = list.size();
+    if (num > 0) {
+      AtomicInteger count = new AtomicInteger();
+      for (Closeable hook : list) {
+        Promise<Void> promise = Promise.promise();
+        promise.future().onComplete(ar -> {
+          if (count.incrementAndGet() == num) {
+            completionHandler.handle(Future.succeededFuture());
           }
+        });
+        try {
+          hook.close(promise);
+        } catch (Throwable t) {
+          log.warn("Failed to run close hook", t);
+          promise.tryFail(t);
         }
-      } else {
-        completionHandler.handle(Future.succeededFuture());
       }
     } else {
       completionHandler.handle(Future.succeededFuture());
     }
   }
-
 }
diff --git a/src/main/java/io/vertx/core/impl/ContextImpl.java b/src/main/java/io/vertx/core/impl/ContextImpl.java
--- a/src/main/java/io/vertx/core/impl/ContextImpl.java
+++ b/src/main/java/io/vertx/core/impl/ContextImpl.java
@@ -12,7 +12,6 @@
 package io.vertx.core.impl;
 
 import io.netty.channel.EventLoop;
-import io.netty.channel.EventLoopGroup;
 import io.vertx.core.*;
 import io.vertx.core.impl.logging.Logger;
 import io.vertx.core.impl.logging.LoggerFactory;
@@ -51,15 +50,6 @@ static void executeIsolated(Handler<Void> task) {
     }
   }
 
-  private static EventLoop getEventLoop(VertxInternal vertx) {
-    EventLoopGroup group = vertx.getEventLoopGroup();
-    if (group != null) {
-      return group.next();
-    } else {
-      return null;
-    }
-  }
-
   private static final Logger log = LoggerFactory.getLogger(ContextImpl.class);
 
   private static final String DISABLE_TIMINGS_PROP_NAME = "vertx.disableContextTimings";
@@ -80,13 +70,14 @@ private static EventLoop getEventLoop(VertxInternal vertx) {
   final WorkerPool workerPool;
   final TaskQueue orderedTasks;
 
-  ContextImpl(VertxInternal vertx, VertxTracer<?, ?> tracer, WorkerPool internalBlockingPool, WorkerPool workerPool, Deployment deployment,
-                        ClassLoader tccl) {
-    this(vertx, tracer, getEventLoop(vertx), internalBlockingPool, workerPool, deployment, tccl);
-  }
-
-  ContextImpl(VertxInternal vertx, VertxTracer<?, ?> tracer, EventLoop eventLoop, WorkerPool internalBlockingPool, WorkerPool workerPool, Deployment deployment,
-                        ClassLoader tccl) {
+  ContextImpl(VertxInternal vertx,
+              VertxTracer<?, ?> tracer,
+              EventLoop eventLoop,
+              WorkerPool internalBlockingPool,
+              WorkerPool workerPool,
+              Deployment deployment,
+              CloseHooks closeHooks,
+              ClassLoader tccl) {
     if (VertxThread.DISABLE_TCCL && tccl != ClassLoader.getSystemClassLoader()) {
       log.warn("You have disabled TCCL checks but you have a custom TCCL to set.");
     }
@@ -97,26 +88,31 @@ private static EventLoop getEventLoop(VertxInternal vertx) {
     this.tccl = tccl;
     this.owner = vertx;
     this.workerPool = workerPool;
+    this.closeHooks = closeHooks;
     this.internalBlockingPool = internalBlockingPool;
     this.orderedTasks = new TaskQueue();
     this.internalOrderedTasks = new TaskQueue();
-    this.closeHooks = new CloseHooks(log);
   }
 
   public Deployment getDeployment() {
     return deployment;
   }
 
-  public void addCloseHook(Closeable hook) {
-    closeHooks.add(hook);
+  @Override
+  public CloseHooks closeHooks() {
+    return closeHooks;
   }
 
-  public boolean removeCloseHook(Closeable hook) {
-    return closeHooks.remove(hook);
+  public void addCloseHook(Closeable hook) {
+    if (closeHooks != null) {
+      closeHooks.add(hook);
+    }
   }
 
-  public void runCloseHooks(Handler<AsyncResult<Void>> completionHandler) {
-    closeHooks.run(completionHandler);
+  public void removeCloseHook(Closeable hook) {
+    if (deployment != null) {
+      closeHooks.remove(hook);
+    }
   }
 
   @Override
@@ -299,8 +295,8 @@ public final void addCloseHook(Closeable hook) {
     }
 
     @Override
-    public final boolean removeCloseHook(Closeable hook) {
-      return delegate.removeCloseHook(hook);
+    public final void removeCloseHook(Closeable hook) {
+      delegate.removeCloseHook(hook);
     }
 
     @Override
diff --git a/src/main/java/io/vertx/core/impl/ContextInternal.java b/src/main/java/io/vertx/core/impl/ContextInternal.java
--- a/src/main/java/io/vertx/core/impl/ContextInternal.java
+++ b/src/main/java/io/vertx/core/impl/ContextInternal.java
@@ -261,4 +261,6 @@ static ContextInternal current() {
    */
   long setTimer(long delay, Handler<Long> handler);
 
+  CloseHooks closeHooks();
+
 }
diff --git a/src/main/java/io/vertx/core/impl/Deployment.java b/src/main/java/io/vertx/core/impl/Deployment.java
--- a/src/main/java/io/vertx/core/impl/Deployment.java
+++ b/src/main/java/io/vertx/core/impl/Deployment.java
@@ -57,4 +57,5 @@ default void doUndeploy(ContextInternal undeployingContext, Handler<AsyncResult<
   void undeployHandler(Handler<Void> handler);
 
   boolean isChild();
+
 }
diff --git a/src/main/java/io/vertx/core/impl/DeploymentManager.java b/src/main/java/io/vertx/core/impl/DeploymentManager.java
--- a/src/main/java/io/vertx/core/impl/DeploymentManager.java
+++ b/src/main/java/io/vertx/core/impl/DeploymentManager.java
@@ -185,14 +185,16 @@ private Future<Deployment> doDeploy(String identifier,
     AtomicInteger deployCount = new AtomicInteger();
     AtomicBoolean failureReported = new AtomicBoolean();
     for (Verticle verticle: verticles) {
+      CloseHooks closeHooks = new CloseHooks(log);
       WorkerExecutorInternal workerExec = poolName != null ? vertx.createSharedWorkerExecutor(poolName, options.getWorkerPoolSize(), options.getMaxWorkerExecuteTime(), options.getMaxWorkerExecuteTimeUnit()) : null;
       WorkerPool pool = workerExec != null ? workerExec.getPool() : null;
-      ContextImpl context = (ContextImpl) (options.isWorker() ? vertx.createWorkerContext(deployment, pool, tccl) :
-        vertx.createEventLoopContext(deployment, pool, tccl));
+      ContextImpl context = (ContextImpl) (options.isWorker() ? vertx.createWorkerContext(deployment, closeHooks, pool, tccl) :
+        vertx.createEventLoopContext(deployment, closeHooks, pool, tccl));
       if (workerExec != null) {
         context.addCloseHook(workerExec);
       }
-      deployment.addVerticle(new VerticleHolder(verticle, context));
+      VerticleHolder holder = new VerticleHolder(verticle, context, closeHooks);
+      deployment.addVerticle(holder);
       context.runOnContext(v -> {
         try {
           verticle.init(vertx, context);
@@ -219,12 +221,12 @@ private Future<Deployment> doDeploy(String identifier,
                 promise.complete(deployment);
               }
             } else if (failureReported.compareAndSet(false, true)) {
-              deployment.rollback(callingContext, promise, context, ar.cause());
+              deployment.rollback(callingContext, promise, context, holder.closeHooks, ar.cause());
             }
           });
         } catch (Throwable t) {
           if (failureReported.compareAndSet(false, true))
-            deployment.rollback(callingContext, promise, context, t);
+            deployment.rollback(callingContext, promise, context, holder.closeHooks, t);
         }
       });
     }
@@ -233,12 +235,15 @@ private Future<Deployment> doDeploy(String identifier,
   }
 
   static class VerticleHolder {
+
     final Verticle verticle;
     final ContextImpl context;
+    final CloseHooks closeHooks;
 
-    VerticleHolder(Verticle verticle, ContextImpl context) {
+    VerticleHolder(Verticle verticle, ContextImpl context, CloseHooks closeHooks) {
       this.verticle = verticle;
       this.context = context;
+      this.closeHooks = closeHooks;
     }
   }
 
@@ -269,7 +274,7 @@ public void addVerticle(VerticleHolder holder) {
       verticles.add(holder);
     }
 
-    private synchronized void rollback(ContextInternal callingContext, Handler<AsyncResult<Deployment>> completionHandler, ContextImpl context, Throwable cause) {
+    private synchronized void rollback(ContextInternal callingContext, Handler<AsyncResult<Deployment>> completionHandler, ContextImpl context, CloseHooks closeHooks, Throwable cause) {
       if (status == ST_DEPLOYED) {
         status = ST_UNDEPLOYING;
         doUndeployChildren(callingContext).onComplete(childrenResult -> {
@@ -289,7 +294,7 @@ private synchronized void rollback(ContextInternal callingContext, Handler<Async
           if (childrenResult.failed()) {
             reportFailure(cause, callingContext, completionHandler);
           } else {
-            context.runCloseHooks(closeHookAsyncResult -> reportFailure(cause, callingContext, completionHandler));
+            closeHooks.run(closeHookAsyncResult -> reportFailure(cause, callingContext, completionHandler));
           }
         });
       }
@@ -344,7 +349,7 @@ public synchronized Future<Void> doUndeploy(ContextInternal undeployingContext)
               if (metrics != null) {
                 metrics.verticleUndeployed(verticleHolder.verticle);
               }
-              context.runCloseHooks(ar2 -> {
+              verticleHolder.closeHooks.run(ar2 -> {
                 if (ar2.failed()) {
                   // Log error but we report success anyway
                   log.error("Failed to run close hook", ar2.cause());
@@ -447,7 +452,6 @@ public boolean isChild() {
     public String deploymentID() {
       return deploymentID;
     }
-
   }
 
 }
diff --git a/src/main/java/io/vertx/core/impl/EventLoopContext.java b/src/main/java/io/vertx/core/impl/EventLoopContext.java
--- a/src/main/java/io/vertx/core/impl/EventLoopContext.java
+++ b/src/main/java/io/vertx/core/impl/EventLoopContext.java
@@ -23,14 +23,15 @@
  */
 public class EventLoopContext extends ContextImpl {
 
-  EventLoopContext(VertxInternal vertx, VertxTracer<?, ?> tracer, WorkerPool internalBlockingPool, WorkerPool workerPool, Deployment deployment,
+  EventLoopContext(VertxInternal vertx,
+                   VertxTracer<?, ?> tracer,
+                   EventLoop eventLoop,
+                   WorkerPool internalBlockingPool,
+                   WorkerPool workerPool,
+                   Deployment deployment,
+                   CloseHooks closeHooks,
                    ClassLoader tccl) {
-    super(vertx, tracer, internalBlockingPool, workerPool, deployment, tccl);
-  }
-
-  EventLoopContext(VertxInternal vertx, VertxTracer<?, ?> tracer, EventLoop eventLoop, WorkerPool internalBlockingPool, WorkerPool workerPool, Deployment deployment,
-                          ClassLoader tccl) {
-    super(vertx, tracer, eventLoop, internalBlockingPool, workerPool, deployment, tccl);
+    super(vertx, tracer, eventLoop, internalBlockingPool, workerPool, deployment, closeHooks, tccl);
   }
 
   @Override
@@ -79,6 +80,11 @@ static class Duplicated extends ContextImpl.Duplicated<EventLoopContext> {
       super(delegate);
     }
 
+    @Override
+    public CloseHooks closeHooks() {
+      return delegate.closeHooks();
+    }
+
     @Override
     <T> void execute(T argument, Handler<T> task) {
       nettyEventLoop().execute(() -> emit(argument, task));
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
@@ -277,7 +277,10 @@ public NetServer createNetServer() {
   }
 
   public NetClient createNetClient(NetClientOptions options) {
-    return new NetClientImpl(this, options);
+    CloseHooks hooks = resolveHooks();
+    NetClientImpl client = new NetClientImpl(this, options, hooks);
+    hooks.add(client);
+    return client;
   }
 
   @Override
@@ -313,7 +316,10 @@ public HttpServer createHttpServer() {
   }
 
   public HttpClient createHttpClient(HttpClientOptions options) {
-    return new HttpClientImpl(this, options);
+    CloseHooks hooks = resolveHooks();
+    HttpClientImpl client = new HttpClientImpl(this, hooks, options);
+    hooks.add(client);
+    return client;
   }
 
   @Override
@@ -382,7 +388,7 @@ public ContextInternal getOrCreateContext() {
     ContextInternal ctx = getContext();
     if (ctx == null) {
       // We are running embedded - Create a context
-      ctx = createEventLoopContext((Deployment) null, null, Thread.currentThread().getContextClassLoader());
+      ctx = createEventLoopContext(null, null, null, Thread.currentThread().getContextClassLoader());
     }
     return ctx;
   }
@@ -427,26 +433,26 @@ public boolean cancelTimer(long id) {
   }
 
   @Override
-  public EventLoopContext createEventLoopContext(Deployment deployment, WorkerPool workerPool, ClassLoader tccl) {
-    return new EventLoopContext(this, tracer, internalBlockingPool, workerPool != null ? workerPool : this.workerPool, deployment, tccl);
+  public EventLoopContext createEventLoopContext(Deployment deployment, CloseHooks closeHooks, WorkerPool workerPool, ClassLoader tccl) {
+    return new EventLoopContext(this, tracer, eventLoopGroup.next(), internalBlockingPool, workerPool != null ? workerPool : this.workerPool, deployment, closeHooks, tccl);
   }
 
   @Override
   public EventLoopContext createEventLoopContext(EventLoop eventLoop, WorkerPool workerPool, ClassLoader tccl) {
-    return new EventLoopContext(this, tracer, eventLoop, internalBlockingPool, workerPool != null ? workerPool : this.workerPool, null, tccl);
+    return new EventLoopContext(this, tracer, eventLoop, internalBlockingPool, workerPool != null ? workerPool : this.workerPool, null, null, tccl);
   }
 
   @Override
-  public ContextInternal createWorkerContext(Deployment deployment, WorkerPool workerPool, ClassLoader tccl) {
+  public ContextInternal createWorkerContext(Deployment deployment, CloseHooks closeHooks, WorkerPool workerPool, ClassLoader tccl) {
     if (workerPool == null) {
       workerPool = this.workerPool;
     }
-    return new WorkerContext(this, tracer, internalBlockingPool, workerPool, deployment, tccl);
+    return new WorkerContext(this, tracer, internalBlockingPool, workerPool, deployment, closeHooks, tccl);
   }
 
   @Override
   public ContextInternal createWorkerContext() {
-    return createWorkerContext(null, null, null);
+    return createWorkerContext(null, null, null, null);
   }
 
   @Override
@@ -532,7 +538,6 @@ public synchronized void close(Handler<AsyncResult<Void>> completionHandler) {
       return;
     }
     closed = true;
-
     closeHooks.run(ar -> {
       deploymentManager.undeployAll().onComplete(ar1 -> {
         HAManager haManager = haManager();
@@ -552,33 +557,7 @@ public synchronized void close(Handler<AsyncResult<Void>> completionHandler) {
             ebClose.future().onComplete(ar4 -> {
               closeClusterManager(ar5 -> {
                 // Copy set to prevent ConcurrentModificationException
-                Set<HttpServerImpl> httpServers = new HashSet<>(sharedHttpServers.values());
-                Set<NetServerImpl> netServers = new HashSet<>(sharedNetServers.values());
-                sharedHttpServers.clear();
-                sharedNetServers.clear();
-
-                int serverCount = httpServers.size() + netServers.size();
-
-                AtomicInteger serverCloseCount = new AtomicInteger();
-
-                Handler<AsyncResult<Void>> serverCloseHandler = res -> {
-                  if (res.failed()) {
-                    log.error("Failure in shutting down server", res.cause());
-                  }
-                  if (serverCloseCount.incrementAndGet() == serverCount) {
-                    deleteCacheDirAndShutdown(completionHandler);
-                  }
-                };
-
-                for (HttpServerImpl server : httpServers) {
-                  server.closeAll(serverCloseHandler);
-                }
-                for (NetServerImpl server : netServers) {
-                  server.closeAll(serverCloseHandler);
-                }
-                if (serverCount == 0) {
-                  deleteCacheDirAndShutdown(completionHandler);
-                }
+                deleteCacheDirAndShutdown(completionHandler);
               });
             });
           });
@@ -1099,9 +1078,13 @@ public synchronized WorkerExecutorImpl createSharedWorkerExecutor(String name, i
     } else {
       sharedWorkerPool.refCount++;
     }
-    ContextInternal context = getOrCreateContext();
-    WorkerExecutorImpl namedExec = new WorkerExecutorImpl(context, sharedWorkerPool);
-    context.addCloseHook(namedExec);
+    ContextInternal ctx = getContext();
+    CloseHooks hooks = ctx != null ? ctx.closeHooks() : null;
+    if (hooks == null) {
+      hooks = closeHooks;
+    }
+    WorkerExecutorImpl namedExec = new WorkerExecutorImpl(this, closeHooks, sharedWorkerPool);
+    hooks.add(namedExec);
     return namedExec;
   }
 
@@ -1125,4 +1108,18 @@ public void addCloseHook(Closeable hook) {
   public void removeCloseHook(Closeable hook) {
     closeHooks.remove(hook);
   }
+
+  @Override
+  public CloseHooks closeHooks() {
+    return closeHooks;
+  }
+
+  private CloseHooks resolveHooks() {
+    ContextInternal context = getContext();
+    CloseHooks hooks = context != null ? context.closeHooks() : null;
+    if (hooks == null) {
+      hooks = closeHooks();
+    }
+    return hooks;
+  }
 }
diff --git a/src/main/java/io/vertx/core/impl/VertxInternal.java b/src/main/java/io/vertx/core/impl/VertxInternal.java
--- a/src/main/java/io/vertx/core/impl/VertxInternal.java
+++ b/src/main/java/io/vertx/core/impl/VertxInternal.java
@@ -83,14 +83,14 @@ public interface VertxInternal extends Vertx {
   /**
    * @return event loop context
    */
-  ContextInternal createEventLoopContext(Deployment deployment, WorkerPool workerPool, ClassLoader tccl);
+  ContextInternal createEventLoopContext(Deployment deployment, CloseHooks closeHooks, WorkerPool workerPool, ClassLoader tccl);
 
   ContextInternal createEventLoopContext(EventLoop eventLoop, WorkerPool workerPool, ClassLoader tccl);
 
   /**
    * @return worker loop context
    */
-  ContextInternal createWorkerContext(Deployment deployment, WorkerPool pool, ClassLoader tccl);
+  ContextInternal createWorkerContext(Deployment deployment, CloseHooks closeHooks, WorkerPool pool, ClassLoader tccl);
 
   ContextInternal createWorkerContext();
 
@@ -151,4 +151,6 @@ public interface VertxInternal extends Vertx {
 
   void removeCloseHook(Closeable hook);
 
+  CloseHooks closeHooks();
+
 }
diff --git a/src/main/java/io/vertx/core/impl/WorkerContext.java b/src/main/java/io/vertx/core/impl/WorkerContext.java
--- a/src/main/java/io/vertx/core/impl/WorkerContext.java
+++ b/src/main/java/io/vertx/core/impl/WorkerContext.java
@@ -26,9 +26,14 @@
  */
 public class WorkerContext extends ContextImpl {
 
-  WorkerContext(VertxInternal vertx, VertxTracer<?, ?> tracer, WorkerPool internalBlockingPool, WorkerPool workerPool, Deployment deployment,
+  WorkerContext(VertxInternal vertx,
+                VertxTracer<?, ?> tracer,
+                WorkerPool internalBlockingPool,
+                WorkerPool workerPool,
+                Deployment deployment,
+                CloseHooks closeHooks,
                 ClassLoader tccl) {
-    super(vertx, tracer, internalBlockingPool, workerPool, deployment, tccl);
+    super(vertx, tracer, vertx.getEventLoopGroup().next(), internalBlockingPool, workerPool, deployment, closeHooks, tccl);
   }
 
   @Override
@@ -131,6 +136,11 @@ static class Duplicated extends ContextImpl.Duplicated<WorkerContext> {
       super(delegate);
     }
 
+    @Override
+    public CloseHooks closeHooks() {
+      return delegate.closeHooks();
+    }
+
     @Override
     <T> void execute(T argument, Handler<T> task) {
       delegate.execute(this, orderedTasks, argument, task);
diff --git a/src/main/java/io/vertx/core/impl/WorkerExecutorImpl.java b/src/main/java/io/vertx/core/impl/WorkerExecutorImpl.java
--- a/src/main/java/io/vertx/core/impl/WorkerExecutorImpl.java
+++ b/src/main/java/io/vertx/core/impl/WorkerExecutorImpl.java
@@ -22,12 +22,14 @@
  */
 class WorkerExecutorImpl implements MetricsProvider, WorkerExecutorInternal {
 
-  private final ContextInternal ctx;
+  private final VertxInternal vertx;
+  private final CloseHooks closeHooks;
   private final VertxImpl.SharedWorkerPool pool;
   private boolean closed;
 
-  public WorkerExecutorImpl(ContextInternal ctx, VertxImpl.SharedWorkerPool pool) {
-    this.ctx = ctx;
+  public WorkerExecutorImpl(VertxInternal vertx, CloseHooks closeHooks, VertxImpl.SharedWorkerPool pool) {
+    this.vertx = vertx;
+    this.closeHooks = closeHooks;
     this.pool = pool;
   }
 
@@ -44,7 +46,7 @@ public boolean isMetricsEnabled() {
 
   @Override
   public Vertx vertx() {
-    return ctx.owner();
+    return vertx;
   }
 
   @Override
@@ -57,7 +59,7 @@ public WorkerPool getPool() {
     if (closed) {
       throw new IllegalStateException("Worker executor closed");
     }
-    ContextInternal context = (ContextInternal) ctx.owner().getOrCreateContext();
+    ContextInternal context = (ContextInternal) vertx.getOrCreateContext();
     ContextImpl impl = context instanceof ContextImpl.Duplicated ? ((ContextImpl.Duplicated)context).delegate : (ContextImpl) context;
     return ContextImpl.executeBlocking(context, blockingCodeHandler, pool, ordered ? impl.orderedTasks : null);
   }
@@ -71,7 +73,7 @@ public synchronized <T> void executeBlocking(Handler<Promise<T>> blockingCodeHan
 
   @Override
   public Future<Void> close() {
-    PromiseInternal<Void> promise = ctx.promise();
+    PromiseInternal<Void> promise = vertx.promise();
     close(promise);
     return promise.future();
   }
@@ -81,7 +83,7 @@ public void close(Promise<Void> completion) {
     synchronized (this) {
       if (!closed) {
         closed = true;
-        ctx.removeCloseHook(this);
+        closeHooks.remove(this);
         pool.release();
       }
     }
diff --git a/src/main/java/io/vertx/core/net/NetClient.java b/src/main/java/io/vertx/core/net/NetClient.java
--- a/src/main/java/io/vertx/core/net/NetClient.java
+++ b/src/main/java/io/vertx/core/net/NetClient.java
@@ -114,4 +114,9 @@ public interface NetClient extends Measured {
    */
   Future<Void> close();
 
+  /**
+   * @return a future closed when the client is closed
+   */
+  Future<Void> closeFuture();
+
 }
diff --git a/src/main/java/io/vertx/core/net/impl/NetClientImpl.java b/src/main/java/io/vertx/core/net/impl/NetClientImpl.java
--- a/src/main/java/io/vertx/core/net/impl/NetClientImpl.java
+++ b/src/main/java/io/vertx/core/net/impl/NetClientImpl.java
@@ -21,12 +21,12 @@
 import io.netty.handler.stream.ChunkedWriteHandler;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.concurrent.GenericFutureListener;
-import io.netty.util.concurrent.GlobalEventExecutor;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Closeable;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Promise;
+import io.vertx.core.impl.CloseHooks;
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.impl.PromiseInternal;
 import io.vertx.core.impl.VertxInternal;
@@ -52,7 +52,7 @@
  *
  * @author <a href="http://tfox.org">Tim Fox</a>
  */
-public class NetClientImpl implements MetricsProvider, NetClient {
+public class NetClientImpl implements MetricsProvider, NetClient, Closeable {
 
   private static final Logger log = LoggerFactory.getLogger(NetClientImpl.class);
   protected final int idleTimeout;
@@ -61,39 +61,33 @@ public class NetClientImpl implements MetricsProvider, NetClient {
 
   private final VertxInternal vertx;
   private final NetClientOptions options;
-  protected final SSLHelper sslHelper;
+  private final SSLHelper sslHelper;
   private final ChannelGroup channelGroup;
-  private final Closeable closeHook;
-  private final ContextInternal creatingContext;
+  private final CloseHooks closeHooks;
   private final TCPMetrics metrics;
+  private final PromiseInternal<Void> closePromise;
+  private final Future<Void> closeFuture;
   private volatile boolean closed;
 
-  public NetClientImpl(VertxInternal vertx, NetClientOptions options) {
-    this(vertx, options, true);
-  }
-
-  public NetClientImpl(VertxInternal vertx, NetClientOptions options, boolean useCreatingContext) {
+  public NetClientImpl(VertxInternal vertx, NetClientOptions options, CloseHooks closeHooks) {
     this.vertx = vertx;
-    this.channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
+    this.channelGroup = new DefaultChannelGroup(vertx.getAcceptorEventLoopGroup().next());
     this.options = new NetClientOptions(options);
     this.sslHelper = new SSLHelper(options, options.getKeyCertOptions(), options.getTrustOptions());
-    this.closeHook = completionHandler -> {
-      NetClientImpl.this.close();
-      completionHandler.handle(Future.succeededFuture());
-    };
-    if (useCreatingContext) {
-      creatingContext = vertx.getContext();
-      if (creatingContext != null) {
-        creatingContext.addCloseHook(closeHook);
-      }
-    } else {
-      creatingContext = null;
-    }
-    VertxMetrics metrics = vertx.metricsSPI();
-    this.metrics = metrics != null ? metrics.createNetClientMetrics(options) : null;
+    this.closeHooks = closeHooks;
+    this.metrics = vertx.metricsSPI() != null ? vertx.metricsSPI().createNetClientMetrics(options) : null;
     logEnabled = options.getLogActivity();
     idleTimeout = options.getIdleTimeout();
     idleTimeoutUnit = options.getIdleTimeoutUnit();
+    closePromise = (PromiseInternal) Promise.promise();
+    if (metrics != null) {
+      closeFuture = closePromise.future().compose(v -> {
+        metrics.close();
+        return Future.succeededFuture();
+      });
+    } else {
+      closeFuture = closePromise.future();
+    }
   }
 
   protected void initChannel(ChannelPipeline pipeline) {
@@ -141,35 +135,43 @@ public NetClient connect(int port, String host, String serverName, Handler<Async
     return connect(SocketAddress.inetSocketAddress(port, host), serverName, connectHandler);
   }
 
+  @Override
+  public Future<Void> closeFuture() {
+    return closePromise.future();
+  }
+
   @Override
   public void close(Handler<AsyncResult<Void>> handler) {
-    close(vertx.getOrCreateContext().promise(handler));
+    if (closeHooks != null) {
+      closeHooks.remove(this);
+    }
+    ContextInternal closingCtx = vertx.getOrCreateContext();
+    close(handler != null ? closingCtx.promise(handler) : null);
   }
 
   @Override
   public Future<Void> close() {
-    PromiseInternal<Void> promise = vertx.getOrCreateContext().promise();
+    if (closeHooks != null) {
+      closeHooks.remove(this);
+    }
+    ContextInternal closingCtx = vertx.getOrCreateContext();
+    PromiseInternal<Void> promise = closingCtx.promise();
     close(promise);
     return promise.future();
   }
 
-  private void close(PromiseInternal<Void> promise) {
-    boolean closed;
+  @Override
+  public void close(Promise<Void> completion) {
+    boolean close;
     synchronized (this) {
-      closed = this.closed;
-      this.closed = true;
+      close = !closed;
+      closed = true;
     }
-    if (closed) {
-      promise.complete();
-      return;
+    if (close) {
+      ChannelGroupFuture fut = channelGroup.close();
+      fut.addListener(closePromise);
     }
-    ChannelGroupFuture fut = channelGroup.close();
-    fut.addListener(promise);
-    promise.future().onComplete(ar -> {
-      if (metrics != null) {
-        metrics.close();
-      }
-    });
+    closeFuture.onComplete(completion);
   }
 
   @Override
@@ -277,7 +279,7 @@ protected void finalize() throws Throwable {
     // Make sure this gets cleaned up if there are no more references to it
     // so as not to leave connections and resources dangling until the system is shutdown
     // which could make the JVM run out of file handles.
-    close((PromiseInternal<Void>) Promise.<Void>promise());
+    close((Handler<AsyncResult<Void>>) null);
     super.finalize();
   }
 }
diff --git a/src/main/java/io/vertx/core/net/impl/NetServerImpl.java b/src/main/java/io/vertx/core/net/impl/NetServerImpl.java
--- a/src/main/java/io/vertx/core/net/impl/NetServerImpl.java
+++ b/src/main/java/io/vertx/core/net/impl/NetServerImpl.java
@@ -228,15 +228,6 @@ public boolean isClosed() {
     return !isListening();
   }
 
-  @Override
-  protected void finalize() throws Throwable {
-    // Make sure this gets cleaned up if there are no more references to it
-    // so as not to leave connections and resources dangling until the system is shutdown
-    // which could make the JVM run out of file handles.
-    close();
-    super.finalize();
-  }
-
   private class NetServerWorker implements Handler<Channel> {
 
     private final ContextInternal context;
diff --git a/src/main/java/io/vertx/core/net/impl/TCPServerBase.java b/src/main/java/io/vertx/core/net/impl/TCPServerBase.java
--- a/src/main/java/io/vertx/core/net/impl/TCPServerBase.java
+++ b/src/main/java/io/vertx/core/net/impl/TCPServerBase.java
@@ -21,6 +21,7 @@
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Promise;
+import io.vertx.core.impl.CloseHooks;
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.impl.PromiseInternal;
 import io.vertx.core.impl.VertxInternal;
@@ -53,6 +54,7 @@ public abstract class TCPServerBase implements Closeable, MetricsProvider {
   protected final NetServerOptions options;
   protected final ContextInternal creatingContext;
   protected final SSLHelper sslHelper;
+  protected final CloseHooks closeHooks;
 
   // Per server
   private EventLoop eventLoop;
@@ -71,14 +73,20 @@ public abstract class TCPServerBase implements Closeable, MetricsProvider {
   private TCPMetrics<?> metrics;
 
   public TCPServerBase(VertxInternal vertx, NetServerOptions options) {
+
+    ContextInternal context = vertx.getContext();
+    CloseHooks hooks = context != null ? context.closeHooks() : null;
+    if (hooks == null) {
+      hooks = vertx.closeHooks();
+    }
+
     this.vertx = vertx;
     this.options = new NetServerOptions(options);
     this.sslHelper = new SSLHelper(options, options.getKeyCertOptions(), options.getTrustOptions());
-    this.creatingContext = vertx.getContext();
+    this.creatingContext = context;
+    this.closeHooks = hooks;
 
-    if (creatingContext != null) {
-      creatingContext.addCloseHook(this);
-    }
+    closeHooks.add(this);
   }
 
   public int actualPort() {
@@ -188,9 +196,7 @@ public synchronized TCPMetrics<?> getMetrics() {
 
   @Override
   public synchronized void close(Promise<Void> completion) {
-    if (creatingContext != null) {
-      creatingContext.removeCloseHook(this);
-    }
+    closeHooks.remove(this);
     if (!listening) {
       completion.complete();
       return;
@@ -240,4 +246,15 @@ public void closeAll(Handler<AsyncResult<Void>> handler) {
     CompositeFuture fut = CompositeFuture.all(futures);
     fut.onComplete(ar -> handler.handle(ar.mapEmpty()));
   }
+
+  public abstract Future<Void> close();
+
+  @Override
+  protected void finalize() throws Throwable {
+    // Make sure this gets cleaned up if there are no more references to it
+    // so as not to leave connections and resources dangling until the system is shutdown
+    // which could make the JVM run out of file handles.
+    close();
+    super.finalize();
+  }
 }


Test patch:
diff --git a/src/test/benchmarks/io/vertx/core/impl/BenchmarkContext.java b/src/test/benchmarks/io/vertx/core/impl/BenchmarkContext.java
--- a/src/test/benchmarks/io/vertx/core/impl/BenchmarkContext.java
+++ b/src/test/benchmarks/io/vertx/core/impl/BenchmarkContext.java
@@ -21,11 +21,16 @@ public class BenchmarkContext extends ContextImpl {
 
   public static BenchmarkContext create(Vertx vertx) {
     VertxImpl impl = (VertxImpl) vertx;
-    return new BenchmarkContext(impl, impl.internalBlockingPool, impl.workerPool, null, Thread.currentThread().getContextClassLoader());
+    return new BenchmarkContext(
+      impl,
+      impl.internalBlockingPool,
+      impl.workerPool,
+      Thread.currentThread().getContextClassLoader()
+    );
   }
 
-  public BenchmarkContext(VertxInternal vertx, WorkerPool internalBlockingPool, WorkerPool workerPool, Deployment deployment, ClassLoader tccl) {
-    super(vertx, null, internalBlockingPool, workerPool, deployment, tccl);
+  public BenchmarkContext(VertxInternal vertx, WorkerPool internalBlockingPool, WorkerPool workerPool, ClassLoader tccl) {
+    super(vertx, null, vertx.getEventLoopGroup().next(), internalBlockingPool, workerPool, null, null, tccl);
   }
 
   @Override
diff --git a/src/test/java/io/vertx/core/ContextTaskTest.java b/src/test/java/io/vertx/core/ContextTaskTest.java
--- a/src/test/java/io/vertx/core/ContextTaskTest.java
+++ b/src/test/java/io/vertx/core/ContextTaskTest.java
@@ -61,7 +61,7 @@ private ContextInternal createEventLoopContext() {
   }
 
   private ContextInternal createWorkerContext() {
-    return ((VertxInternal) vertx).createWorkerContext(null, new WorkerPool(workerExecutor, null), Thread.currentThread().getContextClassLoader());
+    return ((VertxInternal) vertx).createWorkerContext(null, null, new WorkerPool(workerExecutor, null), Thread.currentThread().getContextClassLoader());
   }
 
   // SCHEDULE + DISPATCH
diff --git a/src/test/java/io/vertx/core/ContextTest.java b/src/test/java/io/vertx/core/ContextTest.java
--- a/src/test/java/io/vertx/core/ContextTest.java
+++ b/src/test/java/io/vertx/core/ContextTest.java
@@ -33,7 +33,7 @@ public class ContextTest extends VertxTestBase {
   private ExecutorService workerExecutor;
 
   private ContextInternal createWorkerContext() {
-    return ((VertxInternal) vertx).createWorkerContext(null, new WorkerPool(workerExecutor, null), Thread.currentThread().getContextClassLoader());
+    return ((VertxInternal) vertx).createWorkerContext(null, null, new WorkerPool(workerExecutor, null), Thread.currentThread().getContextClassLoader());
   }
 
   @Override
diff --git a/src/test/java/io/vertx/core/NamedWorkerPoolTest.java b/src/test/java/io/vertx/core/NamedWorkerPoolTest.java
--- a/src/test/java/io/vertx/core/NamedWorkerPoolTest.java
+++ b/src/test/java/io/vertx/core/NamedWorkerPoolTest.java
@@ -328,7 +328,7 @@ public void testCloseWorkerPoolsWhenVertxCloses() {
       try {
         exec.executeBlocking(fut -> fail(), ar -> fail());
         fail();
-      } catch (RejectedExecutionException ignore) {
+      } catch (IllegalStateException ignore) {
       }
       // Check we can still close
       exec.close();
diff --git a/src/test/java/io/vertx/core/VertxTest.java b/src/test/java/io/vertx/core/VertxTest.java
--- a/src/test/java/io/vertx/core/VertxTest.java
+++ b/src/test/java/io/vertx/core/VertxTest.java
@@ -11,22 +11,29 @@
 
 package io.vertx.core;
 
-import io.vertx.core.AsyncResult;
-import io.vertx.core.Closeable;
-import io.vertx.core.Future;
-import io.vertx.core.Handler;
-import io.vertx.core.Vertx;
+import io.vertx.core.http.HttpClient;
+import io.vertx.core.http.HttpClientOptions;
 import io.vertx.core.impl.VertxInternal;
+import io.vertx.core.net.NetClient;
+import io.vertx.core.net.NetSocket;
 import io.vertx.test.core.AsyncTestBase;
 import org.junit.Test;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
 
+import java.lang.ref.WeakReference;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
 public class VertxTest extends AsyncTestBase {
 
+  private final org.openjdk.jmh.runner.Runner RUNNER = new Runner(new OptionsBuilder().shouldDoGC(true).build());
+
   @Test
   public void testCloseHooksCalled() throws Exception {
     AtomicInteger closedCount = new AtomicInteger();
@@ -102,7 +109,7 @@ public void close(Promise<Void> completion) {
   }
 
   @Test
-  public void testCloseFuture() throws Exception {
+  public void testCloseFuture() {
     Vertx vertx = Vertx.vertx();
     Future<Void> fut = vertx.close();
     // Check that we can get a callback on the future as thread pools are closed by the operation
@@ -111,4 +118,89 @@ public void testCloseFuture() throws Exception {
     }));
     await();
   }
+
+  @Test
+  public void testFinalizeHttpClient() throws Exception {
+    Vertx vertx = Vertx.vertx();
+    try {
+      CountDownLatch latch = new CountDownLatch(1);
+      AtomicReference<NetSocket> socketRef = new AtomicReference<>();
+      vertx.createNetServer()
+        .connectHandler(socketRef::set)
+        .listen(8080, "localhost")
+        .onComplete(onSuccess(server -> latch.countDown()));
+      awaitLatch(latch);
+      AtomicBoolean closed = new AtomicBoolean();
+      // No keep alive so the connection is not held in the pool ????
+      HttpClient client = vertx.createHttpClient(new HttpClientOptions().setKeepAlive(false));
+      client.closeFuture().onComplete(ar -> closed.set(true));
+      client.get(8080, "localhost", "/", onFailure(err -> {}));
+      WeakReference<HttpClient> ref = new WeakReference<>(client);
+      client = null;
+      assertWaitUntil(() -> socketRef.get() != null);
+      for (int i = 0;i < 10;i++) {
+        Thread.sleep(10);
+        RUNNER.runSystemGC();
+        assertFalse(closed.get());
+        assertNotNull(ref.get());
+      }
+      socketRef.get().close();
+      long now = System.currentTimeMillis();
+      while (true) {
+        assertTrue(System.currentTimeMillis() - now < 20_000);
+        RUNNER.runSystemGC();
+        if (ref.get() == null) {
+          assertTrue(closed.get());
+          break;
+        }
+      }
+    } finally {
+      vertx.close(ar -> {
+        testComplete();
+      });
+    }
+    await();
+  }
+
+  @Test
+  public void testFinalizeNetClient() throws Exception {
+    Vertx vertx = Vertx.vertx();
+    try {
+      CountDownLatch latch = new CountDownLatch(1);
+      AtomicReference<NetSocket> socketRef = new AtomicReference<>();
+      vertx.createNetServer()
+        .connectHandler(socketRef::set)
+        .listen(1234, "localhost")
+        .onComplete(onSuccess(server -> latch.countDown()));
+      awaitLatch(latch);
+      AtomicBoolean closed = new AtomicBoolean();
+      NetClient client = vertx.createNetClient();
+      client.closeFuture().onComplete(ar -> closed.set(true));
+      client.connect(1234, "localhost", onSuccess(so -> {}));
+      WeakReference<NetClient> ref = new WeakReference<>(client);
+      client = null;
+      assertWaitUntil(() -> socketRef.get() != null);
+      for (int i = 0;i < 10;i++) {
+        Thread.sleep(10);
+        RUNNER.runSystemGC();
+        assertFalse(closed.get());
+        assertNotNull(ref.get());
+      }
+      socketRef.get().close();
+      long now = System.currentTimeMillis();
+      while (true) {
+        assertTrue(System.currentTimeMillis() - now < 20_000);
+        RUNNER.runSystemGC();
+        if (ref.get() == null) {
+          assertTrue(closed.get());
+          break;
+        }
+      }
+    } finally {
+      vertx.close(ar -> {
+        testComplete();
+      });
+    }
+    await();
+  }
 }
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
@@ -30,6 +30,7 @@
 import io.vertx.test.verticles.SimpleServer;
 import io.vertx.test.core.TestUtils;
 import org.junit.Assume;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import java.io.File;
@@ -66,6 +67,12 @@ protected VertxOptions getOptions() {
     return options;
   }
 
+  @Ignore
+  @Override
+  public void testBrokenFormUploadLargeFile() {
+    super.testBrokenFormUploadLargeFile();
+  }
+
   @Test
   public void testClientOptions() {
     HttpClientOptions options = new HttpClientOptions();