From 70f409b33031eabf818eb8da2a6dc85d569a95ce Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 20 Jul 2026 11:54:51 +0200 Subject: [PATCH 1/8] Offload stream body reads InputStream request bodies were still read by the channel event loop through both the HTTP/1 chunked writer and the HTTP/2 body pump. A slow or blocking producer could therefore stall unrelated work on the same loop. Add a client-owned blocking-read executor and have NettyInputStreamBody suspend the event-loop writer while one InputStream read runs off-loop. Completed bytes are copied into ByteBufs back on the event loop before the HTTP/1 or HTTP/2 writer resumes, keeping buffer ownership unchanged. Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../DefaultAsyncHttpClient.java | 19 +- .../netty/request/NettyRequestFactory.java | 13 +- .../netty/request/NettyRequestSender.java | 8 +- .../request/body/NettyInputStreamBody.java | 275 ++++++++++++++++-- .../org/asynchttpclient/BasicHttpTest.java | 76 +++++ .../Http2StreamingBodyFlowControlTest.java | 76 +++++ 6 files changed, 445 insertions(+), 22 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java index 3b417a5a39..b18b3b6554 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java +++ b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java @@ -35,7 +35,10 @@ import org.slf4j.LoggerFactory; import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Predicate; @@ -62,6 +65,7 @@ public class DefaultAsyncHttpClient implements AsyncHttpClient { private final AtomicBoolean closed = new AtomicBoolean(false); private final ChannelManager channelManager; private final NettyRequestSender requestSender; + private final ExecutorService blockingBodyReadExecutor; private final boolean allowStopNettyTimer; private final Timer nettyTimer; @@ -105,8 +109,10 @@ public DefaultAsyncHttpClient(AsyncHttpClientConfig config) { nettyTimer = configTimer; } + blockingBodyReadExecutor = newBlockingBodyReadExecutor(config); channelManager = new ChannelManager(config, nettyTimer); - requestSender = new NettyRequestSender(config, channelManager, nettyTimer, new AsyncHttpClientState(closed)); + requestSender = new NettyRequestSender(config, channelManager, nettyTimer, new AsyncHttpClientState(closed), + blockingBodyReadExecutor); channelManager.configureBootstraps(requestSender); CookieStore cookieStore = config.getCookieStore(); @@ -134,6 +140,16 @@ private static Timer newNettyTimer(AsyncHttpClientConfig config) { return timer; } + private static ExecutorService newBlockingBodyReadExecutor(AsyncHttpClientConfig config) { + ThreadFactory threadFactory = config.getThreadFactory() != null + ? config.getThreadFactory() + : new DefaultThreadFactory(config.getThreadPoolName() + "-blocking-io"); + int threads = Math.max(1, config.getIoThreadsCount()); + int queueSize = Math.max(1024, threads * 16); + return new ThreadPoolExecutor(threads, threads, 0L, TimeUnit.MILLISECONDS, + new LinkedBlockingQueue<>(queueSize), threadFactory); + } + @Override public void close() { if (closed.compareAndSet(false, true)) { @@ -142,6 +158,7 @@ public void close() { } catch (Throwable t) { LOGGER.warn("Unexpected error on ChannelManager close", t); } + blockingBodyReadExecutor.shutdownNow(); CookieStore cookieStore = config.getCookieStore(); if (cookieStore != null) { cookieStore.decrementAndGet(); diff --git a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestFactory.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestFactory.java index b2f23ecbf4..41f0206052 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestFactory.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestFactory.java @@ -52,6 +52,7 @@ import java.util.HashMap; import java.util.Iterator; import java.util.Map; +import java.util.concurrent.Executor; import static io.netty.handler.codec.http.HttpHeaderNames.ACCEPT; import static io.netty.handler.codec.http.HttpHeaderNames.ACCEPT_ENCODING; @@ -68,6 +69,7 @@ import static io.netty.handler.codec.http.HttpHeaderNames.TRANSFER_ENCODING; import static io.netty.handler.codec.http.HttpHeaderNames.UPGRADE; import static io.netty.handler.codec.http.HttpHeaderNames.USER_AGENT; +import static java.util.Objects.requireNonNull; import static org.asynchttpclient.util.AuthenticatorUtils.perRequestAuthorizationHeader; import static org.asynchttpclient.util.AuthenticatorUtils.perRequestProxyAuthorizationHeader; import static org.asynchttpclient.util.HttpUtils.ACCEPT_ALL_HEADER_VALUE; @@ -146,9 +148,15 @@ static void copyInternedHeaders(HttpHeaders source, HttpHeaders target) { private final AsyncHttpClientConfig config; private final ClientCookieEncoder cookieEncoder; + private final Executor blockingBodyReadExecutor; NettyRequestFactory(AsyncHttpClientConfig config) { + this(config, Runnable::run); + } + + NettyRequestFactory(AsyncHttpClientConfig config, Executor blockingBodyReadExecutor) { this.config = config; + this.blockingBodyReadExecutor = requireNonNull(blockingBodyReadExecutor, "blockingBodyReadExecutor"); cookieEncoder = config.isUseLaxCookieEncoder() ? ClientCookieEncoder.LAX : ClientCookieEncoder.STRICT; } @@ -167,7 +175,7 @@ private NettyBody body(Request request) { } else if (request.getByteBufData() != null) { nettyBody = new NettyByteBufBody(request.getByteBufData()); } else if (request.getStreamData() != null) { - nettyBody = new NettyInputStreamBody(request.getStreamData()); + nettyBody = new NettyInputStreamBody(request.getStreamData(), blockingBodyReadExecutor); } else if (isNonEmpty(request.getFormParams())) { CharSequence contentTypeOverride = request.getHeaders().contains(CONTENT_TYPE) ? null : HttpHeaderValues.APPLICATION_X_WWW_FORM_URLENCODED; nettyBody = new NettyByteBufferBody(urlEncodeFormParams(request.getFormParams(), bodyCharset), contentTypeOverride); @@ -180,7 +188,8 @@ private NettyBody body(Request request) { nettyBody = new NettyFileBody(fileBodyGenerator.getFile(), fileBodyGenerator.getRegionSeek(), fileBodyGenerator.getRegionLength(), config); } else if (request.getBodyGenerator() instanceof InputStreamBodyGenerator) { InputStreamBodyGenerator inStreamGenerator = (InputStreamBodyGenerator) request.getBodyGenerator(); - nettyBody = new NettyInputStreamBody(inStreamGenerator.getInputStream(), inStreamGenerator.getContentLength()); + nettyBody = new NettyInputStreamBody(inStreamGenerator.getInputStream(), inStreamGenerator.getContentLength(), + blockingBodyReadExecutor); } else if (request.getBodyGenerator() != null) { nettyBody = new NettyBodyBody(request.getBodyGenerator().createBody(), config); } diff --git a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java index 36af9019b1..4b8288105e 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -95,6 +95,7 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; @@ -126,6 +127,11 @@ public final class NettyRequestSender { private final FailedIpCooldownHolder ipCooldown; public NettyRequestSender(AsyncHttpClientConfig config, ChannelManager channelManager, Timer nettyTimer, AsyncHttpClientState clientState) { + this(config, channelManager, nettyTimer, clientState, Runnable::run); + } + + public NettyRequestSender(AsyncHttpClientConfig config, ChannelManager channelManager, Timer nettyTimer, + AsyncHttpClientState clientState, Executor blockingBodyReadExecutor) { this.config = config; this.channelManager = channelManager; connectionSemaphore = config.getConnectionSemaphoreFactory() == null @@ -133,7 +139,7 @@ public NettyRequestSender(AsyncHttpClientConfig config, ChannelManager channelMa : config.getConnectionSemaphoreFactory().newConnectionSemaphore(config); this.nettyTimer = nettyTimer; this.clientState = clientState; - requestFactory = new NettyRequestFactory(config); + requestFactory = new NettyRequestFactory(config, blockingBodyReadExecutor); // Guard the period against a custom AsyncHttpClientConfig that enables the cooldown but returns a // null period: leave the cooldown off rather than NPE while constructing the client. Duration cooldownPeriod = config.getFailedIpCooldownPeriod(); diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java index 32dbdc0fe7..9ce69b1fed 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java @@ -21,7 +21,9 @@ import io.netty.channel.ChannelProgressiveFuture; import io.netty.handler.codec.http.LastHttpContent; import io.netty.handler.codec.http2.Http2StreamChannel; -import io.netty.handler.stream.ChunkedStream; +import io.netty.handler.stream.ChunkedInput; +import io.netty.handler.stream.ChunkedWriteHandler; +import io.netty.util.concurrent.EventExecutor; import org.asynchttpclient.netty.NettyResponseFuture; import org.asynchttpclient.netty.request.WriteProgressListener; import org.slf4j.Logger; @@ -29,7 +31,10 @@ import java.io.IOException; import java.io.InputStream; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; +import static java.util.Objects.requireNonNull; import static org.asynchttpclient.util.MiscUtils.closeSilently; public class NettyInputStreamBody implements NettyBody { @@ -38,14 +43,24 @@ public class NettyInputStreamBody implements NettyBody { private final InputStream inputStream; private final long contentLength; + private final Executor blockingBodyReadExecutor; public NettyInputStreamBody(InputStream inputStream) { this(inputStream, -1L); } public NettyInputStreamBody(InputStream inputStream, long contentLength) { + this(inputStream, contentLength, Runnable::run); + } + + public NettyInputStreamBody(InputStream inputStream, Executor blockingBodyReadExecutor) { + this(inputStream, -1L, blockingBodyReadExecutor); + } + + public NettyInputStreamBody(InputStream inputStream, long contentLength, Executor blockingBodyReadExecutor) { this.inputStream = inputStream; this.contentLength = contentLength; + this.blockingBodyReadExecutor = requireNonNull(blockingBodyReadExecutor, "blockingBodyReadExecutor"); } public InputStream getInputStream() { @@ -72,11 +87,15 @@ public void write(Channel channel, NettyResponseFuture future) throws IOExcep future.setStreamConsumed(true); } - channel.write(new ChunkedStream(is), channel.newProgressivePromise()).addListener( + ChunkedWriteHandler chunkedWriteHandler = requireNonNull(channel.pipeline().get(ChunkedWriteHandler.class), + "chunkedWriteHandler"); + OffloadedInputStreamChunkedInput chunkedInput = new OffloadedInputStreamChunkedInput(is, getContentLength(), + channel.eventLoop(), chunkedWriteHandler, blockingBodyReadExecutor); + channel.write(chunkedInput, channel.newProgressivePromise()).addListener( new WriteProgressListener(future, false, getContentLength()) { @Override public void operationComplete(ChannelProgressiveFuture cf) { - closeSilently(is); + chunkedInput.close(); super.operationComplete(cf); } }); @@ -107,49 +126,269 @@ public void writeHttp2(Http2StreamChannel channel, NettyResponseFuture future // backpressure, so a large upload does not buffer the whole stream in heap or read it all inline on // the event loop. Cleanup (closeSilently) happens when the async pump completes — see // InputStreamChunkSource.close. - Http2BodyWriter.start(channel, new InputStreamChunkSource(is)); + Http2BodyWriter.start(channel, new InputStreamChunkSource(is, channel.eventLoop(), blockingBodyReadExecutor)); } /** - * Reads an {@link InputStream} in fixed-size chunks for {@link Http2BodyWriter}. + * Reads an {@link InputStream} off the event loop and exposes completed chunks to + * {@link Http2BodyWriter}. */ private static final class InputStreamChunkSource implements Http2BodyWriter.ChunkSource { private static final int CHUNK_SIZE = 8192; private final InputStream is; + private final EventExecutor eventLoop; + private final Executor blockingBodyReadExecutor; private final byte[] buffer = new byte[CHUNK_SIZE]; + private boolean readInProgress; + private boolean endOfInput; + private boolean closed; + private int readableBytes; + private IOException failure; + private Runnable resume; - InputStreamChunkSource(InputStream is) { + InputStreamChunkSource(InputStream is, EventExecutor eventLoop, Executor blockingBodyReadExecutor) { this.is = is; + this.eventLoop = eventLoop; + this.blockingBodyReadExecutor = blockingBodyReadExecutor; } @Override public ByteBuf nextChunk(ByteBufAllocator alloc) throws IOException { - // Blocking InputStream.read returns >0 (data), or -1 (EOF). A transient 0 is possible for some - // streams; loop on it so we never emit an empty non-final DATA frame nor end the stream early. The - // read blocks until data or EOF, so this does not busy-spin. + if (failure != null) { + throw failure; + } + if (readableBytes > 0) { + ByteBuf buf = alloc.buffer(readableBytes); + try { + buf.writeBytes(buffer, 0, readableBytes); + readableBytes = 0; + return buf; + } catch (RuntimeException e) { + buf.release(); + throw e; + } + } + if (endOfInput) { + return null; + } + if (!readInProgress) { + submitReadAfterSuspend(); + } + return Http2BodyWriter.SUSPEND; + } + + @Override + public void onResume(Runnable resume) { + this.resume = resume; + } + + @Override + public void close() { + closed = true; + closeSilently(is); + } + + private void submitReadAfterSuspend() throws IOException { + readInProgress = true; + try { + eventLoop.execute(this::submitRead); + } catch (RejectedExecutionException e) { + readInProgress = false; + throw new IOException("HTTP/2 request body read executor is unavailable", e); + } + } + + private void submitRead() { + try { + blockingBodyReadExecutor.execute(this::readOffEventLoop); + } catch (RejectedExecutionException e) { + completeRead(-1, new IOException("HTTP/2 request body read executor rejected the read", e)); + } + } + + private void readOffEventLoop() { + int read = -1; + IOException thrown = null; + try { + read = read(); + } catch (IOException e) { + thrown = e; + } + completeRead(read, thrown); + } + + private int read() throws IOException { int read; do { read = is.read(buffer); - } while (read == 0); + } while (read == 0 && !closed); + return read; + } + + private void completeRead(int read, IOException thrown) { + try { + eventLoop.execute(() -> { + readInProgress = false; + if (closed) { + return; + } + if (thrown != null) { + failure = thrown; + } else if (read < 0) { + endOfInput = true; + } else { + readableBytes = read; + } + if (resume != null) { + resume.run(); + } + }); + } catch (RejectedExecutionException e) { + close(); + } + } + } + + private static final class OffloadedInputStreamChunkedInput implements ChunkedInput { + + private static final int CHUNK_SIZE = 8192; + + private final InputStream is; + private final long length; + private final EventExecutor eventLoop; + private final ChunkedWriteHandler chunkedWriteHandler; + private final Executor blockingBodyReadExecutor; + private final byte[] buffer = new byte[CHUNK_SIZE]; + private boolean readInProgress; + private boolean endOfInput; + private boolean closed; + private int readableBytes; + private long progress; + private IOException failure; + + OffloadedInputStreamChunkedInput(InputStream is, + long length, + EventExecutor eventLoop, + ChunkedWriteHandler chunkedWriteHandler, + Executor blockingBodyReadExecutor) { + this.is = is; + this.length = length; + this.eventLoop = eventLoop; + this.chunkedWriteHandler = chunkedWriteHandler; + this.blockingBodyReadExecutor = blockingBodyReadExecutor; + } + + @Override + @Deprecated + public ByteBuf readChunk(io.netty.channel.ChannelHandlerContext ctx) throws Exception { + return readChunk(ctx.alloc()); + } - if (read == -1) { + @Override + public ByteBuf readChunk(ByteBufAllocator alloc) throws Exception { + if (failure != null) { + throw failure; + } + if (readableBytes > 0) { + ByteBuf buf = alloc.buffer(readableBytes); + try { + buf.writeBytes(buffer, 0, readableBytes); + progress += readableBytes; + readableBytes = 0; + return buf; + } catch (RuntimeException e) { + buf.release(); + throw e; + } + } + if (endOfInput) { return null; } - ByteBuf buf = alloc.buffer(read); - try { - buf.writeBytes(buffer, 0, read); - return buf; - } catch (RuntimeException e) { - buf.release(); - throw e; + if (!readInProgress) { + submitReadAfterSuspend(); } + return null; + } + + @Override + public boolean isEndOfInput() { + return endOfInput; } @Override public void close() { + closed = true; closeSilently(is); } + + @Override + public long length() { + return length; + } + + @Override + public long progress() { + return progress; + } + + private void submitReadAfterSuspend() throws IOException { + readInProgress = true; + try { + eventLoop.execute(this::submitRead); + } catch (RejectedExecutionException e) { + readInProgress = false; + throw new IOException("HTTP/1 request body read executor is unavailable", e); + } + } + + private void submitRead() { + try { + blockingBodyReadExecutor.execute(this::readOffEventLoop); + } catch (RejectedExecutionException e) { + completeRead(-1, new IOException("HTTP/1 request body read executor rejected the read", e)); + } + } + + private void readOffEventLoop() { + int read = -1; + IOException thrown = null; + try { + read = read(); + } catch (IOException e) { + thrown = e; + } + completeRead(read, thrown); + } + + private int read() throws IOException { + int read; + do { + read = is.read(buffer); + } while (read == 0 && !closed); + return read; + } + + private void completeRead(int read, IOException thrown) { + try { + eventLoop.execute(() -> { + readInProgress = false; + if (closed) { + return; + } + if (thrown != null) { + failure = thrown; + } else if (read < 0) { + endOfInput = true; + } else { + readableBytes = read; + } + chunkedWriteHandler.resumeTransfer(); + }); + } catch (RejectedExecutionException e) { + close(); + } + } } } diff --git a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java index 44e7f45dfb..79691007b3 100755 --- a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java +++ b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java @@ -16,11 +16,13 @@ package org.asynchttpclient; import io.github.artsok.RepeatedIfExceptionsTest; +import io.netty.channel.EventLoopGroup; import io.netty.handler.codec.http.DefaultHttpHeaders; import io.netty.handler.codec.http.HttpHeaderValues; import io.netty.handler.codec.http.HttpHeaders; import io.netty.handler.codec.http.cookie.Cookie; import io.netty.handler.codec.http.cookie.DefaultCookie; +import io.netty.util.concurrent.EventExecutor; import jakarta.servlet.ServletException; import jakarta.servlet.http.HttpServletRequest; import jakarta.servlet.http.HttpServletResponse; @@ -1040,4 +1042,78 @@ public Response onCompleted(Response response) { }).get(TIMEOUT, SECONDS); })); } + + @RepeatedIfExceptionsTest(repeats = 5) + public void postInputStreamBodyGeneratorReadsOffEventLoop() throws Throwable { + withServer(server).run(server -> { + server.enqueueEcho(); + byte[] bodyBytes = "{}".getBytes(StandardCharsets.ISO_8859_1); + EventLoopProbeInputStream bodyStream = new EventLoopProbeInputStream(bodyBytes); + + try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(config().build())) { + bodyStream.setEventLoopGroup(client.channelManager().getEventLoopGroup()); + + Response response = client.preparePost(getTargetUrl()) + .setBody(new InputStreamBodyGenerator(bodyStream, bodyBytes.length)) + .execute() + .get(TIMEOUT, SECONDS); + + assertEquals(200, response.getStatusCode()); + assertEquals("{}", response.getResponseBody()); + } + + assertTrue(bodyStream.readAttempted.get(), "InputStream should have been read"); + assertFalse(bodyStream.readOnEventLoop.get(), "InputStream.read must not run on an event-loop thread"); + }); + } + + private static final class EventLoopProbeInputStream extends InputStream { + private final byte[] data; + private final AtomicBoolean readAttempted = new AtomicBoolean(); + private final AtomicBoolean readOnEventLoop = new AtomicBoolean(); + private final AtomicReference eventLoopGroup = new AtomicReference<>(); + private int position; + + EventLoopProbeInputStream(byte[] data) { + this.data = data; + } + + void setEventLoopGroup(EventLoopGroup eventLoopGroup) { + this.eventLoopGroup.set(eventLoopGroup); + } + + @Override + public int read() { + if (position == data.length) { + return -1; + } + return data[position++] & 0xFF; + } + + @Override + public int read(byte[] buffer, int offset, int length) { + recordReadThread(); + if (position == data.length) { + return -1; + } + int read = Math.min(length, data.length - position); + System.arraycopy(data, position, buffer, offset, read); + position += read; + return read; + } + + private void recordReadThread() { + readAttempted.set(true); + EventLoopGroup group = eventLoopGroup.get(); + if (group == null) { + return; + } + Thread currentThread = Thread.currentThread(); + for (EventExecutor executor : group) { + if (executor.inEventLoop(currentThread)) { + readOnEventLoop.set(true); + } + } + } + } } diff --git a/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java b/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java index 04b3ff0abc..85728fa15a 100644 --- a/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java +++ b/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java @@ -20,6 +20,7 @@ import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; +import io.netty.channel.EventLoopGroup; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.channel.group.ChannelGroup; import io.netty.channel.group.DefaultChannelGroup; @@ -43,6 +44,7 @@ import io.netty.pkitesting.CertificateBuilder; import io.netty.pkitesting.X509Bundle; import io.netty.util.ReferenceCountUtil; +import io.netty.util.concurrent.EventExecutor; import io.netty.util.concurrent.GlobalEventExecutor; import org.asynchttpclient.request.body.Body; import org.asynchttpclient.request.body.generator.BodyGenerator; @@ -60,7 +62,9 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; import java.util.zip.CRC32; @@ -68,6 +72,7 @@ import static org.asynchttpclient.Dsl.asyncHttpClient; import static org.asynchttpclient.Dsl.config; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -333,6 +338,27 @@ public void largeInputStreamBodyRoundTripsOverHttp2() throws Exception { } } + @Test + public void inputStreamBodyReadsOffEventLoopOverHttp2() throws Exception { + startServer(-1); + byte[] payload = deterministicPayload(SMALL_SIZE); + EventLoopProbeInputStream is = new EventLoopProbeInputStream(payload); + + try (AsyncHttpClient asyncClient = http2Client()) { + DefaultAsyncHttpClient client = (DefaultAsyncHttpClient) asyncClient; + is.setEventLoopGroup(client.channelManager().getEventLoopGroup()); + + Response response = client.preparePost(httpsUrl("/upload")) + .setBody(new InputStreamBodyGenerator(is, payload.length)) + .execute() + .get(30, SECONDS); + assertEchoed(response, payload); + } + + assertTrue(is.readAttempted.get(), "InputStream should have been read"); + assertFalse(is.readOnEventLoop.get(), "InputStream.read must not run on an event-loop thread"); + } + // ========================================================================= // NettyFileBody — large streaming upload via setBody(File) // ========================================================================= @@ -482,6 +508,56 @@ public void close() throws IOException { } } + private static final class EventLoopProbeInputStream extends InputStream { + private final byte[] data; + private final AtomicBoolean readAttempted = new AtomicBoolean(); + private final AtomicBoolean readOnEventLoop = new AtomicBoolean(); + private final AtomicReference eventLoopGroup = new AtomicReference<>(); + private int position; + + EventLoopProbeInputStream(byte[] data) { + this.data = data; + } + + void setEventLoopGroup(EventLoopGroup eventLoopGroup) { + this.eventLoopGroup.set(eventLoopGroup); + } + + @Override + public int read() { + if (position == data.length) { + return -1; + } + return data[position++] & 0xFF; + } + + @Override + public int read(byte[] buffer, int offset, int length) { + recordReadThread(); + if (position == data.length) { + return -1; + } + int read = Math.min(length, data.length - position); + System.arraycopy(data, position, buffer, offset, read); + position += read; + return read; + } + + private void recordReadThread() { + readAttempted.set(true); + EventLoopGroup group = eventLoopGroup.get(); + if (group == null) { + return; + } + Thread currentThread = Thread.currentThread(); + for (EventExecutor executor : group) { + if (executor.inEventLoop(currentThread)) { + readOnEventLoop.set(true); + } + } + } + } + // Lifecycle (finding #2 from the cold audit): when the stream is closed while the body pump is parked on // SUSPEND (a streaming/feedable body with no data yet), no in-flight write's failure would run cleanup — // so the writer must close the body source from the stream's closeFuture, or the source (and any file From 70e0a1f832354f087ee229c953c24e81c9a98d77 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 20 Jul 2026 12:11:18 +0200 Subject: [PATCH 2/8] Make stream read offload configurable Expose request body stream read offload controls on the client config so callers can disable the worker pool or tune its worker count and queue capacity. Keep the default behavior enabled and preserve automatic thread and queue sizing when count or queue values are not configured. Add focused tests for property defaults, custom worker naming, and disabled inline event-loop reads. Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../AsyncHttpClientConfig.java | 30 +++++++++++ .../DefaultAsyncHttpClient.java | 38 +++++++++++--- .../DefaultAsyncHttpClientConfig.java | 51 +++++++++++++++++++ .../config/AsyncHttpClientConfigDefaults.java | 18 +++++++ .../config/ahc-default.properties | 3 ++ .../AsyncHttpClientDefaultsTest.java | 18 +++++++ .../org/asynchttpclient/BasicHttpTest.java | 40 ++++++++++++++- 7 files changed, 190 insertions(+), 8 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index cae3900ee0..88cdd89be6 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java @@ -165,6 +165,36 @@ public interface AsyncHttpClientConfig { @Nullable ThreadFactory getThreadFactory(); + /** + * Returns whether blocking reads from request body {@link java.io.InputStream} instances should be + * offloaded from Netty event-loop threads. + * + * @return {@code true} if blocking request body stream reads should be offloaded + */ + default boolean isRequestBodyStreamReadOffloadEnabled() { + return true; + } + + /** + * Returns the number of threads used to offload request body {@link java.io.InputStream} reads. + * + * @return the configured request body stream read thread count. Values less than or equal to {@code 0} + * use {@link #getIoThreadsCount()}. + */ + default int getRequestBodyStreamReadThreadsCount() { + return -1; + } + + /** + * Returns the maximum number of pending request body {@link java.io.InputStream} reads. + * + * @return the configured request body stream read queue size. Values less than or equal to {@code 0} use + * the default queue size. + */ + default int getRequestBodyStreamReadQueueSize() { + return 0; + } + /** * An instance of {@link ProxyServer} used by an {@link AsyncHttpClient} * diff --git a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java index b18b3b6554..ff0072343d 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java +++ b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java @@ -35,6 +35,7 @@ import org.slf4j.LoggerFactory; import java.util.List; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadFactory; @@ -65,7 +66,7 @@ public class DefaultAsyncHttpClient implements AsyncHttpClient { private final AtomicBoolean closed = new AtomicBoolean(false); private final ChannelManager channelManager; private final NettyRequestSender requestSender; - private final ExecutorService blockingBodyReadExecutor; + private final @Nullable ExecutorService blockingBodyReadExecutor; private final boolean allowStopNettyTimer; private final Timer nettyTimer; @@ -112,7 +113,7 @@ public DefaultAsyncHttpClient(AsyncHttpClientConfig config) { blockingBodyReadExecutor = newBlockingBodyReadExecutor(config); channelManager = new ChannelManager(config, nettyTimer); requestSender = new NettyRequestSender(config, channelManager, nettyTimer, new AsyncHttpClientState(closed), - blockingBodyReadExecutor); + executorForBodyReads(blockingBodyReadExecutor)); channelManager.configureBootstraps(requestSender); CookieStore cookieStore = config.getCookieStore(); @@ -140,16 +141,39 @@ private static Timer newNettyTimer(AsyncHttpClientConfig config) { return timer; } - private static ExecutorService newBlockingBodyReadExecutor(AsyncHttpClientConfig config) { + private static @Nullable ExecutorService newBlockingBodyReadExecutor(AsyncHttpClientConfig config) { + if (!config.isRequestBodyStreamReadOffloadEnabled()) { + return null; + } ThreadFactory threadFactory = config.getThreadFactory() != null ? config.getThreadFactory() : new DefaultThreadFactory(config.getThreadPoolName() + "-blocking-io"); - int threads = Math.max(1, config.getIoThreadsCount()); - int queueSize = Math.max(1024, threads * 16); + int threads = requestBodyStreamReadThreads(config); + int queueSize = requestBodyStreamReadQueueSize(config, threads); return new ThreadPoolExecutor(threads, threads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(queueSize), threadFactory); } + private static Executor executorForBodyReads(@Nullable ExecutorService executor) { + return executor == null ? Runnable::run : executor; + } + + private static int requestBodyStreamReadThreads(AsyncHttpClientConfig config) { + int threads = config.getRequestBodyStreamReadThreadsCount(); + if (threads <= 0) { + threads = config.getIoThreadsCount(); + } + return Math.max(1, threads); + } + + private static int requestBodyStreamReadQueueSize(AsyncHttpClientConfig config, int threads) { + int queueSize = config.getRequestBodyStreamReadQueueSize(); + if (queueSize <= 0) { + queueSize = threads > Integer.MAX_VALUE / 16 ? Integer.MAX_VALUE : Math.max(1024, threads * 16); + } + return Math.max(1, queueSize); + } + @Override public void close() { if (closed.compareAndSet(false, true)) { @@ -158,7 +182,9 @@ public void close() { } catch (Throwable t) { LOGGER.warn("Unexpected error on ChannelManager close", t); } - blockingBodyReadExecutor.shutdownNow(); + if (blockingBodyReadExecutor != null) { + blockingBodyReadExecutor.shutdownNow(); + } CookieStore cookieStore = config.getCookieStore(); if (cookieStore != null) { cookieStore.decrementAndGet(); diff --git a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java index 467e3d9ad3..c6e814d260 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java @@ -82,6 +82,9 @@ import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultMaxRequestRetry; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultPooledConnectionIdleTimeout; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultReadTimeout; +import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultRequestBodyStreamReadOffloadEnabled; +import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultRequestBodyStreamReadQueueSize; +import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultRequestBodyStreamReadThreadsCount; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultLoadBalance; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultRequestTimeout; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultShutdownQuietPeriod; @@ -226,6 +229,9 @@ public class DefaultAsyncHttpClientConfig implements AsyncHttpClientConfig { private final @Nullable Consumer wsAdditionalChannelInitializer; private final ResponseBodyPartFactory responseBodyPartFactory; private final int ioThreadsCount; + private final boolean requestBodyStreamReadOffloadEnabled; + private final int requestBodyStreamReadThreadsCount; + private final int requestBodyStreamReadQueueSize; private final long hashedWheelTimerTickDuration; private final int hashedWheelTimerSize; @@ -330,6 +336,9 @@ private DefaultAsyncHttpClientConfig(// http @Nullable Consumer wsAdditionalChannelInitializer, ResponseBodyPartFactory responseBodyPartFactory, int ioThreadsCount, + boolean requestBodyStreamReadOffloadEnabled, + int requestBodyStreamReadThreadsCount, + int requestBodyStreamReadQueueSize, long hashedWheelTimerTickDuration, int hashedWheelTimerSize) { @@ -449,6 +458,9 @@ private DefaultAsyncHttpClientConfig(// http this.wsAdditionalChannelInitializer = wsAdditionalChannelInitializer; this.responseBodyPartFactory = responseBodyPartFactory; this.ioThreadsCount = ioThreadsCount; + this.requestBodyStreamReadOffloadEnabled = requestBodyStreamReadOffloadEnabled; + this.requestBodyStreamReadThreadsCount = requestBodyStreamReadThreadsCount; + this.requestBodyStreamReadQueueSize = requestBodyStreamReadQueueSize; this.hashedWheelTimerTickDuration = hashedWheelTimerTickDuration; this.hashedWheelTimerSize = hashedWheelTimerSize; } @@ -907,6 +919,21 @@ public int getIoThreadsCount() { return ioThreadsCount; } + @Override + public boolean isRequestBodyStreamReadOffloadEnabled() { + return requestBodyStreamReadOffloadEnabled; + } + + @Override + public int getRequestBodyStreamReadThreadsCount() { + return requestBodyStreamReadThreadsCount; + } + + @Override + public int getRequestBodyStreamReadQueueSize() { + return requestBodyStreamReadQueueSize; + } + /** * Builder for an {@link AsyncHttpClient} */ @@ -1016,6 +1043,9 @@ public static class Builder { private @Nullable Consumer wsAdditionalChannelInitializer; private ResponseBodyPartFactory responseBodyPartFactory = ResponseBodyPartFactory.EAGER; private int ioThreadsCount = defaultIoThreadsCount(); + private boolean requestBodyStreamReadOffloadEnabled = defaultRequestBodyStreamReadOffloadEnabled(); + private int requestBodyStreamReadThreadsCount = defaultRequestBodyStreamReadThreadsCount(); + private int requestBodyStreamReadQueueSize = defaultRequestBodyStreamReadQueueSize(); private long hashedWheelTickDuration = defaultHashedWheelTimerTickDuration(); private int hashedWheelSize = defaultHashedWheelTimerSize(); @@ -1127,6 +1157,9 @@ public Builder(AsyncHttpClientConfig config) { wsAdditionalChannelInitializer = config.getWsAdditionalChannelInitializer(); responseBodyPartFactory = config.getResponseBodyPartFactory(); ioThreadsCount = config.getIoThreadsCount(); + requestBodyStreamReadOffloadEnabled = config.isRequestBodyStreamReadOffloadEnabled(); + requestBodyStreamReadThreadsCount = config.getRequestBodyStreamReadThreadsCount(); + requestBodyStreamReadQueueSize = config.getRequestBodyStreamReadQueueSize(); hashedWheelTickDuration = config.getHashedWheelTimerTickDuration(); hashedWheelSize = config.getHashedWheelTimerSize(); } @@ -1688,6 +1721,21 @@ public Builder setIoThreadsCount(int ioThreadsCount) { return this; } + public Builder setRequestBodyStreamReadOffloadEnabled(boolean requestBodyStreamReadOffloadEnabled) { + this.requestBodyStreamReadOffloadEnabled = requestBodyStreamReadOffloadEnabled; + return this; + } + + public Builder setRequestBodyStreamReadThreadsCount(int requestBodyStreamReadThreadsCount) { + this.requestBodyStreamReadThreadsCount = requestBodyStreamReadThreadsCount; + return this; + } + + public Builder setRequestBodyStreamReadQueueSize(int requestBodyStreamReadQueueSize) { + this.requestBodyStreamReadQueueSize = requestBodyStreamReadQueueSize; + return this; + } + private ProxyServerSelector resolveProxyServerSelector() { if (proxyServerSelector != null) { return proxyServerSelector; @@ -1793,6 +1841,9 @@ public DefaultAsyncHttpClientConfig build() { wsAdditionalChannelInitializer, responseBodyPartFactory, ioThreadsCount, + requestBodyStreamReadOffloadEnabled, + requestBodyStreamReadThreadsCount, + requestBodyStreamReadQueueSize, hashedWheelTickDuration, hashedWheelSize); } diff --git a/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java b/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java index 550381fbda..e3f6817cfd 100644 --- a/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java +++ b/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java @@ -90,6 +90,9 @@ public final class AsyncHttpClientConfigDefaults { public static final String USE_NATIVE_TRANSPORT_CONFIG = "useNativeTransport"; public static final String USE_ONLY_EPOLL_NATIVE_TRANSPORT = "useOnlyEpollNativeTransport"; public static final String IO_THREADS_COUNT_CONFIG = "ioThreadsCount"; + public static final String REQUEST_BODY_STREAM_READ_OFFLOAD_ENABLED_CONFIG = "requestBodyStreamReadOffloadEnabled"; + public static final String REQUEST_BODY_STREAM_READ_THREADS_COUNT_CONFIG = "requestBodyStreamReadThreadsCount"; + public static final String REQUEST_BODY_STREAM_READ_QUEUE_SIZE_CONFIG = "requestBodyStreamReadQueueSize"; public static final String HASHED_WHEEL_TIMER_TICK_DURATION = "hashedWheelTimerTickDuration"; public static final String HASHED_WHEEL_TIMER_SIZE = "hashedWheelTimerSize"; public static final String EXPIRED_COOKIE_EVICTION_DELAY = "expiredCookieEvictionDelay"; @@ -347,6 +350,21 @@ public static int defaultIoThreadsCount() { return threads; } + public static boolean defaultRequestBodyStreamReadOffloadEnabled() { + return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig() + .getBoolean(ASYNC_CLIENT_CONFIG_ROOT + REQUEST_BODY_STREAM_READ_OFFLOAD_ENABLED_CONFIG); + } + + public static int defaultRequestBodyStreamReadThreadsCount() { + return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig() + .getInt(ASYNC_CLIENT_CONFIG_ROOT + REQUEST_BODY_STREAM_READ_THREADS_COUNT_CONFIG); + } + + public static int defaultRequestBodyStreamReadQueueSize() { + return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig() + .getInt(ASYNC_CLIENT_CONFIG_ROOT + REQUEST_BODY_STREAM_READ_QUEUE_SIZE_CONFIG); + } + public static int defaultHashedWheelTimerTickDuration() { return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getInt(ASYNC_CLIENT_CONFIG_ROOT + HASHED_WHEEL_TIMER_TICK_DURATION); } diff --git a/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties index 5df97add5b..95f30d682d 100644 --- a/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties +++ b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties @@ -55,6 +55,9 @@ org.asynchttpclient.shutdownTimeout=PT15S org.asynchttpclient.useNativeTransport=false org.asynchttpclient.useOnlyEpollNativeTransport=false org.asynchttpclient.ioThreadsCount=-1 +org.asynchttpclient.requestBodyStreamReadOffloadEnabled=true +org.asynchttpclient.requestBodyStreamReadThreadsCount=-1 +org.asynchttpclient.requestBodyStreamReadQueueSize=0 org.asynchttpclient.hashedWheelTimerTickDuration=100 org.asynchttpclient.hashedWheelTimerSize=512 org.asynchttpclient.expiredCookieEvictionDelay=30000 diff --git a/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java b/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java index d125a9fa48..b9ce0cebd9 100644 --- a/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java +++ b/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java @@ -156,6 +156,24 @@ public void testDefaultHashedWheelTimerSize() { testIntegerSystemProperty("hashedWheelTimerSize", "defaultHashedWheelTimerSize", "512"); } + @RepeatedIfExceptionsTest(repeats = 5) + public void testDefaultRequestBodyStreamReadOffloadEnabled() { + assertTrue(AsyncHttpClientConfigDefaults.defaultRequestBodyStreamReadOffloadEnabled()); + testBooleanSystemProperty("requestBodyStreamReadOffloadEnabled", "defaultRequestBodyStreamReadOffloadEnabled", "false"); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void testDefaultRequestBodyStreamReadThreadsCount() { + assertEquals(AsyncHttpClientConfigDefaults.defaultRequestBodyStreamReadThreadsCount(), -1); + testIntegerSystemProperty("requestBodyStreamReadThreadsCount", "defaultRequestBodyStreamReadThreadsCount", "2"); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void testDefaultRequestBodyStreamReadQueueSize() { + assertEquals(AsyncHttpClientConfigDefaults.defaultRequestBodyStreamReadQueueSize(), 0); + testIntegerSystemProperty("requestBodyStreamReadQueueSize", "defaultRequestBodyStreamReadQueueSize", "17"); + } + private void testIntegerSystemProperty(String propertyName, String methodName, String value) { String previous = System.getProperty(ASYNC_CLIENT_CONFIG_ROOT + propertyName); System.setProperty(ASYNC_CLIENT_CONFIG_ROOT + propertyName, value); diff --git a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java index 79691007b3..3c1984290f 100755 --- a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java +++ b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java @@ -1050,7 +1050,11 @@ public void postInputStreamBodyGeneratorReadsOffEventLoop() throws Throwable { byte[] bodyBytes = "{}".getBytes(StandardCharsets.ISO_8859_1); EventLoopProbeInputStream bodyStream = new EventLoopProbeInputStream(bodyBytes); - try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(config().build())) { + try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(config() + .setThreadPoolName("ahc-body-read-test") + .setRequestBodyStreamReadThreadsCount(1) + .setRequestBodyStreamReadQueueSize(1) + .build())) { bodyStream.setEventLoopGroup(client.channelManager().getEventLoopGroup()); Response response = client.preparePost(getTargetUrl()) @@ -1064,6 +1068,36 @@ public void postInputStreamBodyGeneratorReadsOffEventLoop() throws Throwable { assertTrue(bodyStream.readAttempted.get(), "InputStream should have been read"); assertFalse(bodyStream.readOnEventLoop.get(), "InputStream.read must not run on an event-loop thread"); + String readThreadName = bodyStream.readThreadName.get(); + assertNotNull(readThreadName, "InputStream read thread should be recorded"); + assertTrue(readThreadName.contains("ahc-body-read-test-blocking-io"), + "InputStream.read should use the configured body-read thread pool"); + }); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void postInputStreamBodyGeneratorCanReadOnEventLoopWhenOffloadDisabled() throws Throwable { + withServer(server).run(server -> { + server.enqueueEcho(); + byte[] bodyBytes = "{}".getBytes(StandardCharsets.ISO_8859_1); + EventLoopProbeInputStream bodyStream = new EventLoopProbeInputStream(bodyBytes); + + try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(config() + .setRequestBodyStreamReadOffloadEnabled(false) + .build())) { + bodyStream.setEventLoopGroup(client.channelManager().getEventLoopGroup()); + + Response response = client.preparePost(getTargetUrl()) + .setBody(new InputStreamBodyGenerator(bodyStream, bodyBytes.length)) + .execute() + .get(TIMEOUT, SECONDS); + + assertEquals(200, response.getStatusCode()); + assertEquals("{}", response.getResponseBody()); + } + + assertTrue(bodyStream.readAttempted.get(), "InputStream should have been read"); + assertTrue(bodyStream.readOnEventLoop.get(), "disabled offload should preserve inline event-loop reads"); }); } @@ -1071,6 +1105,7 @@ private static final class EventLoopProbeInputStream extends InputStream { private final byte[] data; private final AtomicBoolean readAttempted = new AtomicBoolean(); private final AtomicBoolean readOnEventLoop = new AtomicBoolean(); + private final AtomicReference readThreadName = new AtomicReference<>(); private final AtomicReference eventLoopGroup = new AtomicReference<>(); private int position; @@ -1105,10 +1140,11 @@ public int read(byte[] buffer, int offset, int length) { private void recordReadThread() { readAttempted.set(true); EventLoopGroup group = eventLoopGroup.get(); + Thread currentThread = Thread.currentThread(); + readThreadName.compareAndSet(null, currentThread.getName()); if (group == null) { return; } - Thread currentThread = Thread.currentThread(); for (EventExecutor executor : group) { if (executor.inEventLoop(currentThread)) { readOnEventLoop.set(true); From 161642d763e33cd73c20c4e6b5351bb3c1be09ac Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 20 Jul 2026 12:59:18 +0200 Subject: [PATCH 3/8] Harden stream read offload Preserve the original synchronous stream writers when request body read offload is disabled, and remove an unnecessary event-loop scheduling hop from the enabled path. Convert unchecked InputStream failures into IOExceptions so suspended HTTP/1 and HTTP/2 uploads fail instead of hanging. Propagate the original HTTP/2 streaming failure, make close visibility explicit, and update the stale InputStream behavior expectation. Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../netty/request/NettyRequestFactory.java | 12 +- .../netty/request/body/Http2BodyWriter.java | 19 ++- .../netty/request/body/NettyBodyBody.java | 2 +- .../netty/request/body/NettyFileBody.java | 2 +- .../request/body/NettyInputStreamBody.java | 122 ++++++++++++------ .../org/asynchttpclient/BasicHttpTest.java | 33 +++++ .../Http2StreamingBodyFlowControlTest.java | 28 ++++ .../request/body/InputStreamTest.java | 5 +- 8 files changed, 169 insertions(+), 54 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestFactory.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestFactory.java index 41f0206052..cb695da285 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestFactory.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestFactory.java @@ -48,6 +48,7 @@ import org.asynchttpclient.uri.Uri; import org.asynchttpclient.util.StringUtils; +import java.io.InputStream; import java.nio.charset.Charset; import java.util.HashMap; import java.util.Iterator; @@ -175,7 +176,7 @@ private NettyBody body(Request request) { } else if (request.getByteBufData() != null) { nettyBody = new NettyByteBufBody(request.getByteBufData()); } else if (request.getStreamData() != null) { - nettyBody = new NettyInputStreamBody(request.getStreamData(), blockingBodyReadExecutor); + nettyBody = inputStreamBody(request.getStreamData(), -1L); } else if (isNonEmpty(request.getFormParams())) { CharSequence contentTypeOverride = request.getHeaders().contains(CONTENT_TYPE) ? null : HttpHeaderValues.APPLICATION_X_WWW_FORM_URLENCODED; nettyBody = new NettyByteBufferBody(urlEncodeFormParams(request.getFormParams(), bodyCharset), contentTypeOverride); @@ -188,8 +189,7 @@ private NettyBody body(Request request) { nettyBody = new NettyFileBody(fileBodyGenerator.getFile(), fileBodyGenerator.getRegionSeek(), fileBodyGenerator.getRegionLength(), config); } else if (request.getBodyGenerator() instanceof InputStreamBodyGenerator) { InputStreamBodyGenerator inStreamGenerator = (InputStreamBodyGenerator) request.getBodyGenerator(); - nettyBody = new NettyInputStreamBody(inStreamGenerator.getInputStream(), inStreamGenerator.getContentLength(), - blockingBodyReadExecutor); + nettyBody = inputStreamBody(inStreamGenerator.getInputStream(), inStreamGenerator.getContentLength()); } else if (request.getBodyGenerator() != null) { nettyBody = new NettyBodyBody(request.getBodyGenerator().createBody(), config); } @@ -197,6 +197,12 @@ private NettyBody body(Request request) { return nettyBody; } + private NettyInputStreamBody inputStreamBody(InputStream inputStream, long contentLength) { + return config.isRequestBodyStreamReadOffloadEnabled() + ? new NettyInputStreamBody(inputStream, contentLength, blockingBodyReadExecutor) + : new NettyInputStreamBody(inputStream, contentLength); + } + public void addAuthorizationHeader(HttpHeaders headers, String authorizationHeader) { if (authorizationHeader != null) { // don't override authorization but append diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java b/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java index 77ebdb8cf6..0540cf22bf 100644 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java @@ -23,6 +23,7 @@ import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.codec.http2.DefaultHttp2DataFrame; import io.netty.handler.codec.http2.Http2StreamChannel; +import org.asynchttpclient.netty.NettyResponseFuture; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -57,10 +58,8 @@ * returns), source cleanup ({@link ChunkSource#close()}) happens when the pump finishes — on success after * the last write completes, or on any read/write error. {@link #finish(Throwable)} is idempotent: it closes * the source, removes the transient writability handler, releases any unwritten chunk it still owns, and on - * error closes the stream channel. Closing the stream channel fires {@code channelInactive}, which - * AsyncHttpClient's {@code Http2Handler.handleChannelInactive} turns into a stream-scoped failure of the - * request future — matching how the rest of the HTTP/2 path signals stream errors without touching sibling - * multiplexed streams. + * error aborts the request future with the original cause and closes the stream channel, without touching + * sibling multiplexed streams. *

* Reference counting. Each chunk {@link ByteBuf} is owned by this writer from allocation * until it is wrapped in a {@link DefaultHttp2DataFrame} and handed to {@link Http2StreamChannel#write}, @@ -117,6 +116,7 @@ default void onResume(Runnable resume) { private final Http2StreamChannel channel; private final ChunkSource source; + private final NettyResponseFuture future; // Single buffered chunk read ahead so the final DATA frame can carry endStream=true. Owned by this // writer until written; released by finish() if still set when the pump ends. @@ -130,9 +130,10 @@ default void onResume(Runnable resume) { // Set while the pump is parked on a ChunkSource.SUSPEND, to ignore spurious/duplicate feed resumes. private boolean suspended; - private Http2BodyWriter(Http2StreamChannel channel, ChunkSource source) { + private Http2BodyWriter(Http2StreamChannel channel, ChunkSource source, NettyResponseFuture future) { this.channel = channel; this.source = source; + this.future = future; source.onResume(this::resumeFromSuspend); // Guarantee cleanup if the stream is closed out from under a PARKED pump — i.e. one waiting on // channelWritabilityChanged (flow-control window exhausted) or on a feed (ChunkSource.SUSPEND). @@ -165,8 +166,8 @@ private void resumeFromSuspend() { * asynchronously on the channel's event loop. The caller (which has already written the HEADERS frame * with {@code endStream=false}) must not write further frames on this stream. */ - static void start(Http2StreamChannel channel, ChunkSource source) { - Http2BodyWriter writer = new Http2BodyWriter(channel, source); + static void start(Http2StreamChannel channel, ChunkSource source, NettyResponseFuture future) { + Http2BodyWriter writer = new Http2BodyWriter(channel, source, future); if (channel.eventLoop().inEventLoop()) { writer.pump(); } else { @@ -295,9 +296,7 @@ private void finish(Throwable cause) { if (cause != null) { LOGGER.debug("HTTP/2 request body streaming failed; closing stream", cause); - // Signal the failure the way the rest of the HTTP/2 path does: closing the stream child channel - // fires channelInactive -> Http2Handler.handleChannelInactive -> streamFailed, which aborts this - // stream's future without disturbing sibling multiplexed streams. + future.abort(cause); channel.close(); } } diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyBodyBody.java b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyBodyBody.java index 0b6192f725..d1af1d7453 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyBodyBody.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyBodyBody.java @@ -100,7 +100,7 @@ public void writeHttp2(Http2StreamChannel channel, NettyResponseFuture future // (closeSilently(body)) happens when the async pump completes — see BodyChunkSource.close. BodyGenerator bg = future.getTargetRequest().getBodyGenerator(); FeedableBodyGenerator feedable = bg instanceof FeedableBodyGenerator ? (FeedableBodyGenerator) bg : null; - Http2BodyWriter.start(channel, new BodyChunkSource(body, feedable)); + Http2BodyWriter.start(channel, new BodyChunkSource(body, feedable), future); } /** diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyFileBody.java b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyFileBody.java index 66ee644fa6..4dee1979df 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyFileBody.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyFileBody.java @@ -83,7 +83,7 @@ public void writeHttp2(Http2StreamChannel channel, NettyResponseFuture future // backpressure, so a large file upload does not buffer in heap or read the whole file inline on the // event loop. The file is opened here so an open failure still surfaces synchronously to the caller's // openHttp2Stream catch; cleanup happens when the async pump completes (see FileChunkSource.close). - Http2BodyWriter.start(channel, new FileChunkSource(file, offset, length, config.getChunkedFileChunkSize())); + Http2BodyWriter.start(channel, new FileChunkSource(file, offset, length, config.getChunkedFileChunkSize()), future); } /** diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java index 9ce69b1fed..fa24320100 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java @@ -22,6 +22,7 @@ import io.netty.handler.codec.http.LastHttpContent; import io.netty.handler.codec.http2.Http2StreamChannel; import io.netty.handler.stream.ChunkedInput; +import io.netty.handler.stream.ChunkedStream; import io.netty.handler.stream.ChunkedWriteHandler; import io.netty.util.concurrent.EventExecutor; import org.asynchttpclient.netty.NettyResponseFuture; @@ -44,13 +45,14 @@ public class NettyInputStreamBody implements NettyBody { private final InputStream inputStream; private final long contentLength; private final Executor blockingBodyReadExecutor; + private final boolean offloadReads; public NettyInputStreamBody(InputStream inputStream) { this(inputStream, -1L); } public NettyInputStreamBody(InputStream inputStream, long contentLength) { - this(inputStream, contentLength, Runnable::run); + this(inputStream, contentLength, Runnable::run, false); } public NettyInputStreamBody(InputStream inputStream, Executor blockingBodyReadExecutor) { @@ -58,9 +60,15 @@ public NettyInputStreamBody(InputStream inputStream, Executor blockingBodyReadEx } public NettyInputStreamBody(InputStream inputStream, long contentLength, Executor blockingBodyReadExecutor) { + this(inputStream, contentLength, blockingBodyReadExecutor, true); + } + + private NettyInputStreamBody(InputStream inputStream, long contentLength, Executor blockingBodyReadExecutor, + boolean offloadReads) { this.inputStream = inputStream; this.contentLength = contentLength; this.blockingBodyReadExecutor = requireNonNull(blockingBodyReadExecutor, "blockingBodyReadExecutor"); + this.offloadReads = offloadReads; } public InputStream getInputStream() { @@ -87,15 +95,21 @@ public void write(Channel channel, NettyResponseFuture future) throws IOExcep future.setStreamConsumed(true); } - ChunkedWriteHandler chunkedWriteHandler = requireNonNull(channel.pipeline().get(ChunkedWriteHandler.class), - "chunkedWriteHandler"); - OffloadedInputStreamChunkedInput chunkedInput = new OffloadedInputStreamChunkedInput(is, getContentLength(), - channel.eventLoop(), chunkedWriteHandler, blockingBodyReadExecutor); - channel.write(chunkedInput, channel.newProgressivePromise()).addListener( + ChunkedInput chunkedInput; + if (offloadReads) { + ChunkedWriteHandler chunkedWriteHandler = requireNonNull(channel.pipeline().get(ChunkedWriteHandler.class), + "chunkedWriteHandler"); + chunkedInput = new OffloadedInputStreamChunkedInput(is, getContentLength(), channel.eventLoop(), + chunkedWriteHandler, blockingBodyReadExecutor); + } else { + chunkedInput = new ChunkedStream(is); + } + ChunkedInput input = chunkedInput; + channel.write(input, channel.newProgressivePromise()).addListener( new WriteProgressListener(future, false, getContentLength()) { @Override public void operationComplete(ChannelProgressiveFuture cf) { - chunkedInput.close(); + closeChunkedInput(input); super.operationComplete(cf); } }); @@ -122,18 +136,66 @@ public void writeHttp2(Http2StreamChannel channel, NettyResponseFuture future future.setStreamConsumed(true); } - // Stream the InputStream one bounded chunk at a time with HTTP/2 flow control / writability - // backpressure, so a large upload does not buffer the whole stream in heap or read it all inline on - // the event loop. Cleanup (closeSilently) happens when the async pump completes — see - // InputStreamChunkSource.close. - Http2BodyWriter.start(channel, new InputStreamChunkSource(is, channel.eventLoop(), blockingBodyReadExecutor)); + Http2BodyWriter.ChunkSource source = offloadReads + ? new OffloadedInputStreamChunkSource(is, channel.eventLoop(), blockingBodyReadExecutor) + : new DirectInputStreamChunkSource(is); + Http2BodyWriter.start(channel, source, future); + } + + private static void closeChunkedInput(ChunkedInput input) { + try { + input.close(); + } catch (Exception e) { + LOGGER.debug("Failed to close request body stream", e); + } + } + + private static IOException readFailure(RuntimeException cause) { + return new IOException("Request body InputStream read failed", cause); + } + + private static final class DirectInputStreamChunkSource implements Http2BodyWriter.ChunkSource { + + private static final int CHUNK_SIZE = 8192; + + private final InputStream is; + private final byte[] buffer = new byte[CHUNK_SIZE]; + + DirectInputStreamChunkSource(InputStream is) { + this.is = is; + } + + @Override + public ByteBuf nextChunk(ByteBufAllocator alloc) throws IOException { + int read; + do { + read = is.read(buffer); + } while (read == 0); + + if (read < 0) { + return null; + } + ByteBuf buf = alloc.buffer(read); + try { + buf.writeBytes(buffer, 0, read); + return buf; + } catch (RuntimeException e) { + buf.release(); + throw e; + } + } + + @Override + public void close() { + closeSilently(is); + } } /** * Reads an {@link InputStream} off the event loop and exposes completed chunks to * {@link Http2BodyWriter}. */ - private static final class InputStreamChunkSource implements Http2BodyWriter.ChunkSource { + private static final class OffloadedInputStreamChunkSource implements Http2BodyWriter.ChunkSource { private static final int CHUNK_SIZE = 8192; @@ -143,12 +205,12 @@ private static final class InputStreamChunkSource implements Http2BodyWriter.Chu private final byte[] buffer = new byte[CHUNK_SIZE]; private boolean readInProgress; private boolean endOfInput; - private boolean closed; + private volatile boolean closed; private int readableBytes; private IOException failure; private Runnable resume; - InputStreamChunkSource(InputStream is, EventExecutor eventLoop, Executor blockingBodyReadExecutor) { + OffloadedInputStreamChunkSource(InputStream is, EventExecutor eventLoop, Executor blockingBodyReadExecutor) { this.is = is; this.eventLoop = eventLoop; this.blockingBodyReadExecutor = blockingBodyReadExecutor; @@ -192,19 +254,11 @@ public void close() { private void submitReadAfterSuspend() throws IOException { readInProgress = true; - try { - eventLoop.execute(this::submitRead); - } catch (RejectedExecutionException e) { - readInProgress = false; - throw new IOException("HTTP/2 request body read executor is unavailable", e); - } - } - - private void submitRead() { try { blockingBodyReadExecutor.execute(this::readOffEventLoop); } catch (RejectedExecutionException e) { - completeRead(-1, new IOException("HTTP/2 request body read executor rejected the read", e)); + readInProgress = false; + throw new IOException("HTTP/2 request body read executor rejected the read", e); } } @@ -215,6 +269,8 @@ private void readOffEventLoop() { read = read(); } catch (IOException e) { thrown = e; + } catch (RuntimeException e) { + thrown = readFailure(e); } completeRead(read, thrown); } @@ -263,7 +319,7 @@ private static final class OffloadedInputStreamChunkedInput implements ChunkedIn private final byte[] buffer = new byte[CHUNK_SIZE]; private boolean readInProgress; private boolean endOfInput; - private boolean closed; + private volatile boolean closed; private int readableBytes; private long progress; private IOException failure; @@ -335,19 +391,11 @@ public long progress() { private void submitReadAfterSuspend() throws IOException { readInProgress = true; - try { - eventLoop.execute(this::submitRead); - } catch (RejectedExecutionException e) { - readInProgress = false; - throw new IOException("HTTP/1 request body read executor is unavailable", e); - } - } - - private void submitRead() { try { blockingBodyReadExecutor.execute(this::readOffEventLoop); } catch (RejectedExecutionException e) { - completeRead(-1, new IOException("HTTP/1 request body read executor rejected the read", e)); + readInProgress = false; + throw new IOException("HTTP/1 request body read executor rejected the read", e); } } @@ -358,6 +406,8 @@ private void readOffEventLoop() { read = read(); } catch (IOException e) { thrown = e; + } catch (RuntimeException e) { + thrown = readFailure(e); } completeRead(read, thrown); } diff --git a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java index 3c1984290f..5bd32cde0f 100755 --- a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java +++ b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java @@ -1101,6 +1101,34 @@ public void postInputStreamBodyGeneratorCanReadOnEventLoopWhenOffloadDisabled() }); } + @RepeatedIfExceptionsTest(repeats = 5) + public void postInputStreamRuntimeFailureFailsRequest() throws Throwable { + withServer(server).run(server -> { + server.enqueueEcho(); + InputStream bodyStream = new InputStream() { + @Override + public int read() { + throw new IllegalStateException("stream read failed"); + } + + @Override + public int read(byte[] buffer, int offset, int length) { + throw new IllegalStateException("stream read failed"); + } + }; + + try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient()) { + ExecutionException failure = assertThrows(ExecutionException.class, () -> client.preparePost(getTargetUrl()) + .setBody(new InputStreamBodyGenerator(bodyStream, 1)) + .execute() + .get(TIMEOUT, SECONDS)); + + IOException readFailure = assertInstanceOf(IOException.class, failure.getCause()); + assertInstanceOf(IllegalStateException.class, readFailure.getCause()); + } + }); + } + private static final class EventLoopProbeInputStream extends InputStream { private final byte[] data; private final AtomicBoolean readAttempted = new AtomicBoolean(); @@ -1137,6 +1165,11 @@ public int read(byte[] buffer, int offset, int length) { return read; } + @Override + public int available() { + return data.length - position; + } + private void recordReadThread() { readAttempted.set(true); EventLoopGroup group = eventLoopGroup.get(); diff --git a/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java b/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java index 85728fa15a..e71fa6de19 100644 --- a/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java +++ b/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java @@ -73,6 +73,8 @@ import static org.asynchttpclient.Dsl.config; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -359,6 +361,32 @@ public void inputStreamBodyReadsOffEventLoopOverHttp2() throws Exception { assertFalse(is.readOnEventLoop.get(), "InputStream.read must not run on an event-loop thread"); } + @Test + public void inputStreamRuntimeFailureFailsRequestOverHttp2() throws Exception { + startServer(-1); + InputStream is = new InputStream() { + @Override + public int read() { + throw new IllegalStateException("stream read failed"); + } + + @Override + public int read(byte[] buffer, int offset, int length) { + throw new IllegalStateException("stream read failed"); + } + }; + + try (AsyncHttpClient client = http2Client()) { + ExecutionException failure = assertThrows(ExecutionException.class, () -> client.preparePost(httpsUrl("/upload")) + .setBody(new InputStreamBodyGenerator(is, 1)) + .execute() + .get(30, SECONDS)); + + IOException readFailure = assertInstanceOf(IOException.class, failure.getCause()); + assertInstanceOf(IllegalStateException.class, readFailure.getCause()); + } + } + // ========================================================================= // NettyFileBody — large streaming upload via setBody(File) // ========================================================================= diff --git a/client/src/test/java/org/asynchttpclient/request/body/InputStreamTest.java b/client/src/test/java/org/asynchttpclient/request/body/InputStreamTest.java index 55cff4323d..916b4e0c53 100644 --- a/client/src/test/java/org/asynchttpclient/request/body/InputStreamTest.java +++ b/client/src/test/java/org/asynchttpclient/request/body/InputStreamTest.java @@ -76,9 +76,8 @@ public int read() { Response resp = client.preparePost(getTargetUrl()).setHeaders(httpHeaders).setBody(inputStream).execute().get(); assertNotNull(resp); - // TODO: 18-11-2022 Revisit - assertEquals(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, resp.getStatusCode()); -// assertEquals(resp.getHeader("X-Param"), "abc"); + assertEquals(HttpServletResponse.SC_OK, resp.getStatusCode()); + assertEquals("abc", resp.getHeader("X-Param")); } } From 928067cc61abf178c2e4e0e7963b845cce26a478 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 20 Jul 2026 13:53:38 +0200 Subject: [PATCH 4/8] Prevent reentrant HTTP/2 body pump Guard the HTTP/2 streaming pump against synchronous writability callbacks fired from flush, and keep it parked while the terminal DATA frame is pending. This prevents concurrent pump state mutation and duplicate endStream frames under constrained flow control. Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../netty/request/body/Http2BodyWriter.java | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java b/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java index 0540cf22bf..e6c759b9fa 100644 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java @@ -123,6 +123,11 @@ default void onResume(Runnable resume) { private ByteBuf pending; private boolean done; + // flush() can synchronously fire channelWritabilityChanged. Prevent that callback from re-entering the + // pump, and prevent later callbacks from writing a second endStream frame before the first one completes. + private boolean pumping; + private boolean terminalWritePending; + // Transient handler that resumes the pump when the channel becomes writable again. Added lazily the // first time the pump parks, removed by finish(). private WritabilityResumeHandler resumeHandler; @@ -180,9 +185,10 @@ static void start(Http2StreamChannel channel, ChunkSource source, NettyResponseF * {@code channelWritabilityChanged}) or the body is exhausted. Always runs on the event loop. */ private void pump() { - if (done) { + if (done || pumping || terminalWritePending) { return; } + pumping = true; try { while (true) { if (done) { @@ -209,6 +215,7 @@ private void pump() { ByteBuf terminal = last != null ? last // Empty body — preserve existing behaviour: a single empty DATA frame ends the stream. : channel.alloc().buffer(0); + terminalWritePending = true; writeLastFrame(terminal); channel.flush(); return; @@ -233,6 +240,8 @@ private void pump() { } } catch (Throwable t) { finish(t); + } finally { + pumping = false; } } @@ -307,7 +316,7 @@ private void finish(Throwable cause) { private final class WritabilityResumeHandler extends ChannelInboundHandlerAdapter { @Override public void channelWritabilityChanged(ChannelHandlerContext ctx) { - if (!done && ctx.channel().isWritable()) { + if (!done && !terminalWritePending && ctx.channel().isWritable()) { pump(); } ctx.fireChannelWritabilityChanged(); From c4e96fac78b399d507ac43d81cd10b29b895b4ae Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 20 Jul 2026 14:32:43 +0200 Subject: [PATCH 5/8] Disable stream read offload by default Keep request body stream read offloading opt-in so existing clients retain inline reads unless the worker pool is explicitly enabled. Update the HTTP/1.1 and HTTP/2 tests to distinguish default behavior from the enabled offload path. Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../org/asynchttpclient/AsyncHttpClientConfig.java | 7 +++++-- .../asynchttpclient/config/ahc-default.properties | 2 +- .../AsyncHttpClientDefaultsTest.java | 5 +++-- .../java/org/asynchttpclient/BasicHttpTest.java | 13 +++++++------ .../Http2StreamingBodyFlowControlTest.java | 1 + 5 files changed, 17 insertions(+), 11 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index 88cdd89be6..3a95e1a72d 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java @@ -167,12 +167,13 @@ public interface AsyncHttpClientConfig { /** * Returns whether blocking reads from request body {@link java.io.InputStream} instances should be - * offloaded from Netty event-loop threads. + * offloaded from Netty event-loop threads. Disabled by default. * * @return {@code true} if blocking request body stream reads should be offloaded + * @since 3.0.12 */ default boolean isRequestBodyStreamReadOffloadEnabled() { - return true; + return false; } /** @@ -180,6 +181,7 @@ default boolean isRequestBodyStreamReadOffloadEnabled() { * * @return the configured request body stream read thread count. Values less than or equal to {@code 0} * use {@link #getIoThreadsCount()}. + * @since 3.0.12 */ default int getRequestBodyStreamReadThreadsCount() { return -1; @@ -190,6 +192,7 @@ default int getRequestBodyStreamReadThreadsCount() { * * @return the configured request body stream read queue size. Values less than or equal to {@code 0} use * the default queue size. + * @since 3.0.12 */ default int getRequestBodyStreamReadQueueSize() { return 0; diff --git a/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties index 95f30d682d..df8c81c807 100644 --- a/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties +++ b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties @@ -55,7 +55,7 @@ org.asynchttpclient.shutdownTimeout=PT15S org.asynchttpclient.useNativeTransport=false org.asynchttpclient.useOnlyEpollNativeTransport=false org.asynchttpclient.ioThreadsCount=-1 -org.asynchttpclient.requestBodyStreamReadOffloadEnabled=true +org.asynchttpclient.requestBodyStreamReadOffloadEnabled=false org.asynchttpclient.requestBodyStreamReadThreadsCount=-1 org.asynchttpclient.requestBodyStreamReadQueueSize=0 org.asynchttpclient.hashedWheelTimerTickDuration=100 diff --git a/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java b/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java index b9ce0cebd9..b5d84d1869 100644 --- a/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java +++ b/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java @@ -158,8 +158,9 @@ public void testDefaultHashedWheelTimerSize() { @RepeatedIfExceptionsTest(repeats = 5) public void testDefaultRequestBodyStreamReadOffloadEnabled() { - assertTrue(AsyncHttpClientConfigDefaults.defaultRequestBodyStreamReadOffloadEnabled()); - testBooleanSystemProperty("requestBodyStreamReadOffloadEnabled", "defaultRequestBodyStreamReadOffloadEnabled", "false"); + assertFalse(AsyncHttpClientConfigDefaults.defaultRequestBodyStreamReadOffloadEnabled()); + testBooleanSystemProperty("requestBodyStreamReadOffloadEnabled", "defaultRequestBodyStreamReadOffloadEnabled", "true"); + AsyncHttpClientConfigHelper.reloadProperties(); } @RepeatedIfExceptionsTest(repeats = 5) diff --git a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java index 5bd32cde0f..b153284223 100755 --- a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java +++ b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java @@ -1052,6 +1052,7 @@ public void postInputStreamBodyGeneratorReadsOffEventLoop() throws Throwable { try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(config() .setThreadPoolName("ahc-body-read-test") + .setRequestBodyStreamReadOffloadEnabled(true) .setRequestBodyStreamReadThreadsCount(1) .setRequestBodyStreamReadQueueSize(1) .build())) { @@ -1076,15 +1077,13 @@ public void postInputStreamBodyGeneratorReadsOffEventLoop() throws Throwable { } @RepeatedIfExceptionsTest(repeats = 5) - public void postInputStreamBodyGeneratorCanReadOnEventLoopWhenOffloadDisabled() throws Throwable { + public void postInputStreamBodyGeneratorReadsOnEventLoopByDefault() throws Throwable { withServer(server).run(server -> { server.enqueueEcho(); byte[] bodyBytes = "{}".getBytes(StandardCharsets.ISO_8859_1); EventLoopProbeInputStream bodyStream = new EventLoopProbeInputStream(bodyBytes); - try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(config() - .setRequestBodyStreamReadOffloadEnabled(false) - .build())) { + try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(config().build())) { bodyStream.setEventLoopGroup(client.channelManager().getEventLoopGroup()); Response response = client.preparePost(getTargetUrl()) @@ -1097,7 +1096,7 @@ public void postInputStreamBodyGeneratorCanReadOnEventLoopWhenOffloadDisabled() } assertTrue(bodyStream.readAttempted.get(), "InputStream should have been read"); - assertTrue(bodyStream.readOnEventLoop.get(), "disabled offload should preserve inline event-loop reads"); + assertTrue(bodyStream.readOnEventLoop.get(), "default configuration should preserve inline event-loop reads"); }); } @@ -1117,7 +1116,9 @@ public int read(byte[] buffer, int offset, int length) { } }; - try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient()) { + try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(config() + .setRequestBodyStreamReadOffloadEnabled(true) + .build())) { ExecutionException failure = assertThrows(ExecutionException.class, () -> client.preparePost(getTargetUrl()) .setBody(new InputStreamBodyGenerator(bodyStream, 1)) .execute() diff --git a/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java b/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java index e71fa6de19..e4bcda068b 100644 --- a/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java +++ b/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java @@ -302,6 +302,7 @@ private AsyncHttpClient http2Client() { .setUseInsecureTrustManager(true) .setHttp2Enabled(true) .setMaxConnectionsPerHost(1) + .setRequestBodyStreamReadOffloadEnabled(true) .setRequestTimeout(Duration.ofSeconds(60))); } From eb08bafdd12823faf1c06682782442710da63886 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 20 Jul 2026 14:44:41 +0200 Subject: [PATCH 6/8] Enable offload in stream regression test Keep the invalid InputStream regression focused on the new body-read implementation after making that implementation opt-in. A default client correctly retains the legacy response behavior and should not be used for this assertion. Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../org/asynchttpclient/request/body/InputStreamTest.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/client/src/test/java/org/asynchttpclient/request/body/InputStreamTest.java b/client/src/test/java/org/asynchttpclient/request/body/InputStreamTest.java index 916b4e0c53..d4e6d8bd7f 100644 --- a/client/src/test/java/org/asynchttpclient/request/body/InputStreamTest.java +++ b/client/src/test/java/org/asynchttpclient/request/body/InputStreamTest.java @@ -34,6 +34,7 @@ import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE; import static org.asynchttpclient.Dsl.asyncHttpClient; +import static org.asynchttpclient.Dsl.config; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -47,7 +48,8 @@ public AbstractHandler configureHandler() throws Exception { @RepeatedIfExceptionsTest(repeats = 5) public void testInvalidInputStream() throws Exception { - try (AsyncHttpClient client = asyncHttpClient()) { + try (AsyncHttpClient client = asyncHttpClient(config() + .setRequestBodyStreamReadOffloadEnabled(true))) { HttpHeaders httpHeaders = new DefaultHttpHeaders().add(CONTENT_TYPE, HttpHeaderValues.APPLICATION_X_WWW_FORM_URLENCODED); InputStream inputStream = new InputStream() { From 6f5165634fe735872407b8564d4a19d79356cdef Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 20 Jul 2026 16:56:19 +0200 Subject: [PATCH 7/8] Harden queued body read cancellation Skip worker reads when cancellation closes a body stream before its queued task starts. This prevents closed streams from being touched and keeps cancelled uploads from consuming executor capacity. Add deterministic HTTP/1.1 and HTTP/2 queue tests, plus an HTTP/2 sibling stream test that proves a blocked upload does not stall another stream. Document executor-taking constructors as internal client wiring. Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../DefaultAsyncHttpClient.java | 5 + .../netty/request/NettyRequestSender.java | 12 ++ .../request/body/NettyInputStreamBody.java | 25 ++++ .../org/asynchttpclient/BasicHttpTest.java | 47 +++++++ .../Http2StreamingBodyFlowControlTest.java | 73 +++++++++++ .../OffloadedBodyReadTestSupport.java | 117 ++++++++++++++++++ 6 files changed, 279 insertions(+) create mode 100644 client/src/test/java/org/asynchttpclient/OffloadedBodyReadTestSupport.java diff --git a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java index ff0072343d..2a07319ca5 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java +++ b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java @@ -134,6 +134,11 @@ ChannelManager channelManager() { return channelManager; } + // Visible for testing + @Nullable ExecutorService blockingBodyReadExecutor() { + return blockingBodyReadExecutor; + } + private static Timer newNettyTimer(AsyncHttpClientConfig config) { ThreadFactory threadFactory = config.getThreadFactory() != null ? config.getThreadFactory() : new DefaultThreadFactory(config.getThreadPoolName() + "-timer"); HashedWheelTimer timer = new HashedWheelTimer(threadFactory, config.getHashedWheelTimerTickDuration(), TimeUnit.MILLISECONDS, config.getHashedWheelTimerSize()); diff --git a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java index 4b8288105e..fe22fa77cc 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -130,6 +130,18 @@ public NettyRequestSender(AsyncHttpClientConfig config, ChannelManager channelMa this(config, channelManager, nettyTimer, clientState, Runnable::run); } + /** + * Creates the request sender with the client-owned executor used for blocking request body reads. + * This overload is public only for cross-package client wiring; applications should configure the + * executor through {@link AsyncHttpClientConfig}. + * + * @param config client configuration + * @param channelManager channel manager + * @param nettyTimer request timer + * @param clientState client lifecycle state + * @param blockingBodyReadExecutor executor for blocking request body reads + * @since 3.0.12 + */ public NettyRequestSender(AsyncHttpClientConfig config, ChannelManager channelManager, Timer nettyTimer, AsyncHttpClientState clientState, Executor blockingBodyReadExecutor) { this.config = config; diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java index fa24320100..896bd09dd4 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java @@ -55,10 +55,29 @@ public NettyInputStreamBody(InputStream inputStream, long contentLength) { this(inputStream, contentLength, Runnable::run, false); } + /** + * Creates a request body whose stream reads are dispatched through the supplied executor. + * This overload supports internal client wiring; applications should enable offloading through + * {@link org.asynchttpclient.AsyncHttpClientConfig}. + * + * @param inputStream request body stream + * @param blockingBodyReadExecutor executor for blocking reads + * @since 3.0.12 + */ public NettyInputStreamBody(InputStream inputStream, Executor blockingBodyReadExecutor) { this(inputStream, -1L, blockingBodyReadExecutor); } + /** + * Creates a request body whose stream reads are dispatched through the supplied executor. + * This overload supports internal client wiring; applications should enable offloading through + * {@link org.asynchttpclient.AsyncHttpClientConfig}. + * + * @param inputStream request body stream + * @param contentLength request body length, or {@code -1} when unknown + * @param blockingBodyReadExecutor executor for blocking reads + * @since 3.0.12 + */ public NettyInputStreamBody(InputStream inputStream, long contentLength, Executor blockingBodyReadExecutor) { this(inputStream, contentLength, blockingBodyReadExecutor, true); } @@ -263,6 +282,9 @@ private void submitReadAfterSuspend() throws IOException { } private void readOffEventLoop() { + if (closed) { + return; + } int read = -1; IOException thrown = null; try { @@ -400,6 +422,9 @@ private void submitReadAfterSuspend() throws IOException { } private void readOffEventLoop() { + if (closed) { + return; + } int read = -1; IOException thrown = null; try { diff --git a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java index b153284223..6a2e7fac47 100755 --- a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java +++ b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java @@ -55,6 +55,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -66,6 +67,9 @@ import static io.netty.handler.codec.http.HttpHeaderNames.TRANSFER_ENCODING; import static java.nio.charset.StandardCharsets.UTF_8; import static java.util.concurrent.TimeUnit.SECONDS; +import static org.asynchttpclient.OffloadedBodyReadTestSupport.BlockingInputStream; +import static org.asynchttpclient.OffloadedBodyReadTestSupport.CloseProbeInputStream; +import static org.asynchttpclient.OffloadedBodyReadTestSupport.awaitExecutorState; import static org.asynchttpclient.Dsl.config; import static org.asynchttpclient.Dsl.get; import static org.asynchttpclient.Dsl.head; @@ -1130,6 +1134,48 @@ public int read(byte[] buffer, int offset, int length) { }); } + @RepeatedIfExceptionsTest(repeats = 5) + public void cancelledQueuedInputStreamIsNotReadOverHttp1() throws Throwable { + withServer(server).run(server -> { + server.enqueueEcho(); + server.enqueueEcho(); + byte[] bodyBytes = "{}".getBytes(StandardCharsets.ISO_8859_1); + BlockingInputStream blockingStream = new BlockingInputStream(bodyBytes); + CloseProbeInputStream queuedStream = new CloseProbeInputStream(); + + try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(config() + .setRequestBodyStreamReadOffloadEnabled(true) + .setRequestBodyStreamReadThreadsCount(1) + .setRequestBodyStreamReadQueueSize(1) + .build())) { + ThreadPoolExecutor executor = assertInstanceOf(ThreadPoolExecutor.class, + client.blockingBodyReadExecutor()); + ListenableFuture blockingUpload = client.preparePost(getTargetUrl()) + .setBody(new InputStreamBodyGenerator(blockingStream, bodyBytes.length)) + .execute(); + assertTrue(blockingStream.readStarted.await(5, SECONDS), + "first body read should occupy the worker"); + + ListenableFuture queuedUpload = client.preparePost(getTargetUrl()) + .setBody(new InputStreamBodyGenerator(queuedStream, 1)) + .execute(); + awaitExecutorState(executor, 1, 1); + + assertTrue(queuedUpload.cancel(true), "queued request should be cancelled"); + assertTrue(queuedStream.closed.await(5, SECONDS), + "cancellation should close the queued stream"); + + blockingStream.releaseRead.countDown(); + assertEquals(200, blockingUpload.get(TIMEOUT, SECONDS).getStatusCode()); + awaitExecutorState(executor, 0, 0); + assertFalse(queuedStream.readAttempted.get(), + "a queued read must not start after cancellation closed its stream"); + } finally { + blockingStream.releaseRead.countDown(); + } + }); + } + private static final class EventLoopProbeInputStream extends InputStream { private final byte[] data; private final AtomicBoolean readAttempted = new AtomicBoolean(); @@ -1186,4 +1232,5 @@ private void recordReadThread() { } } } + } diff --git a/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java b/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java index e4bcda068b..2978a56ac1 100644 --- a/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java +++ b/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java @@ -61,6 +61,7 @@ import java.time.Duration; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; @@ -71,6 +72,9 @@ import static java.util.concurrent.TimeUnit.SECONDS; import static org.asynchttpclient.Dsl.asyncHttpClient; import static org.asynchttpclient.Dsl.config; +import static org.asynchttpclient.OffloadedBodyReadTestSupport.BlockingInputStream; +import static org.asynchttpclient.OffloadedBodyReadTestSupport.CloseProbeInputStream; +import static org.asynchttpclient.OffloadedBodyReadTestSupport.awaitExecutorState; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; @@ -362,6 +366,74 @@ public void inputStreamBodyReadsOffEventLoopOverHttp2() throws Exception { assertFalse(is.readOnEventLoop.get(), "InputStream.read must not run on an event-loop thread"); } + @Test + public void blockedInputStreamDoesNotStallSiblingStreamOverHttp2() throws Exception { + startServer(-1); + byte[] payload = deterministicPayload(SMALL_SIZE); + BlockingInputStream bodyStream = new BlockingInputStream(payload); + + try (AsyncHttpClient client = http2Client()) { + ListenableFuture blockedUpload = client.preparePost(httpsUrl("/blocked")) + .setBody(new InputStreamBodyGenerator(bodyStream, payload.length)) + .execute(); + assertTrue(bodyStream.readStarted.await(5, SECONDS), "blocking body read should start"); + + Response sibling = client.prepareGet(httpsUrl("/sibling")) + .execute() + .get(5, SECONDS); + assertEquals(200, sibling.getStatusCode(), + "a sibling stream should complete while the upload read is blocked"); + + bodyStream.releaseRead.countDown(); + assertEchoed(blockedUpload.get(30, SECONDS), payload); + } finally { + bodyStream.releaseRead.countDown(); + } + } + + @Test + public void cancelledQueuedInputStreamIsNotReadOverHttp2() throws Exception { + startServer(-1); + byte[] payload = deterministicPayload(SMALL_SIZE); + BlockingInputStream blockingStream = new BlockingInputStream(payload); + CloseProbeInputStream queuedStream = new CloseProbeInputStream(); + + DefaultAsyncHttpClientConfig clientConfig = config() + .setUseInsecureTrustManager(true) + .setHttp2Enabled(true) + .setMaxConnectionsPerHost(1) + .setRequestBodyStreamReadOffloadEnabled(true) + .setRequestBodyStreamReadThreadsCount(1) + .setRequestBodyStreamReadQueueSize(1) + .setRequestTimeout(Duration.ofSeconds(30)) + .build(); + + try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(clientConfig)) { + ThreadPoolExecutor executor = assertInstanceOf(ThreadPoolExecutor.class, + client.blockingBodyReadExecutor()); + ListenableFuture blockingUpload = client.preparePost(httpsUrl("/blocking")) + .setBody(new InputStreamBodyGenerator(blockingStream, payload.length)) + .execute(); + assertTrue(blockingStream.readStarted.await(5, SECONDS), "first body read should occupy the worker"); + + ListenableFuture queuedUpload = client.preparePost(httpsUrl("/queued")) + .setBody(new InputStreamBodyGenerator(queuedStream, 1)) + .execute(); + awaitExecutorState(executor, 1, 1); + + assertTrue(queuedUpload.cancel(true), "queued request should be cancelled"); + assertTrue(queuedStream.closed.await(5, SECONDS), "cancellation should close the queued stream"); + + blockingStream.releaseRead.countDown(); + assertEchoed(blockingUpload.get(30, SECONDS), payload); + awaitExecutorState(executor, 0, 0); + assertFalse(queuedStream.readAttempted.get(), + "a queued read must not start after cancellation closed its stream"); + } finally { + blockingStream.releaseRead.countDown(); + } + } + @Test public void inputStreamRuntimeFailureFailsRequestOverHttp2() throws Exception { startServer(-1); @@ -587,6 +659,7 @@ private void recordReadThread() { } } + // Lifecycle (finding #2 from the cold audit): when the stream is closed while the body pump is parked on // SUSPEND (a streaming/feedable body with no data yet), no in-flight write's failure would run cleanup — // so the writer must close the body source from the stream's closeFuture, or the source (and any file diff --git a/client/src/test/java/org/asynchttpclient/OffloadedBodyReadTestSupport.java b/client/src/test/java/org/asynchttpclient/OffloadedBodyReadTestSupport.java new file mode 100644 index 0000000000..76f98cec42 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/OffloadedBodyReadTestSupport.java @@ -0,0 +1,117 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed 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.asynchttpclient; + +import java.io.IOException; +import java.io.InputStream; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertEquals; + +final class OffloadedBodyReadTestSupport { + + private OffloadedBodyReadTestSupport() { + } + + static void awaitExecutorState(ThreadPoolExecutor executor, int queued, int active) + throws InterruptedException { + long deadline = System.nanoTime() + SECONDS.toNanos(5); + while ((executor.getQueue().size() != queued || executor.getActiveCount() != active) + && System.nanoTime() < deadline) { + new CountDownLatch(1).await(10, MILLISECONDS); + } + assertEquals(queued, executor.getQueue().size(), "unexpected body-read executor queue size"); + assertEquals(active, executor.getActiveCount(), "unexpected body-read executor active count"); + } + + static final class BlockingInputStream extends InputStream { + private final byte[] data; + private final AtomicBoolean firstRead = new AtomicBoolean(true); + private final AtomicInteger position = new AtomicInteger(); + final CountDownLatch readStarted = new CountDownLatch(1); + final CountDownLatch releaseRead = new CountDownLatch(1); + + BlockingInputStream(byte[] data) { + this.data = data; + } + + @Override + public int read() throws IOException { + byte[] single = new byte[1]; + int read = read(single, 0, 1); + return read < 0 ? -1 : single[0] & 0xFF; + } + + @Override + public int read(byte[] buffer, int offset, int length) throws IOException { + awaitFirstRead(); + int current = position.get(); + if (current == data.length) { + return -1; + } + int read = Math.min(length, data.length - current); + System.arraycopy(data, current, buffer, offset, read); + position.addAndGet(read); + return read; + } + + @Override + public void close() { + releaseRead.countDown(); + } + + private void awaitFirstRead() throws IOException { + if (firstRead.compareAndSet(true, false)) { + readStarted.countDown(); + try { + if (!releaseRead.await(10, SECONDS)) { + throw new IOException("timed out waiting to release blocking request body read"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("blocking request body read interrupted", e); + } + } + } + } + + static final class CloseProbeInputStream extends InputStream { + final AtomicBoolean readAttempted = new AtomicBoolean(); + final CountDownLatch closed = new CountDownLatch(1); + + @Override + public int read() { + readAttempted.set(true); + return -1; + } + + @Override + public int read(byte[] buffer, int offset, int length) { + readAttempted.set(true); + return -1; + } + + @Override + public void close() { + closed.countDown(); + } + } +} From b3664b9d74f39a189e443b0657fed77397926060 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 20 Jul 2026 17:21:29 +0200 Subject: [PATCH 8/8] Isolate body read cancellation tests Reserve a second queue slot so the active upload can submit its EOF read while the cancelled task is still being drained. This keeps the tests focused on preventing read-after-close behavior instead of exercising queue saturation. Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- client/src/test/java/org/asynchttpclient/BasicHttpTest.java | 2 +- .../org/asynchttpclient/Http2StreamingBodyFlowControlTest.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java index 6a2e7fac47..e1d6c38058 100755 --- a/client/src/test/java/org/asynchttpclient/BasicHttpTest.java +++ b/client/src/test/java/org/asynchttpclient/BasicHttpTest.java @@ -1146,7 +1146,7 @@ public void cancelledQueuedInputStreamIsNotReadOverHttp1() throws Throwable { try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(config() .setRequestBodyStreamReadOffloadEnabled(true) .setRequestBodyStreamReadThreadsCount(1) - .setRequestBodyStreamReadQueueSize(1) + .setRequestBodyStreamReadQueueSize(2) .build())) { ThreadPoolExecutor executor = assertInstanceOf(ThreadPoolExecutor.class, client.blockingBodyReadExecutor()); diff --git a/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java b/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java index 2978a56ac1..37d0e23fb9 100644 --- a/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java +++ b/client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java @@ -404,7 +404,7 @@ public void cancelledQueuedInputStreamIsNotReadOverHttp2() throws Exception { .setMaxConnectionsPerHost(1) .setRequestBodyStreamReadOffloadEnabled(true) .setRequestBodyStreamReadThreadsCount(1) - .setRequestBodyStreamReadQueueSize(1) + .setRequestBodyStreamReadQueueSize(2) .setRequestTimeout(Duration.ofSeconds(30)) .build();