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 2025/03/13 20:08:01 UTC
(httpcomponents-core) branch master updated: Bug fix: Fixed a race condition in async message stream handlers that can occur when a message with no content body is being submitted asynchronously
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
The following commit(s) were added to refs/heads/master by this push:
new 1741a27da Bug fix: Fixed a race condition in async message stream handlers that can occur when a message with no content body is being submitted asynchronously
1741a27da is described below
commit 1741a27da3a2c75f70bb17d2b7e7dede27beaabd
Author: Oleg Kalnichevski <ol...@apache.org>
AuthorDate: Sat Mar 8 19:35:20 2025 +0100
Bug fix: Fixed a race condition in async message stream handlers that can occur when a message with no content body is being submitted asynchronously
---
.../http2/impl/nio/ClientH2StreamHandler.java | 7 +-
.../http2/impl/nio/ServerH2StreamHandler.java | 6 +-
.../hc/core5/testing/nio/HttpIntegrationTest.java | 138 +++++++++++++++++++++
.../http/impl/nio/ClientHttp1StreamHandler.java | 2 +-
.../http/impl/nio/ServerHttp1StreamHandler.java | 5 +-
5 files changed, 150 insertions(+), 8 deletions(-)
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientH2StreamHandler.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientH2StreamHandler.java
index 665c230ab..80e6db1d7 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientH2StreamHandler.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientH2StreamHandler.java
@@ -142,12 +142,13 @@ private void commitRequest(final HttpRequest request, final EntityDetails entity
httpProcessor.process(request, entityDetails, context);
final List<Header> headers = DefaultH2RequestConverter.INSTANCE.convert(request);
- outputChannel.submit(headers, entityDetails == null);
- connMetrics.incrementRequestCount();
-
if (entityDetails == null) {
requestState.set(MessageState.COMPLETE);
+ outputChannel.submit(headers, true);
+ connMetrics.incrementRequestCount();
} else {
+ outputChannel.submit(headers, false);
+ connMetrics.incrementRequestCount();
final Header h = request.getFirstHeader(HttpHeaders.EXPECT);
final boolean expectContinue = h != null && HeaderElements.CONTINUE.equalsIgnoreCase(h.getValue());
if (expectContinue) {
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerH2StreamHandler.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerH2StreamHandler.java
index a490ca68f..efdef785e 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerH2StreamHandler.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerH2StreamHandler.java
@@ -178,11 +178,13 @@ private void commitResponse(
final boolean endStream = responseEntityDetails == null ||
receivedRequest != null && Method.HEAD.isSame(receivedRequest.getMethod());
- outputChannel.submit(responseHeaders, endStream);
- connMetrics.incrementResponseCount();
if (endStream) {
responseState.set(MessageState.COMPLETE);
+ outputChannel.submit(responseHeaders, endStream);
+ connMetrics.incrementResponseCount();
} else {
+ outputChannel.submit(responseHeaders, endStream);
+ connMetrics.incrementResponseCount();
exchangeHandler.produce(outputChannel);
if (responseState.compareAndSet(MessageState.IDLE, MessageState.BODY)) {
outputChannel.requestOutput();
diff --git a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/HttpIntegrationTest.java b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/HttpIntegrationTest.java
index 427d86e9e..5fd703cd7 100644
--- a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/HttpIntegrationTest.java
+++ b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/HttpIntegrationTest.java
@@ -81,6 +81,7 @@
import org.apache.hc.core5.http.nio.entity.StringAsyncEntityConsumer;
import org.apache.hc.core5.http.nio.entity.StringAsyncEntityProducer;
import org.apache.hc.core5.http.nio.support.AbstractServerExchangeHandler;
+import org.apache.hc.core5.http.nio.support.AsyncResponseBuilder;
import org.apache.hc.core5.http.nio.support.BasicAsyncServerExpectationDecorator;
import org.apache.hc.core5.http.nio.support.BasicRequestConsumer;
import org.apache.hc.core5.http.nio.support.BasicRequestProducer;
@@ -97,6 +98,7 @@
import org.apache.hc.core5.util.Timeout;
import org.hamcrest.CoreMatchers;
import org.hamcrest.MatcherAssert;
+import org.hamcrest.Matchers;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -1309,6 +1311,142 @@ public void releaseResources() {
}
}
+ @Test
+ void testDelayedResponseSubmission() throws Exception {
+ final HttpTestServer server = server();
+ final HttpTestClient client = client();
+
+ server.register("/hello", () -> new AbstractServerExchangeHandler<Message<HttpRequest, String>>() {
+
+ private final Random random = new Random(System.currentTimeMillis());
+
+ @Override
+ protected AsyncRequestConsumer<Message<HttpRequest, String>> supplyConsumer(
+ final HttpRequest request,
+ final EntityDetails entityDetails,
+ final HttpContext context) throws HttpException {
+ return new BasicRequestConsumer<>(entityDetails != null ? new StringAsyncEntityConsumer() : null);
+ }
+
+ @Override
+ protected void handle(
+ final Message<HttpRequest, String> requestMessage,
+ final AsyncServerRequestHandler.ResponseTrigger responseTrigger,
+ final HttpContext context) throws HttpException, IOException {
+ executorResource.getExecutorService().execute(() -> {
+ try {
+ Thread.sleep(random.nextInt(200));
+ responseTrigger.submitResponse(AsyncResponseBuilder.create(HttpStatus.SC_OK)
+ .setEntity(new MultiLineEntityProducer("All is well", 100))
+ .build(),
+ context);
+ Thread.sleep(random.nextInt(200));
+ } catch (final Exception ignore) {
+ // ignore
+ }
+ });
+
+ }
+
+ });
+ final InetSocketAddress serverEndpoint = server.start();
+
+ final HttpHost target = target(serverEndpoint);
+
+ client.start();
+
+ final Future<ClientSessionEndpoint> connectFuture = client.connect(
+ "localhost", serverEndpoint.getPort(), TIMEOUT);
+ final ClientSessionEndpoint streamEndpoint = connectFuture.get();
+
+ final Queue<Future<Message<HttpResponse, String>>> queue = new LinkedList<>();
+ for (int i = 0; i < REQ_NUM; i++) {
+ final BasicHttpRequest request = BasicRequestBuilder.get()
+ .setHttpHost(target)
+ .setPath("/hello")
+ .build();
+ queue.add(streamEndpoint.execute(
+ new BasicRequestProducer(request, null),
+ new BasicResponseConsumer<>(new StringAsyncEntityConsumer()), null));
+ }
+ while (!queue.isEmpty()) {
+ final Future<Message<HttpResponse, String>> future = queue.remove();
+ final Message<HttpResponse, String> result = future.get(LONG_TIMEOUT.getDuration(), LONG_TIMEOUT.getTimeUnit());
+ Assertions.assertNotNull(result);
+ final HttpResponse response = result.getHead();
+ Assertions.assertNotNull(response);
+ Assertions.assertEquals(200, response.getCode());
+ MatcherAssert.assertThat(result.getBody(), Matchers.startsWith("All is well"));
+ }
+ }
+
+ @Test
+ void testDelayedResponseSubmissionNoResponseBody() throws Exception {
+ final HttpTestServer server = server();
+ final HttpTestClient client = client();
+
+ server.register("/hello", () -> new AbstractServerExchangeHandler<Message<HttpRequest, String>>() {
+
+ private final Random random = new Random(System.currentTimeMillis());
+
+ @Override
+ protected AsyncRequestConsumer<Message<HttpRequest, String>> supplyConsumer(
+ final HttpRequest request,
+ final EntityDetails entityDetails,
+ final HttpContext context) throws HttpException {
+ return new BasicRequestConsumer<>(entityDetails != null ? new StringAsyncEntityConsumer() : null);
+ }
+
+ @Override
+ protected void handle(
+ final Message<HttpRequest, String> requestMessage,
+ final AsyncServerRequestHandler.ResponseTrigger responseTrigger,
+ final HttpContext context) throws HttpException, IOException {
+ executorResource.getExecutorService().execute(() -> {
+ try {
+ Thread.sleep(random.nextInt(200));
+ responseTrigger.submitResponse(AsyncResponseBuilder.create(200)
+ .build(),
+ context);
+ Thread.sleep(random.nextInt(200));
+ } catch (final Exception ignore) {
+ // ignore
+ }
+ });
+
+ }
+
+ });
+ final InetSocketAddress serverEndpoint = server.start();
+
+ final HttpHost target = target(serverEndpoint);
+
+ client.start();
+
+ final Future<ClientSessionEndpoint> connectFuture = client.connect(
+ "localhost", serverEndpoint.getPort(), TIMEOUT);
+ final ClientSessionEndpoint streamEndpoint = connectFuture.get();
+
+ final Queue<Future<Message<HttpResponse, String>>> queue = new LinkedList<>();
+ for (int i = 0; i < REQ_NUM; i++) {
+ final BasicHttpRequest request = BasicRequestBuilder.get()
+ .setHttpHost(target)
+ .setPath("/hello")
+ .build();
+ queue.add(streamEndpoint.execute(
+ new BasicRequestProducer(request, null),
+ new BasicResponseConsumer<>(new StringAsyncEntityConsumer()), null));
+ }
+ while (!queue.isEmpty()) {
+ final Future<Message<HttpResponse, String>> future = queue.remove();
+ final Message<HttpResponse, String> result = future.get(LONG_TIMEOUT.getDuration(), LONG_TIMEOUT.getTimeUnit());
+ Assertions.assertNotNull(result);
+ final HttpResponse response = result.getHead();
+ Assertions.assertNotNull(response);
+ Assertions.assertEquals(200, response.getCode());
+ }
+ }
+
void testHeaderTooLarge(final String method) throws Exception {
final HttpTestServer server = server();
final HttpTestClient client = client();
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/nio/ClientHttp1StreamHandler.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/nio/ClientHttp1StreamHandler.java
index afd9bf2fe..b54c603db 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/nio/ClientHttp1StreamHandler.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/nio/ClientHttp1StreamHandler.java
@@ -166,9 +166,9 @@ private void commitRequest(final HttpRequest request, final EntityDetails entity
final boolean endStream = entityDetails == null;
if (endStream) {
- outputChannel.submit(request, true, FlushMode.IMMEDIATE);
committedRequest = request;
requestState.set(MessageState.COMPLETE);
+ outputChannel.submit(request, true, FlushMode.IMMEDIATE);
} else {
final Header h = request.getFirstHeader(HttpHeaders.EXPECT);
final boolean expectContinue = h != null && HeaderElements.CONTINUE.equalsIgnoreCase(h.getValue());
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/nio/ServerHttp1StreamHandler.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/nio/ServerHttp1StreamHandler.java
index 9a73652cd..5e2761873 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/nio/ServerHttp1StreamHandler.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/nio/ServerHttp1StreamHandler.java
@@ -189,13 +189,14 @@ private void commitResponse(
keepAlive = false;
}
- outputChannel.submit(response, endStream, endStream ? FlushMode.IMMEDIATE : FlushMode.BUFFER);
if (endStream) {
+ responseState.set(MessageState.COMPLETE);
+ outputChannel.submit(response, true, FlushMode.IMMEDIATE);
if (!keepAlive) {
outputChannel.close();
}
- responseState.set(MessageState.COMPLETE);
} else {
+ outputChannel.submit(response, false, FlushMode.BUFFER);
exchangeHandler.produce(internalDataChannel);
if (responseState.compareAndSet(MessageState.IDLE, MessageState.BODY)) {
outputChannel.requestOutput();