From 06f2fa41ddd83236a8484572e93fb3363c2084ad Mon Sep 17 00:00:00 2001 From: Ludovic Orban Date: Thu, 1 Sep 2022 14:45:24 +0200 Subject: [PATCH] Jetty 9.4.x : fix client remove idle destinations (#8495) Fixes #8493: RemoveIdleDestinations's race condition and improve logging. Signed-off-by: Ludovic Orban --- .../jetty/client/AbstractConnectionPool.java | 8 +- .../org/eclipse/jetty/client/HttpClient.java | 85 ++++++++++-- .../eclipse/jetty/client/HttpDestination.java | 124 +++++++++++++----- .../client/HttpClientProxyProtocolTest.java | 1 - .../eclipse/jetty/client/HttpClientTest.java | 36 +++++ .../http/HttpDestinationOverHTTPTest.java | 16 ++- .../java/org/eclipse/jetty/util/Pool.java | 20 ++- 7 files changed, 236 insertions(+), 54 deletions(-) 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 2a63d45219e2..f95c164e402e 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 @@ -306,7 +306,7 @@ protected void tryCreate(boolean create) { 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; } @@ -516,15 +516,17 @@ public boolean sweep() @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 e6473633ad56..ac33d5554594 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 @@ -77,6 +77,7 @@ 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; /** @@ -148,11 +149,12 @@ public class HttpClient extends ContainerLifeCycle private boolean tcpNoDelay = true; private boolean strictEventOrdering = false; private HttpField encodingField; - private boolean removeIdleDestinations = false; + private long destinationIdleTimeout; private boolean connectBlocking = false; private String name = getClass().getSimpleName() + "@" + Integer.toHexString(hashCode()); private HttpCompliance httpCompliance = HttpCompliance.RFC7230; private String defaultRequestContentType = "application/octet-stream"; + private Sweeper destinationSweeper; /** * Creates a {@link HttpClient} instance that can perform requests to non-TLS destinations only @@ -252,7 +254,14 @@ protected void doStart() throws Exception cookieStore = cookieManager.getCookieStore(); transport.setHttpClient(this); + super.doStart(); + + if (getDestinationIdleTimeout() > 0L) + { + destinationSweeper = new Sweeper(scheduler, 1000L); + destinationSweeper.start(); + } } private CookieManager newCookieManager() @@ -263,6 +272,12 @@ private CookieManager newCookieManager() @Override protected void doStop() throws Exception { + if (destinationSweeper != null) + { + destinationSweeper.stop(); + destinationSweeper = null; + } + decoderFactories.clear(); handlers.clear(); @@ -318,6 +333,11 @@ CookieManager getCookieManager() return cookieManager; } + Sweeper getDestinationSweeper() + { + return destinationSweeper; + } + /** * @return the authentication store associated with this instance */ @@ -570,20 +590,27 @@ protected Origin createOrigin(String scheme, String host, int port, Object tag) */ public HttpDestination resolveDestination(Origin origin) { - return destinations.computeIfAbsent(origin, o -> + return destinations.compute(origin, (k, v) -> { - HttpDestination newDestination = getTransport().newHttpDestination(o); - addManaged(newDestination); - if (LOG.isDebugEnabled()) - LOG.debug("Created {}", newDestination); - return newDestination; + if (v == null || v.stale()) + { + HttpDestination newDestination = getTransport().newHttpDestination(k); + 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; } /** @@ -1080,14 +1107,50 @@ public void setStrictEventOrdering(boolean strictEventOrdering) 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()) + LOG.warn("Calling setDestinationIdleTimeout() while started has no effect"); + 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; } /** @@ -1101,10 +1164,12 @@ public boolean isRemoveIdleDestinations() * * @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 b6f9f499f04c..39a147d125b2 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 @@ -26,6 +26,7 @@ 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; @@ -45,14 +46,16 @@ 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.log.Log; import org.eclipse.jetty.util.log.Logger; import org.eclipse.jetty.util.ssl.SslContextFactory; +import org.eclipse.jetty.util.thread.Locker; import org.eclipse.jetty.util.thread.Scheduler; import org.eclipse.jetty.util.thread.Sweeper; @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 = Log.getLogger(HttpDestination.class); @@ -65,7 +68,10 @@ public abstract class HttpDestination extends ContainerLifeCycle implements Dest private final ClientConnectionFactory connectionFactory; private final HttpField hostField; private final RequestTimeouts requestTimeouts; + private final Locker staleLock = new Locker(); private ConnectionPool connectionPool; + private boolean stale; + private long activeNanos; public HttpDestination(HttpClient client, Origin origin) { @@ -104,23 +110,78 @@ public HttpDestination(HttpClient client, Origin origin) hostField = new HttpField(HttpHeader.HOST, host); } + public boolean stale() + { + try (Locker.Lock 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 (Locker.Lock 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.isLocked(); + 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); } @@ -462,11 +523,7 @@ public boolean remove(Connection connection) { 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 @@ -501,22 +558,6 @@ public void abort(Throwable cause) { 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 @@ -530,16 +571,39 @@ public String asString() return origin.asString(); } + @ManagedAttribute("For how long this destination has been idle in ms") + public long getIdle() + { + if (getHttpClient().getDestinationIdleTimeout() <= 0L) + return -1; + try (Locker.Lock l = staleLock.lock()) + { + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - activeNanos); + } + } + + @ManagedAttribute("Whether this destinations is stale") + public boolean isStale() + { + try (Locker.Lock 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(), asString(), hashCode(), proxy == null ? "" : "(via " + proxy + ")", + getState(), getQueuedRequestCount(), - getConnectionPool()); + getConnectionPool(), + isStale(), + getIdle()); } /** 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 8a3af54cc8ca..1777b04dfabb 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 @@ -73,7 +73,6 @@ private void startClient() throws Exception 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 e40ff243112d..f8cabfd87df0 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 @@ -103,6 +103,7 @@ 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; @@ -538,6 +539,41 @@ public void handle(String target, org.eclipse.jetty.server.Request baseRequest, } } + @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-client/src/test/java/org/eclipse/jetty/client/http/HttpDestinationOverHTTPTest.java b/jetty-client/src/test/java/org/eclipse/jetty/client/http/HttpDestinationOverHTTPTest.java index 7ef524de391d..25587eedc4bf 100644 --- a/jetty-client/src/test/java/org/eclipse/jetty/client/http/HttpDestinationOverHTTPTest.java +++ b/jetty-client/src/test/java/org/eclipse/jetty/client/http/HttpDestinationOverHTTPTest.java @@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit; import java.util.function.Supplier; +import org.awaitility.Awaitility; import org.eclipse.jetty.client.AbstractHttpClientServerTest; import org.eclipse.jetty.client.ConnectionPool; import org.eclipse.jetty.client.ConnectionPoolHelper; @@ -301,6 +302,9 @@ public void testRequestFailedIfMaxRequestsQueuedPerDestinationExceeded(Scenario 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(); @@ -316,7 +320,7 @@ public void testDestinationIsRemoved(Scenario scenario) throws Exception Destination destinationAfter = client.getDestination(scenario.getScheme(), host, port); assertSame(destinationBefore, destinationAfter); - client.setRemoveIdleDestinations(true); + Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> client.getDestinations().isEmpty()); response = client.newRequest(host, port) .scheme(scenario.getScheme()) @@ -334,21 +338,19 @@ public void testDestinationIsRemoved(Scenario scenario) throws Exception 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-util/src/main/java/org/eclipse/jetty/util/Pool.java b/jetty-util/src/main/java/org/eclipse/jetty/util/Pool.java index 4c052c91fd85..1dcca99cbe01 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 @@ -132,7 +132,7 @@ public Pool(StrategyType strategyType, int maxEntries) 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; } @@ -341,14 +341,25 @@ public Entry reserve() try (Locker.Lock l = locker.lock()) { if (closed) + { + if (LOGGER.isDebugEnabled()) + LOGGER.debug("{} is closed, returning null reserved entry", this); return null; + } // If we have no space - if (entries.size() >= maxEntries) + int entriesSize = entries.size(); + if (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; } } @@ -546,6 +557,9 @@ public boolean isClosed() @Override public void close() { + if (LOGGER.isDebugEnabled()) + LOGGER.debug("Closing {}", this); + List copy; try (Locker.Lock l = locker.lock()) { @@ -943,7 +957,7 @@ boolean tryRelease() break; } - int currentMaxUsageCount = maxUsage; + int currentMaxUsageCount = getMaxUsageCount(getPooled()); boolean overUsed = currentMaxUsageCount > 0 && usageCount >= currentMaxUsageCount; return !(overUsed && newMultiplexCount == 0); }