Bug fix: Async message stream handlers to avoid firing data produce events from multiple threads

ok2c · ok2c · commit 1bf8e652549b · 2025-03-08T17:51:13.000+01:00
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
@@ -153,8 +153,10 @@ private void commitRequest(final HttpRequest request, final EntityDetails entity
                 if (expectContinue) {
                     requestState.set(MessageState.ACK);
                 } else {
-                    requestState.set(MessageState.BODY);
                     exchangeHandler.produce(dataChannel);
+                    if (requestState.compareAndSet(MessageState.HEADERS, MessageState.BODY)) {
+                        outputChannel.requestOutput();
+                    }
                 }
             }
         } else {
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
@@ -183,8 +183,10 @@ private void commitResponse(
             if (endStream) {
                 responseState.set(MessageState.COMPLETE);
             } else {
-                responseState.set(MessageState.BODY);
                 exchangeHandler.produce(outputChannel);
+                if (responseState.compareAndSet(MessageState.IDLE, MessageState.BODY)) {
+                    outputChannel.requestOutput();
+                }
             }
         } else {
             throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Response already committed");
diff --git a/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/Http1IntegrationTest.java b/httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/Http1IntegrationTest.java
@@ -56,7 +56,6 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.hc.core5.function.Callback;
 import org.apache.hc.core5.http.ConnectionClosedException;
@@ -1005,7 +1004,6 @@ void testDelayedExpectContinueAck() throws Exception {
             private final Random random = new Random(System.currentTimeMillis());
             private final AsyncEntityProducer entityProducer = AsyncEntityProducers.create(
                     "All is well");
-            private final ReentrantLock lock = new ReentrantLock();
 
             @Override
             public void handleRequest(
@@ -1023,12 +1021,7 @@ public void handleRequest(
                                 responseChannel.sendInformation(new BasicHttpResponse(HttpStatus.SC_CONTINUE), context);
                             }
                             final HttpResponse response = new BasicHttpResponse(200);
-                            lock.lock();
-                            try {
-                                responseChannel.sendResponse(response, entityProducer, context);
-                            } finally {
-                                lock.unlock();
-                            }
+                            responseChannel.sendResponse(response, entityProducer, context);
                         }
                     } catch (final Exception ignore) {
                         // ignore
@@ -1052,22 +1045,12 @@ public void streamEnd(final List<? extends Header> trailers) throws HttpExceptio
 
             @Override
             public int available() {
-                lock.lock();
-                try {
-                    return entityProducer.available();
-                } finally {
-                    lock.unlock();
-                }
+                return entityProducer.available();
             }
 
             @Override
             public void produce(final DataStreamChannel channel) throws IOException {
-                lock.lock();
-                try {
-                    entityProducer.produce(channel);
-                } finally {
-                    lock.unlock();
-                }
+                entityProducer.produce(channel);
             }
 
             @Override
@@ -2060,19 +2043,13 @@ void testDelayedRequestSubmission() throws Exception {
                     new AsyncRequestProducer() {
 
                         private final Random random = new Random(System.currentTimeMillis());
-                        private final ReentrantLock lock = new ReentrantLock();
 
                         @Override
                         public void sendRequest(final RequestChannel channel, final HttpContext context) throws HttpException, IOException {
                             executorResource.getExecutorService().execute(() -> {
                                 try {
                                     Thread.sleep(random.nextInt(200));
-                                    lock.lock();
-                                    try {
-                                        channel.sendRequest(request, entityProducer, context);
-                                    } finally {
-                                        lock.unlock();
-                                    }
+                                    channel.sendRequest(request, entityProducer, context);
                                 } catch (final Exception ignore) {
                                     // ignore
                                 }
@@ -2081,32 +2058,17 @@ public void sendRequest(final RequestChannel channel, final HttpContext context)
 
                         @Override
                         public boolean isRepeatable() {
-                            lock.lock();
-                            try {
-                                return entityProducer.isRepeatable();
-                            } finally {
-                                lock.unlock();
-                            }
+                            return entityProducer.isRepeatable();
                         }
 
                         @Override
                         public int available() {
-                            lock.lock();
-                            try {
-                                return entityProducer.available();
-                            } finally {
-                                lock.unlock();
-                            }
+                            return entityProducer.available();
                         }
 
                         @Override
                         public void produce(final DataStreamChannel channel) throws IOException {
-                            lock.lock();
-                            try {
-                                entityProducer.produce(channel);
-                            } finally {
-                                lock.unlock();
-                            }
+                            entityProducer.produce(channel);
                         }
 
                         @Override
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
@@ -180,8 +180,10 @@ private void commitRequest(final HttpRequest request, final EntityDetails entity
                     final Timeout timeout = http1Config.getWaitForContinueTimeout() != null ? http1Config.getWaitForContinueTimeout() : DEFAULT_WAIT_FOR_CONTINUE;
                     outputChannel.setSocketTimeout(timeout);
                 } else {
-                    requestState.set(MessageState.BODY);
                     exchangeHandler.produce(internalDataChannel);
+                    if (requestState.compareAndSet(MessageState.HEADERS, MessageState.BODY)) {
+                        outputChannel.requestOutput();
+                    }
                 }
             }
         } else {
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
@@ -196,8 +196,10 @@ private void commitResponse(
                 }
                 responseState.set(MessageState.COMPLETE);
             } else {
-                responseState.set(MessageState.BODY);
                 exchangeHandler.produce(internalDataChannel);
+                if (responseState.compareAndSet(MessageState.IDLE, MessageState.BODY)) {
+                    outputChannel.requestOutput();
+                }
             }
         } else {
             throw new HttpException("Response already committed");

Original file line number	Diff line number	Diff line change
`@@ -153,8 +153,10 @@ private void commitRequest(final HttpRequest request, final EntityDetails entity`
`153`	`153`	`if (expectContinue) {`
`154`	`154`	`requestState.set(MessageState.ACK);`
`155`	`155`	`} else {`
`156`		`- requestState.set(MessageState.BODY);`
`157`	`156`	`exchangeHandler.produce(dataChannel);`
	`157`	`+ if (requestState.compareAndSet(MessageState.HEADERS, MessageState.BODY)) {`
	`158`	`+ outputChannel.requestOutput();`
	`159`	`+ }`
`158`	`160`	`}`
`159`	`161`	`}`
`160`	`162`	`} else {`
Original file line number	Diff line number	Diff line change
`@@ -180,8 +180,10 @@ private void commitRequest(final HttpRequest request, final EntityDetails entity`
`180`	`180`	`final Timeout timeout = http1Config.getWaitForContinueTimeout() != null ? http1Config.getWaitForContinueTimeout() : DEFAULT_WAIT_FOR_CONTINUE;`
`181`	`181`	`outputChannel.setSocketTimeout(timeout);`
`182`	`182`	`} else {`
`183`		`- requestState.set(MessageState.BODY);`
`184`	`183`	`exchangeHandler.produce(internalDataChannel);`
	`184`	`+ if (requestState.compareAndSet(MessageState.HEADERS, MessageState.BODY)) {`
	`185`	`+ outputChannel.requestOutput();`
	`186`	`+ }`
`185`	`187`	`}`
`186`	`188`	`}`
`187`	`189`	`} else {`
Original file line number	Diff line number	Diff line change
`@@ -196,8 +196,10 @@ private void commitResponse(`
`196`	`196`	`}`
`197`	`197`	`responseState.set(MessageState.COMPLETE);`
`198`	`198`	`} else {`
`199`		`- responseState.set(MessageState.BODY);`
`200`	`199`	`exchangeHandler.produce(internalDataChannel);`
	`200`	`+ if (responseState.compareAndSet(MessageState.IDLE, MessageState.BODY)) {`
	`201`	`+ outputChannel.requestOutput();`
	`202`	`+ }`
`201`	`203`	`}`
`202`	`204`	`} else {`
`203`	`205`	`throw new HttpException("Response already committed");`