Merge remote-tracking branch 'origin/jetty-11.0.x' into jetty-12.0.x

Signed-off-by: Ludovic Orban <lorban@bitronix.be>
This commit is contained in:
Ludovic Orban 2022-09-02 09:12:24 +02:00
commit bb83e835cf
10 changed files with 246 additions and 64 deletions

View File

@ -277,7 +277,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;
}
@ -533,15 +533,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<Connection>

View File

@ -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;
}
/**
* <p>
* Whether destinations that have no connections (nor active nor idle) and no exchanges
* should be removed after the specified timeout.
* </p>
* <p>
* If the specified {@code destinationIdleTimeout} is 0 or negative, then the destinations
* are not removed.
* </p>
* <p>
* 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.
* </p>
*
* @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);
}
/**

View File

@ -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 class HttpDestination extends ContainerLifeCycle implements Destination, Closeable, Callback, Dumpable
public 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 class HttpDestination extends ContainerLifeCycle implements Destination,
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 class HttpDestination extends ContainerLifeCycle implements Destination,
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 class HttpDestination extends ContainerLifeCycle implements Destination,
{
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 class HttpDestination extends ContainerLifeCycle implements Destination,
{
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 class HttpDestination extends ContainerLifeCycle implements Destination,
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());
}
/**

View File

@ -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");
}

View File

@ -70,7 +70,6 @@ public class HttpClientProxyProtocolTest
clientThreads.setName("client");
client = new HttpClient();
client.setExecutor(clientThreads);
client.setRemoveIdleDestinations(false);
client.start();
}

View File

@ -87,6 +87,7 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ArgumentsSource;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
@ -512,6 +513,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

View File

@ -129,7 +129,7 @@ public class Pool<T> implements AutoCloseable, Dumpable
public Pool(StrategyType strategyType, int maxEntries, boolean cache, ToIntFunction<T> maxMultiplex)
{
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;
this.maxMultiplex = Objects.requireNonNull(maxMultiplex);
@ -205,14 +205,25 @@ public class Pool<T> implements AutoCloseable, Dumpable
try (AutoLock ignored = 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 = new Entry();
entries.add(entry);
if (LOGGER.isDebugEnabled())
LOGGER.debug("{} returning new reserved entry {}", this, entry);
return entry;
}
}
@ -368,6 +379,9 @@ public class Pool<T> implements AutoCloseable, Dumpable
@Override
public void close()
{
if (LOGGER.isDebugEnabled())
LOGGER.debug("Closing {}", this);
List<Entry> copy;
try (AutoLock ignored = lock.lock())
{

View File

@ -13,7 +13,7 @@
<properties>
<bundle-symbolic-name>${project.groupId}.jaas</bundle-symbolic-name>
<apacheds.version>2.0.0.AM26</apacheds.version>
<apache.directory.api.version>2.1.0</apache.directory.api.version>
<apache.directory.api.version>2.1.2</apache.directory.api.version>
<spotbugs.onlyAnalyze>org.eclipse.jetty.jaas.*</spotbugs.onlyAnalyze>
</properties>

View File

@ -13,7 +13,7 @@
<name>Jetty :: GCloud</name>
<properties>
<gcloud.version>2.10.1</gcloud.version>
<gcloud.version>2.11.0</gcloud.version>
</properties>
<modules>

16
pom.xml
View File

@ -41,7 +41,7 @@
<felix.version>7.0.5</felix.version>
<findbugs.jsr305.version>3.0.2</findbugs.jsr305.version>
<google.errorprone.version>2.14.0</google.errorprone.version>
<grpc.version>1.48.0</grpc.version>
<grpc.version>1.49.0</grpc.version>
<gson.version>2.9.1</gson.version>
<guava.version>31.1-jre</guava.version>
<guice.version>5.1.0</guice.version>
@ -55,7 +55,7 @@
<jboss.logging.processor.version>2.2.1.Final</jboss.logging.processor.version>
<jboss.logging.version>3.5.0.Final</jboss.logging.version>
<jboss-logmanager.version>2.1.18.Final</jboss-logmanager.version>
<jboss-threads.version>3.1.0.Final</jboss-threads.version>
<jboss-threads.version>3.5.0.Final</jboss-threads.version>
<jetty-assembly-descriptors.version>1.1</jetty-assembly-descriptors.version>
<jetty.perf-helper.version>1.0.7</jetty.perf-helper.version>
<jetty-quiche-native.version>0.12.0</jetty-quiche-native.version>
@ -73,7 +73,7 @@
<junit.version>5.9.0</junit.version>
<kerb-simplekdc.version>2.0.2</kerb-simplekdc.version>
<log4j2.version>2.18.0</log4j2.version>
<logback.version>1.3.0-beta0</logback.version>
<logback.version>1.4.0</logback.version>
<mariadb.version>3.0.6</mariadb.version>
<mariadb.docker.version>10.3.6</mariadb.docker.version>
<maven.deps.version>3.8.4</maven.deps.version>
@ -93,7 +93,7 @@
<testcontainers.version>1.17.3</testcontainers.version>
<weld.version>4.0.3.Final</weld.version>
<wildfly.common.version>1.6.0.Final</wildfly.common.version>
<wildfly.elytron.version>1.20.0.Final</wildfly.elytron.version>
<wildfly.elytron.version>1.20.1.Final</wildfly.elytron.version>
<xmemcached.version>2.4.7</xmemcached.version>
<weld.version>4.0.2.Final</weld.version>
@ -113,7 +113,7 @@
<maven.assembly.plugin.version>3.4.2</maven.assembly.plugin.version>
<maven.bundle.plugin.version>5.1.8</maven.bundle.plugin.version>
<maven.clean.plugin.version>3.2.0</maven.clean.plugin.version>
<maven.checkstyle.plugin.version>3.1.2</maven.checkstyle.plugin.version>
<maven.checkstyle.plugin.version>3.2.0</maven.checkstyle.plugin.version>
<maven.compiler.plugin.version>3.10.1</maven.compiler.plugin.version>
<maven.dependency.plugin.version>3.3.0</maven.dependency.plugin.version>
<maven.deploy.plugin.version>3.0.0</maven.deploy.plugin.version>
@ -124,8 +124,8 @@
<maven.install.plugin.version>3.0.1</maven.install.plugin.version>
<maven.invoker.plugin.version>3.3.0</maven.invoker.plugin.version>
<maven.jar.plugin.version>3.2.2</maven.jar.plugin.version>
<maven.javadoc.plugin.version>3.3.1</maven.javadoc.plugin.version>
<maven.jxr.plugin.version>3.1.1</maven.jxr.plugin.version>
<maven.javadoc.plugin.version>3.4.1</maven.javadoc.plugin.version>
<maven.jxr.plugin.version>3.3.0</maven.jxr.plugin.version>
<maven.plugin-tools.version>3.6.4</maven.plugin-tools.version>
<maven-plugin.plugin.version>3.6.2</maven-plugin.plugin.version>
<maven.project-info-reports.plugin.version>3.1.2</maven.project-info-reports.plugin.version>
@ -248,7 +248,7 @@
</requireJavaVersion>
<versionTxtRule implementation="org.eclipse.jetty.toolchain.enforcer.rules.VersionTxtRule" />
<versionOsgiRule implementation="org.eclipse.jetty.toolchain.enforcer.rules.RequireOsgiCompatibleVersionRule" />
<!-- do not support alphax as a version number -->
<!-- do not support alphax as a version number -->
<!--
<versionRedhatRule implementation="org.eclipse.jetty.toolchain.enforcer.rules.RequireRedhatCompatibleVersionRule" />
<versionDebianRule implementation="org.eclipse.jetty.toolchain.enforcer.rules.RequireDebianCompatibleVersionRule" />