diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index cae3900ee0..e0c665bbc9 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 467e3d9ad3..6e2efd9c40 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/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index 914e89945e..72a515c231 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,30 +205,47 @@ 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()) { transportFactory = getNativeTransportFactory(config); } else { - transportFactory = NioTransportFactory.INSTANCE; + transportFactory = autoSelectTransportFactory(); + } + 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; + } } - eventLoopGroup = transportFactory.newEventLoopGroup(config.getIoThreadsCount(), threadFactory); } 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); @@ -267,6 +286,24 @@ 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()) { + // Prefer epoll: io_uring needs RLIMIT_MEMLOCK (often constrained in CI/containers). + if (EpollTransportFactory.isAvailable()) { + return new EpollTransportFactory(); + } else if (IoUringTransportFactory.isAvailable()) { + return new IoUringTransportFactory(); + } + } + return NioTransportFactory.INSTANCE; + } + public static boolean isSslHandlerConfigured(ChannelPipeline pipeline) { return pipeline.get(SSL_HANDLER) != null; } 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 20a85b8ac4..7a6eeda1b7 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) { + if (t instanceof ConnectException) { + return t; // Already has proper type; preserve for retry predicates + } + String address = remoteAddress.toString(); + String message = t.getMessage(); + 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 28a0f359de..487afe5072 100755 --- a/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java +++ b/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java @@ -56,8 +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())) { - return 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")) { + return true; + } } try { 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 cf1acb0a78..9a0dc97fc5 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 (position == getContentLength() || read < 0) { + + // 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(); + 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; + } + } + + boolean allDeclaredBytesWritten = contentLength >= 0 && position >= contentLength; + if ((sourceExhausted || allDeclaredBytesWritten) && buffer.position() == 0) { state = MultipartState.POST_CONTENT; if (channel.isOpen()) { channel.close(); diff --git a/client/src/test/java/org/asynchttpclient/DefaultAsyncHttpClientTest.java b/client/src/test/java/org/asynchttpclient/DefaultAsyncHttpClientTest.java index 1b613ac359..0070e06b42 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()); 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 54e0dbf5a4..7230948377 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()) {