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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
