From 553ab87564b5424431cad4a949d8fa84c6601837 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 20 Jul 2026 10:48:27 +0200 Subject: [PATCH 1/6] Batch HTTP/2 upload body flushes Avoid flushing every intermediate HTTP/2 DATA chunk produced by Http2BodyWriter. The pump now flushes when it yields for source suspension, when backpressure makes the stream unwritable, or when it writes the terminal DATA frame. This preserves bounded in-flight write behavior while reducing per-chunk flush overhead for multi-chunk uploads. Add a focused unit test for the single terminal flush path while the stream remains writable. Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../netty/request/body/Http2BodyWriter.java | 14 ++- .../request/body/Http2BodyWriterTest.java | 97 +++++++++++++++++++ 2 files changed, 107 insertions(+), 4 deletions(-) create mode 100644 client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java 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 77ebdb8cf..5e6a03a36 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 @@ -38,8 +38,9 @@ *

* This writer is a one-chunk-at-a-time pump driven entirely on the stream channel's event loop: *

*

* Lifecycle / cleanup. Because the pump completes asynchronously (after {@code writeHttp2} @@ -125,9 +125,10 @@ default void onResume(Runnable resume) { 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. + // pump, and remember permanently once the terminal frame has been emitted so later callbacks cannot write + // a second endStream frame before its write completes. private boolean pumping; - private boolean terminalWritePending; + private boolean terminalWritten; // Transient handler that resumes the pump when the channel becomes writable again. Added lazily the // first time the pump parks, removed by finish(). @@ -186,7 +187,7 @@ static void start(Http2StreamChannel channel, ChunkSource source) { * event loop. Any written frames are flushed before the pump parks or completes. */ private void pump() { - if (done || pumping || terminalWritePending) { + if (done || pumping || terminalWritten) { return; } pumping = true; @@ -204,8 +205,13 @@ private void pump() { // No data available yet but the body is not finished (feedable body). Park until the // source signals more via the onResume callback. Any already-buffered `pending` chunk is // retained (O(1)); we deliberately do not flush an early endStream. - channel.flush(); suspended = true; + channel.flush(); + if (!suspended) { + // A synchronous flush callback resumed the source while pump() was guarded against + // re-entry. Consume that resume here instead of parking indefinitely. + continue; + } return; } @@ -217,7 +223,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; + terminalWritten = true; writeLastFrame(terminal); channel.flush(); return; @@ -323,7 +329,7 @@ private void finish(Throwable cause) { private final class WritabilityResumeHandler extends ChannelInboundHandlerAdapter { @Override public void channelWritabilityChanged(ChannelHandlerContext ctx) { - if (!done && !terminalWritePending && ctx.channel().isWritable()) { + if (!done && !terminalWritten && ctx.channel().isWritable()) { pump(); } ctx.fireChannelWritabilityChanged(); diff --git a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java index c2de0a1f4..af6a274b9 100644 --- a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java @@ -21,11 +21,22 @@ import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.ChannelOutboundHandlerAdapter; import io.netty.channel.ChannelPipeline; +import io.netty.channel.ChannelPromise; import io.netty.channel.DefaultChannelPromise; import io.netty.channel.EventLoop; +import io.netty.channel.WriteBufferWaterMark; +import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.http2.DefaultHttp2DataFrame; +import io.netty.handler.codec.http2.DefaultHttp2Headers; +import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame; +import io.netty.handler.codec.http2.Http2DataFrame; +import io.netty.handler.codec.http2.Http2FrameCodecBuilder; +import io.netty.handler.codec.http2.Http2MultiplexHandler; import io.netty.handler.codec.http2.Http2StreamChannel; +import io.netty.handler.codec.http2.Http2StreamChannelBootstrap; import io.netty.util.ReferenceCountUtil; import io.netty.util.concurrent.ImmediateEventExecutor; import org.junit.jupiter.api.Test; @@ -36,6 +47,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; @@ -83,23 +95,96 @@ public void sourceSuspensionFlushesBeforeParking() { when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT); when(channel.isWritable()).thenReturn(true); when(channel.closeFuture()).thenReturn(new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE)); + AtomicInteger flushes = new AtomicInteger(); when(channel.write(any())).thenAnswer(invocation -> { ReferenceCountUtil.release(invocation.getArgument(0)); return succeededFuture(channel); }); + when(channel.flush()).thenAnswer(invocation -> { + flushes.incrementAndGet(); + return channel; + }); - SuspendingChunkSource source = new SuspendingChunkSource(); + SuspendingChunkSource source = new SuspendingChunkSource(flushes); Http2BodyWriter.start(channel, source); assertEquals(0, source.closed); + assertEquals(0, source.flushesBeforeSuspend); + assertEquals(1, flushes.get()); verify(channel, times(1)).write(any(DefaultHttp2DataFrame.class)); - verify(channel, times(1)).flush(); source.finish(); assertEquals(1, source.closed); verify(channel, times(2)).write(any(DefaultHttp2DataFrame.class)); - verify(channel, times(2)).flush(); + assertEquals(2, flushes.get()); + } + + @Test + public void synchronousResumeDuringSuspensionFlushIsNotLost() { + Http2StreamChannel channel = mock(Http2StreamChannel.class); + EventLoop eventLoop = mock(EventLoop.class); + when(eventLoop.inEventLoop()).thenReturn(true); + when(channel.eventLoop()).thenReturn(eventLoop); + when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT); + when(channel.isWritable()).thenReturn(true); + when(channel.closeFuture()).thenReturn(new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE)); + when(channel.write(any())).thenAnswer(invocation -> { + ReferenceCountUtil.release(invocation.getArgument(0)); + return succeededFuture(channel); + }); + AtomicInteger flushes = new AtomicInteger(); + SuspendingChunkSource source = new SuspendingChunkSource(flushes); + when(channel.flush()).thenAnswer(invocation -> { + if (flushes.incrementAndGet() == 1) { + source.finish(); + } + return channel; + }); + + Http2BodyWriter.start(channel, source); + + assertEquals(1, source.closed); + assertEquals(2, flushes.get()); + verify(channel, times(2)).write(any(DefaultHttp2DataFrame.class)); + } + + @Test + public void realStreamBoundsUnflushedBytesByWaterMark() { + UnflushedBytesTracker tracker = new UnflushedBytesTracker(); + EmbeddedChannel parent = new EmbeddedChannel( + Http2FrameCodecBuilder.forClient().build(), + new Http2MultiplexHandler(new ChannelInboundHandlerAdapter())); + Http2StreamChannel stream = new Http2StreamChannelBootstrap(parent) + .handler(tracker) + .open() + .syncUninterruptibly() + .getNow(); + int chunkSize = 1024; + int highWaterMark = 8 * 1024; + stream.config().setWriteBufferWaterMark(new WriteBufferWaterMark(highWaterMark / 2, highWaterMark)); + FixedChunkSource source = new FixedChunkSource(highWaterMark / chunkSize * 2, chunkSize); + + try { + stream.writeAndFlush(new DefaultHttp2HeadersFrame(new DefaultHttp2Headers() + .method("POST") + .scheme("https") + .authority("localhost") + .path("/"))).syncUninterruptibly(); + + Http2BodyWriter.start(stream, source); + parent.runPendingTasks(); + + assertEquals(1, source.closed); + assertTrue(tracker.peakUnflushedBytes >= highWaterMark, + "the real stream must reach its configured high-water mark"); + assertTrue(tracker.peakUnflushedBytes <= highWaterMark + chunkSize, + "unflushed DATA must stay within the high-water mark plus one chunk"); + } finally { + stream.close().syncUninterruptibly(); + parent.runPendingTasks(); + parent.finishAndReleaseAll(); + } } @Test @@ -116,7 +201,8 @@ public void unwritableChannelResumesWithoutReentrantTerminalWrite() throws Excep when(eventLoop.inEventLoop()).thenReturn(true); when(channel.eventLoop()).thenReturn(eventLoop); when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT); - when(channel.isWritable()).thenReturn(false, false, true); + AtomicBoolean writable = new AtomicBoolean(); + when(channel.isWritable()).thenAnswer(invocation -> writable.get()); when(channel.closeFuture()).thenReturn(new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE)); when(channel.pipeline()).thenReturn(pipeline); when(pipeline.addLast(any(ChannelHandler.class))).thenAnswer(invocation -> { @@ -153,6 +239,7 @@ public void unwritableChannelResumesWithoutReentrantTerminalWrite() throws Excep verify(channel, times(1)).write(any(DefaultHttp2DataFrame.class)); verify(channel, times(1)).flush(); + writable.set(true); ((io.netty.channel.ChannelInboundHandlerAdapter) resumeHandler.get()) .channelWritabilityChanged(eventContext); @@ -172,10 +259,16 @@ private static ChannelFuture succeededFuture(Http2StreamChannel channel) { private static final class FixedChunkSource implements Http2BodyWriter.ChunkSource { private int chunks; + private final int chunkSize; private int closed; FixedChunkSource(int chunks) { + this(chunks, 1); + } + + FixedChunkSource(int chunks, int chunkSize) { this.chunks = chunks; + this.chunkSize = chunkSize; } @Override @@ -184,7 +277,7 @@ public ByteBuf nextChunk(ByteBufAllocator alloc) { return null; } chunks--; - return alloc.buffer(1).writeByte(chunks); + return alloc.buffer(chunkSize).writeZero(chunkSize); } @Override @@ -194,17 +287,27 @@ public void close() { } private static final class SuspendingChunkSource implements Http2BodyWriter.ChunkSource { + private final AtomicInteger flushes; private int state; private int closed; + private int flushesBeforeSuspend; private Runnable resume; + SuspendingChunkSource(AtomicInteger flushes) { + this.flushes = flushes; + } + @Override public ByteBuf nextChunk(ByteBufAllocator alloc) { if (state < 2) { state++; return alloc.buffer(1).writeByte(state); } - return state == 2 ? Http2BodyWriter.SUSPEND : null; + if (state == 2) { + flushesBeforeSuspend = flushes.get(); + return Http2BodyWriter.SUSPEND; + } + return null; } @Override @@ -222,4 +325,24 @@ public void close() { closed++; } } + + private static final class UnflushedBytesTracker extends ChannelOutboundHandlerAdapter { + private int unflushedBytes; + private int peakUnflushedBytes; + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { + if (msg instanceof Http2DataFrame) { + unflushedBytes += ((Http2DataFrame) msg).content().readableBytes(); + peakUnflushedBytes = Math.max(peakUnflushedBytes, unflushedBytes); + } + ctx.write(msg, promise); + } + + @Override + public void flush(ChannelHandlerContext ctx) { + unflushedBytes = 0; + ctx.flush(); + } + } } From 633ac777486d3bf2b2032869c837eed43b3516a9 Mon Sep 17 00:00:00 2001 From: Aayush Atharva <24762260+hyperxpro@users.noreply.github.com> Date: Sat, 1 Aug 2026 14:00:06 +0530 Subject: [PATCH 5/6] Update client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java --- .../asynchttpclient/netty/request/body/Http2BodyWriterTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java index af6a274b9..f59e63720 100644 --- a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java @@ -226,7 +226,7 @@ public void unwritableChannelResumesWithoutReentrantTerminalWrite() throws Excep if (handler != null && terminalFrames.get() != 0 && nestedWritabilityCallbackFired.compareAndSet(false, true)) { - ((io.netty.channel.ChannelInboundHandlerAdapter) handler).channelWritabilityChanged(eventContext); + ((ChannelInboundHandlerAdapter) handler).channelWritabilityChanged(eventContext); } return channel; }); From 75d86f5434c5302ff230980f7b8d49ce9df3f06c Mon Sep 17 00:00:00 2001 From: Aayush Atharva <24762260+hyperxpro@users.noreply.github.com> Date: Sat, 1 Aug 2026 14:00:13 +0530 Subject: [PATCH 6/6] Update client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java --- .../netty/request/body/Http2BodyWriterTest.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java index f59e63720..670c78ad2 100644 --- a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java @@ -240,8 +240,7 @@ public void unwritableChannelResumesWithoutReentrantTerminalWrite() throws Excep verify(channel, times(1)).flush(); writable.set(true); - ((io.netty.channel.ChannelInboundHandlerAdapter) resumeHandler.get()) - .channelWritabilityChanged(eventContext); + ((ChannelInboundHandlerAdapter) resumeHandler.get()).channelWritabilityChanged(eventContext); assertEquals(0, source.closed); assertEquals(1, terminalFrames.get());