From 37b2e6d340d804ba685ef959c0e4b18146776edf Mon Sep 17 00:00:00 2001 From: Prince Kumar Date: Mon, 3 Aug 2026 09:15:36 +0000 Subject: [PATCH 1/2] using fsspec adaptive-readahead cache --- gcsfs/core.py | 60 +-- gcsfs/prefetcher.py | 887 +-------------------------------- gcsfs/tests/test_prefetcher.py | 2 +- gcsfs/zonal_file.py | 2 +- 4 files changed, 31 insertions(+), 920 deletions(-) diff --git a/gcsfs/core.py b/gcsfs/core.py index 1bae97e60..308d98612 100644 --- a/gcsfs/core.py +++ b/gcsfs/core.py @@ -34,7 +34,7 @@ from .credentials import GoogleCredentials from .inventory_report import InventoryReport from .retry import errs, retry_request, validate_response -from .zb_hns_utils import DEFAULT_CONCURRENCY, MAX_PREFETCH_SIZE +from .zb_hns_utils import DEFAULT_CONCURRENCY logger = logging.getLogger("gcsfs") @@ -2049,7 +2049,7 @@ async def default_fetcher(start, size, split_factor=1): fetcher_fn = default_fetcher - from .prefetcher import BackgroundPrefetcher + from fsspec.prefetch import BackgroundPrefetcher prefetcher = BackgroundPrefetcher( fetcher=fetcher_fn, @@ -2313,7 +2313,7 @@ def __init__( mode="rb", block_size=DEFAULT_BLOCK_SIZE, autocommit=True, - cache_type="readahead", + cache_type="adaptive_readahead", cache_options=None, acl=None, consistency="md5", @@ -2376,6 +2376,15 @@ def __init__( raise OSError("Attempt to open a bucket") self.generation = _coalesce_generation(generation, path_generation) self.concurrency = kwargs.get("concurrency", DEFAULT_CONCURRENCY) + + if "r" in mode and cache_type == "adaptive_readahead": + if cache_type not in fsspec.core.caches: + warnings.warn( + "fsspec adaptive_readahead cache is unavailable in this environment; " + "falling back to readahead" + ) + cache_type = "readahead" + super().__init__( gcsfs, path, @@ -2394,34 +2403,6 @@ def __init__( self.consistency = consistency self.checker = get_consistency_checker(consistency) - # Ideally, all of these fields should be part of `cache_options`. Because current - # `fsspec` caches do not accept arbitrary `*args` and `**kwargs`, passing them - # there currently causes instantiation errors. We are holding off on introducing - # them as explicit keyword arguments to ensure existing user workloads are not - # disrupted. This will be refactored once the upstream `fsspec` changes are merged. - use_prefetch_reader = kwargs.get( - "use_experimental_adaptive_prefetching", False - ) or os.environ.get( - "USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING", "false" - ).lower() in ( - "true", - "1", - ) - - if "r" in mode and use_prefetch_reader: - max_prefetch_size = kwargs.get("max_prefetch_size", MAX_PREFETCH_SIZE) - from .prefetcher import BackgroundPrefetcher - - self._prefetch_engine = BackgroundPrefetcher( - self._async_fetch_range, - self.size, - max_prefetch_size=max_prefetch_size, - concurrency=self.concurrency, - loop=self.gcsfs.loop, - ) - else: - self._prefetch_engine = None - # _supports_append is an internal argument not meant to be used directly. # If True, allows opening file in append mode. This is generally not supported # by GCS, but may be supported by subclasses (e.g. ZonalFile). This flag should @@ -2612,8 +2593,6 @@ def _fetch_range(self, start=None, end=None): if not both None, fetch only given range """ try: - if hasattr(self, "_prefetch_engine") and self._prefetch_engine: - return self._prefetch_engine.fetch(start=start, end=end) return self.fs.cat_file( self.path, start=start, @@ -2626,21 +2605,6 @@ def _fetch_range(self, start=None, end=None): return b"" raise - async def _async_fetch_range(self, start_offset, total_size, split_factor=1): - """Async fetcher mapped to the Prefetcher engine for regional buckets.""" - return await self.gcsfs._cat_file_concurrent( - self.path, - start=start_offset, - end=start_offset + total_size, - concurrency=split_factor, - cache_type=self.cache_type, - ) - - def close(self): - super().close() - if hasattr(self, "_prefetch_engine") and self._prefetch_engine: - self._prefetch_engine.close() - def _convert_fixed_key_metadata(metadata, *, from_google=False): """ diff --git a/gcsfs/prefetcher.py b/gcsfs/prefetcher.py index 9f3b52b9e..544e46eea 100644 --- a/gcsfs/prefetcher.py +++ b/gcsfs/prefetcher.py @@ -1,875 +1,22 @@ -import asyncio -import ctypes -import logging -import weakref -from collections import deque +"""Compatibility re-exports for prefetch classes. -import fsspec.asyn +Prefetch logic now lives in filesystem_spec/fsspec. +""" -logger = logging.getLogger(__name__) - -from gcsfs.zb_hns_utils import ( +from fsspec.prefetch import ( # noqa: F401 HAS_CPYTHON_API, - PyBytes_AsString, - PyBytes_FromStringAndSize, + BackgroundPrefetcher, + PrefetchConsumer, + PrefetchProducer, + RunningAverageTracker, + _fast_slice, ) - -# Please refer to following discussion to understand why this is required at this point -# Discussion = https://github.com/fsspec/gcsfs/pull/795#discussion_r3032749881 -def _fast_slice(src_bytes, offset, read_size): - if read_size == 0: - return b"" - if offset < 0 or offset + read_size > len(src_bytes): - raise ValueError("Slice indices out of bounds") - - if HAS_CPYTHON_API: - dest_bytes = PyBytes_FromStringAndSize(None, read_size) - src_ptr = PyBytes_AsString(src_bytes) - dest_ptr = PyBytes_AsString(dest_bytes) - # Releases the GIL - ctypes.memmove(dest_ptr, src_ptr + offset, read_size) - return dest_bytes - else: - # Standard fallback for PyPy/non-CPython - return src_bytes[offset : offset + read_size] - - -class RunningAverageTracker: - """Tracks a running average of values over a sliding window. - - This is used to monitor read sizes and adaptively scale the - prefetching strategy based on recent user behavior. - """ - - def __init__(self, maxlen=10): - """Initializes the tracker with a specific window size. - - Args: - maxlen (int): The maximum number of historical values to keep. - """ - logger.debug("Initializing RunningAverageTracker with maxlen: %d", maxlen) - self._history = deque(maxlen=maxlen) - self._sum = 0 - - def add(self, value: int): - """Adds a new value to the sliding window and updates the rolling sum. - - Args: - value (int): The integer value to add to the history. - """ - if value <= 0: - raise ValueError( - "Internal error, RunningAverageTracker tried inserting negative value" - ) - if len(self._history) == self._history.maxlen: - self._sum -= self._history[0] - - self._history.append(value) - self._sum += value - logger.debug( - "RunningAverageTracker added value: %d, new sum: %d", value, self._sum - ) - - @property - def average(self) -> int: - """Calculates and returns the current running average. - - Returns: - int: The integer average of the current history. - """ - count = len(self._history) - if count == 0: - return 1024 * 1024 # 1MB - return self._sum // count - - @property - def is_variable(self) -> bool: - """Determines if the history contains distinct chunk sizes.""" - count = len(self._history) - if count < 2: - return False - - return len(set(self._history)) > 1 - - @property - def last_value(self) -> int: - """Returns the most recent entry in the history.""" - if not self._history: - raise RuntimeError("No entry found in history") - - return self._history[-1] - - def clear(self): - """Clears the history and resets the sum to zero.""" - logger.debug("Clearing RunningAverageTracker history.") - self._history.clear() - self._sum = 0 - - -class PrefetchProducer: - """Background worker that fetches sequential blocks of data. - - This class handles the network requests. It spawns asynchronous tasks - to fetch data ahead of the user's current reading position and - places those task promises into a queue for the consumer. - """ - - # If the request is too small, and prefetch window is expanded till 5MB - # we then make request in 5MB blocks. - MIN_CHUNK_SIZE = 5 * 1024 * 1024 - - # If user doesn't specify any max_prefetch_size, the prefetcher defaults - # to maximum of 2 * io_size and 128MB - MIN_PREFETCH_SIZE = 128 * 1024 * 1024 - - # The prefetching starts on the third read. - MIN_STREAKS_FOR_PREFETCHING = 3 - - # Threshold for disabling proactive prefetching on large, variable reads. - # - # If the average read size exceeds this value and patterns are variable, - # prefetching shifts from an I/O bottleneck to a memory(CPU) bottleneck. When a user - # requests random massive sizes (e.g., jumping between 64MB and INF), the - # producer still fetches chunks based on the rolling average. The consumer - # then has to pick up multiple chunks and stitch them together to match the - # exact requested size. - # - # For small average read sizes, this byte assembly is fast and the bottleneck - # remains the network I/O. However, for massive reads (>= 64MB), the extra - # step of copying and assembling huge byte strings in memory severely slows - # down the operation. - VARIABLE_IO_THRESHOLD = 64 * 1024 * 1024 - - def __init__( - self, - fetcher, - size: int, - concurrency: int, - queue: asyncio.Queue, - wakeup_event: asyncio.Event, - consumer: "PrefetchConsumer", - tracker: RunningAverageTracker, - orchestrator: "BackgroundPrefetcher", - user_max_prefetch_size=None, - ): - """Initializes the background producer. - - Args: - fetcher (Callable): A coroutine function to fetch bytes from a remote source. - size (int): Total size of the file being fetched. - concurrency (int): Maximum number of concurrent fetch tasks. - queue (asyncio.Queue): The shared queue to push download tasks into. - wakeup_event (asyncio.Event): Event used to wake the producer from an idle state. - consumer (PrefetchConsumer): The consumer reading the prefetched chunks. - tracker (RunningAverageTracker): Tracker for history of read sizes. - orchestrator (BackgroundPrefetcher): The parent object managing the operation. - user_max_prefetch_size (int, optional): A hard limit for prefetch size overrides. - """ - logger.debug( - "Initializing PrefetchProducer: size=%d, concurrency=%d, user_max_prefetch_size=%s", - size, - concurrency, - user_max_prefetch_size, - ) - self.fetcher = fetcher - self.size = size - self.concurrency = concurrency - self.queue = queue - self.wakeup_event = wakeup_event - - self.consumer = consumer - self.tracker = tracker - self.orchestrator = weakref.proxy(orchestrator) - self._user_max_prefetch_size = user_max_prefetch_size - - self.current_offset = 0 - self.is_stopped = False - self._active_tasks = set() - self._producer_task = None - - @property - def max_prefetch_size(self) -> int: - """Calculates the maximum prefetch size based on user intent or io size. - - Returns: - int: The maximum number of bytes to prefetch ahead. - """ - if self._user_max_prefetch_size is not None: - return min( - self._user_max_prefetch_size, - max(2 * self.tracker.average, self.MIN_PREFETCH_SIZE), - ) - return max(2 * self.tracker.average, self.MIN_PREFETCH_SIZE) - - def start(self): - """Starts the background producer loop. - - This clears any previous wakeup events and spawns the main loop task. - """ - logger.debug("Starting PrefetchProducer loop.") - self.is_stopped = False - self.wakeup_event.clear() - self._producer_task = asyncio.create_task(self._loop()) - - async def stop(self): - """Cancels all active fetch tasks and shuts down the producer loop. - - This method ensures the queue is flushed and waits for cancelled - tasks to finish cleaning up. - """ - logger.debug( - "Stopping PrefetchProducer. Active fetch tasks: %d", len(self._active_tasks) - ) - self.is_stopped = True - self.wakeup_event.set() - - tasks_to_wait = [] - if self._producer_task and not self._producer_task.done(): - self._producer_task.cancel() - tasks_to_wait.append(self._producer_task) - - for task in list(self._active_tasks): - if not task.done(): - tasks_to_wait.append(task) - - # We do not cancel the network task, instead we wait on them. - # This is intentionally done to avoid MRD stream disruption. - self._active_tasks.clear() - - # Clear out any leftover items in the queue - cleared_items = 0 - while not self.queue.empty(): - try: - item = self.queue.get_nowait() - if ( - isinstance(item, asyncio.Task) - and item.done() - and not item.cancelled() - ): - item.exception() - cleared_items += 1 - except asyncio.QueueEmpty: - break - - if cleared_items > 0: - logger.debug( - "Cleared %d leftover items from the queue during stop.", cleared_items - ) - - if tasks_to_wait: - logger.debug( - "Waiting for %d cancelled tasks to finish their teardown.", - len(tasks_to_wait), - ) - await asyncio.gather(*tasks_to_wait, return_exceptions=True) - - self.wakeup_event.clear() - - async def restart(self, new_offset: int): - """Stops current tasks and restarts the background loop at a new byte offset. - - Args: - new_offset (int): The new byte position to start prefetching from. - """ - logger.debug("Restarting PrefetchProducer at new offset: %d", new_offset) - await self.stop() - self.current_offset = new_offset - self.start() - - async def _loop(self): - """The main background loop that delegates calculations and spawns tasks.""" - logger.debug("PrefetchProducer internal loop is now running.") - try: - while not self.is_stopped: - await self.wakeup_event.wait() - self.wakeup_event.clear() - - if self.is_stopped: - break - - await self._process_prefetch_cycle() - - except asyncio.CancelledError: - logger.debug("PrefetchProducer loop was cancelled.") - except Exception as e: - logger.error( - "PrefetchProducer loop encountered an unexpected error: %s", - e, - exc_info=True, - ) - self.is_stopped = True - self.orchestrator.set_error(e) - await self.queue.put(e) - - def _calculate_prefetch_params(self) -> tuple[int, int, int]: - """ - Evaluates current trackers and state to determine sizes. - - Returns: - tuple: (prefetch_size, io_size, effective_prefetch_size) - """ - avg_io_size = self.tracker.average - streak = self.consumer.sequential_streak - is_variable = self.tracker.is_variable - last_read_size = self.tracker.last_value - - exceeds_user_max = ( - self._user_max_prefetch_size is not None - and avg_io_size > self._user_max_prefetch_size - ) - - # Disable prefetching ahead if variable AND average > 64MB, or if it exceeds user max - if ( - is_variable and avg_io_size > self.VARIABLE_IO_THRESHOLD - ) or exceeds_user_max: - logger.debug( - "Large IO detected (variable > 64MB or > user max). Disabling background prefetching." - ) - prefetch_multiplier = 1 - elif streak < self.MIN_STREAKS_FOR_PREFETCHING: - prefetch_multiplier = 1 - else: - prefetch_multiplier = streak - self.MIN_STREAKS_FOR_PREFETCHING + 1 - - if self.queue.empty() or prefetch_multiplier == 1: - io_size = last_read_size - else: - io_size = avg_io_size - - prefetch_size = min(prefetch_multiplier * io_size, self.max_prefetch_size) - if self.consumer.offset + prefetch_size < self.consumer.target_offset: - prefetch_size = self.consumer.target_offset - self.consumer.offset - - if is_variable: - effective_prefetch_size = prefetch_size - else: - effective_prefetch_size = (prefetch_size // io_size) * io_size - if effective_prefetch_size == 0: - effective_prefetch_size = prefetch_size - - return prefetch_size, io_size, effective_prefetch_size - - async def _process_prefetch_cycle(self): - """Executes a single cycle of enqueuing fetch tasks.""" - prefetch_size, io_size, effective_prefetch_size = ( - self._calculate_prefetch_params() - ) - - logger.debug( - "Producer awake. Current offset: %d, User offset: %d, Prefetch size: %d", - self.current_offset, - self.consumer.offset, - prefetch_size, - ) - - while ( - not self.is_stopped - and (self.current_offset - self.consumer.offset) < prefetch_size - and self.current_offset < self.size - ): - user_offset = self.consumer.offset - space_remaining = self.size - self.current_offset - prefetch_space_available = prefetch_size - ( - self.current_offset - user_offset - ) - - if prefetch_size >= self.MIN_CHUNK_SIZE: - if prefetch_space_available >= self.MIN_CHUNK_SIZE: - actual_size = min( - max(self.MIN_CHUNK_SIZE, io_size), space_remaining - ) - else: - break - else: - actual_size = min(io_size, space_remaining) - - if prefetch_space_available < actual_size: - if ( - self.tracker.is_variable - or prefetch_space_available == prefetch_size - ): - actual_size = prefetch_space_available - else: - break - - streak = self.consumer.sequential_streak - if streak < self.MIN_STREAKS_FOR_PREFETCHING: - sfactor = self.concurrency - else: - sfactor = min( - self.concurrency, - max( - 1, - actual_size * self.concurrency // effective_prefetch_size, - ), - ) - - logger.debug( - "Spawning fetch task. Offset: %d, Size: %d, Split Factor: %d", - self.current_offset, - actual_size, - sfactor, - ) - - download_task = asyncio.create_task( - self.fetcher(self.current_offset, actual_size, split_factor=sfactor) - ) - self._active_tasks.add(download_task) - download_task.add_done_callback(self._active_tasks.discard) - - await self.queue.put(download_task) - self.current_offset += actual_size - - if self.current_offset >= self.size: - logger.debug("Producer reached EOF. Exiting background loop.") - self.is_stopped = True - - -class PrefetchConsumer: - """Consumes prefetched chunks from the queue and manages byte slicing. - - This class pulls data out of the shared queue and slices it into the - exact byte sizes requested by the user. It also manages the local block buffer. - """ - - def __init__( - self, - queue: asyncio.Queue, - wakeup_event: asyncio.Event, - tracker: RunningAverageTracker, - orchestrator: "BackgroundPrefetcher", - ): - """Initializes the consumer. - - Args: - queue (asyncio.Queue): The shared queue containing fetch tasks. - wakeup_event (asyncio.Event): Event used to wake the producer when more data is needed. - tracker (RunningAverageTracker): Tracker for history of read sizes. - orchestrator (BackgroundPrefetcher): The parent object managing the operation. - """ - logger.debug("Initializing PrefetchConsumer.") - self.queue = queue - self.wakeup_event = wakeup_event - self.tracker = tracker - self.orchestrator = weakref.proxy(orchestrator) - self.sequential_streak = 0 - self.offset = 0 - self.target_offset = 0 - self._current_block = b"" - self._current_block_idx = 0 - - def seek(self, new_offset: int): - """Clears the buffer and resets the internal offset for a hard seek. - - Args: - new_offset (int): The byte position the consumer is jumping to. - """ - logger.debug( - "Consumer executing hard seek to offset %d. Clearing internal buffer.", - new_offset, - ) - self.offset = new_offset - self.target_offset = new_offset - self.sequential_streak = 0 - self._current_block = b"" - self._current_block_idx = 0 - - def clear_buffer(self): - """Discards the local byte buffer. Useful during shutdown or resets.""" - logger.debug("Consumer local block buffer cleared.") - self._current_block = b"" - self._current_block_idx = 0 - - async def _advance(self, size: int, save_data: bool) -> list[bytes]: - """Internal method to advance the offset and optionally extract data. - - Handles queue exhaustion, producer wakeups, and streak tracking. - """ - if size <= 0: - return [] - - chunks = [] - processed = 0 - self.target_offset = self.offset + size - - while processed < size: - available = len(self._current_block) - self._current_block_idx - trigger_wakeup = False - - if not available: - is_producer_stopped = ( - self.orchestrator.producer is None - or self.orchestrator.producer.is_stopped - ) - if is_producer_stopped and self.queue.empty(): - logger.debug("Consumer reached EOF.") - break - - if self.queue.empty(): - logger.debug("Queue is empty. Waking up producer.") - self.wakeup_event.set() - - task = await self.queue.get() - - if isinstance(task, Exception): - logger.error("Consumer retrieved an exception: %s", task) - self.orchestrator.set_error(task) - raise task - - try: - block = await task - - self.sequential_streak += 1 - if ( - self.sequential_streak - >= PrefetchProducer.MIN_STREAKS_FOR_PREFETCHING - ): - exceeds_user_max = ( - self.orchestrator.max_prefetch_size is not None - and self.tracker.average - > self.orchestrator.max_prefetch_size - ) - is_massive_variable = ( - self.tracker.is_variable - and self.tracker.average - > PrefetchProducer.VARIABLE_IO_THRESHOLD - ) - - # Suppress proactive wakeups to prevent large CPU assembly - # on erratic large reads or exceeding max - if not (is_massive_variable or exceeds_user_max): - trigger_wakeup = True - else: - logger.debug( - "Suppressing proactive producer wakeup due to massive variable" - " workload or exceeding user max prefetch." - ) - - self._current_block = block - self._current_block_idx = 0 - available = len(self._current_block) - except asyncio.CancelledError: - raise - except Exception as e: - logger.error("Consumer caught an error: %s", e, exc_info=True) - self.orchestrator.set_error(e) - raise e - - if not self._current_block: - break - - needed = size - processed - take = min(needed, available) - - if save_data: - if take == len(self._current_block) and self._current_block_idx == 0: - chunk = self._current_block - else: - # Native Python slicing was GIL bound in my experiments. - chunk = await asyncio.to_thread( - _fast_slice, self._current_block, self._current_block_idx, take - ) - chunks.append(chunk) - - self._current_block_idx += take - processed += take - self.offset += take - if trigger_wakeup: - self.wakeup_event.set() - - return chunks - - async def consume(self, size: int) -> bytes: - """Pulls exactly 'size' bytes from the local block or the task queue. - - If the local block is exhausted, this will wait on the queue for the next - available chunk of data. - - Args: - size (int): The exact number of bytes to retrieve. - - Returns: - bytes: The requested bytes. This may be shorter than 'size' if EOF is reached. - - Raises: - Exception: Re-raises any exceptions encountered by the producer fetch tasks. - """ - if size <= 0: - return b"" - - chunks = await self._advance(size, save_data=True) - - if not chunks: - return b"" - - if len(chunks) == 1: - return chunks[0] - - return await asyncio.to_thread(b"".join, chunks) - - async def skip(self, size: int) -> None: - """Advances the consumer offset without allocating memory.""" - await self._advance(size, save_data=False) - - -class BackgroundPrefetcher: - """Orchestrator that manages reading behavior and coordinates background work. - - This acts as the main public interface for the file reader. It tracks the - user's reading history, routes seek operations, and links the producer's - network tasks with the consumer's data slicing logic. - """ - - producer = None - - def __init__( - self, fetcher, size: int, concurrency: int, max_prefetch_size=None, loop=None - ): - """Initializes the background prefetcher. - - Args: - fetcher (Callable): A coroutine of the form `f(start, end)` which gets bytes from the remote. - size (int): Total byte size of the file being read. - concurrency (int): Number of concurrent network requests to use for large chunks. - max_prefetch_size (int, optional): Maximum bytes to prefetch ahead of the current user offset. - loop (asyncio.AbstractEventLoop, optional): The event loop to attach the prefetcher to. - If executing synchronously, this should be the fsspec background loop. If executing - asynchronously (asynchronous=True), this should be None so it can automatically - inherit the user's currently running event loop. - - Raises: - ValueError: If max_prefetch_size is provided but is not a positive integer. - """ - logger.debug( - "Starting BackgroundPrefetcher. Size: %d, Concurrency: %d, Max Prefetch: %s", - size, - concurrency, - max_prefetch_size, - ) - self.size = size - self.concurrency = concurrency - self.max_prefetch_size = max_prefetch_size - - if max_prefetch_size is not None and max_prefetch_size <= 0: - logger.error("Invalid max_prefetch_size provided: %s", max_prefetch_size) - raise ValueError( - "max_prefetch_size should be a positive integer to use adaptive prefetching!" - ) - - self.loop = loop - self._error = None - self.is_stopped = False - self.user_offset = 0 - self.read_tracker = RunningAverageTracker(maxlen=10) - - self.queue = None - self.wakeup_event = None - self._async_lock = None - self.consumer = None - self.producer = None - - def _start(): - # Ensures all primitives bind directly to `self.loop` - self.queue = asyncio.Queue() - self.wakeup_event = asyncio.Event() - self._async_lock = asyncio.Lock() - - self.consumer = PrefetchConsumer( - queue=self.queue, - wakeup_event=self.wakeup_event, - tracker=self.read_tracker, - orchestrator=self, - ) - - self.producer = PrefetchProducer( - fetcher=fetcher, - size=self.size, - concurrency=self.concurrency, - queue=self.queue, - wakeup_event=self.wakeup_event, - consumer=self.consumer, - tracker=self.read_tracker, - orchestrator=self, - user_max_prefetch_size=max_prefetch_size, - ) - self.producer.start() - - try: - current_loop = asyncio.get_running_loop() - except RuntimeError: - current_loop = None - - if current_loop is self.loop and self.loop is not None: - # We are already safely running inside the fsspec background loop - _start() - elif self.loop is not None: - # We are on the main thread; schedule setup on the fsspec background loop - async def _start_wrapper(): - _start() - - fsspec.asyn.sync(self.loop, _start_wrapper) - elif current_loop is not None: - # asynchronous=True: use the user's active event loop - self.loop = current_loop - _start() - else: - # asynchronous=True but called completely outside of an async context - raise RuntimeError("No event loop found") - - logger.debug("BackgroundPrefetcher initialization complete.") - - def __enter__(self): - """Context manager entry point.""" - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - """Context manager exit point. Ensures the prefetcher is cleanly closed.""" - self.close() - - async def __aenter__(self): - """Async context manager entry point.""" - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - """Async context manager exit point. Ensures the prefetcher is cleanly closed.""" - await self.aclose() - - def set_error(self, e: Exception): - logger.error("Global error state set in BackgroundPrefetcher: %s", e) - self._error = e - - async def _restart_producer(self, new_offset: int): - logger.debug( - "Handling seek request. Restarting producer at offset: %d", new_offset - ) - self._error = None - await self.producer.restart(new_offset) - self.consumer.seek(new_offset) - self.read_tracker.clear() - - async def _async_fetch(self, start, end): - """Core internal async fetching logic, protected safely by the async lock.""" - async with self._async_lock: - try: - if self.is_stopped: - raise RuntimeError("The file instance has been closed.") - - logger.debug("Executing _async_fetch for range %d - %d.", start, end) - - # If the prefetcher is in error state, let's do a hard seek to start offset. - if self._error: - logger.info( - "Recovering from error state. Restarting producer at offset: %d", - start, - ) - self.user_offset = start - await self._restart_producer(start) - elif start != self.user_offset: - block_offset = ( - self.consumer.offset - self.consumer._current_block_idx - ) - if self.user_offset < start <= self.producer.current_offset: - logger.debug( - "Soft seek detected. Skipping ahead from %d to %d.", - self.user_offset, - start, - ) - skip_amount = start - self.user_offset - await self.consumer.skip(skip_amount) - self.user_offset = start - elif block_offset <= start < self.consumer.offset: - logger.debug( - "Local seek performed. User offset moved from %d to %d. " - "Adjusting buffer index from %d to %d.", - self.user_offset, - start, - self.consumer._current_block_idx, - start - block_offset, - ) - self.consumer._current_block_idx = start - block_offset - self.consumer.offset = start - self.consumer.target_offset = start - self.user_offset = start - else: - logger.debug( - "Hard seek detected. Moving user offset from %d to %d.", - self.user_offset, - start, - ) - self.user_offset = start - await self._restart_producer(start) - - requested_size = end - start - self.read_tracker.add(requested_size) - - chunk = await self.consumer.consume(requested_size) - self.user_offset += len(chunk) - - logger.debug("Completed _async_fetch. Returned %d bytes.", len(chunk)) - return chunk - except asyncio.CancelledError as e: - self._error = e - raise - except Exception as e: - logger.error( - "Exception raised during asynchronous fetch: %s", e, exc_info=True - ) - self._error = e - if self.producer and not self.producer.is_stopped: - await self.producer.stop() - raise - - async def _async_close(self): - """Asynchronous teardown logic protected by the async lock.""" - async with self._async_lock: - if self.is_stopped: - return - - self.is_stopped = True - logger.debug("Acquired async lock. Tearing down producer and buffers.") - - if self.producer: - await self.producer.stop() - - self.consumer.clear_buffer() - logger.debug("BackgroundPrefetcher closed successfully.") - - async def afetch(self, start: int | None, end: int | None) -> bytes: - """Asynchronous API counterpart to `_fetch`.""" - if start is None: - start = 0 - if end is None: - end = self.size - - end = min(end, self.size) - logger.debug( - "Asynchronous afetch called for bounds start=%s, end=%s.", start, end - ) - - if start >= self.size or start >= end: - return b"" - - if self.is_stopped: - logger.error( - "Cannot fetch data: BackgroundPrefetcher is stopped or closed." - ) - raise RuntimeError( - "The file instance has been closed. This can occur if a close operation " - "is executed concurrently while a read operation is still in progress." - ) - - return await self._async_fetch(start, end) - - def fetch(self, start: int | None, end: int | None) -> bytes: - """Synchronous API wrapper delegating to `afetch`.""" - # Delegates all boundaries, checking, and fetching to the async event loop perfectly - return fsspec.asyn.sync(self.loop, self.afetch, start, end) - - async def aclose(self): - """Safely shuts down the prefetcher from an asynchronous context.""" - await self._async_close() - - def close(self): - """Safely shuts down the prefetcher from a synchronous context.""" - fsspec.asyn.sync(self.loop, self._async_close) +__all__ = [ + "HAS_CPYTHON_API", + "BackgroundPrefetcher", + "PrefetchConsumer", + "PrefetchProducer", + "RunningAverageTracker", + "_fast_slice", +] diff --git a/gcsfs/tests/test_prefetcher.py b/gcsfs/tests/test_prefetcher.py index df9fd357a..4df9c448c 100644 --- a/gcsfs/tests/test_prefetcher.py +++ b/gcsfs/tests/test_prefetcher.py @@ -708,7 +708,7 @@ def test_local_seek_optimization(prefetcher_factory): assert fetcher.call_count > calls_after_next -@mock.patch("gcsfs.prefetcher.HAS_CPYTHON_API", False) +@mock.patch("fsspec.prefetch.HAS_CPYTHON_API", False) def test_fast_slice_pypy_fallback(): """ Tests that when HAS_CPYTHON_API is False (e.g., on PyPy), _fast_slice diff --git a/gcsfs/zonal_file.py b/gcsfs/zonal_file.py index 908f82b35..994e9b979 100644 --- a/gcsfs/zonal_file.py +++ b/gcsfs/zonal_file.py @@ -28,7 +28,7 @@ def __init__( mode="rb", block_size=DEFAULT_BLOCK_SIZE, autocommit=True, - cache_type="readahead_chunked", + cache_type="adaptive_readahead", cache_options=None, acl=None, consistency="md5", From 71d35a9af4180090f08bd3e14bfc9164a9f39ccd Mon Sep 17 00:00:00 2001 From: Prince Kumar Date: Fri, 7 Aug 2026 05:07:13 +0000 Subject: [PATCH 2/2] removing prefetcher with their test --- gcsfs/core.py | 11 +- gcsfs/prefetcher.py | 22 - gcsfs/tests/test_core.py | 2 +- gcsfs/tests/test_prefetcher.py | 726 --------------------------------- gcsfs/zonal_file.py | 2 +- 5 files changed, 9 insertions(+), 754 deletions(-) delete mode 100644 gcsfs/prefetcher.py delete mode 100644 gcsfs/tests/test_prefetcher.py diff --git a/gcsfs/core.py b/gcsfs/core.py index 308d98612..c2427b5c3 100644 --- a/gcsfs/core.py +++ b/gcsfs/core.py @@ -2049,7 +2049,7 @@ async def default_fetcher(start, size, split_factor=1): fetcher_fn = default_fetcher - from fsspec.prefetch import BackgroundPrefetcher + from fsspec.prefetcher import BackgroundPrefetcher prefetcher = BackgroundPrefetcher( fetcher=fetcher_fn, @@ -2313,7 +2313,7 @@ def __init__( mode="rb", block_size=DEFAULT_BLOCK_SIZE, autocommit=True, - cache_type="adaptive_readahead", + cache_type="adaptive", cache_options=None, acl=None, consistency="md5", @@ -2377,10 +2377,13 @@ def __init__( self.generation = _coalesce_generation(generation, path_generation) self.concurrency = kwargs.get("concurrency", DEFAULT_CONCURRENCY) - if "r" in mode and cache_type == "adaptive_readahead": + if cache_type is None: + cache_type = "adaptive" + + if "r" in mode and cache_type == "adaptive": if cache_type not in fsspec.core.caches: warnings.warn( - "fsspec adaptive_readahead cache is unavailable in this environment; " + "fsspec adaptive cache is unavailable in this environment; " "falling back to readahead" ) cache_type = "readahead" diff --git a/gcsfs/prefetcher.py b/gcsfs/prefetcher.py deleted file mode 100644 index 544e46eea..000000000 --- a/gcsfs/prefetcher.py +++ /dev/null @@ -1,22 +0,0 @@ -"""Compatibility re-exports for prefetch classes. - -Prefetch logic now lives in filesystem_spec/fsspec. -""" - -from fsspec.prefetch import ( # noqa: F401 - HAS_CPYTHON_API, - BackgroundPrefetcher, - PrefetchConsumer, - PrefetchProducer, - RunningAverageTracker, - _fast_slice, -) - -__all__ = [ - "HAS_CPYTHON_API", - "BackgroundPrefetcher", - "PrefetchConsumer", - "PrefetchProducer", - "RunningAverageTracker", - "_fast_slice", -] diff --git a/gcsfs/tests/test_core.py b/gcsfs/tests/test_core.py index ae42c8d66..a1330a78d 100644 --- a/gcsfs/tests/test_core.py +++ b/gcsfs/tests/test_core.py @@ -3401,7 +3401,7 @@ def test_get_file_concurrent_early_eof(gcs): return_value=-1, ), mock.patch( - "gcsfs.prefetcher.BackgroundPrefetcher.afetch", + "fsspec.prefetcher.BackgroundPrefetcher.afetch", new_callable=mock.AsyncMock, ) as mock_afetch, ): diff --git a/gcsfs/tests/test_prefetcher.py b/gcsfs/tests/test_prefetcher.py deleted file mode 100644 index 4df9c448c..000000000 --- a/gcsfs/tests/test_prefetcher.py +++ /dev/null @@ -1,726 +0,0 @@ -import asyncio -from unittest import mock - -import fsspec.asyn -import pytest - -from gcsfs.prefetcher import BackgroundPrefetcher, RunningAverageTracker, _fast_slice - - -@pytest.fixture -def prefetcher_factory(): - prefetchers = [] - - def _make_prefetcher(**kwargs): - if "loop" not in kwargs: - kwargs["loop"] = fsspec.asyn.get_loop() - - bp = BackgroundPrefetcher(**kwargs) - prefetchers.append(bp) - return bp - - yield _make_prefetcher - - for bp in prefetchers: - bp.is_stopped = False - bp.close() - - -class MockFetcher: - def __init__(self, data, fail_at_call=None, hang_at_call=None): - self.data = data - self.calls = [] - self.fail_at_call = fail_at_call - self.hang_at_call = hang_at_call - self.call_count = 0 - - async def __call__(self, start, size, split_factor=1): - self.call_count += 1 - self.calls.append({"start": start, "size": size, "split_factor": split_factor}) - - await asyncio.sleep(0.001) - - if self.hang_at_call is not None and self.call_count >= self.hang_at_call: - await asyncio.sleep(1000) - - if self.fail_at_call is not None and self.call_count >= self.fail_at_call: - raise OSError("Simulated Network Timeout") - - return self.data[start : start + size] - - -def test_fast_slice_direct(): - src = b"0123456789" - assert _fast_slice(src, 2, 4) == b"2345" - assert _fast_slice(src, 5, 0) == b"" - assert _fast_slice(src, 0, 10) == b"0123456789" - - -def test_running_average_tracker(): - tracker = RunningAverageTracker(maxlen=3) - assert tracker.average == 1024 * 1024 # Default 1MB fallback - - tracker.add(512) - tracker.add(512) - assert tracker.average == 512 - - tracker.add(2048) - assert tracker.average == 1024 # (512 + 512 + 2048) // 3 - - tracker.clear() - assert tracker.average == 1024 * 1024 - - -def test_max_prefetch_size_property(prefetcher_factory): - bp1 = prefetcher_factory(fetcher=MockFetcher(b""), size=10000, concurrency=4) - assert bp1.producer.max_prefetch_size == bp1.producer.MIN_PREFETCH_SIZE - - bp2 = prefetcher_factory(fetcher=MockFetcher(b""), size=1000000000, concurrency=4) - # Give it a history so it calculates 2x the io_size - bp2.read_tracker.add(100 * 1024 * 1024) - assert bp2.producer.max_prefetch_size == 200 * 1024 * 1024 - - -def test_sequential_read_spanning_blocks(prefetcher_factory): - data = b"A" * 100 + b"B" * 100 + b"C" * 100 - fetcher = MockFetcher(data) - bp = prefetcher_factory(fetcher=fetcher, size=300, concurrency=4) - bp.read_tracker.add(100) # Seed the adaptive tracker - - assert bp.fetch(0, 100) == b"A" * 100 - assert bp.fetch(100, 150) == b"B" * 50 - assert bp.consumer._current_block_idx == 50 - assert bp.fetch(150, 250) == b"B" * 50 + b"C" * 50 - assert bp.fetch(250, 300) == b"C" * 50 - assert bp.fetch(300, 310) == b"" - - -def test_fetch_default_args_and_out_of_bounds(prefetcher_factory): - fetcher = MockFetcher(b"12345") - bp = prefetcher_factory(fetcher=fetcher, size=5, concurrency=4) - - assert bp.fetch(None, None) == b"12345" - assert bp.fetch(None, 2) == b"12" - assert bp.fetch(5, 10) == b"" - assert bp.fetch(10, 20) == b"" - assert bp.fetch(2, 2) == b"" - assert bp.fetch(4, 2) == b"" - - -def test_seek_logic(prefetcher_factory): - data = b"0123456789" * 10 - fetcher = MockFetcher(data) - bp = prefetcher_factory(fetcher=fetcher, size=100, concurrency=4) - - assert bp.fetch(0, 10) == data[0:10] - assert bp.fetch(10, 20) == data[10:20] - assert bp.user_offset == 20 - assert bp.fetch(50, 60) == data[50:60] - assert bp.user_offset == 60 - assert bp.fetch(10, 20) == data[10:20] - assert bp.user_offset == 20 - - -def test_exception_placed_in_queue(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 100), size=100, concurrency=4) - - async def inject_error(): - await bp.queue.put(ValueError("Injected Producer Error")) - - fsspec.asyn.sync(bp.loop, inject_error) - - with pytest.raises(ValueError, match="Injected Producer Error"): - bp.fetch(0, 50) - - assert isinstance(bp._error, ValueError) - - -def test_producer_concurrency_streak_and_min_chunk(prefetcher_factory): - data = b"X" * 1000 - fetcher = MockFetcher(data) - - bp = prefetcher_factory(fetcher=fetcher, size=1000, concurrency=4) - bp.read_tracker.add(50) - - # Temporarily lower chunk limit for test - original_min_chunk = bp.producer.MIN_CHUNK_SIZE - bp.producer.MIN_CHUNK_SIZE = 10 - - # Do 6 reads to push the streak well past the MIN_STREAKS threshold - # Update these values as BackgroundPrefetcher constant changes. - target_streak = bp.producer.MIN_STREAKS_FOR_PREFETCHING + 3 - for i in range(target_streak): - bp.fetch(i * 50, (i + 1) * 50) - - fsspec.asyn.sync(bp.loop, asyncio.sleep, 0.1) - - split_factors = [call["split_factor"] for call in fetcher.calls] - assert split_factors[0] == 4 - assert max(split_factors) > 1 - assert max(split_factors) <= 4 - - bp.producer.MIN_CHUNK_SIZE = original_min_chunk - - -def test_producer_loop_space_constraints(prefetcher_factory): - data = b"Y" * 100 - fetcher = MockFetcher(data) - - bp = prefetcher_factory(fetcher=fetcher, size=100, concurrency=4) - bp.read_tracker.add(60) - - original_min_chunk = bp.producer.MIN_CHUNK_SIZE - bp.producer.MIN_CHUNK_SIZE = 200 - - assert bp.fetch(0, 10) == b"Y" * 10 - - fsspec.asyn.sync(bp.loop, asyncio.sleep, 0.1) - sizes = [call["size"] for call in fetcher.calls] - assert all(s <= 100 for s in sizes) - - bp.producer.MIN_CHUNK_SIZE = original_min_chunk - - -def test_producer_error_propagation_and_recovery(prefetcher_factory): - fetcher = MockFetcher(b"A" * 2000, fail_at_call=3) - bp = prefetcher_factory(fetcher=fetcher, size=2000, concurrency=4) - - for i in range(2): - bp.fetch(i * 100, (i + 1) * 100) - - # 3rd read triggers the network timeout - with pytest.raises(OSError, match="Simulated Network Timeout"): - bp.fetch(400, 500) - - # The prefetcher is now in an error state - assert isinstance(bp._error, OSError) - - # Disable the mock failure so it can succeed on retry - fetcher.fail_at_call = None - - # The next fetch should seamlessly recover, wiping the error and returning data - data = bp.fetch(400, 500) - assert data == b"A" * 100 - assert bp._error is None - - -def test_read_after_close(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 100), size=100, concurrency=4) - bp.close() - - assert bp.is_stopped is True - with pytest.raises(RuntimeError, match="The file instance has been closed"): - bp.fetch(0, 10) - - -def test_read_recovers_after_error(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 100), size=100, concurrency=4) - - # Simulate an error state from a previous failed read - bp._error = ValueError("Pre-existing error") - - # The new error-recovery logic allows a subsequent read to clear the error and succeed - assert bp.fetch(0, 10) == b"X" * 10 - assert bp._error is None - - -def test_empty_queue_when_stopped(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 500), size=500, concurrency=4) - bp.is_stopped = True - - with pytest.raises(RuntimeError, match="The file instance has been closed"): - bp.fetch(0, 100) - - -def test_cancel_all_tasks_cleans_queue_with_exceptions(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 100), size=100, concurrency=4) - - async def inject_task(): - async def dummy_exception_task(): - raise ValueError("Hidden error") - - task = asyncio.create_task(dummy_exception_task()) - await bp.queue.put(task) - await asyncio.sleep(0.05) - - fsspec.asyn.sync(bp.loop, inject_task) - bp.close() - assert bp.queue.empty() - - -def test_cleanup_cancels_active_tasks(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"Z" * 1000), size=1000, concurrency=4) - - async def inject_task(): - async def dummy_task(): - await asyncio.sleep(3) - - task = asyncio.create_task(dummy_task()) - bp.producer._active_tasks.add(task) - - fsspec.asyn.sync(bp.loop, inject_task) - - assert len(bp.producer._active_tasks) > 0 - assert bp.is_stopped is False - - bp.close() - - assert bp.is_stopped is True - assert len(bp.producer._active_tasks) == 0 - - -def test_read_task_cancellation(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 1000), size=1000, concurrency=4) - - async def inject_and_read(): - bp.is_stopped = True - while not bp.queue.empty(): - bp.queue.get_nowait() - - cancel_task = asyncio.create_task(asyncio.sleep(10)) - cancel_task.cancel() - await bp.queue.put(cancel_task) - - with pytest.raises(asyncio.CancelledError): - await bp.consumer.consume(10) - - fsspec.asyn.sync(bp.loop, inject_and_read) - - -def test_async_fetch_exception_trapping(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 100), size=100, concurrency=4) - - async def bad_consume(*args, **kwargs): - raise RuntimeError("Simulated async crash") - - bp.consumer.consume = bad_consume - - with pytest.raises(RuntimeError, match="Simulated async crash"): - bp.fetch(0, 10) - - # Orchestrator should capture the error internally and halt producer processing correctly - assert isinstance(bp._error, RuntimeError) - assert bp.producer.is_stopped is True - - -def test_read_past_eof_internal(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 50), size=50, concurrency=4) - bp.user_offset = 50 - res = bp.fetch(50, 60) - assert res == b"" - - -def test_fetch_with_exact_block_matches(prefetcher_factory): - data = b"X" * 100 - bp = prefetcher_factory(fetcher=MockFetcher(data), size=100, concurrency=4) - bp.read_tracker.add(50) - - assert bp.fetch(0, 50) == b"X" * 50 - assert bp.consumer._current_block_idx == 50 - assert bp.fetch(50, 100) == b"X" * 50 - - -def test_queue_empty_race_condition(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 100), size=100, concurrency=4) - - async def inject(): - bp.queue.put_nowait(asyncio.create_task(asyncio.sleep(0))) - with mock.patch.object(bp.queue, "get_nowait", side_effect=asyncio.QueueEmpty): - await bp.producer.stop() - - fsspec.asyn.sync(bp.loop, inject) - - -def test_producer_space_remaining_break(prefetcher_factory): - bp = prefetcher_factory( - fetcher=MockFetcher(b"X" * 1000), - size=1000, - concurrency=4, - max_prefetch_size=150, - ) - bp.fetch(0, 10) - fsspec.asyn.sync(bp.loop, asyncio.sleep, 0.1) - - -def test_producer_min_chunk_logic(prefetcher_factory): - bp1 = prefetcher_factory( - fetcher=MockFetcher(b"X" * 1000), - size=1000, - concurrency=4, - max_prefetch_size=300, - ) - bp1.producer.MIN_CHUNK_SIZE = 100 - - fsspec.asyn.sync(bp1.loop, asyncio.sleep, 0.1) - - bp2 = prefetcher_factory( - fetcher=MockFetcher(b"X" * 1000), - size=1000, - concurrency=4, - max_prefetch_size=150, - ) - bp2.producer.MIN_CHUNK_SIZE = 100 - fsspec.asyn.sync(bp2.loop, asyncio.sleep, 0.1) - - -def test_producer_loop_exception(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"A" * 100), size=100, concurrency=4) - error_object = ValueError("Producer crash") - - with mock.patch( - "gcsfs.prefetcher.RunningAverageTracker.average", new_callable=mock.PropertyMock - ) as mocked_avg: - mocked_avg.side_effect = error_object - with pytest.raises(ValueError, match="Producer crash"): - bp.fetch(0, 10) - - assert bp.is_stopped is False - assert bp._error == error_object - - -def test_seek_same_offset(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b""), size=100, concurrency=4) - bp.fetch(0, 10) - - -def test_read_history_maxlen(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 2000), size=2000, concurrency=4) - for i in range(12): - bp.fetch(i * 10, (i + 1) * 10) - assert len(bp.read_tracker._history) == 10 - - -def test_fast_slice_branch(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 200), size=200, concurrency=4) - assert bp.fetch(0, 10) == b"X" * 10 - assert bp.fetch(10, 20) == b"X" * 10 - - -def test_async_fetch_not_block_break(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b""), size=100, concurrency=4) - - async def fake_consume(size): - return b"" - - bp.consumer.consume = fake_consume - bp.user_offset = 0 - - res = bp.fetch(0, 50) - assert res == b"" - - -def test_fetch_stopped_before_execution(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 100), size=100, concurrency=4) - bp.is_stopped = True - bp._error = None - - with pytest.raises(RuntimeError, match="The file instance has been closed"): - bp.fetch(0, 10) - - -def test_async_fetch_zero_copy_remainder(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X"), size=100, concurrency=4) - bp.consumer._current_block = b"ABCDE" - bp.consumer._current_block_idx = 0 - bp.user_offset = 0 - res = bp.fetch(0, 5) - assert res == b"ABCDE" - assert bp.consumer._current_block_idx == 5 - - -def test_read_runtime_error_on_stopped_empty(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X"), size=100, concurrency=4) - bp.is_stopped = True - bp.producer.is_stopped = True - - while not bp.queue.empty(): - bp.queue.get_nowait() - - res = fsspec.asyn.sync(bp.loop, bp.consumer.consume, 10) - assert res == b"" - - -def test_init_invalid_max_prefetch_size(): - with pytest.raises( - ValueError, - match=r"max_prefetch_size should be a positive integer", - ): - BackgroundPrefetcher( - fetcher=MockFetcher(b""), size=1000, concurrency=4, max_prefetch_size=0 - ) - - -def test_init_valid_max_prefetch_size_edge_case(prefetcher_factory): - bp = prefetcher_factory( - fetcher=MockFetcher(b""), size=1000, concurrency=4, max_prefetch_size=100 - ) - assert bp.producer._user_max_prefetch_size == 100 - - -def test_consumer_zero_size_checks(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 100), size=100, concurrency=4) - - # 1. Test consume size <= 0 - res_consume_zero = fsspec.asyn.sync(bp.loop, bp.consumer.consume, 0) - assert res_consume_zero == b"" - res_consume_neg = fsspec.asyn.sync(bp.loop, bp.consumer.consume, -5) - assert res_consume_neg == b"" - - # 2. Test _advance size <= 0 directly - # (consume catches it early, so we call _advance directly to hit its internal check) - res_advance_zero = fsspec.asyn.sync( - bp.loop, bp.consumer._advance, 0, save_data=True - ) - assert res_advance_zero == [] - res_advance_neg = fsspec.asyn.sync( - bp.loop, bp.consumer._advance, -10, save_data=False - ) - assert res_advance_neg == [] - - -def test_producer_min_chunk_inner_break(prefetcher_factory): - fetcher = MockFetcher(b"X" * 1000) - bp = prefetcher_factory( - fetcher=fetcher, size=1000, concurrency=4, max_prefetch_size=400 - ) - - bp.read_tracker.add(100) - - original_min_chunk = bp.producer.MIN_CHUNK_SIZE - bp.producer.MIN_CHUNK_SIZE = 200 - - async def trigger_loop(): - bp.producer.current_offset = 250 - bp.consumer.offset = 0 - bp.consumer.target_offset = 0 - # streak=6 makes prefetch_multiplier = 4 (6 - 3 + 1) - # prefetch_size = 4 * 100 = 400 - bp.consumer.sequential_streak = 6 - bp.wakeup_event.set() - await asyncio.sleep(0.05) - - fsspec.asyn.sync(bp.loop, trigger_loop) - - assert fetcher.call_count == 0 - - bp.producer.MIN_CHUNK_SIZE = original_min_chunk - - -def test_producer_loop_break_on_stopped_after_wakeup(prefetcher_factory): - fetcher = MockFetcher(b"X" * 1000) - bp = prefetcher_factory(fetcher=fetcher, size=1000, concurrency=4) - - async def trigger_stop_and_wake(): - bp.producer.is_stopped = True - bp.wakeup_event.set() - await asyncio.sleep(0.05) - - fsspec.asyn.sync(bp.loop, trigger_stop_and_wake) - - # Verify the producer gracefully exited without doing work - assert fetcher.call_count == 0 - - -def test_massive_read_disables_proactive_prefetching(prefetcher_factory): - fetcher = MockFetcher(b"X" * 1000) - - # max_prefetch_size = 40 - bp = prefetcher_factory( - fetcher=fetcher, size=1000, concurrency=4, max_prefetch_size=40 - ) - - # Do enough reads to build a sequential streak and trigger large averages - # Reading 60 bytes at a time. Average = 60. Threshold = 50. - for i in range(4): - bp.fetch(i * 60, (i + 1) * 60) - - fsspec.asyn.sync(bp.loop, asyncio.sleep, 0.1) - - # Because average (60) > threshold (40), prefetch_multiplier is pinned to 1. - # The producer should only fetch what the user specifically read (4 * 60 = 240) - # and should NOT have pre-fetched any additional data ahead into the queue. - assert bp.producer.current_offset == 240 - - -def test_normal_read_allows_proactive_prefetching(prefetcher_factory): - fetcher = MockFetcher(b"X" * 1000) - - # max_prefetch_size = 200 makes dynamic threshold = 100 - bp = prefetcher_factory( - fetcher=fetcher, size=1000, concurrency=4, max_prefetch_size=200 - ) - - # Reading 60 bytes at a time. Average = 60. Threshold = 100. - for i in range(4): - bp.fetch(i * 60, (i + 1) * 60) - - fsspec.asyn.sync(bp.loop, asyncio.sleep, 0.1) - - # Because average (60) <= threshold (100), the producer allows prefetching. - # It calculates a normal prefetch_multiplier > 1 and pre-fetches data ahead. - assert bp.producer.current_offset > 240 - - -def test_target_offset_expands_prefetch(prefetcher_factory): - fetcher = MockFetcher(b"X" * 1000) - bp = prefetcher_factory(fetcher=fetcher, size=1000, concurrency=4) - - # Seed tracker to keep the default `max_prefetch_size` calculation small - bp.read_tracker.add(10) - - # The consumer requests a massive chunk (500 bytes), far exceeding normal prefetch windows - bp.fetch(0, 500) - - fsspec.asyn.sync(bp.loop, asyncio.sleep, 0.1) - - # The new target_offset logic should explicitly tell the producer to expand its - # boundary to cover the requested 500 bytes, overriding the tiny multiplier logic. - assert bp.consumer.target_offset == 500 - assert bp.producer.current_offset >= 500 - - -def test_producer_min_chunk_inner_empty_queue_shrink(prefetcher_factory): - fetcher = MockFetcher(b"X" * 1000) - bp = prefetcher_factory( - fetcher=fetcher, size=1000, concurrency=4, max_prefetch_size=400 - ) - - bp.read_tracker.add(100) - - original_min_chunk = bp.producer.MIN_CHUNK_SIZE - bp.producer.MIN_CHUNK_SIZE = 200 - - async def trigger_loop(): - # Setup conditions where the queue is empty and the user is waiting - # This makes prefetch_space_available exactly equal to prefetch_size - bp.producer.current_offset = 0 - bp.consumer.offset = 0 - bp.consumer.target_offset = 0 - bp.consumer.sequential_streak = 6 - bp.wakeup_event.set() - await asyncio.sleep(0.05) - - fsspec.asyn.sync(bp.loop, trigger_loop) - - # Because space_available == prefetch_size, it triggers the shrink condition - # instead of breaking, ensuring the blocked consumer gets its data. - assert fetcher.call_count > 0 - - bp.producer.MIN_CHUNK_SIZE = original_min_chunk - - -def test_async_context_manager_and_afetch(prefetcher_factory): - bp = prefetcher_factory(fetcher=MockFetcher(b"X" * 100), size=100, concurrency=4) - - async def run_async(): - async with bp as ctx: - res = await ctx.afetch(0, 10) - assert res == b"X" * 10 - # Test default bounds -> (0, 100). Validates a backwards hard seek internally. - res_all = await ctx.afetch(None, None) - assert len(res_all) == 100 - - assert bp.is_stopped is True - assert bp.consumer._current_block == b"" # Buffer cleanly cleared on async exit - - fsspec.asyn.sync(bp.loop, run_async) - - -def test_init_with_explicit_loop(prefetcher_factory): - """Verify that passing an explicit loop assigns it correctly.""" - loop = fsspec.asyn.get_loop() - bp = prefetcher_factory( - fetcher=MockFetcher(b"X"), size=100, concurrency=1, loop=loop - ) - assert bp.loop is loop - - -@pytest.mark.asyncio -async def test_init_with_running_loop(): - """Verify asynchronous=True behavior where it inherits the user's running loop.""" - current = asyncio.get_running_loop() - # explicitly passing loop=None simulates asynchronous=True behavior - bp = BackgroundPrefetcher( - fetcher=MockFetcher(b"X"), size=100, concurrency=1, loop=None - ) - assert bp.loop is current - await bp.aclose() - - -def test_init_within_fsspec_loop(prefetcher_factory): - """Verify the edge case where the prefetcher is initialized while already running inside the target loop.""" - loop = fsspec.asyn.get_loop() - - async def init_inside_loop(): - bp = BackgroundPrefetcher( - fetcher=MockFetcher(b"X"), size=100, concurrency=1, loop=loop - ) - assert bp.loop is loop - await bp.aclose() - - # Submit the initialization directly onto the background thread - fsspec.asyn.sync(loop, init_inside_loop) - - -def test_init_no_loop_raises_error(): - """Verify synchronous execution strictly fails if no explicit loop or active loop is found.""" - with pytest.raises(RuntimeError, match="No event loop found"): - BackgroundPrefetcher( - fetcher=MockFetcher(b"X"), size=100, concurrency=1, loop=None - ) - - -def test_local_seek_optimization(prefetcher_factory): - data = b"0123456789" * 10 - fetcher = MockFetcher(data) - bp = prefetcher_factory(fetcher=fetcher, size=100, concurrency=4) - - # First fetch: read first 50 bytes (0-50) - # This will trigger a fetch. The block size fetched will be 50 bytes (covering [0, 50]). - assert bp.fetch(0, 50) == data[0:50] - assert len(bp.consumer._current_block) == 50 - initial_fetch_calls = fetcher.call_count - assert initial_fetch_calls > 0 - - # 1. Perform a backward seek *within* the currently buffered block [0, 50]. - # Seek back to 5 and read 10 bytes (5 to 15). - # This should be a zero-cost local seek and NOT increment the fetcher call count. - assert bp.fetch(5, 15) == data[5:15] - assert fetcher.call_count == initial_fetch_calls - assert bp.user_offset == 15 - - # 2. Perform a forward seek *within* the currently buffered block [0, 50]. - # Seek forward to 30 and read 10 bytes (30 to 40). - # This should also be resolved locally and NOT increment the fetcher call count. - assert bp.fetch(30, 40) == data[30:40] - assert fetcher.call_count == initial_fetch_calls - assert bp.user_offset == 40 - - # 3. Read further to trigger a hard seek and load the next block [60, 100]. - assert bp.fetch(60, 80) == data[60:80] - calls_after_next = fetcher.call_count - assert calls_after_next > initial_fetch_calls - - # The currently buffered block in the consumer is now [60, 100]. - # Seek backward to 10 (which is outside [60, 100]). - # This should trigger a hard seek and increment the fetcher call count. - assert bp.fetch(10, 20) == data[10:20] - assert fetcher.call_count > calls_after_next - - -@mock.patch("fsspec.prefetch.HAS_CPYTHON_API", False) -def test_fast_slice_pypy_fallback(): - """ - Tests that when HAS_CPYTHON_API is False (e.g., on PyPy), _fast_slice - correctly falls back to standard Python slicing and respects boundaries. - """ - src = b"0123456789_pypy_fallback_test" - - # 1. Verify standard extraction works perfectly - assert _fast_slice(src, 11, 13) == b"pypy_fallback" - - # 2. Verify zero-length read - assert _fast_slice(src, 5, 0) == b"" - - # 3. Verify exact bounds - assert _fast_slice(src, 0, len(src)) == src diff --git a/gcsfs/zonal_file.py b/gcsfs/zonal_file.py index 994e9b979..4d08aaefc 100644 --- a/gcsfs/zonal_file.py +++ b/gcsfs/zonal_file.py @@ -28,7 +28,7 @@ def __init__( mode="rb", block_size=DEFAULT_BLOCK_SIZE, autocommit=True, - cache_type="adaptive_readahead", + cache_type="adaptive", cache_options=None, acl=None, consistency="md5",