diff --git a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java index 7ada0f7fa..a28ee7636 100644 --- a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java +++ b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java @@ -21,6 +21,14 @@ * The control is thread-safe and remains valid until its response completes. Calls made after completion have no * effect. *

+ * 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. + *

* 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. diff --git a/client/src/main/java/org/asynchttpclient/netty/NettyResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/netty/NettyResponseBodyControl.java index 65fae4f8b..555bc69c4 100644 --- a/client/src/main/java/org/asynchttpclient/netty/NettyResponseBodyControl.java +++ b/client/src/main/java/org/asynchttpclient/netty/NettyResponseBodyControl.java @@ -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; @@ -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); } @@ -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(); diff --git a/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java index 819b463f1..8c3a9342b 100644 --- a/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java +++ b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java @@ -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; @@ -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 largeResponseWritten = new CompletableFuture<>(); private final CountDownLatch cancelledStreamClosed = new CountDownLatch(1); @@ -139,6 +143,7 @@ protected void initChannel(Http2StreamChannel streamChannel) { @AfterEach public void stopServer() throws InterruptedException { + anotherThread.shutdownNow(); if (serverChildChannels != null) { serverChildChannels.close().sync(); } @@ -177,6 +182,67 @@ public void suspensionAppliesHttp2FlowControlAndParentIsReused() throws Exceptio } } + @Test + public void earlierSuspendFromAnotherThreadDoesNotUndoLaterResume() throws Exception { + AtomicReference connection = new AtomicReference<>(); + try (AsyncHttpClient client = http2Client(connection::set)) { + RecordingHandler handler = new RecordingHandler(); + ListenableFuture 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 connection = new AtomicReference<>(); + try (AsyncHttpClient client = http2Client(connection::set)) { + RecordingHandler handler = new RecordingHandler(); + ListenableFuture 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()) { @@ -293,13 +359,24 @@ public void suspensionCannotDeferBodylessResponse() throws Exception { } private AsyncHttpClient http2Client() { + return http2Client(ignored -> { + }); + } + + private AsyncHttpClient http2Client(Consumer 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) { diff --git a/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java index 846d8a567..7e7afbfa0 100644 --- a/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java +++ b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java @@ -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; @@ -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 responseContext = new CompletableFuture<>(); private final CompletableFuture firstReplayContext = new CompletableFuture<>(); private final CompletableFuture secondReplayContext = new CompletableFuture<>(); @@ -132,6 +136,7 @@ protected void initChannel(Channel channel) { @AfterEach public void stopServer() throws InterruptedException { + anotherThread.shutdownNow(); if (serverChildChannels != null) { serverChildChannels.close().sync(); } @@ -185,6 +190,64 @@ public void suspensionControlsReadsAndCompletedConnectionIsPooled() throws Excep } } + @Test + public void earlierSuspendFromAnotherThreadDoesNotUndoLaterResume() throws Exception { + AtomicReference clientChannel = new AtomicReference<>(); + try (AsyncHttpClient client = asyncHttpClient(config() + .setRequestTimeout(Duration.ofSeconds(10)) + .setHttpAdditionalChannelInitializer(clientChannel::set))) { + RecordingHandler handler = new RecordingHandler(false); + ListenableFuture 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 clientChannel = new AtomicReference<>(); + try (AsyncHttpClient client = asyncHttpClient(config() + .setRequestTimeout(Duration.ofSeconds(10)) + .setHttpAdditionalChannelInitializer(clientChannel::set))) { + RecordingHandler handler = new RecordingHandler(false); + ListenableFuture 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() @@ -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(); diff --git a/client/src/test/java/org/asynchttpclient/netty/NettyResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/netty/NettyResponseBodyControlTest.java new file mode 100644 index 000000000..e1ba71e68 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/NettyResponseBodyControlTest.java @@ -0,0 +1,280 @@ +/* + * 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.netty; + +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.EventLoop; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.channel.local.LocalChannel; +import io.netty.channel.local.LocalIoHandler; +import org.asynchttpclient.AsyncHandler; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Proxy; +import java.util.Queue; +import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; +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.mockito.Mockito.mock; + +public class NettyResponseBodyControlTest { + + private final NettyResponseFuture future = + new NettyResponseFuture<>(null, mock(AsyncHandler.class), null, 3, null, null, null); + private final AtomicInteger reads = new AtomicInteger(); + private final AtomicInteger suspensionsStarted = new AtomicInteger(); + private final AtomicInteger suspensionsEnded = new AtomicInteger(); + private final ExecutorService anotherThread = Executors.newSingleThreadExecutor(); + + private EventLoopGroup group; + private Channel channel; + + @BeforeEach + public void registerChannel() throws InterruptedException { + group = new MultiThreadIoEventLoopGroup(1, LocalIoHandler.newFactory()); + channel = new LocalChannel(); + channel.pipeline().addLast(new ChannelOutboundHandlerAdapter() { + @Override + public void read(ChannelHandlerContext ctx) { + reads.incrementAndGet(); + ctx.read(); + } + }); + group.register(channel).sync(); + } + + @AfterEach + public void closeChannel() throws InterruptedException { + anotherThread.shutdownNow(); + channel.close().sync(); + group.shutdownGracefully(0, 100, MILLISECONDS).sync(); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void earlierSuspendFromAnotherThreadDoesNotUndoLaterResume(boolean autoRead) throws Exception { + NettyResponseBodyControl control = suspendedControl(autoRead); + + onEventLoop(() -> { + callFromAnotherThread(control::suspend); + control.resume(); + return null; + }); + awaitEventLoop(); + + assertFalse(NettyResponseBodyControl.isSuspended(future)); + assertEquals(autoRead, channel.config().isAutoRead()); + assertEquals(1, reads.get(), "resuming must request exactly one read"); + assertEquals(1, suspensionsStarted.get()); + assertEquals(1, suspensionsEnded.get()); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void resumeFromAnotherThreadIsNotLostToALaterSuspend(boolean autoRead) throws Exception { + // This also passes when no queued suspend() is skipped. It fails if a queued resume() is skipped for a later + // suspend(). + NettyResponseBodyControl control = suspendedControl(autoRead); + control.resume(); + awaitEventLoop(); + + onEventLoop(() -> { + // A callback buffered a part and decided to suspend. Before it did, a consumer took that part and resumed. + callFromAnotherThread(control::resume); + control.suspend(); + return null; + }); + awaitEventLoop(); + + assertFalse(NettyResponseBodyControl.isSuspended(future), "nothing else would resume reads"); + assertEquals(autoRead, channel.config().isAutoRead()); + assertEquals(2, reads.get()); + assertEquals(2, suspensionsStarted.get()); + assertEquals(2, suspensionsEnded.get()); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void queuedCallsFromAnotherThreadEndInTheStateOfTheLatestCall(boolean autoRead) throws Exception { + NettyResponseBodyControl control = suspendedControl(autoRead); + control.resume(); + awaitEventLoop(); + + onEventLoop(() -> { + callFromAnotherThread(control::suspend); + callFromAnotherThread(control::resume); + callFromAnotherThread(control::suspend); + return null; + }); + awaitEventLoop(); + assertTrue(NettyResponseBodyControl.isSuspended(future)); + assertFalse(channel.config().isAutoRead()); + assertEquals(3, suspensionsStarted.get()); + assertEquals(2, suspensionsEnded.get()); + + onEventLoop(() -> { + callFromAnotherThread(control::resume); + callFromAnotherThread(control::suspend); + callFromAnotherThread(control::resume); + return null; + }); + awaitEventLoop(); + assertFalse(NettyResponseBodyControl.isSuspended(future)); + assertEquals(autoRead, channel.config().isAutoRead()); + assertEquals(3, suspensionsStarted.get(), "the skipped suspend() must not start a suspension"); + assertEquals(3, suspensionsEnded.get()); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void suspendAfterASkippedOneStillTakesEffect(boolean autoRead) throws Exception { + NettyResponseBodyControl control = suspendedControl(autoRead); + onEventLoop(() -> { + callFromAnotherThread(control::suspend); + control.resume(); + return null; + }); + awaitEventLoop(); + assertFalse(NettyResponseBodyControl.isSuspended(future)); + + control.suspend(); + awaitEventLoop(); + assertTrue(NettyResponseBodyControl.isSuspended(future)); + assertFalse(channel.config().isAutoRead()); + } + + @Test + public void suspendOnTheEventLoopTakesEffectWhenAResumeClearsTheRequestFirst() { + // A resume() from another thread runs exactly when suspend() asks whether it is on the event loop, that is, + // after suspend() has recorded its request and before it acts on it. + Thread loop = Thread.currentThread(); + Queue queued = new ConcurrentLinkedQueue<>(); + AtomicReference onLoopQuery = new AtomicReference<>(); + InterceptedChannel channel = new InterceptedChannel(loop, queued, onLoopQuery); + NettyResponseBodyControl control = NettyResponseBodyControl.create(future, channel, + suspensionsStarted::incrementAndGet, suspensionsEnded::incrementAndGet, () -> { + }, ignored -> { + }); + try { + onLoopQuery.set(() -> { + try { + callFromAnotherThread(control::resume); + } catch (Exception e) { + throw new IllegalStateException(e); + } + }); + control.suspend(); + assertTrue(NettyResponseBodyControl.isSuspended(future), "a suspend() on the event loop must take effect"); + assertEquals(1, suspensionsStarted.get()); + assertFalse(channel.config().isAutoRead()); + + // The resume from the other thread was queued, and still takes effect afterwards. + Runnable task; + while ((task = queued.poll()) != null) { + task.run(); + } + assertFalse(NettyResponseBodyControl.isSuspended(future)); + assertEquals(1, suspensionsEnded.get()); + } finally { + onLoopQuery.set(null); + NettyResponseBodyControl.complete(future); + channel.finishAndReleaseAll(); + } + } + + private NettyResponseBodyControl suspendedControl(boolean autoRead) throws Exception { + return onEventLoop(() -> { + channel.config().setAutoRead(autoRead); + NettyResponseBodyControl control = NettyResponseBodyControl.create(future, channel, + suspensionsStarted::incrementAndGet, suspensionsEnded::incrementAndGet, () -> { + }, ignored -> { + }); + control.suspend(); + return control; + }); + } + + private T onEventLoop(Callable task) throws Exception { + return channel.eventLoop().submit(task).get(5, SECONDS); + } + + private void awaitEventLoop() throws Exception { + onEventLoop(() -> null); + } + + // 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); + } + + /** + * An {@link EmbeddedChannel} whose event loop is the given thread. Tasks from other threads are queued, and a hook + * runs when the event loop thread asks whether it is on the event loop. + */ + private static final class InterceptedChannel extends EmbeddedChannel { + + private final EventLoop eventLoop; + + InterceptedChannel(Thread loop, Queue queued, AtomicReference onLoopQuery) { + EventLoop real = super.eventLoop(); + eventLoop = (EventLoop) Proxy.newProxyInstance(EventLoop.class.getClassLoader(), + new Class[]{EventLoop.class}, (proxy, method, args) -> { + switch (method.getName()) { + case "inEventLoop": + if (Thread.currentThread() == loop) { + Runnable hook = onLoopQuery.getAndSet(null); + if (hook != null) { + hook.run(); + } + } + return (args == null || args.length == 0 ? Thread.currentThread() : args[0]) == loop; + case "execute": + queued.add((Runnable) args[0]); + return null; + default: + try { + return method.invoke(real, args); + } catch (InvocationTargetException e) { + throw e.getCause(); + } + } + }); + } + + @Override + public EventLoop eventLoop() { + return eventLoop != null ? eventLoop : super.eventLoop(); + } + } +}