diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index cae3900ee..3a95e1a72 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java @@ -165,6 +165,39 @@ 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. Disabled by default. + * + * @return {@code true} if blocking request body stream reads should be offloaded + * @since 3.0.12 + */ + default boolean isRequestBodyStreamReadOffloadEnabled() { + return false; + } + + /** + * 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()}. + * @since 3.0.12 + */ + 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. + * @since 3.0.12 + */ + 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 3b417a5a3..2a07319ca 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java +++ b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java @@ -35,7 +35,11 @@ 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; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Predicate; @@ -62,6 +66,7 @@ public class DefaultAsyncHttpClient implements AsyncHttpClient { private final AtomicBoolean closed = new AtomicBoolean(false); private final ChannelManager channelManager; private final NettyRequestSender requestSender; + private final @Nullable ExecutorService blockingBodyReadExecutor; private final boolean allowStopNettyTimer; private final Timer nettyTimer; @@ -105,8 +110,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), + executorForBodyReads(blockingBodyReadExecutor)); channelManager.configureBootstraps(requestSender); CookieStore cookieStore = config.getCookieStore(); @@ -127,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()); @@ -134,6 +146,39 @@ private static Timer newNettyTimer(AsyncHttpClientConfig config) { return timer; } + 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 = 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)) { @@ -142,6 +187,9 @@ public void close() { } catch (Throwable t) { LOGGER.warn("Unexpected error on ChannelManager close", t); } + 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 467e3d9ad..c6e814d26 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 550381fbd..e3f6817cf 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/java/org/asynchttpclient/netty/request/NettyRequestFactory.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestFactory.java index b2f23ecbf..cb695da28 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestFactory.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestFactory.java @@ -48,10 +48,12 @@ 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; 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 +70,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 +149,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 +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()); + 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); @@ -180,7 +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()); + nettyBody = inputStreamBody(inStreamGenerator.getInputStream(), inStreamGenerator.getContentLength()); } else if (request.getBodyGenerator() != null) { nettyBody = new NettyBodyBody(request.getBodyGenerator().createBody(), config); } @@ -188,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/NettyRequestSender.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java index 36af9019b..fe22fa77c 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,23 @@ 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); + } + + /** + * 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; this.channelManager = channelManager; connectionSemaphore = config.getConnectionSemaphoreFactory() == null @@ -133,7 +151,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/Http2BodyWriter.java b/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java index 77ebdb8cf..e6c759b9f 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,12 +116,18 @@ 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. 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; @@ -130,9 +135,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 +171,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 { @@ -179,9 +185,10 @@ static void start(Http2StreamChannel channel, ChunkSource source) { * {@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) { @@ -208,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; @@ -232,6 +240,8 @@ private void pump() { } } catch (Throwable t) { finish(t); + } finally { + pumping = false; } } @@ -295,9 +305,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(); } } @@ -308,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(); 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 0b6192f72..d1af1d745 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 66ee644fa..4dee1979d 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 32dbdc0fe..896bd09dd 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,10 @@ import io.netty.channel.ChannelProgressiveFuture; 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; import org.asynchttpclient.netty.request.WriteProgressListener; import org.slf4j.Logger; @@ -29,7 +32,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 +44,50 @@ 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, 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); + } + + 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() { @@ -72,11 +114,21 @@ public void write(Channel channel, NettyResponseFuture future) throws IOExcep future.setStreamConsumed(true); } - channel.write(new ChunkedStream(is), 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) { - closeSilently(is); + closeChunkedInput(input); super.operationComplete(cf); } }); @@ -103,38 +155,43 @@ 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)); + Http2BodyWriter.ChunkSource source = offloadReads + ? new OffloadedInputStreamChunkSource(is, channel.eventLoop(), blockingBodyReadExecutor) + : new DirectInputStreamChunkSource(is); + Http2BodyWriter.start(channel, source, future); } - /** - * Reads an {@link InputStream} in fixed-size chunks for {@link Http2BodyWriter}. - */ - private static final class InputStreamChunkSource implements Http2BodyWriter.ChunkSource { + 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]; - InputStreamChunkSource(InputStream is) { + DirectInputStreamChunkSource(InputStream is) { this.is = is; } @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. int read; do { read = is.read(buffer); } while (read == 0); - if (read == -1) { + if (read < 0) { return null; } ByteBuf buf = alloc.buffer(read); @@ -152,4 +209,261 @@ public void close() { closeSilently(is); } } + + /** + * Reads an {@link InputStream} off the event loop and exposes completed chunks to + * {@link Http2BodyWriter}. + */ + private static final class OffloadedInputStreamChunkSource 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 volatile boolean closed; + private int readableBytes; + private IOException failure; + private Runnable resume; + + OffloadedInputStreamChunkSource(InputStream is, EventExecutor eventLoop, Executor blockingBodyReadExecutor) { + this.is = is; + this.eventLoop = eventLoop; + this.blockingBodyReadExecutor = blockingBodyReadExecutor; + } + + @Override + public ByteBuf nextChunk(ByteBufAllocator alloc) throws IOException { + 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 { + blockingBodyReadExecutor.execute(this::readOffEventLoop); + } catch (RejectedExecutionException e) { + readInProgress = false; + throw new IOException("HTTP/2 request body read executor rejected the read", e); + } + } + + private void readOffEventLoop() { + if (closed) { + return; + } + int read = -1; + IOException thrown = null; + try { + read = read(); + } catch (IOException e) { + thrown = e; + } catch (RuntimeException e) { + thrown = readFailure(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; + } + 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 volatile 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()); + } + + @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; + } + 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 { + blockingBodyReadExecutor.execute(this::readOffEventLoop); + } catch (RejectedExecutionException e) { + readInProgress = false; + throw new IOException("HTTP/1 request body read executor rejected the read", e); + } + } + + private void readOffEventLoop() { + if (closed) { + return; + } + int read = -1; + IOException thrown = null; + try { + read = read(); + } catch (IOException e) { + thrown = e; + } catch (RuntimeException e) { + thrown = readFailure(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/main/resources/org/asynchttpclient/config/ahc-default.properties b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties index 5df97add5..df8c81c80 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=false +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 d125a9fa4..b5d84d186 100644 --- a/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java +++ b/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java @@ -156,6 +156,25 @@ public void testDefaultHashedWheelTimerSize() { testIntegerSystemProperty("hashedWheelTimerSize", "defaultHashedWheelTimerSize", "512"); } + @RepeatedIfExceptionsTest(repeats = 5) + public void testDefaultRequestBodyStreamReadOffloadEnabled() { + assertFalse(AsyncHttpClientConfigDefaults.defaultRequestBodyStreamReadOffloadEnabled()); + testBooleanSystemProperty("requestBodyStreamReadOffloadEnabled", "defaultRequestBodyStreamReadOffloadEnabled", "true"); + AsyncHttpClientConfigHelper.reloadProperties(); + } + + @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 44e7f45df..e1d6c3805 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; @@ -53,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; @@ -64,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; @@ -1040,4 +1046,191 @@ 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() + .setThreadPoolName("ahc-body-read-test") + .setRequestBodyStreamReadOffloadEnabled(true) + .setRequestBodyStreamReadThreadsCount(1) + .setRequestBodyStreamReadQueueSize(1) + .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"); + 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 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().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(), "default configuration should preserve inline event-loop reads"); + }); + } + + @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(config() + .setRequestBodyStreamReadOffloadEnabled(true) + .build())) { + 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()); + } + }); + } + + @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(2) + .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(); + private final AtomicBoolean readOnEventLoop = new AtomicBoolean(); + private final AtomicReference readThreadName = new AtomicReference<>(); + 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; + } + + @Override + public int available() { + return data.length - position; + } + + private void recordReadThread() { + readAttempted.set(true); + EventLoopGroup group = eventLoopGroup.get(); + Thread currentThread = Thread.currentThread(); + readThreadName.compareAndSet(null, currentThread.getName()); + if (group == null) { + return; + } + 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 04b3ff0ab..37d0e23fb 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; @@ -59,15 +61,24 @@ 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; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; import java.util.zip.CRC32; 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; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -295,6 +306,7 @@ private AsyncHttpClient http2Client() { .setUseInsecureTrustManager(true) .setHttp2Enabled(true) .setMaxConnectionsPerHost(1) + .setRequestBodyStreamReadOffloadEnabled(true) .setRequestTimeout(Duration.ofSeconds(60))); } @@ -333,6 +345,121 @@ 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"); + } + + @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(2) + .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); + 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) // ========================================================================= @@ -482,6 +609,57 @@ 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 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 000000000..76f98cec4 --- /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(); + } + } +} 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 55cff4323..d4e6d8bd7 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() { @@ -76,9 +78,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")); } }