[
https://issues.apache.org/jira/browse/HADOOP-19946?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18141898#comment-18141898
]
ASF GitHub Bot commented on HADOOP-19946:
-----------------------------------------
anmolanmol1234 commented on code in PR #8830:
URL: https://github.com/apache/hadoop/pull/8830#discussion_r4523749945
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/BlobLayoutCache.java:
##########
@@ -0,0 +1,610 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Expiry;
+import com.github.benmanes.caffeine.cache.Ticker;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.fs.azurebfs.contracts.services.BlobLayoutResponse;
+import
org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import static org.apache.hadoop.fs.azurebfs.services.AbfsInputStream.LOG;
+
+/**
+ * BlobLayoutCache is a singleton class that manages caching of BlobLayout
objects for Azure Blob File System (ABFS).
+ * It uses a Caffeine cache to store layout information for blobs, optimizing
range fetches and minimizing redundant network calls.
+ *
+ * <p>Key Features:</p>
+ * <ul>
+ * <li>Evicts cache entries based on idle timeout and cache weight (number
of cached ranges).</li>
+ * <li>Tracks active streams to prevent eviction while a stream is open.</li>
+ * <li>Manages in-flight promises for range fetches to avoid duplicate
requests.</li>
+ * <li>Provides methods to register/deregister streams, retrieve cached
layouts, compute gaps, and manage promises.</li>
+ * </ul>
+ *
+ * <p>Usage:</p>
+ * <ul>
+ * <li>Call {@link #registerStream(String, long)} when a stream is
opened.</li>
+ * <li>Call {@link #deregisterStream(String)} when a stream is closed.</li>
+ * <li>Use {@link #getBlobLayout(String, long, long)} and {@link
#getGaps(String, long, long)} to query cached layout info.</li>
+ * <li>Use {@link #putBlobLayout(String, BlobLayoutResponse, long)} to
update the cache with new layout info.</li>
+ * <li>Use {@link #getBridgeGap(String, long, long)} to determine the next
range to fetch, considering in-flight and cached ranges.</li>
+ * <li>Use {@link #removePromise(String, long, long)} to remove completed
in-flight promises.</li>
+ * </ul>
+ */
+public final class BlobLayoutCache {
+
+ /**
+ * LayoutEntry represents a cache entry for a blob's layout.
+ * It tracks the BlobLayout, active stream count, and last deregistered time
for eviction logic.
+ */
+ private static class LayoutEntry {
+
+ /** The BlobLayout instance for this entry. */
+ private final BlobLayout layout;
+
+ /** Indicates if the layout is present and valid. */
+ private AtomicBoolean isLayoutPresent = new AtomicBoolean(true);
+
+ /** Number of active streams using this layout. */
+ private final AtomicInteger activeStreams = new AtomicInteger(0);
+
+ /** Last time (in nanoseconds) when all streams were deregistered. Used
for idle eviction. */
+ private final AtomicLong lastDeregisteredTimeInNanos = new AtomicLong(
+ Long.MAX_VALUE);
+
+ /**
+ * Constructs a LayoutEntry with the given content length.
+ * @param contentLength the length of the blob content
+ */
+ LayoutEntry(long contentLength) {
+ this.layout = new BlobLayout(contentLength);
+ }
+
+ /**
+ * Checks if the layout is unavailable (not present).
+ * @return true if layout is unavailable, false otherwise
+ */
+ boolean isLayoutUnavailable() {
+ return !isLayoutPresent.get();
+ }
+ }
+
+ /**
+ * Caffeine Cache instance replacing ConcurrentHashMap.
+ * Uses weighted eviction based on the number of ranges stored in each
layout.
+ */
+ private final Cache<String, LayoutEntry> cache;
+
+ /** Idle timeout in milliseconds for cache eviction. */
+ private final long idleTimeoutMs;
+
+ /** Maximum cache weight (total number of cached ranges) before eviction. */
+ private final long maxCacheWeight;
+
+ /** Time source used for idle-eviction bookkeeping (injectable for tests). */
+ private final Ticker ticker;
+
+ /** Singleton instance of BlobLayoutCache. */
+ private static volatile BlobLayoutCache instance = null;
+
+ /** Guards against repeatedly logging the singleton config-mismatch warning.
*/
+ private static volatile boolean configMismatchWarned = false;
+
+ /**
+ * Registry of in-flight promises for range fetches, keyed by file ETag.
+ * Each value is a thread-safe list of InFlightPromise objects.
+ */
+ private final ConcurrentHashMap<String,
CopyOnWriteArrayList<InFlightPromise>>
+ promiseRegistry = new ConcurrentHashMap<>();
+
+ /**
+ * InFlightPromise represents a pending fetch for a specific range.
+ * @param start the start offset of the range
+ * @param end the end offset of the range
+ * @param future the CompletableFuture representing the fetch operation
+ */
+ public record InFlightPromise(long start, long end,
+ CompletableFuture<Void> future) {}
+
+ /**
+ * Executor to asynchronously execute get blob layout calls
+ */
+ private static final ExecutorService FETCH_EXECUTOR = new ThreadPoolExecutor(
Review Comment:
A ThreadPoolExecutor only grows past core when the queue is full, so with a
1024-slot queue the pool stays at 8 threads and the max of 32 is never used.
Set core = max, or use allowCoreThreadTimeOut. Please also use
ShutdownHookManager, not Runtime.addShutdownHook.
> 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]
