HTTPCLIENT-2233- Add Metrics Collection for IOReactor Thread Pool

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2AsyncRequester.java

@@ -44,6 +44,7 @@
 import org.apache.hc.core5.pool.ManagedConnPool;
 import org.apache.hc.core5.reactor.IOEventHandlerFactory;
 import org.apache.hc.core5.reactor.IOReactorConfig;
+import org.apache.hc.core5.reactor.IOReactorMetricsListener;
 import org.apache.hc.core5.reactor.IOSession;
 import org.apache.hc.core5.reactor.IOSessionListener;
 import org.apache.hc.core5.reactor.ProtocolIOSession;
@@ -71,8 +72,9 @@ public H2AsyncRequester(
             final Decorator<IOSession> ioSessionDecorator,
             final Callback<Exception> exceptionCallback,
             final IOSessionListener sessionListener,
-            final ManagedConnPool<HttpHost, IOSession> connPool) {
-        super(ioReactorConfig, eventHandlerFactory, ioSessionDecorator, exceptionCallback, sessionListener, connPool);
+            final ManagedConnPool<HttpHost, IOSession> connPool,
+            final IOReactorMetricsListener threadPoolListener) {
+        super(ioReactorConfig, eventHandlerFactory, ioSessionDecorator, exceptionCallback, sessionListener, connPool, threadPoolListener);
         this.versionPolicy = versionPolicy != null ? versionPolicy : HttpVersionPolicy.NEGOTIATE;
     }
 
@@ -91,9 +93,10 @@ public H2AsyncRequester(
             final IOSessionListener sessionListener,
             final ManagedConnPool<HttpHost, IOSession> connPool,
             final TlsStrategy tlsStrategy,
-            final Timeout handshakeTimeout) {
+            final Timeout handshakeTimeout,
+            final IOReactorMetricsListener threadPoolListener) {
         super(ioReactorConfig, eventHandlerFactory, ioSessionDecorator, exceptionCallback, sessionListener, connPool,
-                tlsStrategy, handshakeTimeout);
+                tlsStrategy, handshakeTimeout, threadPoolListener);
         this.versionPolicy = versionPolicy != null ? versionPolicy : HttpVersionPolicy.NEGOTIATE;
     }
 

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java

@@ -70,6 +70,7 @@
 import org.apache.hc.core5.reactor.Command;
 import org.apache.hc.core5.reactor.IOEventHandlerFactory;
 import org.apache.hc.core5.reactor.IOReactorConfig;
+import org.apache.hc.core5.reactor.IOReactorMetricsListener;
 import org.apache.hc.core5.reactor.IOSession;
 import org.apache.hc.core5.reactor.IOSessionListener;
 import org.apache.hc.core5.util.Args;
@@ -96,9 +97,10 @@ public H2MultiplexingRequester(
             final Callback<Exception> exceptionCallback,
             final IOSessionListener sessionListener,
             final Resolver<HttpHost, InetSocketAddress> addressResolver,
-            final TlsStrategy tlsStrategy) {
+            final TlsStrategy tlsStrategy,
+            final IOReactorMetricsListener threadPoolListener) {
         super(eventHandlerFactory, ioReactorConfig, ioSessionDecorator, exceptionCallback, sessionListener,
-                        ShutdownCommand.GRACEFUL_IMMEDIATE_CALLBACK, DefaultAddressResolver.INSTANCE);
+                        ShutdownCommand.GRACEFUL_IMMEDIATE_CALLBACK, DefaultAddressResolver.INSTANCE, threadPoolListener);
         this.connPool = new H2ConnPool(this, addressResolver, tlsStrategy);
     }
 

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequesterBootstrap.java

@@ -47,6 +47,7 @@
 import org.apache.hc.core5.http2.nio.support.DefaultAsyncPushConsumerFactory;
 import org.apache.hc.core5.http2.ssl.H2ClientTlsStrategy;
 import org.apache.hc.core5.reactor.IOReactorConfig;
+import org.apache.hc.core5.reactor.IOReactorMetricsListener;
 import org.apache.hc.core5.reactor.IOSession;
 import org.apache.hc.core5.reactor.IOSessionListener;
 import org.apache.hc.core5.util.Args;
@@ -71,6 +72,8 @@ public class H2MultiplexingRequesterBootstrap {
     private IOSessionListener sessionListener;
     private H2StreamListener streamListener;
 
+    private IOReactorMetricsListener threadPoolListener;
+
     private H2MultiplexingRequesterBootstrap() {
         this.routeEntries = new ArrayList<>();
     }
@@ -164,6 +167,17 @@ public final H2MultiplexingRequesterBootstrap setIOSessionListener(final IOSessi
         return this;
     }
 
+    /**
+     * Sets {@link IOReactorMetricsListener} instance.
+     *
+     * @return this instance.
+     * @since 5.4
+     */
+    public final H2MultiplexingRequesterBootstrap setIOReactorMetricsListener(final IOReactorMetricsListener threadPoolListener) {
+        this.threadPoolListener = threadPoolListener;
+        return this;
+    }
+
     /**
      * Sets {@link H2StreamListener} instance.
      *
@@ -243,7 +257,8 @@ public H2MultiplexingRequester create() {
                 exceptionCallback,
                 sessionListener,
                 DefaultAddressResolver.INSTANCE,
-                tlsStrategy != null ? tlsStrategy : new H2ClientTlsStrategy());
+                tlsStrategy != null ? tlsStrategy : new H2ClientTlsStrategy(),
+                threadPoolListener);
     }
 
 }

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2RequesterBootstrap.java

@@ -65,6 +65,7 @@
 import org.apache.hc.core5.pool.StrictConnPool;
 import org.apache.hc.core5.reactor.IOEventHandlerFactory;
 import org.apache.hc.core5.reactor.IOReactorConfig;
+import org.apache.hc.core5.reactor.IOReactorMetricsListener;
 import org.apache.hc.core5.reactor.IOSession;
 import org.apache.hc.core5.reactor.IOSessionListener;
 import org.apache.hc.core5.util.Args;
@@ -99,6 +100,8 @@ public class H2RequesterBootstrap {
     private H2StreamListener streamListener;
     private Http1StreamListener http1StreamListener;
     private ConnPoolListener<HttpHost> connPoolListener;
+    private IOReactorMetricsListener threadPoolListener;
+
 
     private H2RequesterBootstrap() {
         this.routeEntries = new ArrayList<>();
@@ -249,6 +252,17 @@ public final H2RequesterBootstrap setIOSessionListener(final IOSessionListener s
         return this;
     }
 
+    /**
+     * Sets {@link IOReactorMetricsListener} instance.
+     *
+     * @return this instance.
+     * @since 5.4
+     */
+    public final H2RequesterBootstrap setIOReactorMetricsListener(final IOReactorMetricsListener threadPoolListener) {
+        this.threadPoolListener = threadPoolListener;
+        return this;
+    }
+
     /**
      * Sets {@link H2StreamListener} instance.
      *
@@ -393,7 +407,8 @@ public H2AsyncRequester create() {
                 sessionListener,
                 connPool,
                 actualTlsStrategy,
-                handshakeTimeout);
+                handshakeTimeout,
+                threadPoolListener);
     }
 
 }

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2ServerBootstrap.java

@@ -71,6 +71,7 @@
 import org.apache.hc.core5.net.URIAuthority;
 import org.apache.hc.core5.reactor.IOEventHandlerFactory;
 import org.apache.hc.core5.reactor.IOReactorConfig;
+import org.apache.hc.core5.reactor.IOReactorMetricsListener;
 import org.apache.hc.core5.reactor.IOSession;
 import org.apache.hc.core5.reactor.IOSessionListener;
 import org.apache.hc.core5.util.Args;
@@ -102,6 +103,7 @@ public class H2ServerBootstrap {
     private IOSessionListener sessionListener;
     private H2StreamListener h2StreamListener;
     private Http1StreamListener http1StreamListener;
+    private IOReactorMetricsListener threadPoolListener;
 
     private H2ServerBootstrap() {
         this.routeEntries = new ArrayList<>();
@@ -208,6 +210,18 @@ public final H2ServerBootstrap setIOSessionDecorator(final Decorator<IOSession>
         return this;
     }
 
+
+    /**
+     * Sets {@link IOReactorMetricsListener} instance.
+     *
+     * @return this instance.
+     * @since 5.4
+     */
+    public final H2ServerBootstrap setIOReactorMetricsListener(final IOReactorMetricsListener threadPoolListener) {
+        this.threadPoolListener = threadPoolListener;
+        return this;
+    }
+
     /**
      * Sets {@link Exception} {@link Callback} instance.
      *
@@ -247,7 +261,6 @@ public final H2ServerBootstrap setStreamListener(final Http1StreamListener http1
         this.http1StreamListener = http1StreamListener;
         return this;
     }
-
     /**
      * @return this instance.
      * @deprecated Use {@link RequestRouter}.
@@ -522,7 +535,7 @@ public HttpAsyncServer create() {
                 handshakeTimeout);
 
         return new HttpAsyncServer(ioEventHandlerFactory, ioReactorConfig, ioSessionDecorator, exceptionCallback,
-                sessionListener, actualCanonicalHostName);
+                sessionListener, actualCanonicalHostName, threadPoolListener);
     }
 
 }

httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/AsyncRequester.java

@@ -65,7 +65,8 @@ DefaultConnectingIOReactor createIOReactor(
                 LoggingIOSessionDecorator.INSTANCE,
                 LoggingExceptionCallback.INSTANCE,
                 LoggingIOSessionListener.INSTANCE,
-                sessionShutdownCallback);
+                sessionShutdownCallback,
+                LoggingReactorMetricsListener.INSTANCE);
     }
 
     private InetSocketAddress toSocketAddress(final HttpHost host) {

httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/AsyncServer.java

@@ -60,7 +60,8 @@ DefaultListeningIOReactor createIOReactor(
                 LoggingIOSessionDecorator.INSTANCE,
                 LoggingExceptionCallback.INSTANCE,
                 LoggingIOSessionListener.INSTANCE,
-                sessionShutdownCallback);
+                sessionShutdownCallback,
+                LoggingReactorMetricsListener.INSTANCE);
     }
 
     public Future<ListenerEndpoint> listen(final InetSocketAddress address) {

httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/LoggingReactorMetricsListener.java

@@ -0,0 +1,68 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.core5.testing.nio;
+
+import org.apache.hc.core5.reactor.IOReactorMetricsListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LoggingReactorMetricsListener implements IOReactorMetricsListener {
+
+    public static final IOReactorMetricsListener INSTANCE = new LoggingReactorMetricsListener();
+
+    private final Logger logger = LoggerFactory.getLogger("org.apache.hc.core5.http.pool");
+
+    @Override
+    public void onThreadPoolStatus(final int activeThreads, final int pendingConnections) {
+        if (logger.isDebugEnabled()) {
+            logger.debug("Active threads: {}, Pending connections: {}", activeThreads, pendingConnections);
+        }
+    }
+
+    @Override
+    public void onThreadPoolSaturation(final double saturationPercentage) {
+        if (logger.isDebugEnabled()) {
+            logger.debug("Thread pool saturation: {}%", saturationPercentage);
+        }
+    }
+
+    @Override
+    public void onResourceStarvationDetected() {
+        if (logger.isDebugEnabled()) {
+            logger.debug("Resource starvation detected!");
+        }
+    }
+
+    @Override
+    public void onQueueWaitTime(final long averageWaitTimeMillis) {
+        if (logger.isDebugEnabled()) {
+            logger.debug("Average queue wait time: {} ms", averageWaitTimeMillis);
+        }
+    }
+}
+

httpcore5-testing/src/test/java/org/apache/hc/core5/testing/compatibility/http2/H2CompatibilityTest.java

@@ -68,6 +68,7 @@
 import org.apache.hc.core5.testing.nio.LoggingHttp1StreamListener;
 import org.apache.hc.core5.testing.nio.LoggingIOSessionDecorator;
 import org.apache.hc.core5.testing.nio.LoggingIOSessionListener;
+import org.apache.hc.core5.testing.nio.LoggingReactorMetricsListener;
 import org.apache.hc.core5.util.TextUtils;
 import org.apache.hc.core5.util.Timeout;
 
@@ -110,6 +111,7 @@ public static void main(final String... args) throws Exception {
                 .setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE)
                 .setExceptionCallback(LoggingExceptionCallback.INSTANCE)
                 .setIOSessionListener(LoggingIOSessionListener.INSTANCE)
+                .setIOReactorMetricsListener(LoggingReactorMetricsListener.INSTANCE)
                 .create();
     }
 

httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2AsyncRequesterResource.java

@@ -40,6 +40,7 @@
 import org.apache.hc.core5.testing.nio.LoggingHttp1StreamListener;
 import org.apache.hc.core5.testing.nio.LoggingIOSessionDecorator;
 import org.apache.hc.core5.testing.nio.LoggingIOSessionListener;
+import org.apache.hc.core5.testing.nio.LoggingReactorMetricsListener;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.extension.AfterEachCallback;
 import org.junit.jupiter.api.extension.BeforeEachCallback;
@@ -69,7 +70,8 @@ public void beforeEach(final ExtensionContext extensionContext) throws Exception
                 .setConnPoolListener(LoggingConnPoolListener.INSTANCE)
                 .setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE)
                 .setExceptionCallback(LoggingExceptionCallback.INSTANCE)
-                .setIOSessionListener(LoggingIOSessionListener.INSTANCE);
+                .setIOSessionListener(LoggingIOSessionListener.INSTANCE)
+                .setIOReactorMetricsListener(LoggingReactorMetricsListener.INSTANCE);
         bootstrapCustomizer.accept(bootstrap);
         requester = bootstrap.create();
     }

httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2AsyncServerResource.java

@@ -37,6 +37,7 @@
 import org.apache.hc.core5.testing.nio.LoggingExceptionCallback;
 import org.apache.hc.core5.testing.nio.LoggingH2StreamListener;
 import org.apache.hc.core5.testing.nio.LoggingHttp1StreamListener;
+import org.apache.hc.core5.testing.nio.LoggingReactorMetricsListener;
 import org.apache.hc.core5.testing.nio.LoggingIOSessionDecorator;
 import org.apache.hc.core5.testing.nio.LoggingIOSessionListener;
 import org.junit.jupiter.api.Assertions;
@@ -68,7 +69,8 @@ public void beforeEach(final ExtensionContext extensionContext) throws Exception
                 .setStreamListener(LoggingH2StreamListener.INSTANCE)
                 .setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE)
                 .setExceptionCallback(LoggingExceptionCallback.INSTANCE)
-                .setIOSessionListener(LoggingIOSessionListener.INSTANCE);
+                .setIOSessionListener(LoggingIOSessionListener.INSTANCE)
+                .setIOReactorMetricsListener(LoggingReactorMetricsListener.INSTANCE);
         bootstrapCustomizer.accept(bootstrap);
         server = bootstrap.create();
     }

httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2MultiplexingRequesterResource.java

@@ -38,6 +38,7 @@
 import org.apache.hc.core5.testing.nio.LoggingH2StreamListener;
 import org.apache.hc.core5.testing.nio.LoggingIOSessionDecorator;
 import org.apache.hc.core5.testing.nio.LoggingIOSessionListener;
+import org.apache.hc.core5.testing.nio.LoggingReactorMetricsListener;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.extension.AfterEachCallback;
 import org.junit.jupiter.api.extension.BeforeEachCallback;
@@ -65,7 +66,8 @@ public void beforeEach(final ExtensionContext extensionContext) throws Exception
                 .setStreamListener(LoggingH2StreamListener.INSTANCE)
                 .setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE)
                 .setExceptionCallback(LoggingExceptionCallback.INSTANCE)
-                .setIOSessionListener(LoggingIOSessionListener.INSTANCE);
+                .setIOSessionListener(LoggingIOSessionListener.INSTANCE)
+                .setIOReactorMetricsListener(LoggingReactorMetricsListener.INSTANCE);
         bootstrapCustomizer.accept(bootstrap);
         requester = bootstrap.create();
     }

httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/HttpAsyncRequesterResource.java

@@ -38,6 +38,7 @@
 import org.apache.hc.core5.testing.nio.LoggingHttp1StreamListener;
 import org.apache.hc.core5.testing.nio.LoggingIOSessionDecorator;
 import org.apache.hc.core5.testing.nio.LoggingIOSessionListener;
+import org.apache.hc.core5.testing.nio.LoggingReactorMetricsListener;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.extension.AfterEachCallback;
 import org.junit.jupiter.api.extension.BeforeEachCallback;
@@ -67,7 +68,8 @@ public void beforeEach(final ExtensionContext extensionContext) throws Exception
                 .setIOSessionListener(LoggingIOSessionListener.INSTANCE)
                 .setStreamListener(LoggingHttp1StreamListener.INSTANCE_CLIENT)
                 .setConnPoolListener(LoggingConnPoolListener.INSTANCE)
-                .setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE);
+                .setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE)
+                .setIOReactorMetricsListener(LoggingReactorMetricsListener.INSTANCE);
         bootstrapCustomizer.accept(bootstrap);
         requester = bootstrap.create();
     }

httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/HttpAsyncServerResource.java

@@ -36,6 +36,7 @@
 import org.apache.hc.core5.testing.SSLTestContexts;
 import org.apache.hc.core5.testing.nio.LoggingExceptionCallback;
 import org.apache.hc.core5.testing.nio.LoggingHttp1StreamListener;
+import org.apache.hc.core5.testing.nio.LoggingReactorMetricsListener;
 import org.apache.hc.core5.testing.nio.LoggingIOSessionDecorator;
 import org.apache.hc.core5.testing.nio.LoggingIOSessionListener;
 import org.junit.jupiter.api.Assertions;
@@ -66,7 +67,8 @@ public void beforeEach(final ExtensionContext extensionContext) throws Exception
                 .setStreamListener(LoggingHttp1StreamListener.INSTANCE_SERVER)
                 .setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE)
                 .setExceptionCallback(LoggingExceptionCallback.INSTANCE)
-                .setIOSessionListener(LoggingIOSessionListener.INSTANCE);
+                .setIOSessionListener(LoggingIOSessionListener.INSTANCE)
+                .setIOReactorMetricsListener(LoggingReactorMetricsListener.INSTANCE);
         bootstrapCustomizer.accept(bootstrap);
         server = bootstrap.create();
     }