This is an automated email from the ASF dual-hosted git repository.
ffang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cxf.git
The following commit(s) were added to refs/heads/main by this push:
new 7771388fdc2 CXF-9268: Wake up a waiting reader before the hc5
SharedInputBuffer s… (#3674)
7771388fdc2 is described below
commit 7771388fdc29af2f70e54dcb427607680d41a5c9
Author: Freeman(Yue) Fang <[email protected]>
AuthorDate: Tue Nov 3 19:47:23 2026 -0500
CXF-9268: Wake up a waiting reader before the hc5 SharedInputBuffer s…
(#3674)
* CXF-9268: Wake up a waiting reader before the hc5 SharedInputBuffer
suspends input
When a single block of response data filled the buffer while a reader was
waiting for data, consumeContent() waited on suspendInput before signalling
the reader. Only the reader signals suspendInput, after draining the buffer,
so both threads waited forever. The receive timeout did not fire, and the
I/O dispatcher thread, with every connection on it, was blocked. This
happens when a block is at least the reader's array plus the buffer size
(chunkLength), e.g. with the 32 KiB default HTTP/1.1 buffer of httpcore5
5.4.4, or with a small chunkLength.
Signal the reader before waiting for the buffer to be drained.
* CXF-9268: Signal the reader only once in SharedInputBuffer.consumeContent
Signal the reader before the buffer may be waited on to be drained, and no
longer after it: nothing is added to the buffer after the wait, so a second
signal only woke up the reader for nothing.
---
.../http/asyncclient/hc5/SharedInputBuffer.java | 6 +-
.../asyncclient/hc5/SharedInputBufferTest.java | 84 ++++++++++++++++++++++
2 files changed, 88 insertions(+), 2 deletions(-)
diff --git
a/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBuffer.java
b/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBuffer.java
index 0ba6b2c9367..dc882dce34a 100644
---
a/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBuffer.java
+++
b/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBuffer.java
@@ -96,6 +96,10 @@ public class SharedInputBuffer extends ExpandableBuffer {
this.endOfStream = true;
}
+ // Wake up a reader waiting for data before waiting for the buffer
to be drained: only
+ // the reader signals suspendInput, once it has drained the buffer
+ this.condition.signalAll();
+
if (!buffer().hasRemaining() && !this.endOfStream) {
try {
suspendInput.await();
@@ -104,8 +108,6 @@ public class SharedInputBuffer extends ExpandableBuffer {
}
}
- this.condition.signalAll();
-
if (totalRead > 0) {
return totalRead;
}
diff --git
a/rt/transports/http-hc5/src/test/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBufferTest.java
b/rt/transports/http-hc5/src/test/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBufferTest.java
new file mode 100644
index 00000000000..1e66acc0f90
--- /dev/null
+++
b/rt/transports/http-hc5/src/test/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBufferTest.java
@@ -0,0 +1,84 @@
+/**
+ * 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.
+ */
+
+package org.apache.cxf.transport.http.asyncclient.hc5;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+
+public class SharedInputBufferTest {
+
+ @Test
+ public void testReaderWokenUpWhenDataBlockFillsBuffer() throws Exception {
+ final CountDownLatch readerWaiting = new CountDownLatch(1);
+ final SharedInputBuffer inbuf = new SharedInputBuffer(1024) {
+ @Override
+ protected void waitForData(int waitPos) throws IOException {
+ readerWaiting.countDown();
+ super.waitForData(waitPos);
+ }
+ };
+ final byte[] data = new byte[4096];
+ for (int i = 0; i < data.length; i++) {
+ data[i] = (byte)i;
+ }
+
+ ExecutorService executor = Executors.newFixedThreadPool(2, r -> {
+ Thread t = new Thread(r);
+ t.setDaemon(true);
+ return t;
+ });
+ try {
+ Future<byte[]> reader = executor.submit(() -> {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ byte[] b = new byte[256];
+ int n;
+ while ((n = inbuf.read(b, 0, b.length)) != -1) {
+ out.write(b, 0, n);
+ }
+ return out.toByteArray();
+ });
+ executor.submit(() -> {
+ // The reader holds the lock until it waits on it, so the data
is consumed while
+ // the reader is waiting
+ readerWaiting.await();
+ // A block larger than the reader's array plus the buffer size
fills the buffer,
+ // and the I/O thread waits until the reader has drained it
+ inbuf.consumeContent(ByteBuffer.wrap(data), false);
+ inbuf.consumeContent(ByteBuffer.allocate(0), true);
+ return null;
+ });
+
+ assertArrayEquals(data, reader.get(10, TimeUnit.SECONDS));
+ } finally {
+ inbuf.shutdown();
+ executor.shutdownNow();
+ }
+ }
+}
