diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/AbstractConnectionPool.java b/jetty-client/src/main/java/org/eclipse/jetty/client/AbstractConnectionPool.java index e6ca06f27ec..327c9c7ee0d 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/AbstractConnectionPool.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/AbstractConnectionPool.java @@ -275,7 +275,7 @@ public abstract class AbstractConnectionPool extends ContainerLifeCycle implemen { pending.decrementAndGet(); if (LOG.isDebugEnabled()) - LOG.debug("Not creating connection as pool is full, pending: {}", pending); + LOG.debug("Not creating connection as pool {} is full, pending: {}", pool, pending); return; } @@ -517,15 +517,17 @@ public abstract class AbstractConnectionPool extends ContainerLifeCycle implemen @Override public String toString() { - return String.format("%s@%x[c=%d/%d/%d,a=%d,i=%d,q=%d]", + return String.format("%s@%x[s=%s,c=%d/%d/%d,a=%d,i=%d,q=%d,p=%s]", getClass().getSimpleName(), hashCode(), + getState(), getPendingConnectionCount(), getConnectionCount(), getMaxConnectionCount(), getActiveConnectionCount(), getIdleConnectionCount(), - destination.getQueuedRequestCount()); + destination.getQueuedRequestCount(), + pool); } private class FutureConnection extends Promise.Completable diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpClient.java b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpClient.java index 93c5ea5c3ee..21faa61bb6f 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpClient.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpClient.java @@ -74,6 +74,7 @@ import org.eclipse.jetty.util.ssl.SslContextFactory; import org.eclipse.jetty.util.thread.QueuedThreadPool; import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler; import org.eclipse.jetty.util.thread.Scheduler; +import org.eclipse.jetty.util.thread.Sweeper; import org.eclipse.jetty.util.thread.ThreadPool; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -142,12 +143,13 @@ public class HttpClient extends ContainerLifeCycle private boolean tcpNoDelay = true; private boolean strictEventOrdering = false; private HttpField encodingField; - private boolean removeIdleDestinations = false; + private long destinationIdleTimeout; private String name = getClass().getSimpleName() + "@" + Integer.toHexString(hashCode()); private HttpCompliance httpCompliance = HttpCompliance.RFC7230; private String defaultRequestContentType = "application/octet-stream"; private boolean useInputDirectByteBuffers = true; private boolean useOutputDirectByteBuffers = true; + private Sweeper destinationSweeper; /** * Creates a HttpClient instance that can perform HTTP/1.1 requests to non-TLS and TLS destinations. @@ -222,7 +224,14 @@ public class HttpClient extends ContainerLifeCycle cookieStore = cookieManager.getCookieStore(); transport.setHttpClient(this); + super.doStart(); + + if (getDestinationIdleTimeout() > 0L) + { + destinationSweeper = new Sweeper(scheduler, 1000L); + destinationSweeper.start(); + } } private CookieManager newCookieManager() @@ -233,6 +242,12 @@ public class HttpClient extends ContainerLifeCycle @Override protected void doStop() throws Exception { + if (destinationSweeper != null) + { + destinationSweeper.stop(); + destinationSweeper = null; + } + decoderFactories.clear(); handlers.clear(); @@ -290,6 +305,11 @@ public class HttpClient extends ContainerLifeCycle return cookieManager; } + Sweeper getDestinationSweeper() + { + return destinationSweeper; + } + /** * @return the authentication store associated with this instance */ @@ -529,21 +549,28 @@ public class HttpClient extends ContainerLifeCycle */ public HttpDestination resolveDestination(Origin origin) { - return destinations.computeIfAbsent(origin, o -> + return destinations.compute(origin, (k, v) -> { - HttpDestination destination = getTransport().newHttpDestination(o); - // Start the destination before it's published to other threads. - addManaged(destination); - if (LOG.isDebugEnabled()) - LOG.debug("Created {}", destination); - return destination; + if (v == null || v.stale()) + { + HttpDestination newDestination = getTransport().newHttpDestination(k); + // Start the destination before it's published to other threads. + addManaged(newDestination); + if (LOG.isDebugEnabled()) + LOG.debug("Created {}; existing: '{}'", newDestination, v); + return newDestination; + } + return v; }); } protected boolean removeDestination(HttpDestination destination) { + boolean removed = destinations.remove(destination.getOrigin(), destination); removeBean(destination); - return destinations.remove(destination.getOrigin(), destination); + if (LOG.isDebugEnabled()) + LOG.debug("Removed {}; result: {}", destination, removed); + return removed; } /** @@ -1010,14 +1037,50 @@ public class HttpClient extends ContainerLifeCycle this.strictEventOrdering = strictEventOrdering; } + /** + * The default value is 0 + * @return the time in ms after which idle destinations are removed + * @see #setDestinationIdleTimeout(long) + */ + @ManagedAttribute("The time in ms after which idle destinations are removed, disabled when zero or negative") + public long getDestinationIdleTimeout() + { + return destinationIdleTimeout; + } + + /** + *

+ * Whether destinations that have no connections (nor active nor idle) and no exchanges + * should be removed after the specified timeout. + *

+ *

+ * If the specified {@code destinationIdleTimeout} is 0 or negative, then the destinations + * are not removed. + *

+ *

+ * Avoids accumulating destinations when applications (e.g. a spider bot or web crawler) + * hit a lot of different destinations that won't be visited again. + *

+ * + * @param destinationIdleTimeout the time in ms after which idle destinations are removed + */ + public void setDestinationIdleTimeout(long destinationIdleTimeout) + { + if (isStarted()) + throw new IllegalStateException(); + this.destinationIdleTimeout = destinationIdleTimeout; + } + /** * @return whether destinations that have no connections should be removed * @see #setRemoveIdleDestinations(boolean) + * @deprecated replaced by {@link #getDestinationIdleTimeout()} */ + @Deprecated @ManagedAttribute("Whether idle destinations are removed") public boolean isRemoveIdleDestinations() { - return removeIdleDestinations; + return destinationIdleTimeout > 0L; } /** @@ -1031,10 +1094,12 @@ public class HttpClient extends ContainerLifeCycle * * @param removeIdleDestinations whether destinations that have no connections should be removed * @see org.eclipse.jetty.client.DuplexConnectionPool + * @deprecated replaced by {@link #setDestinationIdleTimeout(long)}, calls the latter with a value of 10000 ms. */ + @Deprecated public void setRemoveIdleDestinations(boolean removeIdleDestinations) { - this.removeIdleDestinations = removeIdleDestinations; + setDestinationIdleTimeout(removeIdleDestinations ? 10_000L : 0L); } /** diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java index b27819e68ac..b09f495fd28 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java @@ -21,6 +21,7 @@ import java.util.Iterator; import java.util.List; import java.util.Queue; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.eclipse.jetty.client.api.Connection; @@ -40,14 +41,16 @@ import org.eclipse.jetty.util.annotation.ManagedObject; import org.eclipse.jetty.util.component.ContainerLifeCycle; import org.eclipse.jetty.util.component.Dumpable; import org.eclipse.jetty.util.component.DumpableCollection; +import org.eclipse.jetty.util.component.LifeCycle; import org.eclipse.jetty.util.ssl.SslContextFactory; +import org.eclipse.jetty.util.thread.AutoLock; import org.eclipse.jetty.util.thread.Scheduler; import org.eclipse.jetty.util.thread.Sweeper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @ManagedObject -public abstract class HttpDestination extends ContainerLifeCycle implements Destination, Closeable, Callback, Dumpable +public abstract class HttpDestination extends ContainerLifeCycle implements Destination, Closeable, Callback, Dumpable, Sweeper.Sweepable { private static final Logger LOG = LoggerFactory.getLogger(HttpDestination.class); @@ -60,7 +63,10 @@ public abstract class HttpDestination extends ContainerLifeCycle implements Dest private final ClientConnectionFactory connectionFactory; private final HttpField hostField; private final RequestTimeouts requestTimeouts; + private final AutoLock staleLock = new AutoLock(); private ConnectionPool connectionPool; + private boolean stale; + private long activeNanos; public HttpDestination(HttpClient client, Origin origin, boolean intrinsicallySecure) { @@ -104,23 +110,78 @@ public abstract class HttpDestination extends ContainerLifeCycle implements Dest connectionPool.accept(connection); } + public boolean stale() + { + try (AutoLock l = staleLock.lock()) + { + boolean stale = this.stale; + if (!stale) + this.activeNanos = System.nanoTime(); + if (LOG.isDebugEnabled()) + LOG.debug("Stale check done with result {} on {}", stale, this); + return stale; + } + } + + @Override + public boolean sweep() + { + if (LOG.isDebugEnabled()) + LOG.debug("Sweep check in progress on {}", this); + boolean remove = false; + try (AutoLock l = staleLock.lock()) + { + boolean stale = exchanges.isEmpty() && connectionPool.isEmpty(); + if (!stale) + { + this.activeNanos = System.nanoTime(); + } + else if (isStaleDelayExpired()) + { + this.stale = true; + remove = true; + } + } + if (remove) + { + getHttpClient().removeDestination(this); + LifeCycle.stop(this); + } + if (LOG.isDebugEnabled()) + LOG.debug("Sweep check done with result {} on {}", remove, this); + return remove; + } + + private boolean isStaleDelayExpired() + { + assert staleLock.isHeldByCurrentThread(); + long destinationIdleTimeout = TimeUnit.MILLISECONDS.toNanos(getHttpClient().getDestinationIdleTimeout()); + return System.nanoTime() - activeNanos >= destinationIdleTimeout; + } + @Override protected void doStart() throws Exception { this.connectionPool = newConnectionPool(client); addBean(connectionPool, true); super.doStart(); - Sweeper sweeper = client.getBean(Sweeper.class); - if (sweeper != null && connectionPool instanceof Sweeper.Sweepable) - sweeper.offer((Sweeper.Sweepable)connectionPool); + Sweeper connectionPoolSweeper = client.getBean(Sweeper.class); + if (connectionPoolSweeper != null && connectionPool instanceof Sweeper.Sweepable) + connectionPoolSweeper.offer((Sweeper.Sweepable)connectionPool); + Sweeper destinationSweeper = getHttpClient().getDestinationSweeper(); + if (destinationSweeper != null) + destinationSweeper.offer(this); } @Override protected void doStop() throws Exception { - Sweeper sweeper = client.getBean(Sweeper.class); - if (sweeper != null && connectionPool instanceof Sweeper.Sweepable) - sweeper.remove((Sweeper.Sweepable)connectionPool); + Sweeper destinationSweeper = getHttpClient().getDestinationSweeper(); + if (destinationSweeper != null) + destinationSweeper.remove(this); + Sweeper connectionPoolSweeper = client.getBean(Sweeper.class); + if (connectionPoolSweeper != null && connectionPool instanceof Sweeper.Sweepable) + connectionPoolSweeper.remove((Sweeper.Sweepable)connectionPool); super.doStop(); removeBean(connectionPool); } @@ -449,11 +510,7 @@ public abstract class HttpDestination extends ContainerLifeCycle implements Dest { boolean removed = connectionPool.remove(connection); - if (getHttpExchanges().isEmpty()) - { - tryRemoveIdleDestination(); - } - else if (removed) + if (removed) { // Process queued requests that may be waiting. // We may create a connection that is not @@ -478,22 +535,6 @@ public abstract class HttpDestination extends ContainerLifeCycle implements Dest { exchange.getRequest().abort(cause); } - if (exchanges.isEmpty()) - tryRemoveIdleDestination(); - } - - private void tryRemoveIdleDestination() - { - if (getHttpClient().isRemoveIdleDestinations() && connectionPool.isEmpty()) - { - // There is a race condition between this thread removing the destination - // and another thread queueing a request to this same destination. - // If this destination is removed, but the request queued, a new connection - // will be opened, the exchange will be executed and eventually the connection - // will idle timeout and be closed. Meanwhile a new destination will be created - // in HttpClient and will be used for other requests. - getHttpClient().removeDestination(this); - } } @Override @@ -507,16 +548,39 @@ public abstract class HttpDestination extends ContainerLifeCycle implements Dest return getOrigin().asString(); } + @ManagedAttribute("For how long this destination has been idle in ms") + public long getIdle() + { + if (getHttpClient().getDestinationIdleTimeout() <= 0L) + return -1; + try (AutoLock l = staleLock.lock()) + { + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - activeNanos); + } + } + + @ManagedAttribute("Whether this destinations is stale") + public boolean isStale() + { + try (AutoLock l = staleLock.lock()) + { + return this.stale; + } + } + @Override public String toString() { - return String.format("%s[%s]@%x%s,queue=%d,pool=%s", + return String.format("%s[%s]@%x%s,state=%s,queue=%d,pool=%s,stale=%b,idle=%d", HttpDestination.class.getSimpleName(), getOrigin(), hashCode(), proxy == null ? "" : "(via " + proxy + ")", + getState(), getQueuedRequestCount(), - getConnectionPool()); + getConnectionPool(), + isStale(), + getIdle()); } @FunctionalInterface diff --git a/jetty-client/src/test/java/org/eclipse/jetty/client/DuplexHttpDestinationTest.java b/jetty-client/src/test/java/org/eclipse/jetty/client/DuplexHttpDestinationTest.java index 2dcdccf3e88..85073126619 100644 --- a/jetty-client/src/test/java/org/eclipse/jetty/client/DuplexHttpDestinationTest.java +++ b/jetty-client/src/test/java/org/eclipse/jetty/client/DuplexHttpDestinationTest.java @@ -19,6 +19,7 @@ import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; +import org.awaitility.Awaitility; import org.eclipse.jetty.client.api.Connection; import org.eclipse.jetty.client.api.ContentResponse; import org.eclipse.jetty.client.api.Destination; @@ -291,6 +292,9 @@ public class DuplexHttpDestinationTest extends AbstractHttpClientServerTest public void testDestinationIsRemoved(Scenario scenario) throws Exception { start(scenario, new EmptyServerHandler()); + client.stop(); + client.setDestinationIdleTimeout(1000); + client.start(); String host = "localhost"; int port = connector.getLocalPort(); @@ -305,7 +309,7 @@ public class DuplexHttpDestinationTest extends AbstractHttpClientServerTest Destination destinationAfter = client.resolveDestination(request); assertSame(destinationBefore, destinationAfter); - client.setRemoveIdleDestinations(true); + Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> client.getDestinations().isEmpty()); request = client.newRequest(host, port) .scheme(scenario.getScheme()) @@ -323,21 +327,19 @@ public class DuplexHttpDestinationTest extends AbstractHttpClientServerTest public void testDestinationIsRemovedAfterConnectionError(Scenario scenario) throws Exception { start(scenario, new EmptyServerHandler()); + client.stop(); + client.setDestinationIdleTimeout(1000); + client.start(); String host = "localhost"; int port = connector.getLocalPort(); - client.setRemoveIdleDestinations(true); assertTrue(client.getDestinations().isEmpty(), "Destinations of a fresh client must be empty"); server.stop(); Request request = client.newRequest(host, port).scheme(scenario.getScheme()); assertThrows(Exception.class, request::send); - long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(1); - while (!client.getDestinations().isEmpty() && System.nanoTime() < deadline) - { - Thread.sleep(10); - } + Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> client.getDestinations().isEmpty()); assertTrue(client.getDestinations().isEmpty(), "Destination must be removed after connection error"); } diff --git a/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientProxyProtocolTest.java b/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientProxyProtocolTest.java index a54353ec8de..9c882e73948 100644 --- a/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientProxyProtocolTest.java +++ b/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientProxyProtocolTest.java @@ -70,7 +70,6 @@ public class HttpClientProxyProtocolTest clientThreads.setName("client"); client = new HttpClient(); client.setExecutor(clientThreads); - client.setRemoveIdleDestinations(false); client.start(); } diff --git a/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTest.java b/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTest.java index c755dcc3b24..b7ef4e12725 100644 --- a/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTest.java +++ b/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTest.java @@ -94,6 +94,7 @@ import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ArgumentsSource; +import static org.awaitility.Awaitility.await; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; @@ -528,6 +529,41 @@ public class HttpClientTest extends AbstractHttpClientServerTest } } + @ParameterizedTest + @ArgumentsSource(ScenarioProvider.class) + public void testRetryWithDestinationIdleTimeoutEnabled(Scenario scenario) throws Exception + { + start(scenario, new EmptyServerHandler()); + client.stop(); + client.setDestinationIdleTimeout(1000); + client.setIdleTimeout(1000); + client.setMaxConnectionsPerDestination(1); + client.start(); + + try (StacklessLogging ignored = new StacklessLogging(org.eclipse.jetty.server.HttpChannel.class)) + { + client.newRequest("localhost", connector.getLocalPort()) + .scheme(scenario.getScheme()) + .path("/one") + .send(); + + int idleTimeout = 100; + Thread.sleep(idleTimeout * 2); + + // After serving a request over a connection that hasn't timed out, serving a second + // request with a shorter idle timeout will make the connection timeout immediately + // after being taken out of the pool. This triggers the retry mechanism. + client.newRequest("localhost", connector.getLocalPort()) + .scheme(scenario.getScheme()) + .path("/two") + .idleTimeout(idleTimeout, TimeUnit.MILLISECONDS) + .send(); + } + + // Wait for the sweeper to remove the idle HttpDestination. + await().atMost(5, TimeUnit.SECONDS).until(() -> client.getDestinations().isEmpty()); + } + @ParameterizedTest @ArgumentsSource(ScenarioProvider.class) public void testExchangeIsCompleteOnlyWhenBothRequestAndResponseAreComplete(Scenario scenario) throws Exception diff --git a/jetty-util/src/main/java/org/eclipse/jetty/util/Pool.java b/jetty-util/src/main/java/org/eclipse/jetty/util/Pool.java index fa86d040e16..bed066d38e8 100644 --- a/jetty-util/src/main/java/org/eclipse/jetty/util/Pool.java +++ b/jetty-util/src/main/java/org/eclipse/jetty/util/Pool.java @@ -127,7 +127,7 @@ public class Pool implements AutoCloseable, Dumpable public Pool(StrategyType strategyType, int maxEntries, boolean cache) { this.maxEntries = maxEntries; - this.strategyType = strategyType; + this.strategyType = Objects.requireNonNull(strategyType); this.cache = cache ? new ThreadLocal<>() : null; this.nextIndex = strategyType == StrategyType.ROUND_ROBIN ? new AtomicInteger() : null; } @@ -336,14 +336,25 @@ public class Pool implements AutoCloseable, Dumpable try (AutoLock l = lock.lock()) { if (closed) + { + if (LOGGER.isDebugEnabled()) + LOGGER.debug("{} is closed, returning null reserved entry", this); return null; + } // If we have no space - if (maxEntries > 0 && entries.size() >= maxEntries) + int entriesSize = entries.size(); + if (maxEntries > 0 && entriesSize >= maxEntries) + { + if (LOGGER.isDebugEnabled()) + LOGGER.debug("{} has no space: {} >= {}, returning null reserved entry", this, entriesSize, maxEntries); return null; + } Entry entry = newEntry(); entries.add(entry); + if (LOGGER.isDebugEnabled()) + LOGGER.debug("{} returning new reserved entry {}", this, entry); return entry; } } @@ -514,6 +525,9 @@ public class Pool implements AutoCloseable, Dumpable @Override public void close() { + if (LOGGER.isDebugEnabled()) + LOGGER.debug("Closing {}", this); + List copy; try (AutoLock l = lock.lock()) {