Skip to content

Commit

Permalink
Merge remote-tracking branch 'origin/jetty-10.0.x' into jetty-11.0.x
Browse files Browse the repository at this point in the history 
Signed-off-by: Ludovic Orban <lorban@bitronix.be>
  • Loading branch information
@lorban
lorban committed Sep 1, 2022
2 parents 027de89 + de13cef commit ac5416f
Show file tree
Hide file tree
Showing 10 changed files with 241 additions and 59 deletions.
8 changes: 5 additions & 3 deletions jetty-client/src/main/java/org/eclipse/jetty/client/AbstractConnectionPool.java
View file
Expand Up @@ -275,7 +275,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;
}

Expand Down Expand Up @@ -517,15 +517,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<Connection>
Expand Down
87 changes: 76 additions & 11 deletions jetty-client/src/main/java/org/eclipse/jetty/client/HttpClient.java
View file
Expand Up @@ -74,6 +74,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;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -222,7 +224,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()
Expand All @@ -233,6 +242,12 @@ private CookieManager newCookieManager()
@Override
protected void doStop() throws Exception
{
if (destinationSweeper != null)
{
destinationSweeper.stop();
destinationSweeper = null;
}

decoderFactories.clear();
handlers.clear();

Expand Down Expand Up @@ -290,6 +305,11 @@ CookieManager getCookieManager()
return cookieManager;
}

Sweeper getDestinationSweeper()
{
return destinationSweeper;
}

/**
* @return the authentication store associated with this instance
*/
Expand Down Expand Up @@ -529,21 +549,28 @@ public Origin createOrigin(HttpRequest request, Origin.Protocol protocol)
*/
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;
}

/**
Expand Down Expand Up @@ -1010,14 +1037,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;
}

/**
* <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;
}

/**
Expand All @@ -1031,10 +1094,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);
}

/**
Expand Down
124 changes: 94 additions & 30 deletions jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java
View file
Expand Up @@ -21,6 +21,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;
Expand All @@ -40,14 +41,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.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);

Expand All @@ -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)
{
Expand Down Expand Up @@ -104,23 +110,78 @@ public void accept(Connection connection)
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);
}
Expand Down Expand Up @@ -449,11 +510,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
Expand All @@ -478,22 +535,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
Expand All @@ -507,16 +548,39 @@ public String asString()
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
Expand Down

0 comments on commit ac5416f

Please sign in to comment.