From 56be6a057e6be4323b145e68f0764c320e3a8d53 Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Sun, 28 Jun 2026 19:17:08 +0000 Subject: [PATCH 1/5] Auto-detect native transport by default when its library is available --- .../netty/channel/ChannelManager.java | 19 ++++++++++++++++++- .../DefaultAsyncHttpClientTest.java | 15 +++++++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index 914e89945..c5fc8e3e8 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -208,7 +208,7 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) { if (config.isUseNativeTransport()) { transportFactory = getNativeTransportFactory(config); } else { - transportFactory = NioTransportFactory.INSTANCE; + transportFactory = autoSelectTransportFactory(); } eventLoopGroup = transportFactory.newEventLoopGroup(config.getIoThreadsCount(), threadFactory); } else { @@ -267,6 +267,23 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) { return NioTransportFactory.INSTANCE; } + // Default when useNativeTransport is unset: native transport if its lib is on the classpath (silently), + // else NIO. Use -Dio.netty.transport.noNative=true to force NIO. + private static TransportFactory autoSelectTransportFactory() { + if (PlatformDependent.isOsx()) { + if (KQueueTransportFactory.isAvailable()) { + return new KQueueTransportFactory(); + } + } else if (!PlatformDependent.isWindows()) { + if (IoUringTransportFactory.isAvailable()) { + return new IoUringTransportFactory(); + } else if (EpollTransportFactory.isAvailable()) { + return new EpollTransportFactory(); + } + } + return NioTransportFactory.INSTANCE; + } + public static boolean isSslHandlerConfigured(ChannelPipeline pipeline) { return pipeline.get(SSL_HANDLER) != null; } diff --git a/client/src/test/java/org/asynchttpclient/DefaultAsyncHttpClientTest.java b/client/src/test/java/org/asynchttpclient/DefaultAsyncHttpClientTest.java index 1b613ac35..0070e06b4 100644 --- a/client/src/test/java/org/asynchttpclient/DefaultAsyncHttpClientTest.java +++ b/client/src/test/java/org/asynchttpclient/DefaultAsyncHttpClientTest.java @@ -105,6 +105,21 @@ public void testNativeTransportFallsBackToNioWhenNativeUnavailable() throws IOEx } } + @RepeatedIfExceptionsTest(repeats = 5) + @EnabledOnOs(OS.LINUX) + public void testAutoSelectsNativeTransportByDefaultWhenAvailable() throws IOException { + AsyncHttpClientConfig config = config().build(); + try (DefaultAsyncHttpClient client = (DefaultAsyncHttpClient) asyncHttpClient(config)) { + EventLoopGroup group = client.channelManager().getEventLoopGroup(); + boolean nativeAvailable = Epoll.isAvailable() || IoUring.isAvailable(); + if (nativeAvailable) { + assertFalse(group instanceof NioEventLoopGroup, "default config must auto-select native transport when available"); + } else { + assertInstanceOf(NioEventLoopGroup.class, group, "no native transport available -> NIO"); + } + } + } + @RepeatedIfExceptionsTest(repeats = 5) public void testUseOnlyEpollNativeTransportButNativeTransportIsDisabled() { assertThrows(IllegalArgumentException.class, () -> config().setUseNativeTransport(false).setUseOnlyEpollNativeTransport(true).build()); From 0e31b496eaecdb512d2ffc7397e57f49a83d97d0 Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Sun, 2 Aug 2026 13:56:58 +0000 Subject: [PATCH 2/5] Native transport fixes: multipart backpressure, exception annotation, io_uring fallback Fixes for PR #2216 CI failures on io_uring: - InputStreamMultipartPart: prevent infinite loop with slowTarget backpressure - StackTraceInspector: detect native transport connection resets via errno matching - NettyChannelConnector: annotate connect failures with host:port for io_uring - ChannelManager: fallback from io_uring to epoll on ring allocation failure (io_uring rings charge against RLIMIT_MEMLOCK, not always available) Rebased on #2288 (Fix per-host connection permit leak) for proper semaphore handling. --- .../netty/channel/ChannelManager.java | 41 ++++++++++++++----- .../netty/channel/NettyChannelConnector.java | 14 ++++++- .../netty/future/StackTraceInspector.java | 25 ++++++++++- .../part/InputStreamMultipartPart.java | 38 ++++++++++++++--- 4 files changed, 100 insertions(+), 18 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index c5fc8e3e8..a359b297e 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -129,6 +129,8 @@ public class ChannelManager { // Guards the one-time WARN emitted when a native transport was requested but is unavailable and we // fall back to NIO. Logged once per JVM to avoid spamming logs when many clients are created. private static final AtomicBoolean NATIVE_FALLBACK_WARNED = new AtomicBoolean(); + // Guards the one-time WARN emitted when io_uring allocation fails and we fall back to epoll. + private static final AtomicBoolean IO_URING_FALLBACK_WARNED = new AtomicBoolean(); private final AsyncHttpClientConfig config; private final SslEngineFactory sslEngineFactory; private final EventLoopGroup eventLoopGroup; @@ -203,6 +205,7 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) { ThreadFactory threadFactory = config.getThreadFactory() != null ? config.getThreadFactory() : new DefaultThreadFactory(config.getThreadPoolName()); allowReleaseEventLoopGroup = config.getEventLoopGroup() == null; TransportFactory transportFactory; + EventLoopGroup localEventLoopGroup; if (allowReleaseEventLoopGroup) { if (config.isUseNativeTransport()) { @@ -210,23 +213,39 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) { } else { transportFactory = autoSelectTransportFactory(); } - eventLoopGroup = transportFactory.newEventLoopGroup(config.getIoThreadsCount(), threadFactory); + try { + localEventLoopGroup = transportFactory.newEventLoopGroup(config.getIoThreadsCount(), threadFactory); + } catch (Throwable t) { + if (transportFactory instanceof IoUringTransportFactory && EpollTransportFactory.isAvailable()) { + if (IO_URING_FALLBACK_WARNED.compareAndSet(false, true)) { + LOGGER.warn("io_uring event loop group creation failed ({}); falling back to epoll. " + + "io_uring rings count against RLIMIT_MEMLOCK (~76 KB per io thread, " + + "{} threads requested); raise 'ulimit -l' to use io_uring.", + t, config.getIoThreadsCount()); + } + transportFactory = new EpollTransportFactory(); + localEventLoopGroup = transportFactory.newEventLoopGroup(config.getIoThreadsCount(), threadFactory); + } else { + throw t; + } + } } else { - eventLoopGroup = config.getEventLoopGroup(); + localEventLoopGroup = config.getEventLoopGroup(); - if (eventLoopGroup instanceof NioEventLoopGroup) { + if (localEventLoopGroup instanceof NioEventLoopGroup) { transportFactory = NioTransportFactory.INSTANCE; - } else if (isInstanceof(eventLoopGroup, "io.netty.channel.epoll.EpollEventLoopGroup")) { + } else if (isInstanceof(localEventLoopGroup, "io.netty.channel.epoll.EpollEventLoopGroup")) { transportFactory = new EpollTransportFactory(); - } else if (isInstanceof(eventLoopGroup, "io.netty.channel.kqueue.KQueueEventLoopGroup")) { + } else if (isInstanceof(localEventLoopGroup, "io.netty.channel.kqueue.KQueueEventLoopGroup")) { transportFactory = new KQueueTransportFactory(); - } else if (isInstanceof(eventLoopGroup, "io.netty.channel.uring.IOUringEventLoopGroup")) { + } else if (isInstanceof(localEventLoopGroup, "io.netty.channel.uring.IOUringEventLoopGroup")) { transportFactory = new IoUringTransportFactory(); } else { - throw new IllegalArgumentException("Unknown event loop group " + eventLoopGroup.getClass().getSimpleName()); + throw new IllegalArgumentException("Unknown event loop group " + localEventLoopGroup.getClass().getSimpleName()); } } + this.eventLoopGroup = localEventLoopGroup; channelOptions = buildChannelOptions(config); httpBootstrap = newBootstrap(transportFactory, eventLoopGroup); wsBootstrap = newBootstrap(transportFactory, eventLoopGroup); @@ -275,10 +294,12 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) { return new KQueueTransportFactory(); } } else if (!PlatformDependent.isWindows()) { - if (IoUringTransportFactory.isAvailable()) { - return new IoUringTransportFactory(); - } else if (EpollTransportFactory.isAvailable()) { + // Prefer epoll over io_uring: io_uring requires RLIMIT_MEMLOCK (76 KB per io thread) + // which is often constrained in containers/CI. epoll is native transport without that overhead. + if (EpollTransportFactory.isAvailable()) { return new EpollTransportFactory(); + } else if (IoUringTransportFactory.isAvailable()) { + return new IoUringTransportFactory(); } } return NioTransportFactory.INSTANCE; diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java b/client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java index 20a85b8ac..2b14d7d83 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java @@ -23,6 +23,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.net.ConnectException; import java.net.InetAddress; import java.net.InetSocketAddress; import java.util.List; @@ -118,9 +119,20 @@ public void onFailure(Channel channel, Throwable t) { if (retry) { connect(bootstrap, connectListener); } else { - connectListener.onFailure(channel, t); + connectListener.onFailure(channel, annotateConnectException(t, remoteAddress)); } } }); } + + private Throwable annotateConnectException(Throwable t, InetSocketAddress remoteAddress) { + String address = remoteAddress.toString(); + String message = t.getMessage(); + if (message != null && message.contains(address)) { + return t; + } + ConnectException annotated = new ConnectException((message != null ? message : t.getClass().getSimpleName()) + ": " + address); + annotated.initCause(t); + return annotated; + } } diff --git a/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java b/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java index 28a0f359d..b7b1c3141 100755 --- a/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java +++ b/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java @@ -56,7 +56,14 @@ public static boolean recoverOnNettyDisconnectException(Throwable t) { public static boolean recoverOnReadOrWriteException(Throwable t) { while (true) { - if (t instanceof IOException && "Connection reset by peer".equalsIgnoreCase(t.getMessage())) { + if (t instanceof IOException) { + String msg = t.getMessage(); + if (msg != null && msg.contains("Connection reset")) { + return true; + } + } + + if (isNativeIoException(t)) { return true; } @@ -77,4 +84,20 @@ public static boolean recoverOnReadOrWriteException(Throwable t) { t = t.getCause(); } } + + private static boolean isNativeIoException(Throwable t) { + if (t == null) { + return false; + } + String className = t.getClass().getName(); + if ("io.netty.channel.unix.Errors$NativeIoException".equals(className)) { + try { + java.lang.reflect.Method expectedErr = t.getClass().getMethod("expectedErr"); + int errno = (Integer) expectedErr.invoke(t); + return errno == -54 || errno == -104 || errno == -111; + } catch (Exception ignore) { + } + } + return false; + } } diff --git a/client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java b/client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java index cf1acb0a7..932abe10a 100644 --- a/client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java +++ b/client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java @@ -33,6 +33,7 @@ public class InputStreamMultipartPart extends FileLikeMultipartPart 0) { + if (buffer.position() > 0) { buffer.flip(); - while (buffer.hasRemaining()) { - transferred += target.write(buffer); + int written = target.write(buffer); + if (written > 0) { + transferred += written; + position += written; } buffer.compact(); - position += transferred; + if (written == 0) { + slowTarget = true; + return 0; + } + } + + if (!sourceExhausted) { + int read = channel.read(buffer); + if (read > 0) { + buffer.flip(); + int written = target.write(buffer); + if (written > 0) { + transferred += written; + position += written; + } + buffer.compact(); + if (written == 0) { + slowTarget = true; + } + } else if (read < 0) { + sourceExhausted = true; + } } - if (position == getContentLength() || read < 0) { + + // Only transition to POST_CONTENT when source is exhausted AND buffer is fully drained. + // Never transition while unwritten bytes remain, to avoid data loss. + if (sourceExhausted && buffer.position() == 0) { state = MultipartState.POST_CONTENT; if (channel.isOpen()) { channel.close(); From c9326b588177e499e5ad3a48d62c9d8125cb6e90 Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Sun, 2 Aug 2026 18:15:12 +0000 Subject: [PATCH 3/5] Fix C1+C2 critical regressions: hang and retry break C1: InputStreamMultipartPart - restore position check to prevent hang on streams that don't EOF after declared length (socket-backed, etc) C2: annotateConnectException - preserve ConnectException type to avoid breaking retry predicates that key on ClosedChannelException or SslHandler.disconnect stack frames --- .../org/asynchttpclient/netty/channel/ChannelManager.java | 3 +-- .../netty/channel/NettyChannelConnector.java | 6 +++--- .../body/multipart/part/InputStreamMultipartPart.java | 6 +++--- 3 files changed, 7 insertions(+), 8 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index a359b297e..72a515c23 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -294,8 +294,7 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) { return new KQueueTransportFactory(); } } else if (!PlatformDependent.isWindows()) { - // Prefer epoll over io_uring: io_uring requires RLIMIT_MEMLOCK (76 KB per io thread) - // which is often constrained in containers/CI. epoll is native transport without that overhead. + // Prefer epoll: io_uring needs RLIMIT_MEMLOCK (often constrained in CI/containers). if (EpollTransportFactory.isAvailable()) { return new EpollTransportFactory(); } else if (IoUringTransportFactory.isAvailable()) { diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java b/client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java index 2b14d7d83..7a6eeda1b 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java @@ -126,11 +126,11 @@ public void onFailure(Channel channel, Throwable t) { } private Throwable annotateConnectException(Throwable t, InetSocketAddress remoteAddress) { + if (t instanceof ConnectException) { + return t; // Already has proper type; preserve for retry predicates + } String address = remoteAddress.toString(); String message = t.getMessage(); - if (message != null && message.contains(address)) { - return t; - } ConnectException annotated = new ConnectException((message != null ? message : t.getClass().getSimpleName()) + ": " + address); annotated.initCause(t); return annotated; diff --git a/client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java b/client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java index 932abe10a..365bf7305 100644 --- a/client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java +++ b/client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java @@ -111,9 +111,9 @@ protected long transferContentTo(WritableByteChannel target) throws IOException } } - // Only transition to POST_CONTENT when source is exhausted AND buffer is fully drained. - // Never transition while unwritten bytes remain, to avoid data loss. - if (sourceExhausted && buffer.position() == 0) { + // Don't close until all declared bytes are written, even if source doesn't EOF (socket-backed streams). + boolean allDeclaredBytesWritten = getContentLength() >= 0 && position >= getContentLength(); + if ((sourceExhausted || allDeclaredBytesWritten) && buffer.position() == 0) { state = MultipartState.POST_CONTENT; if (channel.isOpen()) { channel.close(); From afbb92fb4fa44998168921d6791b230fdd8a25fd Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Sun, 2 Aug 2026 19:55:34 +0000 Subject: [PATCH 4/5] Drop errno matching and document native transport default --- .../AsyncHttpClientConfig.java | 7 ++++++ .../DefaultAsyncHttpClientConfig.java | 9 ++++++++ .../netty/future/StackTraceInspector.java | 23 +++---------------- 3 files changed, 19 insertions(+), 20 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index cae3900ee..e0c665bbc 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java @@ -436,6 +436,13 @@ default AddressResolverGroup getAddressResolverGroup() { return null; } + /** + * Whether a native transport was explicitly requested. Note that {@code false} no longer forces NIO: + * a native transport is auto-selected whenever its library is on the classpath. Set + * {@code -Dio.netty.transport.noNative=true} to force NIO. + * + * @return true if a native transport was explicitly requested + */ boolean isUseNativeTransport(); boolean isUseOnlyEpollNativeTransport(); diff --git a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java index 467e3d9ad..6e2efd9c4 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java @@ -1643,6 +1643,15 @@ public Builder setAddressResolverGroup(@Nullable AddressResolverGroup + * Passing {@code false} does not force NIO: when unset, a native transport is still auto-selected + * if its library is on the classpath. Use {@code -Dio.netty.transport.noNative=true} to force NIO. + * + * @param useNativeTransport whether to explicitly request a native transport + * @return the same builder instance + */ public Builder setUseNativeTransport(boolean useNativeTransport) { this.useNativeTransport = useNativeTransport; return this; diff --git a/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java b/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java index b7b1c3141..487afe507 100755 --- a/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java +++ b/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java @@ -56,6 +56,9 @@ public static boolean recoverOnNettyDisconnectException(Throwable t) { public static boolean recoverOnReadOrWriteException(Throwable t) { while (true) { + // Native transports (epoll, io_uring, kqueue) report resets as NativeIoException with the + // strerror text baked into the message, e.g. "recvAddress(..) failed with error(-104): + // Connection reset by peer". Modern JDKs drop the "by peer" suffix, hence the substring match. if (t instanceof IOException) { String msg = t.getMessage(); if (msg != null && msg.contains("Connection reset")) { @@ -63,10 +66,6 @@ public static boolean recoverOnReadOrWriteException(Throwable t) { } } - if (isNativeIoException(t)) { - return true; - } - try { for (StackTraceElement element : t.getStackTrace()) { String className = element.getClassName(); @@ -84,20 +83,4 @@ public static boolean recoverOnReadOrWriteException(Throwable t) { t = t.getCause(); } } - - private static boolean isNativeIoException(Throwable t) { - if (t == null) { - return false; - } - String className = t.getClass().getName(); - if ("io.netty.channel.unix.Errors$NativeIoException".equals(className)) { - try { - java.lang.reflect.Method expectedErr = t.getClass().getMethod("expectedErr"); - int errno = (Integer) expectedErr.invoke(t); - return errno == -54 || errno == -104 || errno == -111; - } catch (Exception ignore) { - } - } - return false; - } } From 74d8c0f9d11ca906389e179ec69dd9d51d8e8cb8 Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Sun, 2 Aug 2026 20:44:12 +0000 Subject: [PATCH 5/5] Cover multipart transfer to a target that refuses writes The existing zero-copy test drains every write, so nothing exercised the path io_uring takes when its staging buffer fills. Adds a bounded channel that returns 0 until the caller yields, which is what #2216 spun on. Catches a second bug it turned up: the part still read once past the declared length, so a socket-backed stream would block there. --- .../part/InputStreamMultipartPart.java | 10 +- .../body/multipart/MultipartBodyTest.java | 167 +++++++++++++++++- 2 files changed, 173 insertions(+), 4 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java b/client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java index 365bf7305..9a0dc97fc 100644 --- a/client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java +++ b/client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java @@ -93,7 +93,12 @@ protected long transferContentTo(WritableByteChannel target) throws IOException } } - if (!sourceExhausted) { + // Stop reading once the declared length is accounted for: a socket-backed stream has nothing more + // to give and would block that read until the request times out. + long contentLength = getContentLength(); + boolean allBytesInHand = contentLength >= 0 && position + buffer.position() >= contentLength; + + if (!sourceExhausted && !allBytesInHand) { int read = channel.read(buffer); if (read > 0) { buffer.flip(); @@ -111,8 +116,7 @@ protected long transferContentTo(WritableByteChannel target) throws IOException } } - // Don't close until all declared bytes are written, even if source doesn't EOF (socket-backed streams). - boolean allDeclaredBytesWritten = getContentLength() >= 0 && position >= getContentLength(); + boolean allDeclaredBytesWritten = contentLength >= 0 && position >= contentLength; if ((sourceExhausted || allDeclaredBytesWritten) && buffer.position() == 0) { state = MultipartState.POST_CONTENT; if (channel.isOpen()) { diff --git a/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartBodyTest.java b/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartBodyTest.java index 54e0dbf5a..723094837 100644 --- a/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartBodyTest.java +++ b/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartBodyTest.java @@ -18,9 +18,13 @@ import io.github.artsok.RepeatedIfExceptionsTest; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; +import io.netty.handler.codec.http.DefaultHttpHeaders; import io.netty.handler.codec.http.EmptyHttpHeaders; +import io.netty.handler.codec.http.HttpHeaders; import org.asynchttpclient.request.body.Body.BodyState; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; import java.io.BufferedInputStream; import java.io.File; import java.io.FileInputStream; @@ -31,13 +35,19 @@ import java.net.URL; import java.nio.ByteBuffer; import java.nio.channels.WritableByteChannel; +import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.concurrent.atomic.AtomicLong; +import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE; import static java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; public class MultipartBodyTest { @@ -69,6 +79,10 @@ private static File getTestfile() throws URISyntaxException { } private static MultipartBody buildMultipart() { + return buildMultipart(EmptyHttpHeaders.INSTANCE); + } + + private static MultipartBody buildMultipart(HttpHeaders requestHeaders) { List parts = new ArrayList<>(PARTS); try { File testFile = getTestfile(); @@ -77,7 +91,14 @@ private static MultipartBody buildMultipart() { } catch (URISyntaxException | FileNotFoundException e) { throw new ExceptionInInitializerError(e); } - return MultipartUtils.newMultipartBody(parts, EmptyHttpHeaders.INSTANCE); + return MultipartUtils.newMultipartBody(parts, requestHeaders); + } + + /** + * Pins the boundary so two serializations of the same parts can be compared byte for byte. + */ + private static HttpHeaders pinnedBoundary() { + return new DefaultHttpHeaders().add(CONTENT_TYPE, "multipart/form-data; boundary=pinnedTestBoundary"); } private static long transferWithCopy(MultipartBody multipartBody, int bufferSize) throws IOException { @@ -150,6 +171,150 @@ public void transferZeroCopy() throws Exception { } } + /** + * Mimics io_uring's ByteBufWritableByteChannel: it stages into a fixed-size buffer and returns 0 once + * that buffer is full, rather than blocking until the socket drains. The unbounded mock used by + * {@link #transferZeroCopy} never exercises that path, which is how issue #2216 shipped. + */ + private static final class BoundedChannel implements WritableByteChannel { + + private final ByteArrayOutputStream written = new ByteArrayOutputStream(); + private final int chunkCapacity; + private int remainingInChunk; + + BoundedChannel(int chunkCapacity) { + this.chunkCapacity = chunkCapacity; + remainingInChunk = chunkCapacity; + } + + @Override + public int write(ByteBuffer src) { + if (remainingInChunk == 0) { + // Stays refused for the rest of this transferTo, exactly like io_uring: the staging buffer + // is only flushed once we hand control back, so spinning here never makes progress. + return 0; + } + int count = Math.min(src.remaining(), remainingInChunk); + byte[] chunk = new byte[count]; + src.get(chunk); + written.write(chunk, 0, count); + remainingInChunk -= count; + return count; + } + + /** + * Models Netty flushing the staging buffer between transferTo calls. + */ + void refill() { + remainingInChunk = chunkCapacity; + } + + @Override + public boolean isOpen() { + return true; + } + + @Override + public void close() { + } + + byte[] toByteArray() { + return written.toByteArray(); + } + } + + private static byte[] drain(MultipartBody body, BoundedChannel target, int maxIterations) throws IOException { + int iterations = 0; + while (body.transferTo(target) != -1L) { + target.refill(); + assertTrue(++iterations < maxIterations, + "transferTo did not finish within " + maxIterations + " calls; it is not making progress"); + } + return target.toByteArray(); + } + + /** + * A target that refuses writes must not cost bytes and must not spin. Sweeps the chunk size so the + * refusal lands at a different offset each time, including mid-part and mid-boundary. + */ + @RepeatedIfExceptionsTest(repeats = 5) + public void transferZeroCopyToTargetThatRefusesWrites() { + // A part that spins on a refusing target never returns, so bound the whole sweep in wall time: + // that is the shape issue #2216 took, and an assertion cannot observe it from the inside. + assertTimeoutPreemptively(Duration.ofSeconds(30), () -> { + byte[] expected; + try (MultipartBody reference = buildMultipart(pinnedBoundary())) { + BoundedChannel unbounded = new BoundedChannel(Integer.MAX_VALUE); + expected = drain(reference, unbounded, 10_000); + assertEquals(reference.getContentLength(), expected.length); + } + + for (int chunkCapacity : new int[]{1, 2, 7, 64, 511, 4096, 65536}) { + try (MultipartBody multipartBody = buildMultipart(pinnedBoundary())) { + BoundedChannel target = new BoundedChannel(chunkCapacity); + // Worst case is one byte per call plus a refusal between every chunk, hence the generous bound. + byte[] actual = drain(multipartBody, target, (int) (multipartBody.getContentLength() * 2 + 1000)); + assertEquals(multipartBody.getContentLength(), actual.length, + "chunkCapacity=" + chunkCapacity + ": wrong number of bytes reached the target"); + assertArrayEquals(expected, actual, + "chunkCapacity=" + chunkCapacity + ": body differs from the reference serialization"); + } + } + }); + } + + /** + * A stream that hands over exactly its declared length must finish without the part reading again for + * EOF. Socket-backed streams have nothing more to give and would block that extra read forever. + */ + @RepeatedIfExceptionsTest(repeats = 5) + public void inputStreamPartFinishesOnDeclaredLengthWithoutWaitingForEof() { + assertTimeoutPreemptively(Duration.ofSeconds(30), () -> { + byte[] content = "declared length, no EOF to follow".getBytes(UTF_8); + NoEofStream stream = new NoEofStream(content); + + List parts = new ArrayList<>(); + parts.add(new InputStreamPart("isPart", stream, "fileName", content.length)); + + try (MultipartBody multipartBody = MultipartUtils.newMultipartBody(parts, pinnedBoundary())) { + BoundedChannel target = new BoundedChannel(8); + byte[] actual = drain(multipartBody, target, (int) (multipartBody.getContentLength() * 2 + 1000)); + assertEquals(multipartBody.getContentLength(), actual.length); + assertFalse(stream.readPastDeclaredLength, + "the part read past the declared length; a socket-backed stream would block there"); + } + }); + } + + /** + * Returns EOF past its content so a regression fails an assertion instead of hanging the build, but + * records that it was asked. + */ + private static final class NoEofStream extends ByteArrayInputStream { + + private final int declaredLength; + private int delivered; + boolean readPastDeclaredLength; + + NoEofStream(byte[] content) { + super(content); + declaredLength = content.length; + } + + @Override + public synchronized int read(byte[] b, int off, int len) { + if (delivered >= declaredLength) { + readPastDeclaredLength = true; + return -1; + } + int read = super.read(b, off, len); + if (read > 0) { + delivered += read; + } + return read; + } + } + @RepeatedIfExceptionsTest(repeats = 5) public void finishingChunkReportsStopAndCarriesAllBytes() throws Exception { try (MultipartBody multipartBody = buildMultipart()) {