[
https://issues.apache.org/jira/browse/HADOOP-19946?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18141853#comment-18141853
]
ASF GitHub Bot commented on HADOOP-19946:
-----------------------------------------
anmolanmol1234 commented on code in PR #8830:
URL: https://github.com/apache/hadoop/pull/8830#discussion_r4522691128
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/KeepAliveCache.java:
##########
@@ -61,216 +65,405 @@ class KeepAliveCache extends
LinkedBlockingDeque<HttpClientConnection>
KeepAliveCache.class);
/**
- * Flag to indicate if the cache is closed.
+ * Indicates whether the cache has been closed.
+ * Used to prevent further operations after closure.
+ */
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ /**
+ * Maximum number of connections allowed in the default host cache.
*/
- private final AtomicBoolean isClosed = new AtomicBoolean(false);
+ private final int maxDefaultConnections;
/**
- * Maximum number of connections that can be cached.
+ * Maximum number of connections allowed in the cluster host cache.
*/
- private final int maxCacheConnections;
+ private final int maxClusterConnections;
/**
- * Account name for which the cache is created. To be used only in exception
- * messages.
+ * The account name path associated with this cache instance.
*/
private final String accountNamePath;
/**
- * Executor server to trigger connection refresh from cache manager.
+ * Executor service for single-threaded cache refresh operations.
*/
- private ExecutorService singleThreadPool = null;
+ private ExecutorService singleThreadPool;
/**
- * Executor service to trigger async cache warmup.
+ * Executor service for fixed-threaded cache warmup and refresh operations.
+ */
+ private ExecutorService fixedThreadPool;
+
+ /**
+ * Represents a pooled HTTP connection with a usage flag and cache key.
+ * <p>
+ * Each instance wraps a {@link HttpClientConnection} and tracks whether it
is currently in use.
+ * The {@code cacheKey} identifies the host this connection is associated
with.
+ * The {@code inUse} flag is used to prevent race conditions during eviction
or retrieval.
*/
- private ExecutorService fixedThreadPool = null;
+ @VisibleForTesting
+ public static final class PooledConnection {
+
+ final HttpClientConnection conn;
+
+ final String cacheKey;
+
+ final AtomicBoolean inUse = new AtomicBoolean(false); // New Flag
+ PooledConnection(HttpClientConnection conn, String cacheKey) {
+ this.conn = conn;
+ this.cacheKey = cacheKey;
+ }
+ }
/**
- * Creates an {@link KeepAliveCache} instance using filesystem's
configuration.
+ * HostQueue is a thread-safe queue for storing pooled HTTP connections per
host.
* <p>
- * The size of the cache is determined by the configuration
- * {@value
org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys#FS_AZURE_APACHE_HTTP_CLIENT_MAX_CACHE_SIZE}.
- * If the configuration is not set, the default value is
- * {@value
org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations#DEFAULT_APACHE_HTTP_CLIENT_MAX_CACHE_SIZE}.
- * <p>.
+ * Uses a ConcurrentLinkedDeque to allow concurrent access and lock-free
operations.
+ * Each HostQueue instance manages the connections for a specific host in
the cache.
+ */
+ @VisibleForTesting
+ public static final class HostQueue {
+
+ // ConcurrentLinkedDeque is thread-safe; no manual lock needed
+ final ConcurrentLinkedDeque<PooledConnection> queue
+ = new ConcurrentLinkedDeque<>();
+ }
+
+ private final ConcurrentHashMap<String, HostQueue> hostCaches
+ = new ConcurrentHashMap<>();
+
+ private final ConcurrentLinkedQueue<PooledConnection> clusterQueue
+ = new ConcurrentLinkedQueue<>();
+
+ private final AtomicInteger defaultSize = new AtomicInteger(0);
+
+ private final AtomicInteger clusterSize = new AtomicInteger(0);
+
+ /**
+ * Checks if the cache is closed.
+ *
+ * @return true if the cache is closed, false otherwise
+ */
+ boolean isClosed() {
+ return closed.get();
+ }
+
+ /**
+ * Gets the current size of the default host cache.
+ *
+ * @return the number of connections cached for the default host
+ */
+ int getCachedDefaultSize() {
+ return defaultSize.get();
+ }
+
+ /**
+ * Constructs a KeepAliveCache instance with the specified configuration.
+ *
+ * @param abfsConfiguration the ABFS configuration
*/
KeepAliveCache(AbfsConfiguration abfsConfiguration) {
- this.accountNamePath =
- abfsConfiguration.getAccountName();
- this.maxCacheConnections =
- abfsConfiguration.getApacheMaxCacheSize();
- // Initialise singleThreadPool if cache refresh is enabled.
+ this.accountNamePath = abfsConfiguration.getAccountName();
+ this.maxDefaultConnections
+ = abfsConfiguration.getApacheMaxDefaultCacheSize();
+ this.maxClusterConnections
+ = abfsConfiguration.getApacheMaxNonDefaultCacheSize();
+
+ // Fix: Always use Daemon threads to prevent JVM hang on exit
if (abfsConfiguration.getApacheCacheRefreshCount() > 0) {
this.singleThreadPool = Executors.newSingleThreadExecutor(r -> {
- Thread thread = new Thread(r);
- thread.setName("CacheRefreshThread");
- thread.setDaemon(true);
- return thread;
+ Thread t = new Thread(r, "CacheRefreshThread");
+ t.setDaemon(true);
+ return t;
});
}
- // Initialise fixedThreadPool if cache warmup or cache refresh is enabled.
- if (abfsConfiguration.getApacheCacheWarmupCount() > 0
- || abfsConfiguration.getApacheCacheRefreshCount() > 0) {
- this.fixedThreadPool = Executors.newFixedThreadPool(Math.min(5,
- Math.max(abfsConfiguration.getApacheCacheWarmupCount(),
- abfsConfiguration.getApacheCacheRefreshCount())), r -> {
- Thread thread = new Thread(r);
- thread.setName("AsyncCacheConnectionThread");
- thread.setDaemon(true);
- return thread;
- });
+ int warmup = abfsConfiguration.getApacheCacheWarmupCount();
+ int refresh = abfsConfiguration.getApacheCacheRefreshCount();
+ if (warmup > 0 || refresh > 0) {
+ this.fixedThreadPool = Executors.newFixedThreadPool(
+ Math.min(5, Math.max(warmup, refresh)), r -> {
+ Thread t = new Thread(r, "AsyncCacheConnectionThread");
+ t.setDaemon(true);
+ return t;
+ });
}
}
/**
- * Safe close of the HttpClientConnection.
+ * Retrieves a pooled HTTP connection for the specified host.
+ * <p>
+ * If the cache is closed, throws ClosedIOException. If no connection is
available, returns null.
+ * Decrements the appropriate counter and closes stale connections.
*
- * @param hc HttpClientConnection to be closed
+ * @param host the host to retrieve a connection for
+ * @param isDefaultHost true if the host is the default host
+ * @return a valid HttpClientConnection or null if unavailable
+ * @throws IOException if the cache is closed
*/
- private void closeHttpClientConnection(final HttpClientConnection hc) {
- try {
- hc.close();
- } catch (IOException ex) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Close failed for connection: {}", hc, ex);
+ public HttpClientConnection get(String host, boolean isDefaultHost)
+ throws IOException {
+ if (closed.get()) {
+ throw new ClosedIOException(accountNamePath, KEEP_ALIVE_CACHE_CLOSED);
+ }
+ HostQueue hq = hostCaches.get(host);
+ if (hq == null) {
+ return null;
+ }
+
+ PooledConnection pooled;
+ // poll() is now an atomic, non-blocking operation
+ while ((pooled = hq.queue.poll()) != null) {
+ if (!pooled.inUse.compareAndSet(false, true)) {
+ continue; // Already claimed by eviction
+ }
+
+ if (!isDefaultHost) {
+ clusterSize.decrementAndGet();
+ } else {
+ defaultSize.decrementAndGet();
+ }
+
+ if (pooled.conn.isOpen() && !pooled.conn.isStale()) {
+ return pooled.conn;
}
+ closeQuietly(pooled.conn);
}
+ return null;
}
/**
- * Close all connections in cache.
+ * Adds a connection to the cache for the specified host type.
+ * <p>
+ * Only valid, open, non-stale connections are cached. Defensive casting
ensures only
+ * AbfsManagedApacheHttpConnection instances are accepted. Returns true if
the connection
+ * was successfully cached, false otherwise.
+ *
+ * @param conn the connection to cache
+ * @param isDefaultHost true if the host is the default host
+ * @return true if the connection was cached, false otherwise
*/
- @Override
- public void close() {
- boolean closed = isClosed.getAndSet(true);
- if (closed) {
- return;
+ public boolean put(HttpClientConnection conn, boolean isDefaultHost) {
+ if (conn == null || closed.get() || !conn.isOpen() || conn.isStale()) {
+ closeQuietly(conn);
+ return false;
}
- closeInternal();
- if (singleThreadPool != null && !singleThreadPool.isShutdown()) {
- singleThreadPool.shutdownNow();
+
+ // Defensive casting check
+ if (!(conn instanceof AbfsManagedApacheHttpConnection)) {
+ closeQuietly(conn);
+ return false;
}
- if (fixedThreadPool != null && !fixedThreadPool.isShutdown()) {
- fixedThreadPool.shutdownNow();
+ String host = ((AbfsManagedApacheHttpConnection) conn).getTargetHost()
+ .toHostString();
+ PooledConnection pooled = new PooledConnection(conn, host);
+ HostQueue hq = hostCaches.computeIfAbsent(host, k -> new HostQueue());
+
+ return isDefaultHost ? putDefault(hq, pooled) : putCluster(hq, pooled);
+ }
+
+ /**
+ * Adds a connection to the default host cache, evicting the oldest if the
cache exceeds its limit.
+ *
+ * @param hq the HostQueue for the default host
+ * @param pooled the pooled connection to add
+ * @return true if the connection was cached, false otherwise
+ */
+ private boolean putDefault(HostQueue hq, PooledConnection pooled) {
+ if (maxDefaultConnections <= 0) {
+ closeQuietly(pooled.conn);
+ return false;
}
+
+ hq.queue.offer(pooled);
+ if (defaultSize.incrementAndGet() > maxDefaultConnections) {
+ PooledConnection evicted = hq.queue.poll(); // Evict oldest
+ if (evicted != null) {
+ // Double-check 'inUse' to ensure we don't close a connection just
handed out
+ if (evicted.inUse.compareAndSet(false, true)) {
+ defaultSize.decrementAndGet();
+ closeQuietly(evicted.conn);
+ } else {
+ // If we couldn't claim it for eviction, it was just grabbed by get()
+ // defaultSize was already decremented by get(), so just move on.
+ }
+ }
+ }
+ return true;
}
/**
- * @return true if the cache is closed, false otherwise.
+ * Adds a connection to the cluster host cache, evicting connections if the
cache exceeds its limit.
+ *
+ * @param hq the HostQueue for the cluster host
+ * @param pooled the pooled connection to add
+ * @return true if the connection was cached, false otherwise
*/
- public boolean getIsClosed() {
- return isClosed.get();
+ private boolean putCluster(HostQueue hq, PooledConnection pooled) {
+ if (maxClusterConnections <= 0) {
+ closeQuietly(pooled.conn);
+ return false;
+ }
+
+ while (clusterSize.get() >= maxClusterConnections) {
+ PooledConnection candidate = clusterQueue.poll();
+ if (candidate == null) {break;}
+
+ if (candidate.inUse.compareAndSet(false, true)) {
+ clusterSize.decrementAndGet();
+ HostQueue evictHq = hostCaches.get(candidate.cacheKey);
+ if (evictHq != null) {
+ // remove() on ConcurrentLinkedDeque is thread-safe and lock-free
+ evictHq.queue.remove(candidate);
+ }
+ closeQuietly(candidate.conn);
+ }
+ }
+
+ hq.queue.offer(pooled);
+ clusterQueue.offer(pooled);
Review Comment:
Every put adds the same PooledConnection to both hq.queue and clusterQueue,
but get() only polls hq.queue. The entry remains in clusterQueue until the
eviction loop runs, which only happens when clusterSize >=
maxClusterConnections.
In normal reuse cycles, the size stays below the limit, so stale entries
accumulate in clusterQueue, retaining references to connections unnecessarily.
> ABFS: Enable data locality aware reads for Azure Blob Storage
> -------------------------------------------------------------
>
> Key: HADOOP-19946
> URL: https://issues.apache.org/jira/browse/HADOOP-19946
> Project: Hadoop Common
> Issue Type: New Feature
> Components: fs/azure
> Reporter: Manish Bhatt
> Assignee: Manish Bhatt
> Priority: Major
> Labels: pull-request-available
>
> Introduce data locality support in ABFS to improve read performance for Azure
> Blob Storage by routing read requests to the endpoint that hosts the required
> data range. The goal is to reduce latency, improve throughput, and make read
> operations more efficient for large files and distributed workloads. This
> work adds blob layout discovery, range-to-endpoint mapping, locality-aware
> range selection, and caching of layout metadata to avoid repeated lookups. It
> also enables segmented read-ahead aligned with locality boundaries and adds
> safeguards for concurrency, eviction, timeout handling, and failure
> propagation needed for reliable production-scale reads.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
