Refactor the pool metrics to be reusable for other kind of pools that…

… can exposed to the metrics implementation

src/main/java/io/vertx/core/impl/ContextImpl.java

@@ -23,7 +23,7 @@
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.logging.Logger;
 import io.vertx.core.logging.LoggerFactory;
-import io.vertx.core.spi.metrics.ThreadPoolMetrics;
+import io.vertx.core.spi.metrics.PoolMetrics;
 
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
@@ -294,12 +294,12 @@ public <T> void executeBlocking(Handler<Future<T>> blockingCodeHandler, Handler<
 
   <T> void executeBlocking(Action<T> action, Handler<Future<T>> blockingCodeHandler,
       Handler<AsyncResult<T>> resultHandler,
-      Executor exec, ThreadPoolMetrics metrics) {
+      Executor exec, PoolMetrics metrics) {
     Object metric = metrics != null ? metrics.taskSubmitted() : null;
     try {
       exec.execute(() -> {
         if (metrics != null) {
-          metrics.taskExecuting(metric);
+          metrics.taskBegin(metric);
         }
         Future<T> res = Future.future();
         try {
@@ -314,7 +314,7 @@ <T> void executeBlocking(Action<T> action, Handler<Future<T>> blockingCodeHandle
           res.fail(e);
         }
         if (metrics != null) {
-          metrics.taskCompleted(metric, res.succeeded());
+          metrics.taskEnd(metric, res.succeeded());
         }
         if (resultHandler != null) {
           runOnContext(v -> res.setHandler(resultHandler));
@@ -335,7 +335,7 @@ protected synchronized Map<String, Object> contextData() {
     return contextData;
   }
 
-  protected Runnable wrapTask(ContextTask cTask, Handler<Void> hTask, boolean checkThread, ThreadPoolMetrics metrics) {
+  protected Runnable wrapTask(ContextTask cTask, Handler<Void> hTask, boolean checkThread, PoolMetrics metrics) {
     Object metric = metrics != null ? metrics.taskSubmitted() : null;
     return () -> {
       Thread th = Thread.currentThread();
@@ -351,7 +351,7 @@ protected Runnable wrapTask(ContextTask cTask, Handler<Void> hTask, boolean chec
         }
       }
       if (metrics != null) {
-        metrics.taskExecuting(metric);
+        metrics.taskBegin(metric);
       }
       if (!DISABLE_TIMINGS) {
         current.executeStart();
@@ -364,7 +364,7 @@ protected Runnable wrapTask(ContextTask cTask, Handler<Void> hTask, boolean chec
           hTask.handle(null);
         }
         if (metrics != null) {
-          metrics.taskCompleted(metric, true);
+          metrics.taskEnd(metric, true);
         }
       } catch (Throwable t) {
         log.error("Unhandled exception", t);
@@ -376,7 +376,7 @@ protected Runnable wrapTask(ContextTask cTask, Handler<Void> hTask, boolean chec
           handler.handle(t);
         }
         if (metrics != null) {
-          metrics.taskCompleted(metric, false);
+          metrics.taskEnd(metric, false);
         }
       } finally {
         // We don't unset the context after execution - this is done later when the context is closed via

src/main/java/io/vertx/core/impl/VertxImpl.java

@@ -58,7 +58,7 @@
 import io.vertx.core.spi.cluster.ClusterManager;
 import io.vertx.core.spi.metrics.Metrics;
 import io.vertx.core.spi.metrics.MetricsProvider;
-import io.vertx.core.spi.metrics.ThreadPoolMetrics;
+import io.vertx.core.spi.metrics.PoolMetrics;
 import io.vertx.core.spi.metrics.VertxMetrics;
 
 import java.io.File;
@@ -136,12 +136,12 @@ public class VertxImpl implements VertxInternal, MetricsProvider {
 
     metrics = initialiseMetrics(options);
 
-    ThreadPoolMetrics workerPoolMetrics = isMetricsEnabled() ? metrics.createMetrics("vert.x-worker-thread", options.getWorkerPoolSize()) : null;
     ExecutorService workerExec = Executors.newFixedThreadPool(options.getWorkerPoolSize(),
-                                              new VertxThreadFactory("vert.x-worker-thread-", checker, true, options.getMaxWorkerExecuteTime()));
-    ThreadPoolMetrics internalBlockingPoolMetrics = isMetricsEnabled() ? metrics.createMetrics("vert.x-internal-blocking", options.getInternalBlockingPoolSize()) : null;
+        new VertxThreadFactory("vert.x-worker-thread-", checker, true, options.getMaxWorkerExecuteTime()));
+    PoolMetrics workerPoolMetrics = isMetricsEnabled() ? metrics.createMetrics(workerExec, "vert.x-worker-thread", options.getWorkerPoolSize()) : null;
     ExecutorService internalBlockingExec = Executors.newFixedThreadPool(options.getInternalBlockingPoolSize(),
-                                                        new VertxThreadFactory("vert.x-internal-blocking-", checker, true, options.getMaxWorkerExecuteTime()));
+        new VertxThreadFactory("vert.x-internal-blocking-", checker, true, options.getMaxWorkerExecuteTime()));
+    PoolMetrics internalBlockingPoolMetrics = isMetricsEnabled() ? metrics.createMetrics(internalBlockingExec, "vert.x-internal-blocking", options.getInternalBlockingPoolSize()) : null;
     internalBlockingPool = new WorkerPool(internalBlockingExec, internalBlockingPoolMetrics);
     namedWorkerPools = new HashMap<>();
     workerPool = new WorkerPool(workerExec, workerPoolMetrics);
@@ -872,7 +872,7 @@ class SharedWorkerPool extends WorkerPool {
     private final String name;
     private int refCount = 1;
 
-    public SharedWorkerPool(String name, ExecutorService workerExec, ThreadPoolMetrics workerMetrics) {
+    public SharedWorkerPool(String name, ExecutorService workerExec, PoolMetrics workerMetrics) {
       super(workerExec, workerMetrics);
       this.workerExec = workerExec;
       this.name = name;
@@ -911,8 +911,8 @@ public synchronized NamedWorkerExecutor createWorkerExecutor(String name, int po
     }
     SharedWorkerPool sharedWorkerPool = namedWorkerPools.get(name);
     if (sharedWorkerPool == null) {
-      ThreadPoolMetrics workerMetrics = isMetricsEnabled() ? metrics.createMetrics(name, poolSize) : null;
       ExecutorService workerExec = Executors.newFixedThreadPool(poolSize, new VertxThreadFactory(name + "-", checker, true, maxExecuteTime));
+      PoolMetrics workerMetrics = isMetricsEnabled() ? metrics.createMetrics(workerExec, name, poolSize) : null;
       namedWorkerPools.put(name, sharedWorkerPool = new SharedWorkerPool(name, workerExec, workerMetrics));
     } else {
       sharedWorkerPool.refCount++;

src/main/java/io/vertx/core/impl/VertxThreadFactory.java

@@ -16,8 +16,6 @@
 
 package io.vertx.core.impl;
 
-import io.vertx.core.spi.metrics.ThreadPoolMetrics;
-
 import java.util.Map;
 import java.util.WeakHashMap;
 import java.util.concurrent.ThreadFactory;

src/main/java/io/vertx/core/impl/WorkerPool.java

@@ -16,7 +16,7 @@
 
 package io.vertx.core.impl;
 
-import io.vertx.core.spi.metrics.ThreadPoolMetrics;
+import io.vertx.core.spi.metrics.PoolMetrics;
 
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
@@ -28,9 +28,9 @@ class WorkerPool {
 
   private final OrderedExecutorFactory orderedFact;
   private final ExecutorService pool;
-  private final ThreadPoolMetrics metrics;
+  private final PoolMetrics metrics;
 
-  public WorkerPool(ExecutorService pool, ThreadPoolMetrics metrics) {
+  public WorkerPool(ExecutorService pool, PoolMetrics metrics) {
     this.orderedFact = new OrderedExecutorFactory(pool);
     this.pool = pool;
     this.metrics = metrics;
@@ -44,7 +44,7 @@ Executor createOrderedExecutor() {
     return orderedFact.getExecutor();
   }
 
-  ThreadPoolMetrics metrics() {
+  PoolMetrics metrics() {
     return metrics;
   }
 

src/main/java/io/vertx/core/metrics/impl/DummyVertxMetrics.java

@@ -91,11 +91,10 @@ public DatagramSocketMetrics createMetrics(DatagramSocket socket, DatagramSocket
   }
 
   @Override
-  public ThreadPoolMetrics createMetrics(String name, int poolSize) {
+  public <P> PoolMetrics<?> createMetrics(P pool, String poolName, int maxPoolSize) {
     return new DummyWorkerPoolMetrics();
   }
 
-
   @Override
   public void close() {
   }
@@ -345,7 +344,7 @@ public boolean isEnabled() {
     }
   }
 
-  private class DummyWorkerPoolMetrics implements ThreadPoolMetrics<Void> {
+  private class DummyWorkerPoolMetrics implements PoolMetrics<Void> {
 
     @Override
     public Void taskSubmitted() {
@@ -357,11 +356,11 @@ public void taskRejected(Void task) {
     }
 
     @Override
-    public void taskExecuting(Void task) {
+    public void taskBegin(Void task) {
     }
 
     @Override
-    public void taskCompleted(Void task, boolean succeeded) {
+    public void taskEnd(Void task, boolean succeeded) {
     }
 
     @Override

src/main/java/io/vertx/core/spi/metrics/ThreadPoolMetrics.java → src/main/java/io/vertx/core/spi/metrics/PoolMetrics.java

@@ -17,35 +17,35 @@
 package io.vertx.core.spi.metrics;
 
 /**
- * An SPI used internally by Vert.x to gather metrics on a worker thread (execute blocking, worker verticle).
+ * An SPI used internally by Vert.x to gather metrics on pools used by Vert.x  (execute blocking, worker verticle).
  *
  * @author <a href="http://escoffier.me">Clement Escoffier</a>
  */
-public interface ThreadPoolMetrics<T> extends Metrics {
+public interface PoolMetrics<T> extends Metrics {
 
   /**
-   * A new task has been submitted to the worker queue.
+   * A new task has been submitted to access the resource.
    * This method is called from the submitter context.
    *
    * @return the submitted task.
    */
   T taskSubmitted();
 
   /**
-   * The task has been rejected. The underlying thread pool has probably be shutdown.
+   * The task has been rejected. The underlying resource has probably be shutdown.
    */
   void taskRejected(T task);
 
   /**
-   * The submitted task start its execution.
+   * The submitted task start to use the resource.
    */
-  void taskExecuting(T task);
+  void taskBegin(T task);
 
   /**
-   * The submitted tasks has completed its execution.
+   * The submitted tasks has completed its execution and release the resource.
    *
    * @param succeeded whether or not the task has gracefully completed
    */
-  void taskCompleted(T task, boolean succeeded);
+  void taskEnd(T task, boolean succeeded);
 
 }

src/main/java/io/vertx/core/spi/metrics/VertxMetrics.java

@@ -17,7 +17,6 @@
 package io.vertx.core.spi.metrics;
 
 import io.vertx.core.Verticle;
-import io.vertx.core.Vertx;
 import io.vertx.core.datagram.DatagramSocket;
 import io.vertx.core.datagram.DatagramSocketOptions;
 import io.vertx.core.eventbus.EventBus;
@@ -168,11 +167,12 @@ default void eventBusInitialized(EventBus bus) {
   }
 
   /**
-   * Provides the thread pool metrics SPI.
+   * Provides the pool metrics SPI.
    *
+   * @param pool the pool of resource, it can be used by the metrics implementation to gather extra statistics
    * @param poolName the name of the thread pool
-   * @param poolSize the size of the pool
+   * @param maxPoolSize the pool max size, or -1 if the number cannot be determined
    * @return the thread pool metrics SPI
    */
-  ThreadPoolMetrics<?> createMetrics(String poolName, int poolSize);
+  <P> PoolMetrics<?> createMetrics(P pool, String poolName, int maxPoolSize);
 }

src/test/java/io/vertx/test/core/MetricsTest.java

@@ -27,7 +27,7 @@
 import io.vertx.core.http.*;
 import io.vertx.core.metrics.MetricsOptions;
 import io.vertx.core.net.NetSocket;
-import io.vertx.core.spi.metrics.ThreadPoolMetrics;
+import io.vertx.core.spi.metrics.PoolMetrics;
 import io.vertx.test.fakemetrics.*;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -673,7 +673,7 @@ private void testDatagram(String host, Consumer<PacketMetric> checker) throws Ex
 
   @Test
   public void testThreadPoolMetricsWithExecuteBlocking() {
-    Map<String, ThreadPoolMetrics> all = FakeThreadPoolMetrics.getThreadPoolMetrics();
+    Map<String, PoolMetrics> all = FakeThreadPoolMetrics.getThreadPoolMetrics();
 
     FakeThreadPoolMetrics metrics = (FakeThreadPoolMetrics) all.get("vert.x-worker-thread");
 
@@ -723,7 +723,7 @@ public void testThreadPoolMetricsWithExecuteBlocking() {
   public void testThreadPoolMetricsWithInternalExecuteBlocking() {
     // Internal blocking thread pool is used by blocking file system actions.
 
-    Map<String, ThreadPoolMetrics> all = FakeThreadPoolMetrics.getThreadPoolMetrics();
+    Map<String, PoolMetrics> all = FakeThreadPoolMetrics.getThreadPoolMetrics();
     FakeThreadPoolMetrics metrics = (FakeThreadPoolMetrics) all.get("vert.x-internal-blocking");
 
     assertThat(metrics.getPoolSize(), is(getOptions().getInternalBlockingPoolSize()));
@@ -784,7 +784,7 @@ public void testThreadPoolMetricsWithWorkerVerticleAndMultiThread() {
 
   private void testWithWorkerVerticle(DeploymentOptions options) {
     AtomicInteger counter = new AtomicInteger();
-    Map<String, ThreadPoolMetrics> all = FakeThreadPoolMetrics.getThreadPoolMetrics();
+    Map<String, PoolMetrics> all = FakeThreadPoolMetrics.getThreadPoolMetrics();
     FakeThreadPoolMetrics metrics = (FakeThreadPoolMetrics) all.get("vert.x-worker-thread");
 
     assertThat(metrics.getPoolSize(), is(getOptions().getInternalBlockingPoolSize()));
@@ -854,7 +854,7 @@ public void testThreadPoolMetricsWithNamedExecuteBlocking() {
 
     WorkerExecutor workerExec = vertx.createWorkerExecutor("my-pool", 10);
 
-    Map<String, ThreadPoolMetrics> all = FakeThreadPoolMetrics.getThreadPoolMetrics();
+    Map<String, PoolMetrics> all = FakeThreadPoolMetrics.getThreadPoolMetrics();
 
     FakeThreadPoolMetrics metrics = (FakeThreadPoolMetrics) all.get("my-pool");
 

src/test/java/io/vertx/test/fakemetrics/FakeThreadPoolMetrics.java

@@ -16,20 +16,20 @@
 
 package io.vertx.test.fakemetrics;
 
-import io.vertx.core.spi.metrics.ThreadPoolMetrics;
+import io.vertx.core.spi.metrics.PoolMetrics;
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
- * A fake implementation of the {@link ThreadPoolMetrics} SPI.
+ * A fake implementation of the {@link PoolMetrics} SPI.
  *
  * @author <a href="http://escoffier.me">Clement Escoffier</a>
  */
-public class FakeThreadPoolMetrics implements ThreadPoolMetrics<Void> {
-  private final static Map<String, ThreadPoolMetrics> METRICS = new ConcurrentHashMap<>();
+public class FakeThreadPoolMetrics implements PoolMetrics<Void> {
+  private final static Map<String, PoolMetrics> METRICS = new ConcurrentHashMap<>();
 
   private final int poolSize;
 
@@ -69,14 +69,14 @@ public void taskRejected(Void task) {
   }
 
   @Override
-  public void taskExecuting(Void task) {
+  public void taskBegin(Void task) {
     waiting.decrementAndGet();
     idle.decrementAndGet();
     running.incrementAndGet();
   }
 
   @Override
-  public void taskCompleted(Void task, boolean succeeded) {
+  public void taskEnd(Void task, boolean succeeded) {
     running.decrementAndGet();
     idle.incrementAndGet();
     completed.incrementAndGet();
@@ -117,7 +117,7 @@ public int numberOfRunningTasks() {
     return running.get();
   }
 
-  public static Map<String, ThreadPoolMetrics> getThreadPoolMetrics() {
+  public static Map<String, PoolMetrics> getThreadPoolMetrics() {
     return METRICS;
   }
 

src/test/java/io/vertx/test/fakemetrics/FakeVertxMetrics.java

@@ -32,6 +32,7 @@
 import io.vertx.core.net.SocketAddress;
 import io.vertx.core.spi.metrics.*;
 
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
@@ -135,8 +136,8 @@ public DatagramSocketMetrics createMetrics(DatagramSocket socket, DatagramSocket
   }
 
   @Override
-  public ThreadPoolMetrics createMetrics(String poolName, int poolSize) {
-    return new FakeThreadPoolMetrics(poolName, poolSize);
+  public <P> PoolMetrics<?> createMetrics(P pool, String poolName, int maxPoolSize) {
+    return new FakeThreadPoolMetrics(poolName, maxPoolSize);
   }
 
   public boolean isEnabled() {