Offload request body stream reads

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}
      *

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,13 +134,51 @@ 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());
         timer.start();
         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();

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<Channel> 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<Channel> 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<Channel> 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);
         }

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);
     }

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,14 +189,20 @@ 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);
         }
 
         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

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,14 +127,31 @@ 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
                 ? new DefaultConnectionSemaphoreFactory().newConnectionSemaphore(config)
                 : 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();