You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hc.apache.org by ol...@apache.org on 2024/10/03 18:20:37 UTC
(httpcomponents-core) 01/01: HTTPCLIENT-2233- Add metrics listener for IOReactor thread pool monitoring (#490)
This is an automated email from the ASF dual-hosted git repository.
olegk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/httpcomponents-core.git
commit b633366bec57e113bd1e9679a8f9d0535ea9e3c1
Author: Arturo Bernal <ab...@apache.org>
AuthorDate: Thu Oct 3 19:20:40 2024 +0200
HTTPCLIENT-2233- Add metrics listener for IOReactor thread pool monitoring (#490)
This change enhances monitoring capabilities for the IOReactor thread pool, providing better insights into performance and potential bottlenecks.
---
.../http2/impl/nio/bootstrap/H2AsyncRequester.java | 11 ++--
.../nio/bootstrap/H2MultiplexingRequester.java | 6 +-
.../H2MultiplexingRequesterBootstrap.java | 17 +++++-
.../impl/nio/bootstrap/H2RequesterBootstrap.java | 17 +++++-
.../impl/nio/bootstrap/H2ServerBootstrap.java | 17 +++++-
.../hc/core5/testing/nio/AsyncRequester.java | 1 +
.../apache/hc/core5/testing/nio/AsyncServer.java | 1 +
.../testing/nio/LoggingReactorMetricsListener.java | 68 +++++++++++++++++++++
.../compatibility/http2/H2CompatibilityTest.java | 2 +
.../extension/nio/H2AsyncRequesterResource.java | 4 +-
.../extension/nio/H2AsyncServerResource.java | 4 +-
.../nio/H2MultiplexingRequesterResource.java | 4 +-
.../extension/nio/HttpAsyncRequesterResource.java | 4 +-
.../extension/nio/HttpAsyncServerResource.java | 4 +-
.../hc/core5/testing/nio/TLSIntegrationTest.java | 2 +
.../core5/http/impl/bootstrap/AsyncRequester.java | 6 +-
.../impl/bootstrap/AsyncRequesterBootstrap.java | 16 ++++-
.../hc/core5/http/impl/bootstrap/AsyncServer.java | 3 +
.../http/impl/bootstrap/AsyncServerBootstrap.java | 15 ++++-
.../http/impl/bootstrap/HttpAsyncRequester.java | 11 ++--
.../core5/http/impl/bootstrap/HttpAsyncServer.java | 9 ++-
.../core5/reactor/DefaultConnectingIOReactor.java | 16 +++++
.../core5/reactor/DefaultListeningIOReactor.java | 17 ++++++
.../hc/core5/reactor/IOReactorMetricsListener.java | 71 ++++++++++++++++++++++
.../apache/hc/core5/reactor/IOSessionRequest.java | 14 +++++
.../hc/core5/reactor/SingleCoreIOReactor.java | 71 ++++++++++++++++++++++
.../org/apache/hc/core5/reactor/IOWorkersTest.java | 2 +-
27 files changed, 387 insertions(+), 26 deletions(-)
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2AsyncRequester.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2AsyncRequester.java
index e88a11f89..9fe585028 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2AsyncRequester.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2AsyncRequester.java
@@ -44,6 +44,7 @@ import org.apache.hc.core5.net.NamedEndpoint;
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 class H2AsyncRequester extends HttpAsyncRequester {
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 class H2AsyncRequester extends HttpAsyncRequester {
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;
}
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java
index 90103e65b..22cf0ecac 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java
@@ -70,6 +70,7 @@ import org.apache.hc.core5.net.URIAuthority;
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 class H2MultiplexingRequester extends AsyncRequester {
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);
}
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequesterBootstrap.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequesterBootstrap.java
index 207bea198..c1f40eb4a 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequesterBootstrap.java
+++ b/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.impl.nio.H2StreamListener;
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 class H2MultiplexingRequesterBootstrap {
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 class H2MultiplexingRequesterBootstrap {
exceptionCallback,
sessionListener,
DefaultAddressResolver.INSTANCE,
- tlsStrategy != null ? tlsStrategy : new H2ClientTlsStrategy());
+ tlsStrategy != null ? tlsStrategy : new H2ClientTlsStrategy(),
+ threadPoolListener);
}
}
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2RequesterBootstrap.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2RequesterBootstrap.java
index c0ea93a0d..d0b0bd496 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2RequesterBootstrap.java
+++ b/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.PoolReusePolicy;
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 class H2RequesterBootstrap {
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 class H2RequesterBootstrap {
sessionListener,
connPool,
actualTlsStrategy,
- handshakeTimeout);
+ handshakeTimeout,
+ threadPoolListener);
}
}
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2ServerBootstrap.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2ServerBootstrap.java
index 822cbe7f4..b54c16564 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2ServerBootstrap.java
+++ b/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.InetAddressUtils;
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 class H2ServerBootstrap {
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 class H2ServerBootstrap {
this.http1StreamListener = http1StreamListener;
return this;
}
-
/**
* @return this instance.
* @deprecated Use {@link RequestRouter}.
@@ -522,7 +535,7 @@ public class H2ServerBootstrap {
handshakeTimeout);
return new HttpAsyncServer(ioEventHandlerFactory, ioReactorConfig, ioSessionDecorator, exceptionCallback,
- sessionListener, actualCanonicalHostName);
+ sessionListener, threadPoolListener, actualCanonicalHostName);
}
}
diff --git a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/AsyncRequester.java b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/AsyncRequester.java
index 3c229034d..7d06a0899 100644
--- a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/AsyncRequester.java
+++ b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/AsyncRequester.java
@@ -65,6 +65,7 @@ public class AsyncRequester extends IOReactorExecutor<DefaultConnectingIOReactor
LoggingIOSessionDecorator.INSTANCE,
LoggingExceptionCallback.INSTANCE,
LoggingIOSessionListener.INSTANCE,
+ LoggingReactorMetricsListener.INSTANCE,
sessionShutdownCallback);
}
diff --git a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/AsyncServer.java b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/AsyncServer.java
index 562d33e8f..3149de2eb 100644
--- a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/AsyncServer.java
+++ b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/AsyncServer.java
@@ -60,6 +60,7 @@ public class AsyncServer extends IOReactorExecutor<DefaultListeningIOReactor> {
LoggingIOSessionDecorator.INSTANCE,
LoggingExceptionCallback.INSTANCE,
LoggingIOSessionListener.INSTANCE,
+ LoggingReactorMetricsListener.INSTANCE,
sessionShutdownCallback);
}
diff --git a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/LoggingReactorMetricsListener.java b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/LoggingReactorMetricsListener.java
new file mode 100644
index 000000000..2202afe9a
--- /dev/null
+++ b/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);
+ }
+ }
+}
+
diff --git a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/compatibility/http2/H2CompatibilityTest.java b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/compatibility/http2/H2CompatibilityTest.java
index 41cadedc9..915b4c4b3 100644
--- a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/compatibility/http2/H2CompatibilityTest.java
+++ b/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.LoggingH2StreamListener;
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 class H2CompatibilityTest {
.setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE)
.setExceptionCallback(LoggingExceptionCallback.INSTANCE)
.setIOSessionListener(LoggingIOSessionListener.INSTANCE)
+ .setIOReactorMetricsListener(LoggingReactorMetricsListener.INSTANCE)
.create();
}
diff --git a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2AsyncRequesterResource.java b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2AsyncRequesterResource.java
index 6ed7cfc9f..c343de763 100644
--- a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2AsyncRequesterResource.java
+++ b/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.LoggingH2StreamListener;
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 class H2AsyncRequesterResource implements BeforeEachCallback, AfterEachCa
.setConnPoolListener(LoggingConnPoolListener.INSTANCE)
.setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE)
.setExceptionCallback(LoggingExceptionCallback.INSTANCE)
- .setIOSessionListener(LoggingIOSessionListener.INSTANCE);
+ .setIOSessionListener(LoggingIOSessionListener.INSTANCE)
+ .setIOReactorMetricsListener(LoggingReactorMetricsListener.INSTANCE);
bootstrapCustomizer.accept(bootstrap);
requester = bootstrap.create();
}
diff --git a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2AsyncServerResource.java b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2AsyncServerResource.java
index 188eac814..91e4ae56e 100644
--- a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2AsyncServerResource.java
+++ b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2AsyncServerResource.java
@@ -37,6 +37,7 @@ import org.apache.hc.core5.testing.SSLTestContexts;
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 class H2AsyncServerResource implements BeforeEachCallback, AfterEachCallb
.setStreamListener(LoggingH2StreamListener.INSTANCE)
.setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE)
.setExceptionCallback(LoggingExceptionCallback.INSTANCE)
- .setIOSessionListener(LoggingIOSessionListener.INSTANCE);
+ .setIOSessionListener(LoggingIOSessionListener.INSTANCE)
+ .setIOReactorMetricsListener(LoggingReactorMetricsListener.INSTANCE);
bootstrapCustomizer.accept(bootstrap);
server = bootstrap.create();
}
diff --git a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2MultiplexingRequesterResource.java b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2MultiplexingRequesterResource.java
index 1739ef893..144d80571 100644
--- a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/H2MultiplexingRequesterResource.java
+++ b/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.LoggingExceptionCallback;
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 class H2MultiplexingRequesterResource implements BeforeEachCallback, Afte
.setStreamListener(LoggingH2StreamListener.INSTANCE)
.setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE)
.setExceptionCallback(LoggingExceptionCallback.INSTANCE)
- .setIOSessionListener(LoggingIOSessionListener.INSTANCE);
+ .setIOSessionListener(LoggingIOSessionListener.INSTANCE)
+ .setIOReactorMetricsListener(LoggingReactorMetricsListener.INSTANCE);
bootstrapCustomizer.accept(bootstrap);
requester = bootstrap.create();
}
diff --git a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/HttpAsyncRequesterResource.java b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/HttpAsyncRequesterResource.java
index 18a5ac67f..9eadba556 100644
--- a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/HttpAsyncRequesterResource.java
+++ b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/HttpAsyncRequesterResource.java
@@ -38,6 +38,7 @@ import org.apache.hc.core5.testing.classic.LoggingConnPoolListener;
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 class HttpAsyncRequesterResource implements BeforeEachCallback, AfterEach
.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();
}
diff --git a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/HttpAsyncServerResource.java b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/HttpAsyncServerResource.java
index d7ee307e4..a522d7e76 100644
--- a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/HttpAsyncServerResource.java
+++ b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/extension/nio/HttpAsyncServerResource.java
@@ -36,6 +36,7 @@ import org.apache.hc.core5.io.CloseMode;
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 class HttpAsyncServerResource implements BeforeEachCallback, AfterEachCal
.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();
}
diff --git a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/TLSIntegrationTest.java b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/TLSIntegrationTest.java
index d1d54687c..e84581832 100644
--- a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/TLSIntegrationTest.java
+++ b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/TLSIntegrationTest.java
@@ -147,6 +147,7 @@ class TLSIntegrationTest {
.setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE)
.setExceptionCallback(LoggingExceptionCallback.INSTANCE)
.setIOSessionListener(LoggingIOSessionListener.INSTANCE)
+ .setIOReactorMetricsListener(LoggingReactorMetricsListener.INSTANCE)
.setRequestRouter(RequestRouter.<Supplier<AsyncServerExchangeHandler>>builder()
.addRoute(RequestRouter.LOCAL_AUTHORITY, "*", () -> new EchoHandler(2048))
.resolveAuthority(RequestRouter.LOCAL_AUTHORITY_RESOLVER)
@@ -165,6 +166,7 @@ class TLSIntegrationTest {
.setIOSessionDecorator(LoggingIOSessionDecorator.INSTANCE)
.setExceptionCallback(LoggingExceptionCallback.INSTANCE)
.setIOSessionListener(LoggingIOSessionListener.INSTANCE)
+ .setIOReactorMetricsListener(LoggingReactorMetricsListener.INSTANCE)
.create();
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncRequester.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncRequester.java
index 2e91765c7..399ca923c 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncRequester.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncRequester.java
@@ -49,6 +49,7 @@ import org.apache.hc.core5.reactor.IOEventHandlerFactory;
import org.apache.hc.core5.reactor.IOReactorConfig;
import org.apache.hc.core5.reactor.IOReactorService;
import org.apache.hc.core5.reactor.IOReactorStatus;
+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;
@@ -73,7 +74,9 @@ public class AsyncRequester extends AbstractConnectionInitiatorBase implements I
final Callback<Exception> exceptionCallback,
final IOSessionListener sessionListener,
final Callback<IOSession> sessionShutdownCallback,
- final Resolver<HttpHost, InetSocketAddress> addressResolver) {
+ final Resolver<HttpHost, InetSocketAddress> addressResolver,
+ final IOReactorMetricsListener threadPoolListener
+ ) {
this.ioReactor = new DefaultConnectingIOReactor(
eventHandlerFactory,
ioReactorConfig,
@@ -81,6 +84,7 @@ public class AsyncRequester extends AbstractConnectionInitiatorBase implements I
ioSessionDecorator,
exceptionCallback,
sessionListener,
+ threadPoolListener,
sessionShutdownCallback);
this.addressResolver = addressResolver != null ? addressResolver : DefaultAddressResolver.INSTANCE;
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncRequesterBootstrap.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncRequesterBootstrap.java
index 12611bf5a..2c16641a4 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncRequesterBootstrap.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncRequesterBootstrap.java
@@ -49,6 +49,7 @@ import org.apache.hc.core5.pool.PoolReusePolicy;
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.Timeout;
@@ -77,6 +78,7 @@ public class AsyncRequesterBootstrap {
private IOSessionListener sessionListener;
private Http1StreamListener streamListener;
private ConnPoolListener<HttpHost> connPoolListener;
+ private IOReactorMetricsListener threadPoolListener;
private AsyncRequesterBootstrap() {
}
@@ -216,6 +218,17 @@ public class AsyncRequesterBootstrap {
return this;
}
+ /**
+ * Sets {@link IOReactorMetricsListener} instance.
+ *
+ * @return this instance.
+ * @since 5.4
+ */
+ public final AsyncRequesterBootstrap setIOReactorMetricsListener(final IOReactorMetricsListener threadPoolListener) {
+ this.threadPoolListener = threadPoolListener;
+ return this;
+ }
+
/**
* Sets {@link Http1StreamListener} instance.
*
@@ -279,7 +292,8 @@ public class AsyncRequesterBootstrap {
sessionListener,
connPool,
tlsStrategyCopy,
- handshakeTimeout);
+ handshakeTimeout,
+ threadPoolListener);
}
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncServer.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncServer.java
index 97b241d4c..f4c34b45f 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncServer.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncServer.java
@@ -45,6 +45,7 @@ import org.apache.hc.core5.reactor.IOEventHandlerFactory;
import org.apache.hc.core5.reactor.IOReactorConfig;
import org.apache.hc.core5.reactor.IOReactorService;
import org.apache.hc.core5.reactor.IOReactorStatus;
+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.ListenerEndpoint;
@@ -64,6 +65,7 @@ public class AsyncServer extends AbstractConnectionInitiatorBase implements IORe
final Decorator<IOSession> ioSessionDecorator,
final Callback<Exception> exceptionCallback,
final IOSessionListener sessionListener,
+ final IOReactorMetricsListener threadPoolListener,
final Callback<IOSession> sessionShutdownCallback) {
this.ioReactor = new DefaultListeningIOReactor(
eventHandlerFactory,
@@ -73,6 +75,7 @@ public class AsyncServer extends AbstractConnectionInitiatorBase implements IORe
ioSessionDecorator,
exceptionCallback,
sessionListener,
+ threadPoolListener,
sessionShutdownCallback);
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncServerBootstrap.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncServerBootstrap.java
index fae3749eb..10b821c59 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncServerBootstrap.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncServerBootstrap.java
@@ -64,6 +64,7 @@ import org.apache.hc.core5.net.InetAddressUtils;
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;
@@ -93,6 +94,7 @@ public class AsyncServerBootstrap {
private Callback<Exception> exceptionCallback;
private IOSessionListener sessionListener;
private Http1StreamListener streamListener;
+ private IOReactorMetricsListener threadPoolListener;
private AsyncServerBootstrap() {
this.routeEntries = new ArrayList<>();
@@ -207,6 +209,17 @@ public class AsyncServerBootstrap {
return this;
}
+ /**
+ * Sets {@link IOReactorMetricsListener} instance.
+ *
+ * @return this instance.
+ * @since 5.4
+ */
+ public final AsyncServerBootstrap setIOReactorMetricsListener(final IOReactorMetricsListener threadPoolListener) {
+ this.threadPoolListener = threadPoolListener;
+ return this;
+ }
+
/**
* Sets {@link Exception} {@link Callback} instance.
*
@@ -504,7 +517,7 @@ public class AsyncServerBootstrap {
tlsStrategy,
handshakeTimeout);
return new HttpAsyncServer(ioEventHandlerFactory, ioReactorConfig, ioSessionDecorator, exceptionCallback,
- sessionListener);
+ sessionListener, threadPoolListener);
}
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java
index 059ee8f44..b8dd4f28f 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java
@@ -78,6 +78,7 @@ import org.apache.hc.core5.reactor.EndpointParameters;
import org.apache.hc.core5.reactor.IOEventHandler;
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;
@@ -111,9 +112,10 @@ public class HttpAsyncRequester extends AsyncRequester implements ConnPoolContro
final IOSessionListener sessionListener,
final ManagedConnPool<HttpHost, IOSession> connPool,
final TlsStrategy tlsStrategy,
- final Timeout handshakeTimeout) {
+ final Timeout handshakeTimeout,
+ final IOReactorMetricsListener threadPoolListener) {
super(eventHandlerFactory, ioReactorConfig, ioSessionDecorator, exceptionCallback, sessionListener,
- ShutdownCommand.GRACEFUL_IMMEDIATE_CALLBACK, DefaultAddressResolver.INSTANCE);
+ ShutdownCommand.GRACEFUL_IMMEDIATE_CALLBACK, DefaultAddressResolver.INSTANCE, threadPoolListener);
this.connPool = Args.notNull(connPool, "Connection pool");
this.tlsStrategy = tlsStrategy;
this.handshakeTimeout = handshakeTimeout;
@@ -129,9 +131,10 @@ public class HttpAsyncRequester extends AsyncRequester implements ConnPoolContro
final Decorator<IOSession> ioSessionDecorator,
final Callback<Exception> exceptionCallback,
final IOSessionListener sessionListener,
- final ManagedConnPool<HttpHost, IOSession> connPool) {
+ final ManagedConnPool<HttpHost, IOSession> connPool,
+ final IOReactorMetricsListener threadPoolListener) {
this(ioReactorConfig, eventHandlerFactory, ioSessionDecorator, exceptionCallback, sessionListener, connPool,
- null, null);
+ null, null, threadPoolListener);
}
@Override
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncServer.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncServer.java
index 32dde4c33..40b19bb65 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncServer.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncServer.java
@@ -39,6 +39,7 @@ import org.apache.hc.core5.reactor.EndpointParameters;
import org.apache.hc.core5.http.nio.command.ShutdownCommand;
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.ListenerEndpoint;
@@ -64,9 +65,10 @@ public class HttpAsyncServer extends AsyncServer {
final Decorator<IOSession> ioSessionDecorator,
final Callback<Exception> exceptionCallback,
final IOSessionListener sessionListener,
+ final IOReactorMetricsListener threadPoolListener,
final String canonicalName) {
super(eventHandlerFactory, ioReactorConfig, ioSessionDecorator, exceptionCallback, sessionListener,
- ShutdownCommand.GRACEFUL_NORMAL_CALLBACK);
+ threadPoolListener, ShutdownCommand.GRACEFUL_NORMAL_CALLBACK);
this.canonicalName = canonicalName;
}
@@ -79,8 +81,9 @@ public class HttpAsyncServer extends AsyncServer {
final IOReactorConfig ioReactorConfig,
final Decorator<IOSession> ioSessionDecorator,
final Callback<Exception> exceptionCallback,
- final IOSessionListener sessionListener) {
- this(eventHandlerFactory, ioReactorConfig, ioSessionDecorator, exceptionCallback, sessionListener, null);
+ final IOSessionListener sessionListener,
+ final IOReactorMetricsListener threadPoolListener) {
+ this(eventHandlerFactory, ioReactorConfig, ioSessionDecorator, exceptionCallback, sessionListener, threadPoolListener, null);
}
/**
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultConnectingIOReactor.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultConnectingIOReactor.java
index bcb1a30fd..4d89dc200 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultConnectingIOReactor.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultConnectingIOReactor.java
@@ -55,6 +55,9 @@ public class DefaultConnectingIOReactor extends AbstractIOReactorBase {
private final static ThreadFactory THREAD_FACTORY = new DefaultThreadFactory("I/O client dispatch", true);
+ /**
+ * @since 5.4
+ */
public DefaultConnectingIOReactor(
final IOEventHandlerFactory eventHandlerFactory,
final IOReactorConfig ioReactorConfig,
@@ -62,6 +65,7 @@ public class DefaultConnectingIOReactor extends AbstractIOReactorBase {
final Decorator<IOSession> ioSessionDecorator,
final Callback<Exception> exceptionCallback,
final IOSessionListener sessionListener,
+ final IOReactorMetricsListener threadPoolListener,
final Callback<IOSession> sessionShutdownCallback) {
Args.notNull(eventHandlerFactory, "Event handler factory");
this.workerCount = ioReactorConfig != null ? ioReactorConfig.getIoThreadCount() : IOReactorConfig.DEFAULT.getIoThreadCount();
@@ -74,6 +78,7 @@ public class DefaultConnectingIOReactor extends AbstractIOReactorBase {
ioReactorConfig != null ? ioReactorConfig : IOReactorConfig.DEFAULT,
ioSessionDecorator,
sessionListener,
+ threadPoolListener,
sessionShutdownCallback);
this.workers[i] = dispatcher;
threads[i] = (threadFactory != null ? threadFactory : THREAD_FACTORY).newThread(new IOReactorWorker(dispatcher));
@@ -82,6 +87,17 @@ public class DefaultConnectingIOReactor extends AbstractIOReactorBase {
this.workerSelector = IOWorkers.newSelector(workers);
}
+ public DefaultConnectingIOReactor(
+ final IOEventHandlerFactory eventHandlerFactory,
+ final IOReactorConfig ioReactorConfig,
+ final ThreadFactory threadFactory,
+ final Decorator<IOSession> ioSessionDecorator,
+ final Callback<Exception> exceptionCallback,
+ final IOSessionListener sessionListener,
+ final Callback<IOSession> sessionShutdownCallback) {
+ this(eventHandlerFactory,ioReactorConfig, threadFactory,ioSessionDecorator, exceptionCallback, sessionListener, null, sessionShutdownCallback);
+ }
+
public DefaultConnectingIOReactor(
final IOEventHandlerFactory eventHandlerFactory,
final IOReactorConfig config,
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultListeningIOReactor.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultListeningIOReactor.java
index fe9677511..feb6b662a 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultListeningIOReactor.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/DefaultListeningIOReactor.java
@@ -80,6 +80,22 @@ public class DefaultListeningIOReactor extends AbstractIOReactorBase implements
final Callback<Exception> exceptionCallback,
final IOSessionListener sessionListener,
final Callback<IOSession> sessionShutdownCallback) {
+ this(eventHandlerFactory, ioReactorConfig, dispatchThreadFactory, listenerThreadFactory, ioSessionDecorator, exceptionCallback, sessionListener, null, sessionShutdownCallback);
+ }
+
+ /**
+ * @since 5.4
+ */
+ public DefaultListeningIOReactor(
+ final IOEventHandlerFactory eventHandlerFactory,
+ final IOReactorConfig ioReactorConfig,
+ final ThreadFactory dispatchThreadFactory,
+ final ThreadFactory listenerThreadFactory,
+ final Decorator<IOSession> ioSessionDecorator,
+ final Callback<Exception> exceptionCallback,
+ final IOSessionListener sessionListener,
+ final IOReactorMetricsListener threadPoolListener,
+ final Callback<IOSession> sessionShutdownCallback) {
Args.notNull(eventHandlerFactory, "Event handler factory");
this.workerCount = ioReactorConfig != null ? ioReactorConfig.getIoThreadCount() : IOReactorConfig.DEFAULT.getIoThreadCount();
this.workers = new SingleCoreIOReactor[workerCount];
@@ -91,6 +107,7 @@ public class DefaultListeningIOReactor extends AbstractIOReactorBase implements
ioReactorConfig != null ? ioReactorConfig : IOReactorConfig.DEFAULT,
ioSessionDecorator,
sessionListener,
+ threadPoolListener,
sessionShutdownCallback);
this.workers[i] = dispatcher;
threads[i + 1] = (dispatchThreadFactory != null ? dispatchThreadFactory : DISPATCH_THREAD_FACTORY).newThread(new IOReactorWorker(dispatcher));
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOReactorMetricsListener.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOReactorMetricsListener.java
new file mode 100644
index 000000000..a94e67577
--- /dev/null
+++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOReactorMetricsListener.java
@@ -0,0 +1,71 @@
+/*
+ * ====================================================================
+ * 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.reactor;
+
+/**
+ * A listener interface for receiving metrics related to the I/O reactor's
+ * thread pool performance and status.
+ *
+ * <p>The implementing class can monitor and act upon important metrics such as
+ * active threads, saturation levels, and resource starvation.</p>
+ *
+ * @since 5.4
+ */
+public interface IOReactorMetricsListener {
+
+ /**
+ * Invoked to report the current status of the thread pool, including
+ * active threads and pending connections.
+ *
+ * @param activeThreads The number of active threads handling connections.
+ * @param pendingConnections The number of pending connection requests in the queue.
+ */
+ void onThreadPoolStatus(int activeThreads, int pendingConnections);
+
+ /**
+ * Invoked to report the saturation level of the thread pool as a percentage
+ * of the active threads to the maximum allowed connections.
+ *
+ * @param saturationPercentage The percentage indicating thread pool saturation.
+ */
+ void onThreadPoolSaturation(double saturationPercentage);
+
+ /**
+ * Invoked when the number of pending connection requests exceeds the
+ * maximum allowed connections, indicating possible resource starvation.
+ */
+ void onResourceStarvationDetected();
+
+ /**
+ * Notifies about the average wait time for connection requests in the queue.
+ *
+ * @param averageWaitTimeMillis average time in milliseconds that connection requests spend in the queue.
+ */
+ void onQueueWaitTime(long averageWaitTimeMillis);
+}
+
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionRequest.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionRequest.java
index c808e0ac8..57f727194 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionRequest.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionRequest.java
@@ -34,6 +34,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hc.core5.annotation.Internal;
import org.apache.hc.core5.concurrent.BasicFuture;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.io.CloseMode;
@@ -50,6 +51,9 @@ final class IOSessionRequest implements Future<IOSession> {
final Object attachment;
final BasicFuture<IOSession> future;
+ private final long enqueueTime;
+
+
private final AtomicReference<ModalCloseable> closeableRef;
public IOSessionRequest(
@@ -67,6 +71,9 @@ final class IOSessionRequest implements Future<IOSession> {
this.attachment = attachment;
this.future = new BasicFuture<>(callback);
this.closeableRef = new AtomicReference<>();
+
+ // Set the time when this request is created
+ this.enqueueTime = System.currentTimeMillis();
}
public void completed(final ProtocolIOSession ioSession) {
@@ -127,4 +134,11 @@ final class IOSessionRequest implements Future<IOSession> {
']';
}
+ // Getter for enqueueTime
+ @Internal
+ public long getEnqueueTime() {
+ return enqueueTime;
+ }
+
+
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java
index 8661d67f2..2d4f4f903 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java
@@ -42,6 +42,8 @@ import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.function.Callback;
@@ -68,6 +70,11 @@ class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements Connect
private final AtomicBoolean shutdownInitiated;
private final long selectTimeoutMillis;
private volatile long lastTimeoutCheckMillis;
+ private final IOReactorMetricsListener threadPoolListener;
+
+ // Atomic variables for tracking total wait time and count of processed requests
+ private final AtomicLong totalWaitTime = new AtomicLong(0);
+ private final AtomicInteger processedRequestCount = new AtomicInteger(0);
SingleCoreIOReactor(
final Callback<Exception> exceptionCallback,
@@ -75,12 +82,14 @@ class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements Connect
final IOReactorConfig reactorConfig,
final Decorator<IOSession> ioSessionDecorator,
final IOSessionListener sessionListener,
+ final IOReactorMetricsListener threadPoolListener,
final Callback<IOSession> sessionShutdownCallback) {
super(exceptionCallback);
this.eventHandlerFactory = Args.notNull(eventHandlerFactory, "Event handler factory");
this.reactorConfig = Args.notNull(reactorConfig, "I/O reactor config");
this.ioSessionDecorator = ioSessionDecorator;
this.sessionListener = sessionListener;
+ this.threadPoolListener = threadPoolListener;
this.sessionShutdownCallback = sessionShutdownCallback;
this.shutdownInitiated = new AtomicBoolean();
this.closedSessions = new ConcurrentLinkedQueue<>();
@@ -136,6 +145,8 @@ class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements Connect
processPendingConnectionRequests();
}
+ reportStatusToThreadPoolListener();
+
// Exit select loop if graceful shutdown has been completed
if (getStatus() == IOReactorStatus.SHUTTING_DOWN && this.selector.keys().isEmpty()) {
break;
@@ -315,6 +326,15 @@ class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements Connect
private void processPendingConnectionRequests() {
IOSessionRequest sessionRequest;
for (int i = 0; i < MAX_CHANNEL_REQUESTS && (sessionRequest = this.requestQueue.poll()) != null; i++) {
+ if (threadPoolListener != null) {
+ // Calculate wait time safely without keeping long-lived state
+ final long waitTimeMillis = System.currentTimeMillis() - sessionRequest.getEnqueueTime();
+
+ // Accumulate total wait time and increment count atomically
+ totalWaitTime.addAndGet(waitTimeMillis);
+ processedRequestCount.incrementAndGet();
+ threadPoolListener.onQueueWaitTime(waitTimeMillis);
+ }
if (!sessionRequest.isCancelled()) {
final SocketChannel socketChannel;
try {
@@ -394,4 +414,55 @@ class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements Connect
}
}
+ /**
+ * Reports the current status of the I/O reactor's thread pool to the
+ * configured metrics listener.
+ *
+ * <p>This method gathers three key metrics:</p>
+ * <ul>
+ * <li><strong>Active Threads:</strong> The number of currently active threads
+ * handling I/O sessions.</li>
+ * <li><strong>Pending Connections:</strong> The number of connection requests
+ * waiting to be processed.</li>
+ * <li><strong>Saturation Percentage:</strong> The ratio of active threads to the
+ * maximum allowed connections (defined by {@code MAX_CHANNEL_REQUESTS}),
+ * expressed as a percentage. It provides insight into how saturated the thread
+ * pool is relative to its maximum capacity. The formula for calculating saturation
+ * is:
+ * <pre>
+ * saturationPercentage = (activeThreads / MAX_CHANNEL_REQUESTS) * 100.0
+ * </pre></li>
+ * </ul>
+ * <p>
+ * If the number of pending connections exceeds {@code MAX_CHANNEL_REQUESTS},
+ * resource starvation is detected, and an appropriate event is reported.
+ *
+ */
+ private void reportStatusToThreadPoolListener() {
+ if (threadPoolListener != null) {
+
+ // Calculate the number of active threads (connections)
+ final int activeThreads = (int) this.selector.keys().stream()
+ .filter(key -> key.isValid() && key.attachment() instanceof InternalChannel)
+ .count();
+
+ // Calculate the number of pending connection requests
+ final int pendingConnections = this.requestQueue.size();
+
+ // Calculate saturation as a percentage of active connections to max allowed connections
+ final double saturationPercentage = ((double) activeThreads / MAX_CHANNEL_REQUESTS) * 100.0;
+
+ // Report thread pool status: active sessions and pending connections
+ threadPoolListener.onThreadPoolStatus(activeThreads, pendingConnections);
+
+ // Report thread pool saturation
+ threadPoolListener.onThreadPoolSaturation(saturationPercentage);
+
+ // Detect resource starvation if pending connections exceed threshold
+ if (pendingConnections > MAX_CHANNEL_REQUESTS) {
+ threadPoolListener.onResourceStarvationDetected();
+ }
+ }
+ }
+
}
diff --git a/httpcore5/src/test/java/org/apache/hc/core5/reactor/IOWorkersTest.java b/httpcore5/src/test/java/org/apache/hc/core5/reactor/IOWorkersTest.java
index 1de15a4b0..c087af432 100644
--- a/httpcore5/src/test/java/org/apache/hc/core5/reactor/IOWorkersTest.java
+++ b/httpcore5/src/test/java/org/apache/hc/core5/reactor/IOWorkersTest.java
@@ -34,7 +34,7 @@ class IOWorkersTest {
@Test
void testIndexOverflow() {
- final SingleCoreIOReactor reactor = new SingleCoreIOReactor(null, mock(IOEventHandlerFactory.class), IOReactorConfig.DEFAULT, null, null, null);
+ final SingleCoreIOReactor reactor = new SingleCoreIOReactor(null, mock(IOEventHandlerFactory.class), IOReactorConfig.DEFAULT, null, null, null, null);
final IOWorkers.Selector selector = IOWorkers.newSelector(new SingleCoreIOReactor[]{reactor, reactor, reactor});
for (long i = Integer.MAX_VALUE - 10; i < (long) Integer.MAX_VALUE + 10; i++) {
selector.next();