Skip to content

fix(netty): never let a queued suspend override a later resume #2361

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mkurz wants to merge 3 commits into
base: main
Choose a base branch
from
+455 −2
Open

fix(netty): never let a queued suspend override a later resume #2361

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b2f6e15
Never let a queued suspend override a later resume
mkurz Oct 6, 2026
1479ab2
Apply a suspend made on the event loop at once
mkurz Oct 21, 2026
7e26abc
Tighten the suspend/resume ordering tests
mkurz Oct 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions client/src/main/java/org/asynchttpclient/ResponseBodyControl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,14 @@
* The control is thread-safe and remains valid until its response completes. Calls made after completion have no
* effect.
* <p>
* Calls take effect on the event loop that delivers the response body callbacks to the {@link AsyncHandler}. A call
* made on that event loop, for example from one of those callbacks, takes effect before it returns. A call made from
* any other thread takes effect asynchronously on the event loop, possibly after further callbacks. A
* {@link #resume()} always takes effect, even after a {@link #suspend()} that a callback makes before the resume is
* applied, so a resume is never lost. A {@link #suspend()} from another thread is skipped if, by the time it would take
* effect, the most recent call to {@link #suspend()} or {@link #resume()} was a resume. Calls made concurrently on
* different threads are not ordered otherwise.
* <p>
* A control is also supplied when the final response headers end the response without a body. In that case,
* {@link #suspend()} cannot defer completion: the control becomes inactive when
* {@link AsyncHandler#onResponseBodyStart(ResponseBodyControl)} returns, and later calls have no effect.
Expand Down
18 changes: 17 additions & 1 deletion client/src/main/java/org/asynchttpclient/netty/NettyResponseBodyControl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ public final class NettyResponseBodyControl implements ResponseBodyControl {
private final boolean previousAutoRead;

private final AtomicBoolean active = new AtomicBoolean(true);
private volatile boolean suspendRequested;
private volatile boolean suspended;
private volatile boolean bodyFullyRead;

Expand Down Expand Up @@ -119,11 +120,17 @@ private NettyResponseBodyControl(NettyResponseFuture<?> future, Channel channel,

@Override
public void suspend() {
execute(this::suspend0);
suspendRequested = true;
if (channel.eventLoop().inEventLoop()) {
suspend0();
} else {
execute(this::applySuspend);
}
}

@Override
public void resume() {
suspendRequested = false;
execute(this::resume0);
}

Expand All @@ -132,6 +139,15 @@ public void cancel() {
execute(this::cancel0);
}

private void applySuspend() {
// A queued suspend() is skipped if resume() was called after the latest suspend(), so it cannot undo a newer
// resume(). A resume() is never skipped, so it is not lost to a suspend() that a callback decided on an older
// view.
if (suspendRequested) {
suspend0();
}
}

private void suspend0() {
if (active.get() && !suspended) {
suspensionStartedAction.run();
Expand Down
79 changes: 78 additions & 1 deletion client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,12 +55,15 @@
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
Expand All @@ -84,6 +87,7 @@ public class Http2ResponseBodyControlTest {
private static final int CANCELLATION_ATTEMPTS = 8;

private final AtomicInteger connectionCount = new AtomicInteger();
private final ExecutorService anotherThread = Executors.newSingleThreadExecutor();
private final CountDownLatch largeResponseQueued = new CountDownLatch(1);
private final CompletableFuture<Void> largeResponseWritten = new CompletableFuture<>();
private final CountDownLatch cancelledStreamClosed = new CountDownLatch(1);
Expand Down Expand Up @@ -139,6 +143,7 @@ protected void initChannel(Http2StreamChannel streamChannel) {

@AfterEach
public void stopServer() throws InterruptedException {
anotherThread.shutdownNow();
if (serverChildChannels != null) {
serverChildChannels.close().sync();
}
Expand Down Expand Up @@ -177,6 +182,67 @@ public void suspensionAppliesHttp2FlowControlAndParentIsReused() throws Exceptio
}
}

@Test
public void earlierSuspendFromAnotherThreadDoesNotUndoLaterResume() throws Exception {

@hyperxpro hyperxpro Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The twin test below also fetches /large-sibling and checks connectionCount. Do the same here. A skipped suspend that still ran the started hook would leave the flow controller in refill mode and this test would not notice.

@mkurz mkurz Oct 22, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done (7e26abc). The test now fetches /large-sibling afterwards and checks connectionCount, as its twin does.

To be precise about what it catches: I tried a variant where a skipped suspend still runs the start hook, and this test still passes. The sibling stream completes over the same connection anyway, so the flow controller left in refill mode doesn't show up there. NettyResponseBodyControlTest does catch that variant, through the exact counts of started and ended suspensions (comment 3).

AtomicReference<Channel> connection = new AtomicReference<>();
try (AsyncHttpClient client = http2Client(connection::set)) {
RecordingHandler handler = new RecordingHandler();
ListenableFuture<RecordingHandler> request = client.prepareGet(url("/large")).execute(handler);
ResponseBodyControl control = handler.control.get(5, SECONDS);
assertTrue(largeResponseQueued.await(5, SECONDS));

// HTTP/2 streams share the event loop of their parent connection.
connection.get().eventLoop().submit(() -> {
callFromAnotherThread(control::suspend);
control.resume();
return null;
}).get(5, SECONDS);

largeResponseWritten.get(5, SECONDS);
assertSame(handler, request.get(5, SECONDS));
assertEquals((long) FRAME_SIZE * FRAME_COUNT, handler.bodyBytes.get());
assertNull(handler.throwable.get());

Response sibling = client.prepareGet(url("/large-sibling"))
.setReadTimeout(Duration.ofSeconds(5))
.execute()
.get(10, SECONDS);
assertEquals((long) FRAME_SIZE * SIBLING_FRAME_COUNT, sibling.getResponseBodyAsBytes().length);
assertEquals(1, connectionCount.get());
}
}

@Test
public void resumeFromAnotherThreadIsNotLostToALaterSuspend() throws Exception {
// This also passes when no queued suspend() is skipped. It fails if a queued resume() is skipped for a later
// suspend().
AtomicReference<Channel> connection = new AtomicReference<>();
try (AsyncHttpClient client = http2Client(connection::set)) {
RecordingHandler handler = new RecordingHandler();
ListenableFuture<RecordingHandler> request = client.prepareGet(url("/large")).execute(handler);
ResponseBodyControl control = handler.control.get(5, SECONDS);
assertTrue(largeResponseQueued.await(5, SECONDS));

connection.get().eventLoop().submit(() -> {
callFromAnotherThread(control::resume);
control.suspend();
return null;
}).get(5, SECONDS);

largeResponseWritten.get(5, SECONDS);
assertSame(handler, request.get(5, SECONDS));
assertEquals((long) FRAME_SIZE * FRAME_COUNT, handler.bodyBytes.get());
assertNull(handler.throwable.get());

Response sibling = client.prepareGet(url("/large-sibling"))
.setReadTimeout(Duration.ofSeconds(5))
.execute()
.get(10, SECONDS);
assertEquals((long) FRAME_SIZE * SIBLING_FRAME_COUNT, sibling.getResponseBodyAsBytes().length);
assertEquals(1, connectionCount.get());
}
}

@Test
public void cancellationResetsOnlyTheHttp2Stream() throws Exception {
try (AsyncHttpClient client = http2Client()) {
Expand Down Expand Up @@ -293,13 +359,24 @@ public void suspensionCannotDeferBodylessResponse() throws Exception {
}

private AsyncHttpClient http2Client() {
return http2Client(ignored -> {
});
}

private AsyncHttpClient http2Client(Consumer<Channel> connectionInitializer) {
return asyncHttpClient(config()
.setUseInsecureTrustManager(true)
.setHttp2Enabled(true)
.setHttp2InitialWindowSize(32 * 1024)
.setMaxConnectionsPerHost(1)
.setReadTimeout(Duration.ofMillis(100))
.setRequestTimeout(Duration.ofSeconds(10)));
.setRequestTimeout(Duration.ofSeconds(10))
.setHttpAdditionalChannelInitializer(connectionInitializer));
}

// The calling event loop stays busy until the call returns, so the control has to queue the call behind it.
private void callFromAnotherThread(Runnable call) throws Exception {
anotherThread.submit(call).get(5, SECONDS);
}

private String url(String path) {
Expand Down
72 changes: 72 additions & 0 deletions client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,10 +57,13 @@
import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
Expand Down Expand Up @@ -91,6 +94,7 @@ public class ResponseBodyControlTest {
AttributeKey.valueOf("response-body-control-connection-id");

private final AtomicInteger connectionCount = new AtomicInteger();
private final ExecutorService anotherThread = Executors.newSingleThreadExecutor();
private final CompletableFuture<ChannelHandlerContext> responseContext = new CompletableFuture<>();
private final CompletableFuture<ChannelHandlerContext> firstReplayContext = new CompletableFuture<>();
private final CompletableFuture<ChannelHandlerContext> secondReplayContext = new CompletableFuture<>();
Expand Down Expand Up @@ -132,6 +136,7 @@ protected void initChannel(Channel channel) {

@AfterEach
public void stopServer() throws InterruptedException {
anotherThread.shutdownNow();
if (serverChildChannels != null) {
serverChildChannels.close().sync();
}
Expand Down Expand Up @@ -185,6 +190,64 @@ public void suspensionControlsReadsAndCompletedConnectionIsPooled() throws Excep
}
}

@Test
public void earlierSuspendFromAnotherThreadDoesNotUndoLaterResume() throws Exception {
AtomicReference<Channel> clientChannel = new AtomicReference<>();
try (AsyncHttpClient client = asyncHttpClient(config()
.setRequestTimeout(Duration.ofSeconds(10))
.setHttpAdditionalChannelInitializer(clientChannel::set))) {
RecordingHandler handler = new RecordingHandler(false);
ListenableFuture<RecordingHandler> request = client.prepareGet(url("/controlled")).execute(handler);
ResponseBodyControl control = handler.control.get(5, SECONDS);
ChannelHandlerContext server = responseContext.get(5, SECONDS);
Channel channel = clientChannel.get();

runOnEventLoop(channel, () -> {
callFromAnotherThread(control::suspend);
control.resume();
return null;
});
awaitEventLoop(channel);

assertTrue(channel.config().isAutoRead(), "the queued suspend() must not override the later resume()");
writeChunk(server, "one");
assertEquals("one", handler.items.poll(5, SECONDS));
writeLast(server);
assertSame(handler, request.get(5, SECONDS));
assertNull(handler.throwable.get());
}
}

@Test
public void resumeFromAnotherThreadIsNotLostToALaterSuspend() throws Exception {
// This also passes when no queued suspend() is skipped. It fails if a queued resume() is skipped for a later
// suspend().
AtomicReference<Channel> clientChannel = new AtomicReference<>();
try (AsyncHttpClient client = asyncHttpClient(config()
.setRequestTimeout(Duration.ofSeconds(10))
.setHttpAdditionalChannelInitializer(clientChannel::set))) {
RecordingHandler handler = new RecordingHandler(false);
ListenableFuture<RecordingHandler> request = client.prepareGet(url("/controlled")).execute(handler);
ResponseBodyControl control = handler.control.get(5, SECONDS);
ChannelHandlerContext server = responseContext.get(5, SECONDS);
Channel channel = clientChannel.get();

runOnEventLoop(channel, () -> {
callFromAnotherThread(control::resume);
control.suspend();
return null;
});
awaitEventLoop(channel);

assertTrue(channel.config().isAutoRead(), "the later suspend() must not drop the queued resume()");
writeChunk(server, "one");
assertEquals("one", handler.items.poll(5, SECONDS));
writeLast(server);
assertSame(handler, request.get(5, SECONDS));
assertNull(handler.throwable.get());
}
}

@Test
public void suspensionPausesReadTimeoutAndCancellationClosesConnection() throws Exception {
try (AsyncHttpClient client = asyncHttpClient(config()
Expand Down Expand Up @@ -562,6 +625,15 @@ private static void awaitEventLoop(Channel channel) throws InterruptedException
}).sync();
}

private static void runOnEventLoop(Channel channel, Callable<?> task) throws Exception {
channel.eventLoop().submit(task).get(5, SECONDS);
}

// The calling event loop stays busy until the call returns, so the control has to queue the call behind it.
private void callFromAnotherThread(Runnable call) throws Exception {
anotherThread.submit(call).get(5, SECONDS);
}

private static void writeChunk(ChannelHandlerContext ctx, String value) throws InterruptedException {
ctx.executor().submit(() -> ctx.writeAndFlush(
new DefaultHttpContent(Unpooled.copiedBuffer(value, CharsetUtil.US_ASCII)))).sync();
Expand Down
Loading
Loading