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:36 UTC
(httpcomponents-core) branch master updated (0a6966806 -> b633366be)
This is an automated email from the ASF dual-hosted git repository.
olegk pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/httpcomponents-core.git
discard 0a6966806 HTTPCLIENT-2233- Add metrics listener for IOReactor thread pool monitoring (#490)
new b633366be HTTPCLIENT-2233- Add metrics listener for IOReactor thread pool monitoring (#490)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (0a6966806)
\
N -- N -- N refs/heads/master (b633366be)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../core5/http2/impl/nio/bootstrap/H2ServerBootstrap.java | 2 +-
.../org/apache/hc/core5/testing/nio/AsyncRequester.java | 4 ++--
.../java/org/apache/hc/core5/testing/nio/AsyncServer.java | 4 ++--
.../apache/hc/core5/http/impl/bootstrap/AsyncRequester.java | 4 ++--
.../apache/hc/core5/http/impl/bootstrap/AsyncServer.java | 8 ++++----
.../hc/core5/http/impl/bootstrap/HttpAsyncServer.java | 8 ++++----
.../apache/hc/core5/reactor/DefaultConnectingIOReactor.java | 13 ++++++++-----
.../apache/hc/core5/reactor/DefaultListeningIOReactor.java | 13 ++++++++-----
.../org/apache/hc/core5/reactor/SingleCoreIOReactor.java | 6 +++---
9 files changed, 34 insertions(+), 28 deletions(-)
(httpcomponents-core) 01/01: HTTPCLIENT-2233- Add metrics listener for IOReactor thread pool monitoring (#490)
Posted by ol...@apache.org.
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();