feat: add ResponseBodyControl.execute to decide in sequence with body delivery by mkurz · Pull Request #2359 · AsyncHttpClient/async-http-client

mkurz · 2026-10-07T08:09:05Z

Summary

Add ResponseBodyControl.execute(Runnable), which runs a task on the event loop that delivers the response body, in sequence with the callbacks AHC makes there, such as onBodyPartReceived. Calls to suspend(), resume() and cancel() from the task are applied before they return, not queued.
This lets a consumer that buffers body parts on another thread decide whether to resume based on everything delivered so far, instead of on a decision that a part delivered in the meantime made stale.
NettyResponseBodyControl already applied its own control calls this way; that method is now public. The default implementation runs the task on the calling thread.

Problem

A typical streaming consumer buffers body parts from onBodyPartReceived and hands them to a reader on another thread. When that reader has emptied the buffer and still wants more, it calls resume(). Called from that thread, resume() only takes effect later, on the event loop. In between, the event loop can still deliver parts from the current socket read, and those parts may already meet the reader's demand. The queued resume() then lets one more socket read through that nobody asked for.

With a highly compressible body, that one extra read can inflate to many megabytes beyond demand. For a gzipped stream with early demand from another thread, Play WS measured about 35 MB buffered beyond demand, against about 1.9 MB when the decision is re-checked on the event loop.

AHC cannot fix this by re-checking the decision itself, because only the consumer knows which state its decision depends on. Play WS currently works around it by reaching the event loop through Netty's internal ThreadExecutorMap, and then making its resume decision there.

Change

default void execute(Runnable task) {
    Objects.requireNonNull(task, "task");
    task.run();
}

On the event loop, for example from within a body callback, the task runs before execute returns.
From another thread, it is queued to the event loop and runs there after the work already pending on it. The event loop can deliver further body parts before the task runs; the task sees all of them.
Either way, suspend(), resume() and cancel() called from the task are applied before they return, not queued. As suspend() already says, parts that were already read can still be delivered after it. resume() can deliver parts, and resume() and cancel() can complete the response, before they return: on HTTP/2, resume() reads the frames queued on the stream, and cancel() runs onCompleted.
onThrowable runs once the control is inactive, so control calls from a task passed from there have no effect. For most failures, such as an I/O error, a closed connection or a reset stream, it runs on the event loop, and the task runs before execute returns. A cancel or abort from another thread calls it on that thread, and a timeout on the timer thread unless useEventLoopTimeouts is enabled. From there, the task is queued and can run while that callback is still running.
The task must not throw. An exception from a task that runs before execute returns is thrown from execute, so a callback that lets it propagate fails the response. An exception from a queued task is logged by the event loop and does not fail the response. If that task was to resume reads, the response stays suspended until it is resumed or cancelled, or the request timeout fires; with the request timeout disabled, indefinitely, since the read timeout is paused while reads are suspended. The Javadoc says so.
The task must not block. Unlike calls to the other methods, it also runs after the response has completed (the interface Javadoc now says so). If the event loop no longer accepts tasks, it does not run.
The default implementation runs the task on the calling thread. It suits implementations that apply control calls synchronously on the calling thread, and does not order the task with callbacks on other threads. An implementation that delegates to another control must delegate execute too.

NettyResponseBodyControl.execute is the existing private method made public, with a null check. It works for HTTP/1.1 and for HTTP/2 stream channels.

Alternatives considered

resumeIf(BooleanSupplier), a resume whose condition AHC evaluates on the event loop. It needs subtle ordering rules: a true condition counts as a resume at that moment and overrides an earlier suspend. A consumer whose own drain loop may run on either thread then has to fit its state machine to those rules, and we found that hard to get right. With execute, such a consumer can write control.execute(() -> { if (cond) control.resume(); }).
Demand counting in AHC (request(n)). HTTP/1.1 decodes one socket read into several parts regardless of demand, so AHC would need its own buffer. That's a much larger change.

execute only exposes what the control already does internally. Callbacks already run on the event loop under the same rule not to block.

Compatibility

A new default method on ResponseBodyControl. Revapi passed. Existing implementations keep compiling and behave as before, with two exceptions:
- an implementation that declares a non-public execute(Runnable) no longer compiles;
- one that already has a public void execute(Runnable) now implements this method, with whatever semantics it has.
No behavior change for existing callers.

AI disclosure

Claude Code on behalf of Matthias Kurz. The commits include Co-Authored-By per AGENTS.md.

Test plan

New tests:

ResponseBodyControlTest:
- execute runs inline in a callback, runs on the event loop when called from another thread, and runs after control calls queued before it;
- a consumer thread that decides inside execute sees the part delivered in the meantime and does not resume;
- execute after the response has completed still runs the task on the event loop;
- a queued task that throws does not fail the response: it stays suspended, a later task resumes it, it completes, and its connection is reused;
- a queued task that throws after the response has completed leaves the pooled connection open and reused;
- a task that throws on the event loop throws from execute, and the response completes when the caller catches it;
- the default implementation runs the task on the calling thread, and rejects a null task with the message task.
Http2ResponseBodyControlTest:
- a task from another thread runs on the thread that delivers the stream's body parts, and resumes the suspended stream;
- a queued task that throws does not fail the stream, and a later task resumes it.
NettyResponseBodyControlExecuteTest: once the event loop has shut down, execute neither runs the task nor throws.

Broken variants, at the first revision, each run against these tests and the existing control tests (24 tests in all):

The task always runs on the calling thread: 9 failures, including "a task from another thread must run on the channel event loop".
The task is always queued: 3 failures, including "a task from a callback must run before execute returns".
A rejected task throws: the shutdown test fails.
No task after completion: the completion test times out (plus failures of existing tests, since the control's own calls use the same method).

Also at the first revision, deciding on the other thread without execute failed "the task must see the part buffered after it was queued".

Broken variants for the review changes, each run against the control tests:

An exception from a queued task fails the response: 2 failures, the HTTP/1.1 and HTTP/2 tests for a queued task that throws.
An exception from a task that runs before execute returns is swallowed: 1 failure, the inline test.
Tasks from another thread run on a pool thread, while the control's own calls stay on the event loop: 4 failures, including the HTTP/2 thread check. With the previous revision's assertNotSame, the HTTP/2 test passed.

Runs:

Each new commit compiles with its tests.
./mvnw -B -ntp -Dgpg.skip=true clean verify on JDK 11: 1,802 tests, 0 failures, 0 errors, 11 skipped; Revapi passed. No test-skipping flags were used.
The control tests (*ResponseBodyControl*Test and NettyResponseFutureTest), compiled on JDK 11 and run on JDK 17, 21 and 25 (Maven Surefire's -Djvm): 40 tests each, 0 failures.
Play WS, with its resume check moved from ThreadExecutorMap to execute, against a local build of the first revision of this branch alone (Scala 2.13):
- all Play WS tests pass, except a new cookie test that needs the separate raw-Cookie fix;
- its gzip streaming test buffers 1,875,916 bytes beyond demand, as with the ThreadExecutorMap workaround;
- with Play WS resuming directly instead of through execute, the same test buffers 35,589,292 bytes and fails.
The review changes are Javadoc, the null check in the default implementation, and tests.
All four open AHC changes merged together (raw Cookie headers on retries, ResponseBodyControl.execute, suspend/resume order, demand-bounded decompression): they merge without conflicts, together and in every pair, and ./mvnw -B -ntp -Dgpg.skip=true clean install on JDK 11 passes with 1,872 tests, 0 failures, 0 errors, 11 skipped; Revapi passed.
Play WS against that merged build, Scala 2.13 and 3.3, JDK 17: all tests pass. The shaded AHC jar contains this revision's default execute.

A consumer that buffers body parts on another thread decides there whether to resume reads, but resume() from that thread only takes effect later on the event loop. In between, the event loop can deliver another part that already meets the consumer's demand, and the resume then lets one more read through. With a highly compressible body, that read can inflate to many megabytes beyond demand. Only the caller knows which state its decision depends on, so AHC cannot re-check it. Add execute(Runnable), which runs a task on the event loop that delivers the response body, in sequence with the callbacks made there: before returning when called on that event loop, and queued to it otherwise. suspend(), resume() and cancel() called from the task take effect immediately, so a consumer can decide there on everything delivered so far. onThrowable can run on other threads, after a cancel or a timeout, so a task is not ordered with it. Unlike the other methods, the task also runs after completion; it does not run once the event loop rejects tasks. NettyResponseBodyControl already applied its own control calls this way; that method is now public. The default implementation runs the task on the calling thread, for implementations that apply control calls synchronously. Play WS needs this: it currently reaches the event loop through Netty's internal ThreadExecutorMap to avoid the extra read. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

hyperxpro · 2026-10-22T23:08:31Z

+     * <p>
+     * Not every callback runs on that event loop. {@link AsyncHandler#onThrowable(Throwable)} can run on the thread
+     * that cancels the request, or on a timer thread when a timeout fires. A task passed to this method from there is
+     * queued like one from any other thread, and can run while that callback is still running.


This is not what happens on most of the paths that end in onThrowable. Timeouts are armed on the channel's loop in TimeoutsHolder.arm, and exceptionCaught, channelInactive and Http2Handler.streamFailed all reach abort on the loop, so inEventLoop() is true and the task runs inline before execute returns, after terminateAndExit has already deactivated the control. We should say the task can run inline from onThrowable with the control already inactive, otherwise a caller that publishes state after execute returns runs the task on stale state.

Agreed, fixed in 188250b. One correction: timeouts run on the event loop only with useEventLoopTimeouts, which is off by default; otherwise they fire on the client's timer thread. But exceptionCaught, channelInactive and streamFailed do reach abort on the event loop, and on every path the control is deactivated before onThrowable runs: abort and done through terminateAndExit, and cancel directly.

The Javadoc now says that onThrowable runs once the control is inactive, so control calls from a task passed from there have no effect. For most failures, such as an I/O error, a closed connection or a reset stream, it runs on the event loop, and the task runs before execute returns. A cancel or abort from another thread calls it on that thread, and a timeout on the timer thread; from there the task is queued and can run while the callback is still running.

hyperxpro · 2026-10-22T23:08:31Z

+     * Called on that event loop, for example from one of those callbacks, the task runs before this method returns.
+     * Called from any other thread, the task is queued to the event loop and runs there after the work already pending
+     * on it; the event loop can deliver further body parts before the task runs. Either way, {@link #suspend()},
+     * {@link #resume()} and {@link #cancel()} called from the task take effect immediately, and the task sees every


suspend() from the task only flips the flag, it does not hold back what is already decoded: on HTTP/1.1 every HttpContent from the same read is still delivered, and on an h2 stream nothing is held back once the peer's END_STREAM is queued, doBeginRead drains it regardless of autoRead. The other direction is missing too: resume() on an h2 stream calls read() synchronously, so the remaining frames and even onCompleted can run inside the task before resume() returns. I'd say "are applied before the task returns", point at the caveat suspend() already has, and add that resume() may deliver parts and complete the response before it returns.

Agreed, changed in 00568c3. The calls are now "applied before they return, not queued", followed by suspend()'s caveat that parts already read can still be delivered, and by the fact that resume() can deliver parts, and resume() and cancel() can complete the response, before they return.

I checked the HTTP/2 case in Netty 4.2.18: when the stream isn't already reading, doBeginRead drains its queued frames synchronously. cancel() from a task runs finishUpdate, so onCompleted runs inside it too.

hyperxpro · 2026-10-22T23:08:31Z

+    public void execute(Runnable task) {
+        Objects.requireNonNull(task, "task");
        if (channel.eventLoop().inEventLoop()) {
            task.run();


What should happen when task throws? From here it escapes into HttpHandler.handleRead, which fails the request and closes the connection (resets the stream on h2), but from another thread Netty's safeExecute just logs a WARN and drops it, and the default on the interface propagates to the caller. Either catch in the queued branch and route it to the same failure path, or document that the task must not throw. Right now a caller can't write a correct task without reading this class.

I documented it, in 1a055b0. The Javadoc now says that the task must not throw, and what happens if it does:

A task that runs before execute returns throws from execute, so a callback that lets it propagate fails the response.

An exception from a queued task is logged by the event loop and does not fail the response. If that task was to resume reads, the response stays suspended until it is resumed or cancelled, or the request timeout fires. With the request timeout disabled, that can be indefinitely, since the read timeout is paused while reads are suspended.

Tests cover queued-task exceptions on HTTP/1.1 and HTTP/2: the response is not failed automatically, and a later task can still resume it. Separate HTTP/1.1 tests verify that inline exceptions propagate from execute and that a queued exception after completion leaves the pooled connection alone.

I also tried routing it to the handler's failure path. That works, but create then needs readFailed or streamFailed from the handler, and it changes the dispatch that suspend(), resume() and cancel() go through as well. Documenting it keeps this PR to the new method. If you'd rather have the response fail, I can add that here or separately.

hyperxpro · 2026-10-22T23:08:31Z

+     * event loop no longer accepts tasks, for example because it is shutting down, the task does not run.
+     * <p>
+     * The default implementation runs the task on the calling thread. That suits implementations that apply control
+     * calls synchronously on the calling thread; it does not order the task with callbacks made on other threads.


A decorator written against 3.0.14 that forwards suspend, resume and cancel to a NettyResponseBodyControl and does not override this one gets the calling-thread default and silently loses the ordering this method exists for. Worth a sentence that implementations delegating to another control must delegate execute too.

Added in 172e27b: "An implementation that delegates to another control must delegate this method too, or the task loses the ordering that control provides."

hyperxpro · 2026-10-22T23:08:31Z

+     * @since 3.0.15
+     */
+    default void execute(Runnable task) {
+        task.run();


NettyResponseBodyControl does requireNonNull(task, "task"), this one NPEs from task.run() without a message. Same check here?

Done in 172e27b, with a test for the message.

hyperxpro · 2026-10-22T23:08:31Z

+            });
+
+            assertSame(handler, request.get(10, SECONDS));
+            assertNotSame(Thread.currentThread(), taskThread.get(5, SECONDS),


assertNotSame passes for any thread that is not the test thread, so this stays green if the task ran on a timer thread, the resume() inside gets queued to the loop anyway and the request still completes. The HTTP/1.1 tests check channel.eventLoop().inEventLoop(taskThread), can we do the same here? setHttpAdditionalChannelInitializer hands out the parent channel for h2 too and the stream channel shares its loop, or record the thread in onBodyPartReceived and assertSame against it.

Fixed in 70a0b0f, with your second suggestion: the handler records the thread that delivers body parts, and the test asserts that the task ran on it. A variant that runs off-loop tasks on a pool thread, while the control's own calls stay on the event loop, now fails the test; with the old assertion it passed.

The Javadoc said a task passed to execute from onThrowable is queued, as if that callback only ran on a cancelling thread or the timer thread. Most failures, such as an I/O error, a closed connection or a reset stream, call it on the event loop, so the task runs before execute returns. On every path the control is inactive by then, so control calls from the task have no effect. Timeouts run on the event loop only with useEventLoopTimeouts, which is off by default. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

The Javadoc said suspend, resume and cancel called from a task take effect immediately. They are applied before they return, but suspend does not hold back parts that were already read, as its own Javadoc says. On HTTP/2, resume reads the frames queued on the stream before it returns, so onBodyPartReceived and onCompleted can run inside it. cancel completes the response before it returns too. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

A task that execute queued from another thread runs as a plain event loop task: Netty logs an exception from it, and the response carries on. A task that runs before execute returns throws to its caller, and from a callback that fails the response. Neither was documented. Say that the task must not throw, and what happens if it does. A queued task meant to resume reads that throws leaves the response suspended until it is resumed or cancelled, or the request timeout fires. Without a request timeout that can be indefinitely, since the read timeout is paused while reads are suspended. The tests pin both cases, and that a task throwing after completion leaves the pooled connection alone. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

A control written against 3.0.14 that forwards suspend, resume and cancel to another control gets the default execute, which runs the task on the calling thread and so loses the ordering execute is for. Say that a delegating control must delegate execute too. The default also rejects a null task with a message now, as NettyResponseBodyControl does. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

assertNotSame against the test thread passed for a task run on any other thread. Had execute run it on a pool thread, the resume() inside would still have been queued to the event loop, and the request would still have completed. Check that the task ran on the thread that delivered the body parts, the stream's event loop. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

mkurz · 2026-10-23T16:44:25Z

Thanks for the review. I've pushed five commits on top and updated the description. Replies are inline.

mkurz mentioned this pull request Oct 7, 2026

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

Open

8 tasks done

mkurz force-pushed the feature/response-body-control-execute branch from bcee2e0 to 0fd1488 Compare October 7, 2026 08:45

mkurz mentioned this pull request Oct 7, 2026

Upgrade to AHC 3.0.15 and decide streamed-body resumes through ResponseBodyControl.execute playframework/play-ws#1487

Draft

18 of 22 tasks

hyperxpro requested changes Oct 22, 2026

View reviewed changes

mkurz and others added 5 commits October 23, 2026 16:33

mkurz requested a review from hyperxpro October 23, 2026 16:45

Conversation

mkurz commented Oct 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Problem

Change

Alternatives considered

Compatibility

AI disclosure

Test plan

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mkurz commented Oct 23, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

mkurz commented Oct 7, 2026 •

edited

Loading