Repository: eclipse-vertx/vert.x

Problem statement:
VertxThread factory
A runtime embedding might want to have Vert.x uses a specific subclass of `VertxThread`. This is a port of https://github.com/eclipse-vertx/vert.x/pull/3749 using the Vert.x 4 service provider mechanism and with a simplification of the factory responsibility.


Hints:

Patch:
diff --git a/src/main/java/io/vertx/core/impl/VertxBuilder.java b/src/main/java/io/vertx/core/impl/VertxBuilder.java
--- a/src/main/java/io/vertx/core/impl/VertxBuilder.java
+++ b/src/main/java/io/vertx/core/impl/VertxBuilder.java
@@ -18,6 +18,7 @@
 import io.vertx.core.net.impl.transport.Transport;
 import io.vertx.core.spi.VertxMetricsFactory;
 import io.vertx.core.spi.VertxServiceProvider;
+import io.vertx.core.spi.VertxThreadFactory;
 import io.vertx.core.spi.VertxTracerFactory;
 import io.vertx.core.spi.cluster.ClusterManager;
 import io.vertx.core.spi.cluster.NodeSelector;
@@ -28,7 +29,6 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Objects;
 
 /**
  * Vertx builder for creating vertx instances with SPI overrides.
@@ -43,6 +43,7 @@ public class VertxBuilder {
   private ClusterManager clusterManager;
   private NodeSelector clusterNodeSelector;
   private VertxTracer tracer;
+  private VertxThreadFactory threadFactory;
   private VertxMetrics metrics;
   private FileResolver fileResolver;
 
@@ -158,6 +159,23 @@ public VertxBuilder metrics(VertxMetrics metrics) {
     return this;
   }
 
+  /**
+   * @return the {@code VertxThreadFactory} to use
+   */
+  public VertxThreadFactory threadFactory() {
+    return threadFactory;
+  }
+
+  /**
+   * Set the {@code VertxThreadFactory} instance to use.
+   * @param factory the metrics
+   * @return this builder instance
+   */
+  public VertxBuilder threadFactory(VertxThreadFactory factory) {
+    this.threadFactory = factory;
+    return this;
+  }
+
   /**
    * Build and return the vertx instance
    */
@@ -169,7 +187,8 @@ public Vertx vertx() {
       metrics,
       tracer,
       transport,
-      fileResolver);
+      fileResolver,
+      threadFactory);
     vertx.init();
     return vertx;
   }
@@ -188,7 +207,8 @@ public void clusteredVertx(Handler<AsyncResult<Vertx>> handler) {
       metrics,
       tracer,
       transport,
-      fileResolver);
+      fileResolver,
+      threadFactory);
     vertx.initClustered(options, handler);
   }
 
@@ -205,6 +225,7 @@ public VertxBuilder init() {
     initClusterManager(options, providers);
     providers.addAll(ServiceHelper.loadFactories(VertxServiceProvider.class));
     initProviders(providers);
+    initThreadFactory();
     return this;
   }
 
@@ -266,4 +287,11 @@ private void initFileResolver() {
     }
     fileResolver = new FileResolver(options.getFileSystemOptions());
   }
+
+  private void initThreadFactory() {
+    if (threadFactory != null) {
+      return;
+    }
+    threadFactory = VertxThreadFactory.INSTANCE;
+  }
 }
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
@@ -56,6 +56,7 @@
 import io.vertx.core.shareddata.SharedData;
 import io.vertx.core.shareddata.impl.SharedDataImpl;
 import io.vertx.core.spi.VerticleFactory;
+import io.vertx.core.spi.VertxThreadFactory;
 import io.vertx.core.spi.cluster.ClusterManager;
 import io.vertx.core.spi.cluster.NodeSelector;
 import io.vertx.core.spi.metrics.Metrics;
@@ -71,6 +72,7 @@
 import java.net.InetSocketAddress;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
 
@@ -111,6 +113,7 @@ public class VertxImpl implements VertxInternal, MetricsProvider {
   private final Map<ServerID, NetServerImpl> sharedNetServers = new HashMap<>();
   final WorkerPool workerPool;
   final WorkerPool internalBlockingPool;
+  private final VertxThreadFactory threadFactory;
   private final ThreadFactory eventLoopThreadFactory;
   private final EventLoopGroup eventLoopGroup;
   private final EventLoopGroup acceptorEventLoopGroup;
@@ -132,7 +135,7 @@ public class VertxImpl implements VertxInternal, MetricsProvider {
   private final VertxTracer tracer;
   private final ThreadLocal<WeakReference<AbstractContext>> stickyContext = new ThreadLocal<>();
 
-  VertxImpl(VertxOptions options, ClusterManager clusterManager, NodeSelector nodeSelector, VertxMetrics metrics, VertxTracer<?, ?> tracer, Transport transport, FileResolver fileResolver) {
+  VertxImpl(VertxOptions options, ClusterManager clusterManager, NodeSelector nodeSelector, VertxMetrics metrics, VertxTracer<?, ?> tracer, Transport transport, FileResolver fileResolver, VertxThreadFactory threadFactory) {
     // Sanity check
     if (Vertx.currentContext() != null) {
       log.warn("You're already on a Vert.x context, are you sure you want to create a new Vertx instance?");
@@ -141,9 +144,9 @@ public class VertxImpl implements VertxInternal, MetricsProvider {
     maxEventLoopExecTime = options.getMaxEventLoopExecuteTime();
     maxEventLoopExecTimeUnit = options.getMaxEventLoopExecuteTimeUnit();
     checker = new BlockedThreadChecker(options.getBlockedThreadCheckInterval(), options.getBlockedThreadCheckIntervalUnit(), options.getWarningExceptionTime(), options.getWarningExceptionTimeUnit());
-    eventLoopThreadFactory = new VertxThreadFactory("vert.x-eventloop-thread-", checker, false, maxEventLoopExecTime, maxEventLoopExecTimeUnit);
+    eventLoopThreadFactory = createThreadFactory(maxEventLoopExecTime, maxEventLoopExecTimeUnit, "vert.x-eventloop-thread-", false);
     eventLoopGroup = transport.eventLoopGroup(Transport.IO_EVENT_LOOP_GROUP, options.getEventLoopPoolSize(), eventLoopThreadFactory, NETTY_IO_RATIO);
-    ThreadFactory acceptorEventLoopThreadFactory = new VertxThreadFactory("vert.x-acceptor-thread-", checker, false, options.getMaxEventLoopExecuteTime(), options.getMaxEventLoopExecuteTimeUnit());
+    ThreadFactory acceptorEventLoopThreadFactory = createThreadFactory(options.getMaxEventLoopExecuteTime(), options.getMaxEventLoopExecuteTimeUnit(), "vert.x-acceptor-thread-", false);
     // The acceptor event loop thread needs to be from a different pool otherwise can get lags in accepted connections
     // under a lot of load
     acceptorEventLoopGroup = transport.eventLoopGroup(Transport.ACCEPTOR_EVENT_LOOP_GROUP, 1, acceptorEventLoopThreadFactory, 100);
@@ -151,10 +154,10 @@ public class VertxImpl implements VertxInternal, MetricsProvider {
     int workerPoolSize = options.getWorkerPoolSize();
     ExecutorService workerExec = new ThreadPoolExecutor(workerPoolSize, workerPoolSize,
       0L, TimeUnit.MILLISECONDS, new LinkedTransferQueue<>(),
-      new VertxThreadFactory("vert.x-worker-thread-", checker, true, options.getMaxWorkerExecuteTime(), options.getMaxWorkerExecuteTimeUnit()));
+      createThreadFactory(options.getMaxWorkerExecuteTime(), options.getMaxWorkerExecuteTimeUnit(), "vert.x-worker-thread-", true));
     PoolMetrics workerPoolMetrics = metrics != null ? metrics.createPoolMetrics("worker", "vert.x-worker-thread", options.getWorkerPoolSize()) : null;
     ExecutorService internalBlockingExec = Executors.newFixedThreadPool(options.getInternalBlockingPoolSize(),
-        new VertxThreadFactory("vert.x-internal-blocking-", checker, true, options.getMaxWorkerExecuteTime(), options.getMaxWorkerExecuteTimeUnit()));
+      createThreadFactory(options.getMaxWorkerExecuteTime(), options.getMaxWorkerExecuteTimeUnit(), "vert.x-internal-blocking-", true));
     PoolMetrics internalBlockingPoolMetrics = metrics != null ? metrics.createPoolMetrics("worker", "vert.x-internal-blocking", options.getInternalBlockingPoolSize()) : null;
     internalBlockingPool = new WorkerPool(internalBlockingExec, internalBlockingPoolMetrics);
     namedWorkerPools = new HashMap<>();
@@ -163,6 +166,7 @@ public class VertxImpl implements VertxInternal, MetricsProvider {
     maxWorkerExecTime = options.getMaxWorkerExecuteTime();
     maxWorkerExecTimeUnit = options.getMaxWorkerExecuteTimeUnit();
 
+    this.threadFactory = threadFactory;
     this.metrics = metrics;
     this.transport = transport;
     this.fileResolver = fileResolver;
@@ -1115,7 +1119,7 @@ public synchronized SharedWorkerPool createSharedWorkerPool(String name, int poo
     }
     SharedWorkerPool sharedWorkerPool = namedWorkerPools.get(name);
     if (sharedWorkerPool == null) {
-      ExecutorService workerExec = Executors.newFixedThreadPool(poolSize, new VertxThreadFactory(name + "-", checker, true, maxExecuteTime, maxExecuteTimeUnit));
+      ExecutorService workerExec = Executors.newFixedThreadPool(poolSize, createThreadFactory(maxExecuteTime, maxExecuteTimeUnit, name + "-", true));
       PoolMetrics workerMetrics = metrics != null ? metrics.createPoolMetrics("worker", name, poolSize) : null;
       namedWorkerPools.put(name, sharedWorkerPool = new SharedWorkerPool(name, workerExec, workerMetrics));
     } else {
@@ -1124,6 +1128,18 @@ public synchronized SharedWorkerPool createSharedWorkerPool(String name, int poo
     return sharedWorkerPool;
   }
 
+  private ThreadFactory createThreadFactory(long maxExecuteTime, TimeUnit maxExecuteTimeUnit, String prefix, boolean worker) {
+    AtomicInteger threadCount = new AtomicInteger(0);
+    return runnable -> {
+      VertxThread thread = threadFactory.newVertxThread(runnable, prefix + threadCount.getAndIncrement(), worker, maxExecuteTime, maxExecuteTimeUnit);
+      checker.registerThread(thread, thread);
+      // Vert.x threads are NOT daemons - we want them to prevent JVM exit so embedded user doesn't
+      // have to explicitly prevent JVM from exiting.
+      thread.setDaemon(false);
+      return thread;
+    };
+  }
+
   @Override
   public Vertx exceptionHandler(Handler<Throwable> handler) {
     exceptionHandler = handler;
diff --git a/src/main/java/io/vertx/core/impl/VertxThread.java b/src/main/java/io/vertx/core/impl/VertxThread.java
--- a/src/main/java/io/vertx/core/impl/VertxThread.java
+++ b/src/main/java/io/vertx/core/impl/VertxThread.java
@@ -18,7 +18,7 @@
 /**
  * @author <a href="mailto:nmaurer@redhat.com">Norman Maurer</a>
  */
-public final class VertxThread extends FastThreadLocalThread implements BlockedThreadChecker.Task {
+public class VertxThread extends FastThreadLocalThread implements BlockedThreadChecker.Task {
 
   static final String DISABLE_TCCL_PROP_NAME = "vertx.disableTCCL";
   static final boolean DISABLE_TCCL = Boolean.getBoolean(DISABLE_TCCL_PROP_NAME);
diff --git a/src/main/java/io/vertx/core/impl/VertxThreadFactory.java b/src/main/java/io/vertx/core/impl/VertxThreadFactory.java
deleted file mode 100644
--- a/src/main/java/io/vertx/core/impl/VertxThreadFactory.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Copyright (c) 2011-2019 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.impl;
-
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * @author <a href="http://tfox.org">Tim Fox</a>
- */
-public class VertxThreadFactory implements ThreadFactory {
-
-  private final String prefix;
-  private final AtomicInteger threadCount = new AtomicInteger(0);
-  private final BlockedThreadChecker checker;
-  private final boolean worker;
-  private final long maxExecTime;
-  private final TimeUnit maxExecTimeUnit;
-
-  VertxThreadFactory(String prefix, BlockedThreadChecker checker, boolean worker, long maxExecTime, TimeUnit maxExecTimeUnit) {
-    this.prefix = prefix;
-    this.checker = checker;
-    this.worker = worker;
-    this.maxExecTime = maxExecTime;
-    this.maxExecTimeUnit = maxExecTimeUnit;
-  }
-
-  public Thread newThread(Runnable runnable) {
-    VertxThread t = new VertxThread(runnable, prefix + threadCount.getAndIncrement(), worker, maxExecTime, maxExecTimeUnit);
-    // Vert.x threads are NOT daemons - we want them to prevent JVM exit so embededd user doesn't
-    // have to explicitly prevent JVM from exiting.
-    if (checker != null) {
-      checker.registerThread(t, t);
-    }
-    // I know the default is false anyway, but just to be explicit-  Vert.x threads are NOT daemons
-    // we want to prevent the JVM from exiting until Vert.x instances are closed
-    t.setDaemon(false);
-    return t;
-  }
-}
diff --git a/src/main/java/io/vertx/core/spi/VertxThreadFactory.java b/src/main/java/io/vertx/core/spi/VertxThreadFactory.java
new file mode 100644
--- /dev/null
+++ b/src/main/java/io/vertx/core/spi/VertxThreadFactory.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright (c) 2011-2019 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.spi;
+
+import io.vertx.core.impl.VertxBuilder;
+import io.vertx.core.impl.VertxThread;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author <a href="http://tfox.org">Tim Fox</a>
+ */
+public interface VertxThreadFactory extends VertxServiceProvider {
+
+  VertxThreadFactory INSTANCE = new VertxThreadFactory() {
+  };
+
+  @Override
+  default void init(VertxBuilder builder) {
+    if (builder.threadFactory() == null) {
+      builder.threadFactory(this);
+    }
+  }
+
+  default VertxThread newVertxThread(Runnable target, String name, boolean worker, long maxExecTime, TimeUnit maxExecTimeUnit) {
+    return new VertxThread(target, name, worker, maxExecTime, maxExecTimeUnit);
+  }
+}


Test patch:
diff --git a/src/test/benchmarks/io/vertx/core/impl/VertxExecutorService.java b/src/test/benchmarks/io/vertx/core/impl/VertxExecutorService.java
--- a/src/test/benchmarks/io/vertx/core/impl/VertxExecutorService.java
+++ b/src/test/benchmarks/io/vertx/core/impl/VertxExecutorService.java
@@ -11,6 +11,8 @@
 
 package io.vertx.core.impl;
 
+import io.vertx.core.spi.VertxThreadFactory;
+
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -22,8 +24,8 @@ public class VertxExecutorService extends ThreadPoolExecutor {
 
   public VertxExecutorService(int maxThreads, String prefix) {
     super(maxThreads, maxThreads,
-        0L, TimeUnit.MILLISECONDS,
-        new LinkedBlockingQueue<>(),
-        new VertxThreadFactory(prefix, new BlockedThreadChecker(10000, TimeUnit.MILLISECONDS, 10000, TimeUnit.MILLISECONDS), false, 10000, TimeUnit.NANOSECONDS));
+      0L, TimeUnit.MILLISECONDS,
+      new LinkedBlockingQueue<>(),
+      r -> VertxThreadFactory.INSTANCE.newVertxThread(r, prefix, false, 10000, TimeUnit.NANOSECONDS));
   }
 }
diff --git a/src/test/classpath/threadfactory/META-INF/services/io.vertx.core.spi.VertxServiceProvider b/src/test/classpath/threadfactory/META-INF/services/io.vertx.core.spi.VertxServiceProvider
new file mode 100644
--- /dev/null
+++ b/src/test/classpath/threadfactory/META-INF/services/io.vertx.core.spi.VertxServiceProvider
@@ -0,0 +1 @@
+io.vertx.it.CustomVertxThreadFactory
diff --git a/src/test/java/io/vertx/core/impl/GlobalEventExecutorNotificationTest.java b/src/test/java/io/vertx/core/impl/GlobalEventExecutorNotificationTest.java
--- a/src/test/java/io/vertx/core/impl/GlobalEventExecutorNotificationTest.java
+++ b/src/test/java/io/vertx/core/impl/GlobalEventExecutorNotificationTest.java
@@ -60,7 +60,7 @@ public ChannelFactory<? extends Channel> channelFactory(boolean domainSocket) {
           throw cause;
         };
       }
-    }).vertx();
+    }).init().vertx();
 
     vertx.createNetServer().connectHandler(so -> {
       fail();
@@ -83,7 +83,7 @@ public ChannelFactory<? extends ServerChannel> serverChannelFactory(boolean doma
           throw cause;
         };
       }
-    }).vertx();
+    }).init().vertx();
 
     vertx.createNetServer()
       .connectHandler(so -> fail())
@@ -103,7 +103,7 @@ public ChannelFactory<? extends ServerChannel> serverChannelFactory(boolean doma
           throw cause;
         };
       }
-    }).vertx();
+    }).init().vertx();
 
     vertx.createHttpServer()
       .requestHandler(req -> fail())
diff --git a/src/test/java/io/vertx/it/CustomVertxThread.java b/src/test/java/io/vertx/it/CustomVertxThread.java
new file mode 100644
--- /dev/null
+++ b/src/test/java/io/vertx/it/CustomVertxThread.java
@@ -0,0 +1,11 @@
+package io.vertx.it;
+
+import io.vertx.core.impl.VertxThread;
+
+import java.util.concurrent.TimeUnit;
+
+public class CustomVertxThread extends VertxThread  {
+  public CustomVertxThread(Runnable target, String name, boolean worker, long maxExecTime, TimeUnit maxExecTimeUnit) {
+    super(target, name, worker, maxExecTime, maxExecTimeUnit);
+  }
+}
diff --git a/src/test/java/io/vertx/it/CustomVertxThreadFactory.java b/src/test/java/io/vertx/it/CustomVertxThreadFactory.java
new file mode 100644
--- /dev/null
+++ b/src/test/java/io/vertx/it/CustomVertxThreadFactory.java
@@ -0,0 +1,14 @@
+package io.vertx.it;
+
+import io.vertx.core.impl.VertxThread;
+import io.vertx.core.spi.VertxThreadFactory;
+
+import java.util.concurrent.TimeUnit;
+
+public class CustomVertxThreadFactory implements VertxThreadFactory {
+
+  @Override
+  public VertxThread newVertxThread(Runnable target, String name, boolean worker, long maxExecTime, TimeUnit maxExecTimeUnit) {
+    return new CustomVertxThread(target, name, worker, maxExecTime, maxExecTimeUnit);
+  }
+}
diff --git a/src/test/java/io/vertx/it/VertxThreadFactoryTest.java b/src/test/java/io/vertx/it/VertxThreadFactoryTest.java
new file mode 100644
--- /dev/null
+++ b/src/test/java/io/vertx/it/VertxThreadFactoryTest.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright (c) 2011-2019 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.it;
+
+import io.vertx.test.core.VertxTestBase;
+import org.junit.Test;
+
+/**
+ * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
+ */
+public class VertxThreadFactoryTest extends VertxTestBase {
+
+  @Test
+  public void testJsonObject() {
+    vertx.runOnContext(v -> {
+      Thread current = Thread.currentThread();
+      assertEquals(CustomVertxThread.class, current.getClass());
+      testComplete();
+    });
+    await();
+  }
+}