Apache
/
httpcomponents-client
mirror of https://github.com/apache/httpcomponents-client.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor HttpClient synchronized sections for virtual threads (#476) 

- Replaced `synchronized` blocks with `ReentrantLock` in `LeaseRequest` to better support virtual threads introduced in JDK 21.
- Ensured each `LeaseRequest` instance has its own unique lock for maintaining original synchronization semantics.
- Addressed potential performance and deadlock issues with virtual threads by using explicit lock primitives from `java.util.concurrent.locks`.
This commit is contained in:
Arturo Bernal 2023-08-27 12:57:15 +02:00 committed by Oleg Kalnichevski
parent 097c17b78b
commit 8466b19861
9 changed files with 416 additions and 241 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
httpclient5-cache/src/main/java/org/apache/hc/client5/http/impl/cache/BasicHttpCacheStorage.java vendored
View File

@ -29,6 +29,7 @@ package org.apache.hc.client5.http.impl.cache;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.hc.client5.http.cache.HttpCacheCASOperation;
import org.apache.hc.client5.http.cache.HttpCacheEntry;
@ -53,9 +54,12 @@ public class BasicHttpCacheStorage implements HttpCacheStorage {
private final CacheMap entries;
private final ReentrantLock lock;
public BasicHttpCacheStorage(final CacheConfig config) {
super();
this.entries = new CacheMap(config.getMaxCacheEntries());
this.lock = new ReentrantLock();
}
/**
@ -67,9 +71,14 @@ public class BasicHttpCacheStorage implements HttpCacheStorage {
* HttpCacheEntry to place in the cache
*/
@Override
public synchronized void putEntry(
public void putEntry(
final String url, final HttpCacheEntry entry) throws ResourceIOException {
entries.put(url, entry);
lock.lock();
try {
entries.put(url, entry);
} finally {
lock.unlock();
}
}
/**
@ -80,8 +89,13 @@ public class BasicHttpCacheStorage implements HttpCacheStorage {
* @return HttpCacheEntry if one exists, or null for cache miss
*/
@Override
public synchronized HttpCacheEntry getEntry(final String url) throws ResourceIOException {
return entries.get(url);
public HttpCacheEntry getEntry(final String url) throws ResourceIOException {
lock.lock();
try {
return entries.get(url);
} finally {
lock.unlock();
}
}
/**
@ -91,15 +105,26 @@ public class BasicHttpCacheStorage implements HttpCacheStorage {
* Url that is the cache key
*/
@Override
public synchronized void removeEntry(final String url) throws ResourceIOException {
entries.remove(url);
public void removeEntry(final String url) throws ResourceIOException {
lock.lock();
try {
entries.remove(url);
} finally {
lock.unlock();
}
}
@Override
public synchronized void updateEntry(
public void updateEntry(
final String url, final HttpCacheCASOperation casOperation) throws ResourceIOException {
final HttpCacheEntry existingEntry = entries.get(url);
entries.put(url, casOperation.execute(existingEntry));
lock.lock();
try {
final HttpCacheEntry existingEntry = entries.get(url);
entries.put(url, casOperation.execute(existingEntry));
} finally {
lock.unlock();
}
}
@Override

27
httpclient5-cache/src/main/java/org/apache/hc/client5/http/impl/cache/BasicIdGenerator.java vendored
View File

@ -32,6 +32,7 @@ import java.security.NoSuchAlgorithmException;
import java.security.SecureRandom;
import java.util.Formatter;
import java.util.Locale;
import java.util.concurrent.locks.ReentrantLock;
/**
* Should produce reasonably unique tokens.
@ -43,6 +44,8 @@ class BasicIdGenerator {
private long count;
private final ReentrantLock lock;
public BasicIdGenerator() {
super();
String hostname;
@ -58,18 +61,24 @@ class BasicIdGenerator {
throw new Error(ex);
}
this.rnd.setSeed(System.currentTimeMillis());
this.lock = new ReentrantLock();
}
public synchronized void generate(final StringBuilder buffer) {
this.count++;
final int rndnum = this.rnd.nextInt();
buffer.append(System.currentTimeMillis());
buffer.append('.');
try (Formatter formatter = new Formatter(buffer, Locale.ROOT)) {
formatter.format("%1$016x-%2$08x", this.count, rndnum);
public void generate(final StringBuilder buffer) {
lock.lock();
try {
this.count++;
final int rndnum = this.rnd.nextInt();
buffer.append(System.currentTimeMillis());
buffer.append('.');
try (Formatter formatter = new Formatter(buffer, Locale.ROOT)) {
formatter.format("%1$016x-%2$08x", this.count, rndnum);
}
buffer.append('.');
buffer.append(this.hostname);
} finally {
lock.unlock();
}
buffer.append('.');
buffer.append(this.hostname);
}
public String generate() {

24
httpclient5-cache/src/main/java/org/apache/hc/client5/http/impl/cache/CacheRevalidatorBase.java vendored
View File

@ -35,6 +35,7 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.hc.client5.http.schedule.ConcurrentCountMap;
import org.apache.hc.client5.http.schedule.SchedulingStrategy;
@ -49,6 +50,8 @@ import org.slf4j.LoggerFactory;
*/
class CacheRevalidatorBase implements Closeable {
private final ReentrantLock lock;
interface ScheduledExecutor {
Future<?> schedule(Runnable command, TimeValue timeValue) throws RejectedExecutionException;
@ -103,6 +106,7 @@ class CacheRevalidatorBase implements Closeable {
this.schedulingStrategy = schedulingStrategy;
this.pendingRequest = new HashSet<>();
this.failureCache = new ConcurrentCountMap<>();
this.lock = new ReentrantLock();
}
/**
@ -119,7 +123,8 @@ class CacheRevalidatorBase implements Closeable {
* Schedules an asynchronous re-validation
*/
void scheduleRevalidation(final String cacheKey, final Runnable command) {
synchronized (pendingRequest) {
lock.lock();
try {
if (!pendingRequest.contains(cacheKey)) {
final int consecutiveFailedAttempts = failureCache.getCount(cacheKey);
final TimeValue executionTime = schedulingStrategy.schedule(consecutiveFailedAttempts);
@ -130,6 +135,8 @@ class CacheRevalidatorBase implements Closeable {
LOG.debug("Revalidation of cache entry with key {} could not be scheduled", cacheKey, ex);
}
}
} finally {
lock.unlock();
}
}
@ -145,21 +152,30 @@ class CacheRevalidatorBase implements Closeable {
void jobSuccessful(final String identifier) {
failureCache.resetCount(identifier);
synchronized (pendingRequest) {
lock.lock();
try {
pendingRequest.remove(identifier);
} finally {
lock.unlock();
}
}
void jobFailed(final String identifier) {
failureCache.increaseCount(identifier);
synchronized (pendingRequest) {
lock.lock();
try {
pendingRequest.remove(identifier);
} finally {
lock.unlock();
}
}
Set<String> getScheduledIdentifiers() {
synchronized (pendingRequest) {
lock.lock();
try {
return new HashSet<>(pendingRequest);
} finally {
lock.unlock();
}
}

39
httpclient5-cache/src/main/java/org/apache/hc/client5/http/impl/cache/ManagedHttpCacheStorage.java vendored
View File

@ -34,6 +34,7 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.hc.client5.http.cache.HttpCacheCASOperation;
import org.apache.hc.client5.http.cache.HttpCacheEntry;
@ -82,12 +83,15 @@ public class ManagedHttpCacheStorage implements HttpCacheStorage, Closeable {
private final Set<ResourceReference> resources;
private final AtomicBoolean active;
private final ReentrantLock lock;
public ManagedHttpCacheStorage(final CacheConfig config) {
super();
this.entries = new CacheMap(config.getMaxCacheEntries());
this.morque = new ReferenceQueue<>();
this.resources = new HashSet<>();
this.active = new AtomicBoolean(true);
this.lock = new ReentrantLock();
}
private void ensureValidState() {
@ -110,9 +114,12 @@ public class ManagedHttpCacheStorage implements HttpCacheStorage, Closeable {
Args.notNull(url, "URL");
Args.notNull(entry, "Cache entry");
ensureValidState();
synchronized (this) {
lock.lock();
try {
this.entries.put(url, entry);
keepResourceReference(entry);
} finally {
lock.unlock();
}
}
@ -120,8 +127,11 @@ public class ManagedHttpCacheStorage implements HttpCacheStorage, Closeable {
public HttpCacheEntry getEntry(final String url) throws ResourceIOException {
Args.notNull(url, "URL");
ensureValidState();
synchronized (this) {
lock.lock();
try {
return this.entries.get(url);
} finally {
lock.unlock();
}
}
@ -129,10 +139,13 @@ public class ManagedHttpCacheStorage implements HttpCacheStorage, Closeable {
public void removeEntry(final String url) throws ResourceIOException {
Args.notNull(url, "URL");
ensureValidState();
synchronized (this) {
lock.lock();
try {
// Cannot deallocate the associated resources immediately as the
// cache entry may still be in use
this.entries.remove(url);
} finally {
lock.unlock();
}
}
@ -143,13 +156,16 @@ public class ManagedHttpCacheStorage implements HttpCacheStorage, Closeable {
Args.notNull(url, "URL");
Args.notNull(casOperation, "CAS operation");
ensureValidState();
synchronized (this) {
lock.lock();
try {
final HttpCacheEntry existing = this.entries.get(url);
final HttpCacheEntry updated = casOperation.execute(existing);
this.entries.put(url, updated);
if (existing != updated) {
keepResourceReference(updated);
}
} finally {
lock.unlock();
}
}
@ -170,8 +186,11 @@ public class ManagedHttpCacheStorage implements HttpCacheStorage, Closeable {
if (isActive()) {
ResourceReference ref;
while ((ref = (ResourceReference) this.morque.poll()) != null) {
synchronized (this) {
lock.lock();
try {
this.resources.remove(ref);
} finally {
lock.unlock();
}
ref.getResource().dispose();
}
@ -180,7 +199,8 @@ public class ManagedHttpCacheStorage implements HttpCacheStorage, Closeable {
public void shutdown() {
if (compareAndSet()) {
synchronized (this) {
lock.lock();
try {
this.entries.clear();
for (final ResourceReference ref: this.resources) {
ref.getResource().dispose();
@ -188,6 +208,8 @@ public class ManagedHttpCacheStorage implements HttpCacheStorage, Closeable {
this.resources.clear();
while (this.morque.poll() != null) {
}
} finally {
lock.unlock();
}
}
}
@ -195,12 +217,15 @@ public class ManagedHttpCacheStorage implements HttpCacheStorage, Closeable {
@Override
public void close() {
if (compareAndSet()) {
synchronized (this) {
lock.lock();
try {
ResourceReference ref;
while ((ref = (ResourceReference) this.morque.poll()) != null) {
this.resources.remove(ref);
ref.getResource().dispose();
}
} finally {
lock.unlock();
}
}
}

10
httpclient5/src/main/java/org/apache/hc/client5/http/impl/cookie/IgnoreCookieSpecFactory.java
View File

@ -27,6 +27,8 @@
package org.apache.hc.client5.http.impl.cookie;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.hc.client5.http.cookie.CookieSpec;
import org.apache.hc.client5.http.cookie.CookieSpecFactory;
import org.apache.hc.core5.annotation.Contract;
@ -43,17 +45,23 @@ public class IgnoreCookieSpecFactory implements CookieSpecFactory {
private volatile CookieSpec cookieSpec;
private final ReentrantLock lock;
public IgnoreCookieSpecFactory() {
super();
this.lock = new ReentrantLock();
}
@Override
public CookieSpec create(final HttpContext context) {
if (cookieSpec == null) {
synchronized (this) {
lock.lock();
try {
if (cookieSpec == null) {
this.cookieSpec = IgnoreSpecSpec.INSTANCE;
}
} finally {
lock.unlock();
}
}
return this.cookieSpec;

10
httpclient5/src/main/java/org/apache/hc/client5/http/impl/cookie/RFC6265CookieSpecFactory.java
View File

@ -27,6 +27,8 @@
package org.apache.hc.client5.http.impl.cookie;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.hc.client5.http.cookie.Cookie;
import org.apache.hc.client5.http.cookie.CookieOrigin;
import org.apache.hc.client5.http.cookie.CookieSpec;
@ -48,6 +50,8 @@ import org.apache.hc.core5.http.protocol.HttpContext;
@Contract(threading = ThreadingBehavior.SAFE)
public class RFC6265CookieSpecFactory implements CookieSpecFactory {
private final ReentrantLock lock;
public enum CompatibilityLevel {
STRICT,
RELAXED,
@ -65,6 +69,7 @@ public class RFC6265CookieSpecFactory implements CookieSpecFactory {
super();
this.compatibilityLevel = compatibilityLevel != null ? compatibilityLevel : CompatibilityLevel.RELAXED;
this.publicSuffixMatcher = publicSuffixMatcher;
this.lock = new ReentrantLock();
}
public RFC6265CookieSpecFactory(final PublicSuffixMatcher publicSuffixMatcher) {
@ -78,7 +83,8 @@ public class RFC6265CookieSpecFactory implements CookieSpecFactory {
@Override
public CookieSpec create(final HttpContext context) {
if (cookieSpec == null) {
synchronized (this) {
lock.lock();
try {
if (cookieSpec == null) {
switch (this.compatibilityLevel) {
case STRICT:
@ -118,6 +124,8 @@ public class RFC6265CookieSpecFactory implements CookieSpecFactory {
LaxExpiresHandler.INSTANCE);
}
}
} finally {
lock.unlock();
}
}
return this.cookieSpec;

374
httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/BasicHttpClientConnectionManager.java
View File

@ -35,6 +35,7 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.hc.client5.http.DnsResolver;
import org.apache.hc.client5.http.HttpRoute;
@ -103,6 +104,8 @@ public class BasicHttpClientConnectionManager implements HttpClientConnectionMan
private final HttpConnectionFactory<ManagedHttpClientConnection> connFactory;
private final String id;
private final ReentrantLock lock;
private ManagedHttpClientConnection conn;
private HttpRoute route;
private Object state;
@ -147,6 +150,7 @@ public class BasicHttpClientConnectionManager implements HttpClientConnectionMan
this.connectionConfig = ConnectionConfig.DEFAULT;
this.tlsConfig = TlsConfig.DEFAULT;
this.closed = new AtomicBoolean(false);
this.lock = new ReentrantLock();
}
public BasicHttpClientConnectionManager(
@ -184,40 +188,71 @@ public class BasicHttpClientConnectionManager implements HttpClientConnectionMan
return state;
}
public synchronized SocketConfig getSocketConfig() {
return socketConfig;
public SocketConfig getSocketConfig() {
lock.lock();
try {
return socketConfig;
} finally {
lock.unlock();
}
}
public synchronized void setSocketConfig(final SocketConfig socketConfig) {
this.socketConfig = socketConfig != null ? socketConfig : SocketConfig.DEFAULT;
public void setSocketConfig(final SocketConfig socketConfig) {
lock.lock();
try {
this.socketConfig = socketConfig != null ? socketConfig : SocketConfig.DEFAULT;
} finally {
lock.unlock();
}
}
/**
* @since 5.2
*/
public synchronized ConnectionConfig getConnectionConfig() {
return connectionConfig;
public ConnectionConfig getConnectionConfig() {
lock.lock();
try {
return connectionConfig;
} finally {
lock.unlock();
}
}
/**
* @since 5.2
*/
public synchronized void setConnectionConfig(final ConnectionConfig connectionConfig) {
this.connectionConfig = connectionConfig != null ? connectionConfig : ConnectionConfig.DEFAULT;
public void setConnectionConfig(final ConnectionConfig connectionConfig) {
lock.lock();
try {
this.connectionConfig = connectionConfig != null ? connectionConfig : ConnectionConfig.DEFAULT;
} finally {
lock.unlock();
}
}
/**
* @since 5.2
*/
public synchronized TlsConfig getTlsConfig() {
return tlsConfig;
public TlsConfig getTlsConfig() {
lock.lock();
try {
return tlsConfig;
} finally {
lock.unlock();
}
}
/**
* @since 5.2
*/
public synchronized void setTlsConfig(final TlsConfig tlsConfig) {
this.tlsConfig = tlsConfig != null ? tlsConfig : TlsConfig.DEFAULT;
public void setTlsConfig(final TlsConfig tlsConfig) {
lock.lock();
try {
this.tlsConfig = tlsConfig != null ? tlsConfig : TlsConfig.DEFAULT;
} finally {
lock.unlock();
}
}
public LeaseRequest lease(final String id, final HttpRoute route, final Object state) {
@ -246,13 +281,18 @@ public class BasicHttpClientConnectionManager implements HttpClientConnectionMan
};
}
private synchronized void closeConnection(final CloseMode closeMode) {
if (this.conn != null) {
if (LOG.isDebugEnabled()) {
LOG.debug("{} Closing connection {}", id, closeMode);
private void closeConnection(final CloseMode closeMode) {
lock.lock();
try {
if (this.conn != null) {
if (LOG.isDebugEnabled()) {
LOG.debug("{} Closing connection {}", id, closeMode);
}
this.conn.close(closeMode);
this.conn = null;
}
this.conn.close(closeMode);
this.conn = null;
} finally {
lock.unlock();
}
}
@ -298,30 +338,35 @@ public class BasicHttpClientConnectionManager implements HttpClientConnectionMan
}
}
synchronized ManagedHttpClientConnection getConnection(final HttpRoute route, final Object state) throws IOException {
Asserts.check(!isClosed(), "Connection manager has been shut down");
if (LOG.isDebugEnabled()) {
LOG.debug("{} Get connection for route {}", id, route);
ManagedHttpClientConnection getConnection(final HttpRoute route, final Object state) throws IOException {
lock.lock();
try {
Asserts.check(!isClosed(), "Connection manager has been shut down");
if (LOG.isDebugEnabled()) {
LOG.debug("{} Get connection for route {}", id, route);
}
Asserts.check(!this.leased, "Connection %s is still allocated", conn);
if (!Objects.equals(this.route, route) || !Objects.equals(this.state, state)) {
closeConnection(CloseMode.GRACEFUL);
}
this.route = route;
this.state = state;
checkExpiry();
validate();
if (this.conn == null) {
this.conn = this.connFactory.createConnection(null);
this.created = System.currentTimeMillis();
} else {
this.conn.activate();
}
this.leased = true;
if (LOG.isDebugEnabled()) {
LOG.debug("{} Using connection {}", id, conn);
}
return this.conn;
} finally {
lock.unlock();
}
Asserts.check(!this.leased, "Connection %s is still allocated", conn);
if (!Objects.equals(this.route, route) || !Objects.equals(this.state, state)) {
closeConnection(CloseMode.GRACEFUL);
}
this.route = route;
this.state = state;
checkExpiry();
validate();
if (this.conn == null) {
this.conn = this.connFactory.createConnection(null);
this.created = System.currentTimeMillis();
} else {
this.conn.activate();
}
this.leased = true;
if (LOG.isDebugEnabled()) {
LOG.debug("{} Using connection {}", id, conn);
}
return this.conn;
}
private InternalConnectionEndpoint cast(final ConnectionEndpoint endpoint) {
@ -332,124 +377,149 @@ public class BasicHttpClientConnectionManager implements HttpClientConnectionMan
}
@Override
public synchronized void release(final ConnectionEndpoint endpoint, final Object state, final TimeValue keepAlive) {
Args.notNull(endpoint, "Managed endpoint");
final InternalConnectionEndpoint internalEndpoint = cast(endpoint);
final ManagedHttpClientConnection conn = internalEndpoint.detach();
if (LOG.isDebugEnabled()) {
LOG.debug("{} Releasing connection {}", id, conn);
}
if (isClosed()) {
return;
}
public void release(final ConnectionEndpoint endpoint, final Object state, final TimeValue keepAlive) {
lock.lock();
try {
if (keepAlive == null) {
this.conn.close(CloseMode.GRACEFUL);
Args.notNull(endpoint, "Managed endpoint");
final InternalConnectionEndpoint internalEndpoint = cast(endpoint);
final ManagedHttpClientConnection conn = internalEndpoint.detach();
if (LOG.isDebugEnabled()) {
LOG.debug("{} Releasing connection {}", id, conn);
}
this.updated = System.currentTimeMillis();
if (!this.conn.isOpen() && !this.conn.isConsistent()) {
this.route = null;
this.conn = null;
this.expiry = Long.MAX_VALUE;
if (LOG.isDebugEnabled()) {
LOG.debug("{} Connection is not kept alive", id);
if (isClosed()) {
return;
}
try {
if (keepAlive == null) {
this.conn.close(CloseMode.GRACEFUL);
}
} else {
this.state = state;
if (conn != null) {
conn.passivate();
}
if (TimeValue.isPositive(keepAlive)) {
if (LOG.isDebugEnabled()) {
LOG.debug("{} Connection can be kept alive for {}", id, keepAlive);
}
this.expiry = this.updated + keepAlive.toMilliseconds();
} else {
if (LOG.isDebugEnabled()) {
LOG.debug("{} Connection can be kept alive indefinitely", id);
}
this.updated = System.currentTimeMillis();
if (!this.conn.isOpen() && !this.conn.isConsistent()) {
this.route = null;
this.conn = null;
this.expiry = Long.MAX_VALUE;
if (LOG.isDebugEnabled()) {
LOG.debug("{} Connection is not kept alive", id);
}
} else {
this.state = state;
if (conn != null) {
conn.passivate();
}
if (TimeValue.isPositive(keepAlive)) {
if (LOG.isDebugEnabled()) {
LOG.debug("{} Connection can be kept alive for {}", id, keepAlive);
}
this.expiry = this.updated + keepAlive.toMilliseconds();
} else {
if (LOG.isDebugEnabled()) {
LOG.debug("{} Connection can be kept alive indefinitely", id);
}
this.expiry = Long.MAX_VALUE;
}
}
} finally {
this.leased = false;
}
} finally {
lock.unlock();
}
}
@Override
public void connect(final ConnectionEndpoint endpoint, final TimeValue timeout, final HttpContext context) throws IOException {
lock.lock();
try {
Args.notNull(endpoint, "Endpoint");
final InternalConnectionEndpoint internalEndpoint = cast(endpoint);
if (internalEndpoint.isConnected()) {
return;
}
final HttpRoute route = internalEndpoint.getRoute();
final HttpHost host;
if (route.getProxyHost() != null) {
host = route.getProxyHost();
} else {
host = route.getTargetHost();
}
final Timeout connectTimeout = timeout != null ? Timeout.of(timeout.getDuration(), timeout.getTimeUnit()) : connectionConfig.getConnectTimeout();
final ManagedHttpClientConnection connection = internalEndpoint.getConnection();
if (LOG.isDebugEnabled()) {
LOG.debug("{} connecting endpoint to {} ({})", ConnPoolSupport.getId(endpoint), host, connectTimeout);
}
this.connectionOperator.connect(
connection,
host,
route.getLocalSocketAddress(),
connectTimeout,
socketConfig,
tlsConfig,
context);
if (LOG.isDebugEnabled()) {
LOG.debug("{} connected {}", ConnPoolSupport.getId(endpoint), ConnPoolSupport.getId(conn));
}
final Timeout socketTimeout = connectionConfig.getSocketTimeout();
if (socketTimeout != null) {
connection.setSocketTimeout(socketTimeout);
}
} finally {
lock.unlock();
}
}
@Override
public void upgrade(
final ConnectionEndpoint endpoint,
final HttpContext context) throws IOException {
lock.lock();
try {
Args.notNull(endpoint, "Endpoint");
Args.notNull(route, "HTTP route");
final InternalConnectionEndpoint internalEndpoint = cast(endpoint);
this.connectionOperator.upgrade(
internalEndpoint.getConnection(),
internalEndpoint.getRoute().getTargetHost(),
tlsConfig,
context);
} finally {
lock.unlock();
}
}
public void closeExpired() {
lock.lock();
try {
if (isClosed()) {
return;
}
if (!this.leased) {
checkExpiry();
}
} finally {
lock.unlock();
}
}
public void closeIdle(final TimeValue idleTime) {
lock.lock();
try {
Args.notNull(idleTime, "Idle time");
if (isClosed()) {
return;
}
if (!this.leased) {
long time = idleTime.toMilliseconds();
if (time < 0) {
time = 0;
}
final long deadline = System.currentTimeMillis() - time;
if (this.updated <= deadline) {
closeConnection(CloseMode.GRACEFUL);
}
}
} finally {
this.leased = false;
}
}
@Override
public synchronized void connect(final ConnectionEndpoint endpoint, final TimeValue timeout, final HttpContext context) throws IOException {
Args.notNull(endpoint, "Endpoint");
final InternalConnectionEndpoint internalEndpoint = cast(endpoint);
if (internalEndpoint.isConnected()) {
return;
}
final HttpRoute route = internalEndpoint.getRoute();
final HttpHost host;
if (route.getProxyHost() != null) {
host = route.getProxyHost();
} else {
host = route.getTargetHost();
}
final Timeout connectTimeout = timeout != null ? Timeout.of(timeout.getDuration(), timeout.getTimeUnit()) : connectionConfig.getConnectTimeout();
final ManagedHttpClientConnection connection = internalEndpoint.getConnection();
if (LOG.isDebugEnabled()) {
LOG.debug("{} connecting endpoint to {} ({})", ConnPoolSupport.getId(endpoint), host, connectTimeout);
}
this.connectionOperator.connect(
connection,
host,
route.getLocalSocketAddress(),
connectTimeout,
socketConfig,
tlsConfig,
context);
if (LOG.isDebugEnabled()) {
LOG.debug("{} connected {}", ConnPoolSupport.getId(endpoint), ConnPoolSupport.getId(conn));
}
final Timeout socketTimeout = connectionConfig.getSocketTimeout();
if (socketTimeout != null) {
connection.setSocketTimeout(socketTimeout);
}
}
@Override
public synchronized void upgrade(
final ConnectionEndpoint endpoint,
final HttpContext context) throws IOException {
Args.notNull(endpoint, "Endpoint");
Args.notNull(route, "HTTP route");
final InternalConnectionEndpoint internalEndpoint = cast(endpoint);
this.connectionOperator.upgrade(
internalEndpoint.getConnection(),
internalEndpoint.getRoute().getTargetHost(),
tlsConfig,
context);
}
public synchronized void closeExpired() {
if (isClosed()) {
return;
}
if (!this.leased) {
checkExpiry();
}
}
public synchronized void closeIdle(final TimeValue idleTime) {
Args.notNull(idleTime, "Idle time");
if (isClosed()) {
return;
}
if (!this.leased) {
long time = idleTime.toMilliseconds();
if (time < 0) {
time = 0;
}
final long deadline = System.currentTimeMillis() - time;
if (this.updated <= deadline) {
closeConnection(CloseMode.GRACEFUL);
}
lock.unlock();
}
}

122
httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/PoolingHttpClientConnectionManager.java
View File

@ -33,6 +33,7 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.hc.client5.http.DnsResolver;
import org.apache.hc.client5.http.HttpRoute;
@ -297,75 +298,82 @@ public class PoolingHttpClientConnectionManager
}
final Future<PoolEntry<HttpRoute, ManagedHttpClientConnection>> leaseFuture = this.pool.lease(route, state, requestTimeout, null);
return new LeaseRequest() {
// Using a ReentrantLock specific to each LeaseRequest instance to maintain the original
// synchronization semantics. This ensures that each LeaseRequest has its own unique lock.
private final ReentrantLock lock = new ReentrantLock();
private volatile ConnectionEndpoint endpoint;
@Override
public synchronized ConnectionEndpoint get(
public ConnectionEndpoint get(
final Timeout timeout) throws InterruptedException, ExecutionException, TimeoutException {
Args.notNull(timeout, "Operation timeout");
if (this.endpoint != null) {
return this.endpoint;
}
final PoolEntry<HttpRoute, ManagedHttpClientConnection> poolEntry;
lock.lock();
try {
poolEntry = leaseFuture.get(timeout.getDuration(), timeout.getTimeUnit());
} catch (final TimeoutException ex) {
leaseFuture.cancel(true);
throw ex;
}
if (LOG.isDebugEnabled()) {
LOG.debug("{} endpoint leased {}", id, ConnPoolSupport.formatStats(route, state, pool));
}
final ConnectionConfig connectionConfig = resolveConnectionConfig(route);
try {
if (poolEntry.hasConnection()) {
final TimeValue timeToLive = connectionConfig.getTimeToLive();
if (TimeValue.isNonNegative(timeToLive)) {
if (timeToLive.getDuration() == 0
|| Deadline.calculate(poolEntry.getCreated(), timeToLive).isExpired()) {
poolEntry.discardConnection(CloseMode.GRACEFUL);
Args.notNull(timeout, "Operation timeout");
if (this.endpoint != null) {
return this.endpoint;
}
final PoolEntry<HttpRoute, ManagedHttpClientConnection> poolEntry;
try {
poolEntry = leaseFuture.get(timeout.getDuration(), timeout.getTimeUnit());
} catch (final TimeoutException ex) {
leaseFuture.cancel(true);
throw ex;
}
if (LOG.isDebugEnabled()) {
LOG.debug("{} endpoint leased {}", id, ConnPoolSupport.formatStats(route, state, pool));
}
final ConnectionConfig connectionConfig = resolveConnectionConfig(route);
try {
if (poolEntry.hasConnection()) {
final TimeValue timeToLive = connectionConfig.getTimeToLive();
if (TimeValue.isNonNegative(timeToLive)) {
if (timeToLive.getDuration() == 0
|| Deadline.calculate(poolEntry.getCreated(), timeToLive).isExpired()) {
poolEntry.discardConnection(CloseMode.GRACEFUL);
}
}
}
}
if (poolEntry.hasConnection()) {
final TimeValue timeValue = resolveValidateAfterInactivity(connectionConfig);
if (TimeValue.isNonNegative(timeValue)) {
if (timeValue.getDuration() == 0
|| Deadline.calculate(poolEntry.getUpdated(), timeValue).isExpired()) {
final ManagedHttpClientConnection conn = poolEntry.getConnection();
boolean stale;
try {
stale = conn.isStale();
} catch (final IOException ignore) {
stale = true;
}
if (stale) {
if (LOG.isDebugEnabled()) {
LOG.debug("{} connection {} is stale", id, ConnPoolSupport.getId(conn));
if (poolEntry.hasConnection()) {
final TimeValue timeValue = resolveValidateAfterInactivity(connectionConfig);
if (TimeValue.isNonNegative(timeValue)) {
if (timeValue.getDuration() == 0
|| Deadline.calculate(poolEntry.getUpdated(), timeValue).isExpired()) {
final ManagedHttpClientConnection conn = poolEntry.getConnection();
boolean stale;
try {
stale = conn.isStale();
} catch (final IOException ignore) {
stale = true;
}
if (stale) {
if (LOG.isDebugEnabled()) {
LOG.debug("{} connection {} is stale", id, ConnPoolSupport.getId(conn));
}
poolEntry.discardConnection(CloseMode.IMMEDIATE);
}
poolEntry.discardConnection(CloseMode.IMMEDIATE);
}
}
}
final ManagedHttpClientConnection conn = poolEntry.getConnection();
if (conn != null) {
conn.activate();
} else {
poolEntry.assignConnection(connFactory.createConnection(null));
}
this.endpoint = new InternalConnectionEndpoint(poolEntry);
if (LOG.isDebugEnabled()) {
LOG.debug("{} acquired {}", id, ConnPoolSupport.getId(endpoint));
}
return this.endpoint;
} catch (final Exception ex) {
if (LOG.isDebugEnabled()) {
LOG.debug("{} endpoint lease failed", id);
}
pool.release(poolEntry, false);
throw new ExecutionException(ex.getMessage(), ex);
}
final ManagedHttpClientConnection conn = poolEntry.getConnection();
if (conn != null) {
conn.activate();
} else {
poolEntry.assignConnection(connFactory.createConnection(null));
}
this.endpoint = new InternalConnectionEndpoint(poolEntry);
if (LOG.isDebugEnabled()) {
LOG.debug("{} acquired {}", id, ConnPoolSupport.getId(endpoint));
}
return this.endpoint;
} catch (final Exception ex) {
if (LOG.isDebugEnabled()) {
LOG.debug("{} endpoint lease failed", id);
}
pool.release(poolEntry, false);
throw new ExecutionException(ex.getMessage(), ex);
} finally {
lock.unlock();
}
}

8
httpclient5/src/main/java/org/apache/hc/client5/http/psl/PublicSuffixMatcherLoader.java
View File

@ -35,6 +35,7 @@ import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.hc.core5.annotation.Contract;
import org.apache.hc.core5.annotation.ThreadingBehavior;
@ -52,6 +53,8 @@ public final class PublicSuffixMatcherLoader {
private static final Logger LOG = LoggerFactory.getLogger(PublicSuffixMatcherLoader.class);
private static final ReentrantLock lock = new ReentrantLock();
private static PublicSuffixMatcher load(final InputStream in) throws IOException {
final List<PublicSuffixList> lists = PublicSuffixListParser.INSTANCE.parseByType(
new InputStreamReader(in, StandardCharsets.UTF_8));
@ -76,7 +79,8 @@ public final class PublicSuffixMatcherLoader {
public static PublicSuffixMatcher getDefault() {
if (DEFAULT_INSTANCE == null) {
synchronized (PublicSuffixMatcherLoader.class) {
lock.lock();
try {
if (DEFAULT_INSTANCE == null){
final URL url = PublicSuffixMatcherLoader.class.getResource(
"/mozilla/public-suffix-list.txt");
@ -91,6 +95,8 @@ public final class PublicSuffixMatcherLoader {
DEFAULT_INSTANCE = new PublicSuffixMatcher(DomainType.ICANN, Collections.singletonList("com"), null);
}
}
} finally {
lock.unlock();
}
}
return DEFAULT_INSTANCE;