diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperator.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperator.java index a5f616865f..2517e9b44e 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperator.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperator.java @@ -89,11 +89,24 @@ public class DefaultHttpClientConnectionOperator implements HttpClientConnection static final DetachedSocketFactory PLAIN_SOCKET_FACTORY = socksProxy -> socksProxy == null ? new Socket() : new Socket(socksProxy); + /** + * RFC 8305, Section 8: "When a reasonable value is unknown, implementations SHOULD use a + * Connection Attempt Delay of 250 milliseconds". + */ + static final TimeValue DEFAULT_CONNECTION_ATTEMPT_DELAY = TimeValue.ofMilliseconds(250); + + /** + * RFC 8305, Section 8: "this value SHOULD NOT be below 10 ms". + */ + private static final long MIN_CONNECTION_ATTEMPT_DELAY_MS = 10; + private final DetachedSocketFactory detachedSocketFactory; private final UnixDomainSocketFactory unixDomainSocketFactory = UnixDomainSocketFactory.getSocketFactory(); private final Lookup tlsSocketStrategyLookup; private final SchemePortResolver schemePortResolver; private final DnsResolver dnsResolver; + private final boolean happyEyeballsEnabled; + private final TimeValue connectionAttemptDelay; /** * @deprecated Provided for backward compatibility @@ -115,6 +128,19 @@ public DefaultHttpClientConnectionOperator( final SchemePortResolver schemePortResolver, final DnsResolver dnsResolver, final Lookup tlsSocketStrategyLookup) { + this(detachedSocketFactory, schemePortResolver, dnsResolver, tlsSocketStrategyLookup, false, null); + } + + /** + * @since 5.7 + */ + public DefaultHttpClientConnectionOperator( + final DetachedSocketFactory detachedSocketFactory, + final SchemePortResolver schemePortResolver, + final DnsResolver dnsResolver, + final Lookup tlsSocketStrategyLookup, + final boolean happyEyeballsEnabled, + final TimeValue connectionAttemptDelay) { super(); this.detachedSocketFactory = Args.notNull(detachedSocketFactory, "Plain socket factory"); this.tlsSocketStrategyLookup = Args.notNull(tlsSocketStrategyLookup, "Socket factory registry"); @@ -122,6 +148,13 @@ public DefaultHttpClientConnectionOperator( DefaultSchemePortResolver.INSTANCE; this.dnsResolver = dnsResolver != null ? dnsResolver : SystemDefaultDnsResolver.INSTANCE; + this.happyEyeballsEnabled = happyEyeballsEnabled; + final TimeValue delay = connectionAttemptDelay != null ? connectionAttemptDelay : DEFAULT_CONNECTION_ATTEMPT_DELAY; + if (happyEyeballsEnabled) { + Args.check(delay.toMilliseconds() >= MIN_CONNECTION_ATTEMPT_DELAY_MS, + "Connection attempt delay must be at least %s ms (RFC 8305, Section 8)", MIN_CONNECTION_ATTEMPT_DELAY_MS); + } + this.connectionAttemptDelay = delay; } /** @@ -200,6 +233,51 @@ public void connect( final int port = this.schemePortResolver.resolve(endpointHost.getSchemeName(), endpointHost); remoteAddresses = this.dnsResolver.resolve(endpointHost.getHostName(), port); } + + // A specific, non-ephemeral local port can only ever be bound by one socket at a + // time; racing would bind several attempts to it concurrently, and the OS would + // reject all but one with a bind conflict instead of a real connect failure. Fall + // back to the existing sequential path in that case, exactly as before this feature, + // since only one attempt is ever in flight there. + final boolean canRace = localAddress == null || localAddress.getPort() == 0; + if (happyEyeballsEnabled && canRace && remoteAddresses.size() > 1) { + onBeforeSocketConnect(context, endpointHost); + if (LOG.isDebugEnabled()) { + LOG.debug("{} racing connection attempts across {} ({})", endpointHost, remoteAddresses, connectTimeout); + } + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(connectionAttemptDelay); + final Socket socket = connector.connect(endpointHost, remoteAddresses, (remoteAddress, socketConsumer) -> { + final Socket attemptSocket = detachedSocketFactory.create(endpointHost.getSchemeName(), socksProxy); + socketConsumer.accept(attemptSocket); + try { + if (localAddress != null) { + attemptSocket.bind(localAddress); + } + configureSocket(attemptSocket, socketConfig, true); + attemptSocket.connect(remoteAddress, TimeValue.isPositive(connectTimeout) ? connectTimeout.toMillisecondsIntBound() : 0); + return attemptSocket; + } catch (final IOException | RuntimeException ex) { + Closer.closeQuietly(attemptSocket); + throw ex; + } + }); + try { + conn.bind(socket); + onAfterSocketConnect(context, endpointHost); + if (LOG.isDebugEnabled()) { + LOG.debug("{} {} connected {}->{}", ConnPoolSupport.getId(conn), endpointHost, conn.getLocalAddress(), conn.getRemoteAddress()); + } + final TlsSocketStrategy tlsSocketStrategy = tlsSocketStrategyLookup != null ? tlsSocketStrategyLookup.lookup(endpointHost.getSchemeName()) : null; + if (tlsSocketStrategy != null) { + upgradeToTls(conn, endpointHost, endpointName, connectTimeout, attachment, context, tlsSocketStrategy, socket); + } + } catch (final IOException | RuntimeException ex) { + Closer.closeQuietly(socket); + throw ex; + } + return; + } + for (int i = 0; i < remoteAddresses.size(); i++) { final InetSocketAddress remoteAddress = remoteAddresses.get(i); final boolean last = i == remoteAddresses.size() - 1; diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnector.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnector.java new file mode 100644 index 0000000000..ffbc0cf640 --- /dev/null +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnector.java @@ -0,0 +1,293 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.impl.io; + +import java.io.IOException; +import java.io.InterruptedIOException; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; + +import org.apache.hc.client5.http.ConnectExceptionSupport; +import org.apache.hc.core5.concurrent.DefaultThreadFactory; +import org.apache.hc.core5.io.Closer; +import org.apache.hc.core5.net.NamedEndpoint; +import org.apache.hc.core5.util.TimeValue; + +/** + * Races a blocking connect attempt against each of a host's resolved addresses (RFC 8305, + * Happy Eyeballs v2) for the Classic (blocking) client. A single dispatcher, running on the + * calling thread, starts the first address immediately and each subsequent one on its own + * worker thread, bounded to the number of resolved addresses: it waits up to + * {@link #connectionAttemptDelay} for an outstanding attempt to finish before starting the + * next one, but starts the next one immediately, at the RFC 8305 Section 5 minimum spacing + * (10ms), as soon as an attempt fails. No thread ever sleeps out a delay itself. + * + *

The first attempt to connect successfully wins the race. Because closing a socket is + * the only reliable way to abort a blocking {@link Socket#connect}, every other socket, be it + * already connecting, about to be registered, or not yet created, must end up closed; a + * worker thread registers its socket through a shared, atomic registry rather than a bare + * reference, so a socket created in the brief window around the race being decided is always + * either excluded from cleanup because it does not exist yet, or closed by the registry the + * instant it is registered. {@code ExecutorService#shutdownNow()} only interrupts threads; it + * does not itself abort a blocking connect, so it is never relied on for that.

+ * + * @since 5.7 + */ +final class HappyEyeballsSocketConnector { + + /** + * RFC 8305, Section 5: the minimum spacing between the start of successive connection + * attempts, including one started promptly after an earlier attempt's failure. + */ + private static final long MIN_CONNECTION_ATTEMPT_DELAY_MS = 10; + + /** + * Connects a single socket to {@code remoteAddress}, registering the socket via + * {@code socketConsumer} as soon as it is created and before the (blocking) connect + * call, so that a concurrent winner can close it even while still connecting. + */ + @FunctionalInterface + interface ConnectAttempt { + Socket connect(InetSocketAddress remoteAddress, Consumer socketConsumer) throws IOException; + } + + /** + * Registers each attempt's socket so it can be closed by whichever happens first: this + * race deciding a winner, or the socket being registered. {@link #decide} and + * {@link #register} coordinate purely through {@link #decided} and the per-slot + * {@link AtomicReference}, so neither a socket registered just before {@link #decide} + * runs nor one registered just after is ever left open: {@link #decide} closes whatever + * it finds already registered, and {@link #register} closes its own socket immediately + * if it finds the registry already decided, including re-checking right after + * registering to catch a {@link #decide} that ran in between. + */ + private static final class SocketRegistry { + + private final AtomicReference[] slots; + private final AtomicBoolean decided = new AtomicBoolean(false); + + @SuppressWarnings("unchecked") + SocketRegistry(final int n) { + slots = new AtomicReference[n]; + for (int i = 0; i < n; i++) { + slots[i] = new AtomicReference<>(); + } + } + + void register(final int index, final Socket socket) { + if (decided.get()) { + Closer.closeQuietly(socket); + return; + } + slots[index].set(socket); + if (decided.get() && slots[index].compareAndSet(socket, null)) { + Closer.closeQuietly(socket); + } + } + + /** Closes every registered socket except {@code winner} (may be {@code null}). */ + void decide(final Socket winner) { + decided.set(true); + for (final AtomicReference slot : slots) { + final Socket socket = slot.getAndSet(null); + if (socket != null && socket != winner) { + Closer.closeQuietly(socket); + } + } + } + } + + private final TimeValue connectionAttemptDelay; + + HappyEyeballsSocketConnector(final TimeValue connectionAttemptDelay) { + this.connectionAttemptDelay = connectionAttemptDelay; + } + + Socket connect( + final NamedEndpoint remoteEndpoint, + final List remoteAddresses, + final ConnectAttempt connectAttempt) throws IOException { + final int n = remoteAddresses.size(); + final ExecutorService executor = Executors.newFixedThreadPool( + n, new DefaultThreadFactory("Happy-eyeballs-classic-connect", true)); + final SocketRegistry registry = new SocketRegistry(n); + final List> futures = new ArrayList<>(n); + Socket winner = null; + try { + final ExecutorCompletionService completionService = new ExecutorCompletionService<>(executor); + futures.add(submit(completionService, 0, remoteAddresses, connectAttempt, registry)); + int started = 1; + long lastStartNanos = System.nanoTime(); + IOException lastFailure = null; + + for (int received = 0; received < n && winner == null; ) { + final Future completed; + if (started < n) { + final long waitMs = pendingDelayMs(lastStartNanos, connectionAttemptDelay.toMilliseconds()); + completed = poll(completionService, waitMs); + if (completed == null) { + // Nothing finished within the delay: start the next address now. + futures.add(submit(completionService, started, remoteAddresses, connectAttempt, registry)); + started++; + lastStartNanos = System.nanoTime(); + continue; + } + } else { + completed = take(completionService); + } + received++; + + final Outcome outcome; + try { + outcome = completed.get(); + } catch (final ExecutionException ex) { + final Throwable cause = ex.getCause(); + lastFailure = cause instanceof IOException ? (IOException) cause : new IOException(cause); + continue; + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted while racing connection attempts"); + } + if (outcome.socket != null) { + winner = outcome.socket; + for (final Future future : futures) { + future.cancel(true); + } + break; + } + if (outcome.failure != null) { + lastFailure = outcome.failure; + } + if (started < n) { + // Advance as promptly as RFC 8305 allows, but never closer than the + // minimum spacing to the most recently started attempt. + final long waitMs = pendingDelayMs(lastStartNanos, MIN_CONNECTION_ATTEMPT_DELAY_MS); + if (waitMs > 0) { + sleepOrAbort(waitMs); + } + futures.add(submit(completionService, started, remoteAddresses, connectAttempt, registry)); + started++; + lastStartNanos = System.nanoTime(); + } + } + if (winner != null) { + return winner; + } + throw lastFailure != null ? + ConnectExceptionSupport.enhance(lastFailure, remoteEndpoint) : + new IOException("Connection to " + remoteEndpoint + " failed for an unknown reason"); + } finally { + // The single point of truth for cleanup, covering every exit path (winner found, + // every address failed, or this method itself was interrupted): closes every + // socket except the winner (if any), including ones registered concurrently with + // this call. shutdownNow() only interrupts worker threads; it cannot itself abort + // a blocking Socket#connect, so the registry above is what actually does that. + registry.decide(winner); + executor.shutdownNow(); + } + } + + private static long pendingDelayMs(final long lastStartNanos, final long targetDelayMs) { + final long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastStartNanos); + return Math.max(0, targetDelayMs - elapsedMs); + } + + /** + * Blocks the dispatcher thread itself for a bounded, short minimum-spacing gap (at most + * {@link #MIN_CONNECTION_ATTEMPT_DELAY_MS}). Unlike the per-address stagger this replaces, + * this never delays a worker thread's connect attempt and never exceeds a few + * milliseconds, so it does not amount to the blocking the rest of this class avoids. An + * interruption here must abort racing immediately, exactly like one during {@link #poll} + * or {@link #take}: it is propagated rather than swallowed, so the caller's {@code finally} + * block runs cleanup and no further attempt is started. + */ + private static void sleepOrAbort(final long millis) throws InterruptedIOException { + try { + Thread.sleep(millis); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted while racing connection attempts"); + } + } + + private static Future poll(final ExecutorCompletionService completionService, final long timeoutMs) throws InterruptedIOException { + try { + return completionService.poll(timeoutMs, TimeUnit.MILLISECONDS); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted while racing connection attempts"); + } + } + + private static Future take(final ExecutorCompletionService completionService) throws InterruptedIOException { + try { + return completionService.take(); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted while racing connection attempts"); + } + } + + private static Future submit( + final ExecutorCompletionService completionService, + final int index, + final List remoteAddresses, + final ConnectAttempt connectAttempt, + final SocketRegistry registry) { + final InetSocketAddress remoteAddress = remoteAddresses.get(index); + return completionService.submit(() -> { + try { + final Socket socket = connectAttempt.connect(remoteAddress, socket1 -> registry.register(index, socket1)); + return new Outcome(socket, null); + } catch (final IOException ex) { + return new Outcome(null, ex); + } + }); + } + + private static final class Outcome { + final Socket socket; + final IOException failure; + + Outcome(final Socket socket, final IOException failure) { + this.socket = socket; + this.failure = failure; + } + } + +} diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/PoolingHttpClientConnectionManagerBuilder.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/PoolingHttpClientConnectionManagerBuilder.java index 5256d32033..796eed53c2 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/PoolingHttpClientConnectionManagerBuilder.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/PoolingHttpClientConnectionManagerBuilder.java @@ -65,6 +65,8 @@ public class PoolingHttpClientConnectionManagerBuilder { private TlsSocketStrategy tlsSocketStrategy; private SchemePortResolver schemePortResolver; private DnsResolver dnsResolver; + private boolean happyEyeballsEnabled; + private TimeValue connectionAttemptDelay; private PoolConcurrencyPolicy poolConcurrencyPolicy; private PoolReusePolicy poolReusePolicy; private Resolver socketConfigResolver; @@ -145,6 +147,38 @@ public final PoolingHttpClientConnectionManagerBuilder setDnsResolver(final DnsR return this; } + /** + * Enables RFC 8305 (Happy Eyeballs v2) connection racing for hosts that resolve to more + * than one address: the first address is tried immediately, on its own thread, and each + * subsequent one on its own thread after {@link #setConnectionAttemptDelay(TimeValue)} + * unless an outstanding attempt fails first. The first successful connection wins; every + * other attempt's socket is closed. Disabled ({@code false}) by default, in which case + * addresses continue to be tried sequentially on the calling thread. + * + * @param happyEyeballsEnabled whether to race connection attempts. + * @return this instance. + * @since 5.7 + */ + public final PoolingHttpClientConnectionManagerBuilder setHappyEyeballsEnabled(final boolean happyEyeballsEnabled) { + this.happyEyeballsEnabled = happyEyeballsEnabled; + return this; + } + + /** + * Sets the delay between the start of successive connection attempts when + * {@link #setHappyEyeballsEnabled(boolean)} is enabled. RFC 8305 Section 8 recommends + * 250ms and requires at least 10ms. Has no effect unless Happy Eyeballs racing is enabled. + * + * @param connectionAttemptDelay the delay, or {@code null} to use the RFC 8305 + * recommended default of 250ms. + * @return this instance. + * @since 5.7 + */ + public final PoolingHttpClientConnectionManagerBuilder setConnectionAttemptDelay(final TimeValue connectionAttemptDelay) { + this.connectionAttemptDelay = connectionAttemptDelay; + return this; + } + /** * Sets {@link SchemePortResolver} instance. * @@ -346,7 +380,9 @@ protected HttpClientConnectionOperator createConnectionOperator( dnsResolver, RegistryBuilder.create() .register(URIScheme.HTTPS.id, tlsSocketStrategy) - .build()); + .build(), + happyEyeballsEnabled, + connectionAttemptDelay); } public PoolingHttpClientConnectionManager build() { diff --git a/httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperatorTest.java b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperatorTest.java new file mode 100644 index 0000000000..95bd1ca6f9 --- /dev/null +++ b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperatorTest.java @@ -0,0 +1,194 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.impl.io; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Proxy; +import java.net.ServerSocket; +import java.net.Socket; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.hc.client5.http.DnsResolver; +import org.apache.hc.client5.http.impl.DefaultSchemePortResolver; +import org.apache.hc.client5.http.io.DetachedSocketFactory; +import org.apache.hc.client5.http.ssl.TlsSocketStrategy; +import org.apache.hc.core5.http.HttpHost; +import org.apache.hc.core5.http.config.Lookup; +import org.apache.hc.core5.http.io.SocketConfig; +import org.apache.hc.core5.http.protocol.BasicHttpContext; +import org.apache.hc.core5.util.TimeValue; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +class DefaultHttpClientConnectionOperatorTest { + + /** + * Regression test: when {@code localAddress} specifies a non-zero (explicit) local port, + * Happy Eyeballs racing must not bind more than one socket to it concurrently, which the + * OS would reject for all but one attempt as a bind conflict rather than a real connect + * failure. The operator falls back to the existing sequential path in that case, so a + * dead first address followed by a real, reachable second address must still succeed, + * exactly as it would against master without this feature. + */ + @Test + void happyEyeballsWithExplicitLocalPort_fallsBackToSequential_stillSucceeds() throws Exception { + try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); + ServerSocket deadPortProbe = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + final int listeningPort = server.getLocalPort(); + final int deadPort = deadPortProbe.getLocalPort(); + deadPortProbe.close(); // nothing listens here once closed: instant refusal + + final int explicitLocalPort; + try (ServerSocket localPortProbe = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + explicitLocalPort = localPortProbe.getLocalPort(); + } + + final List remoteAddresses = Arrays.asList( + new InetSocketAddress(InetAddress.getLoopbackAddress(), deadPort), + new InetSocketAddress(InetAddress.getLoopbackAddress(), listeningPort)); + final DnsResolver dnsResolver = Mockito.mock(DnsResolver.class); + Mockito.when(dnsResolver.resolve(Mockito.anyString(), Mockito.anyInt())).thenReturn(remoteAddresses); + + final DefaultHttpClientConnectionOperator operator = new DefaultHttpClientConnectionOperator( + DefaultHttpClientConnectionOperator.PLAIN_SOCKET_FACTORY, + DefaultSchemePortResolver.INSTANCE, + dnsResolver, + (Lookup) name -> null, + true, + TimeValue.ofMilliseconds(20)); + + final DefaultManagedHttpClientConnection conn = new DefaultManagedHttpClientConnection( + "test", null, null, null, null, null, null, null, null); + final InetSocketAddress localAddress = new InetSocketAddress( + InetAddress.getLoopbackAddress(), explicitLocalPort); + + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + operator.connect(conn, new HttpHost("http", "somehost", 80), localAddress, + TimeValue.ofSeconds(5), SocketConfig.DEFAULT, new BasicHttpContext())); + + assertEquals(listeningPort, ((InetSocketAddress) conn.getRemoteAddress()).getPort()); + assertEquals(explicitLocalPort, ((InetSocketAddress) conn.getLocalAddress()).getPort()); + conn.close(); + } + } + + /** + * Regression test: on the Happy Eyeballs racing path, a TLS handshake failure after the + * TCP race has already picked a winner must still close that winning socket, exactly as + * the sequential path's own try/catch around bind + TLS upgrade already does. + */ + @Test + void happyEyeballsRacing_tlsHandshakeFailure_closesWinningSocket() throws Exception { + try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); + ServerSocket deadPortProbe = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + final int listeningPort = server.getLocalPort(); + final int deadPort = deadPortProbe.getLocalPort(); + deadPortProbe.close(); + + final List remoteAddresses = Arrays.asList( + new InetSocketAddress(InetAddress.getLoopbackAddress(), deadPort), + new InetSocketAddress(InetAddress.getLoopbackAddress(), listeningPort)); + final DnsResolver dnsResolver = Mockito.mock(DnsResolver.class); + Mockito.when(dnsResolver.resolve(Mockito.anyString(), Mockito.anyInt())).thenReturn(remoteAddresses); + + final AtomicReference winningSocket = new AtomicReference<>(); + final DetachedSocketFactory trackingFactory = (final Proxy proxy) -> { + final Socket s = new Socket(); + winningSocket.set(s); + return s; + }; + + final TlsSocketStrategy failingTls = Mockito.mock(TlsSocketStrategy.class); + Mockito.when(failingTls.upgrade(Mockito.any(), Mockito.anyString(), Mockito.anyInt(), Mockito.any(), Mockito.any())) + .thenThrow(new IOException("handshake failed")); + final Lookup tlsLookup = scheme -> failingTls; + + final DefaultHttpClientConnectionOperator operator = new DefaultHttpClientConnectionOperator( + trackingFactory, DefaultSchemePortResolver.INSTANCE, dnsResolver, tlsLookup, true, TimeValue.ofMilliseconds(20)); + final DefaultManagedHttpClientConnection conn = new DefaultManagedHttpClientConnection( + "test", null, null, null, null, null, null, null, null); + + assertThrows(IOException.class, () -> + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + operator.connect(conn, new HttpHost("https", "somehost", 443), null, + TimeValue.ofSeconds(5), SocketConfig.DEFAULT, new BasicHttpContext()))); + + assertTrue(winningSocket.get().isClosed(), "the winning socket must be closed when the TLS handshake fails"); + } + } + + /** Same as above, but the TLS strategy throws a RuntimeException rather than an IOException. */ + @Test + void happyEyeballsRacing_tlsHandshakeRuntimeFailure_closesWinningSocket() throws Exception { + try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); + ServerSocket deadPortProbe = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + final int listeningPort = server.getLocalPort(); + final int deadPort = deadPortProbe.getLocalPort(); + deadPortProbe.close(); + + final List remoteAddresses = Arrays.asList( + new InetSocketAddress(InetAddress.getLoopbackAddress(), deadPort), + new InetSocketAddress(InetAddress.getLoopbackAddress(), listeningPort)); + final DnsResolver dnsResolver = Mockito.mock(DnsResolver.class); + Mockito.when(dnsResolver.resolve(Mockito.anyString(), Mockito.anyInt())).thenReturn(remoteAddresses); + + final AtomicReference winningSocket = new AtomicReference<>(); + final DetachedSocketFactory trackingFactory = (final Proxy proxy) -> { + final Socket s = new Socket(); + winningSocket.set(s); + return s; + }; + + final TlsSocketStrategy failingTls = Mockito.mock(TlsSocketStrategy.class); + Mockito.when(failingTls.upgrade(Mockito.any(), Mockito.anyString(), Mockito.anyInt(), Mockito.any(), Mockito.any())) + .thenThrow(new IllegalStateException("handshake blew up")); + final Lookup tlsLookup = scheme -> failingTls; + + final DefaultHttpClientConnectionOperator operator = new DefaultHttpClientConnectionOperator( + trackingFactory, DefaultSchemePortResolver.INSTANCE, dnsResolver, tlsLookup, true, TimeValue.ofMilliseconds(20)); + final DefaultManagedHttpClientConnection conn = new DefaultManagedHttpClientConnection( + "test", null, null, null, null, null, null, null, null); + + assertThrows(IllegalStateException.class, () -> + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + operator.connect(conn, new HttpHost("https", "somehost", 443), null, + TimeValue.ofSeconds(5), SocketConfig.DEFAULT, new BasicHttpContext()))); + + assertTrue(winningSocket.get().isClosed(), "the winning socket must be closed when the TLS handshake throws unexpectedly"); + } + } +} diff --git a/httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnectorTest.java b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnectorTest.java new file mode 100644 index 0000000000..eec39052ec --- /dev/null +++ b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnectorTest.java @@ -0,0 +1,340 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.impl.io; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.io.InterruptedIOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketAddress; +import java.net.SocketException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.hc.core5.http.HttpHost; +import org.apache.hc.core5.util.TimeValue; +import org.junit.jupiter.api.Test; + +class HappyEyeballsSocketConnectorTest { + + /** A socket whose {@link #connect} blocks until {@link #close} is called, then fails, matching the real behavior of closing a socket mid-connect. */ + private static final class BlockingSocket extends Socket { + + private final CountDownLatch closed = new CountDownLatch(1); + volatile boolean connectAttempted; + + @Override + public void connect(final SocketAddress endpoint, final int timeout) throws IOException { + connectAttempted = true; + try { + closed.await(); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + } + throw new SocketException("Socket closed"); + } + + @Override + public synchronized void close() throws IOException { + closed.countDown(); + super.close(); + } + } + + private static List addresses(final int count) throws Exception { + final List out = new ArrayList<>(count); + for (int i = 1; i <= count; i++) { + out.add(new InetSocketAddress(InetAddress.getByAddress(new byte[]{10, 0, 0, (byte) i}), 8080)); + } + return out; + } + + @Test + void singleAddress_connectsWithoutRacing() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final Socket expected = new Socket(); + final AtomicInteger attempts = new AtomicInteger(); + + final Socket actual = connector.connect(new HttpHost("somehost"), addresses(1), (remoteAddress, socketConsumer) -> { + attempts.incrementAndGet(); + socketConsumer.accept(expected); + return expected; + }); + + assertEquals(expected, actual); + assertEquals(1, attempts.get()); + } + + @Test + void secondAddressWinsWhileFirstStalls_loserSocketIsClosed() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final BlockingSocket firstSocket = new BlockingSocket(); + final Socket secondSocket = new Socket(); + + final Socket winner = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + connector.connect(new HttpHost("somehost"), addresses(2), (remoteAddress, socketConsumer) -> { + if (remoteAddress.getAddress().getAddress()[3] == 1) { + socketConsumer.accept(firstSocket); + firstSocket.connect(remoteAddress, 0); // blocks until closed by the race, then throws + return firstSocket; + } + socketConsumer.accept(secondSocket); + return secondSocket; + })); + + assertEquals(secondSocket, winner); + assertTrue(firstSocket.connectAttempted, "the first attempt must have genuinely started connecting"); + assertTrue(firstSocket.isClosed(), "the losing, still-connecting socket must be closed rather than leaked"); + } + + @Test + void allAddressesFail_aggregatedFailure() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + + final IOException ex = assertThrows(IOException.class, () -> + connector.connect(new HttpHost("somehost"), addresses(2), (remoteAddress, socketConsumer) -> { + throw new IOException("refused: " + remoteAddress); + })); + assertTrue(ex.getMessage() != null); + } + + @Test + void firstAddressFailsFast_secondStartsWithoutWaitingOutTheFullDelay() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(250)); + final Socket winnerSocket = new Socket(); + + final long start = System.nanoTime(); + final Socket winner = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + connector.connect(new HttpHost("somehost"), addresses(2), (remoteAddress, socketConsumer) -> { + if (remoteAddress.getAddress().getAddress()[3] == 1) { + throw new IOException("refused"); + } + socketConsumer.accept(winnerSocket); + return winnerSocket; + })); + final long elapsedMs = (System.nanoTime() - start) / 1_000_000; + + // The 2nd address must not have had to wait out the 250ms stagger delay, since the + // 1st address failed fast rather than merely being slow: it only waits the RFC 8305 + // minimum inter-attempt spacing (10ms). + assertEquals(winnerSocket, winner); + assertTrue(elapsedMs < 200, "must advance well before the full stagger delay elapses, took " + elapsedMs + "ms"); + } + + @Test + void interruptionWhileStillRacing_closesLoserSocket() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final BlockingSocket firstSocket = new BlockingSocket(); + final Socket secondSocket = new Socket(); + + final Socket winner = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + connector.connect(new HttpHost("somehost"), addresses(2), (remoteAddress, socketConsumer) -> { + if (remoteAddress.getAddress().getAddress()[3] == 1) { + socketConsumer.accept(firstSocket); + firstSocket.connect(remoteAddress, 0); + return firstSocket; + } + socketConsumer.accept(secondSocket); + return secondSocket; + })); + + assertEquals(secondSocket, winner); + // The executor's shutdownNow() (called in the finally block) interrupts the still + // "connecting" worker; its socket must still end up closed, not leaked. + assertTrue(firstSocket.isClosed()); + } + + @Test + void interruptionDuringMinimumSpacingWait_abortsBeforeStartingNextAddress() throws Exception { + // Regression test: sleepOrAbort() (the minimum-spacing wait after a fast failure) must + // propagate an interruption instead of swallowing it and continuing the race. + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(250)); + final AtomicInteger attempts = new AtomicInteger(); + final CountDownLatch firstFailed = new CountDownLatch(1); + final AtomicReference thrown = new AtomicReference<>(); + + final Thread caller = new Thread(() -> { + try { + connector.connect(new HttpHost("somehost"), addresses(2), (remoteAddress, socketConsumer) -> { + attempts.incrementAndGet(); + if (remoteAddress.getAddress().getAddress()[3] == 1) { + firstFailed.countDown(); + throw new IOException("refused"); + } + return new Socket(); + }); + } catch (final Throwable ex) { + thrown.set(ex); + } + }); + caller.start(); + assertTrue(firstFailed.await(5, TimeUnit.SECONDS)); + caller.interrupt(); // races the <=10ms minimum-spacing wait that follows the failure + caller.join(5000); + + assertFalse(caller.isAlive()); + // Best-effort timing: if the interruption landed while connect() was still running, it + // must be reported as InterruptedIOException, never silently swallowed. + if (thrown.get() != null) { + assertTrue(thrown.get() instanceof InterruptedIOException, + "an interruption during racing must surface as InterruptedIOException: " + thrown.get()); + } + } + + @Test + void interruptingTheCallingThread_abortsAndClosesTheStillConnectingSocket() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final BlockingSocket onlySocket = new BlockingSocket(); + final CountDownLatch connecting = new CountDownLatch(1); + final AtomicReference thrown = new AtomicReference<>(); + + final Thread caller = new Thread(() -> { + try { + connector.connect(new HttpHost("somehost"), addresses(1), (remoteAddress, socketConsumer) -> { + socketConsumer.accept(onlySocket); + connecting.countDown(); + onlySocket.connect(remoteAddress, 0); // blocks until closed or interrupted + return onlySocket; + }); + } catch (final Throwable ex) { + thrown.set(ex); + } + }); + caller.start(); + assertTrue(connecting.await(5, TimeUnit.SECONDS), "the attempt must have genuinely started connecting"); + caller.interrupt(); + caller.join(5000); + + assertFalse(caller.isAlive(), "the calling thread must have returned after being interrupted"); + assertTrue(thrown.get() instanceof InterruptedIOException, + "connect() must report the interruption rather than silently returning: was " + thrown.get()); + assertTrue(onlySocket.isClosed(), "the still-connecting socket must be closed, not leaked, on interruption"); + } + + @Test + void registrationAfterRaceDecided_closesSocketImmediately() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final Socket winnerSocket = new Socket(); + final CountDownLatch registerNow = new CountDownLatch(1); + final CountDownLatch registered = new CountDownLatch(1); + final AtomicReference lateSocketRef = new AtomicReference<>(); + + final Socket winner = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + connector.connect(new HttpHost("somehost"), addresses(2), (remoteAddress, socketConsumer) -> { + if (remoteAddress.getAddress().getAddress()[3] == 1) { + // Registers strictly after the race below has already picked a winner. + awaitUninterruptibly(registerNow); + final Socket lateSocket = new Socket(); + lateSocketRef.set(lateSocket); + socketConsumer.accept(lateSocket); + registered.countDown(); + throw new IOException("too late"); + } + socketConsumer.accept(winnerSocket); + return winnerSocket; + })); + + assertEquals(winnerSocket, winner); + registerNow.countDown(); + assertTrue(registered.await(5, TimeUnit.SECONDS)); + assertTrue(lateSocketRef.get().isClosed(), + "a socket registered strictly after the race is decided must be closed immediately, not left open"); + } + + private static void awaitUninterruptibly(final CountDownLatch latch) { + try { + latch.await(); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + } + } + + @Test + void doesNotRejectSingleAddressBelowStaggerIndex() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final Socket socket = new Socket(); + + final Socket result = connector.connect(new HttpHost("somehost"), addresses(1), (remoteAddress, socketConsumer) -> { + socketConsumer.accept(socket); + return socket; + }); + + assertEquals(socket, result); + assertFalse(socket.isClosed()); + } + + /** + * Real-socket integration test (no mocks, no fakes): one address has nothing listening + * (genuine, OS-level immediate connection refusal) and the other is a real local server + * that accepts. Proves actual fallover over real loopback TCP, not just that the first + * address happens to succeed. + */ + @Test + void realSockets_unreachableAddressFailsOver_toRealListeningServer() throws Exception { + try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + final int listeningPort = server.getLocalPort(); + final int deadPort = findUnusedLoopbackPort(); + + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final List targets = new ArrayList<>(); + targets.add(new InetSocketAddress(InetAddress.getLoopbackAddress(), deadPort)); + targets.add(new InetSocketAddress(InetAddress.getLoopbackAddress(), listeningPort)); + + final Socket winner = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + connector.connect(new HttpHost("localhost"), targets, (remoteAddress, socketConsumer) -> { + final Socket s = new Socket(); + socketConsumer.accept(s); + s.connect(remoteAddress, 2000); + return s; + })); + + assertEquals(listeningPort, winner.getPort()); + assertFalse(winner.isClosed()); + winner.close(); + } + } + + /** Finds a loopback TCP port nothing is listening on, by briefly binding then closing it. */ + private static int findUnusedLoopbackPort() throws IOException { + try (ServerSocket probe = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + return probe.getLocalPort(); + } + } +}