feat(AdaptiveReadaheadCache): introducing a new cache-type adaptive

[raj-prince]

Descriprtion
This PR introduces the adaptive prefetching functionality as a first-class fsspec cache implementation, with prefetch logic and tests consolidated under fsspec.

To save the review time - providing the diff of already reviewed moved code across the repo.

#!/bin/sh

prefetcher_gcsfs="https://raw.githubusercontent.com/fsspec/gcsfs/refs/heads/main/gcsfs/prefetcher.py"
prefetcher_fsspec="https://raw.githubusercontent.com/raj-prince/filesystem_spec/refs/heads/adaptive-readahead/fsspec/prefetcher.py"
prefetcher_test_gcsfs="https://raw.githubusercontent.com/fsspec/gcsfs/refs/heads/main/gcsfs/tests/test_prefetcher.py"
prefetcher_test_fsspec="https://raw.githubusercontent.com/raj-prince/filesystem_spec/refs/heads/adaptive-readahead/fsspec/tests/test_prefetcher.py"

tmp1=$(mktemp "${TMPDIR:-/tmp}/prefetcher.gcsfs.XXXXXX.py")
tmp2=$(mktemp "${TMPDIR:-/tmp}/prefetcher.fsspec.XXXXXX.py")
tmp3=$(mktemp "${TMPDIR:-/tmp}/test_prefetcher.gcsfs.XXXXXX.py")
tmp4=$(mktemp "${TMPDIR:-/tmp}/test_prefetcher.fsspec.XXXXXX.py")

cleanup() {
	rm -f "$tmp1" "$tmp2" "$tmp3" "$tmp4"
}
trap cleanup EXIT INT TERM

curl -s "$prefetcher_gcsfs" > "$tmp1"
curl -s "$prefetcher_fsspec" > "$tmp2"
diff -u "$tmp1" "$tmp2"

curl -s "$prefetcher_test_gcsfs" > "$tmp3"
curl -s "$prefetcher_test_fsspec" > "$tmp4"
diff -u "$tmp3" "$tmp4"

Diff:

--- /tmp/prefetcher.gcsfs.5KiKLH.py     2026-09-13 00:00:46.052236250 +0000
+++ /tmp/prefetcher.fsspec.e88GLQ.py    2026-09-13 00:00:46.243236208 +0000
@@ -1,41 +1,15 @@
 import asyncio
-import ctypes
 import logging
 import weakref
 from collections import deque
 
-import fsspec.asyn
-
-from gcsfs.zb_hns_utils import (
-    HAS_CPYTHON_API,
-    PyBytes_AsString,
-    PyBytes_FromStringAndSize,
-    sync_teardown,
-)
+from . import asyn as fsspec_asyn
+from .asyn import sync_teardown
+from .utils import HAS_CPYTHON_API, _fast_slice
 
 logger = logging.getLogger(__name__)
 
 
-# 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.
 
@@ -230,9 +204,7 @@
             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)
+        tasks_to_wait.extend(task for task in list(self._active_tasks) if not task.done())
 
         # We do not cancel the network task, instead we wait on them.
         # This is intentionally done to avoid MRD stream disruption.
@@ -294,10 +266,8 @@
         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,
+            logger.exception(
+                "PrefetchProducer loop encountered an unexpected error."
             )
             self.is_stopped = True
             self.orchestrator.set_error(e)
@@ -553,7 +523,7 @@
                 except asyncio.CancelledError:
                     raise
                 except Exception as e:
-                    logger.error("Consumer caught an error: %s", e, exc_info=True)
+                    logger.exception("Consumer caught an error.")
                     self.orchestrator.set_error(e)
                     raise e
 
@@ -709,7 +679,7 @@
             async def _start_wrapper():
                 _start()
 
-            fsspec.asyn.sync(self.loop, _start_wrapper)
+            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
@@ -813,9 +783,7 @@
                 self._error = e
                 raise
             except Exception as e:
-                logger.error(
-                    "Exception raised during asynchronous fetch: %s", e, exc_info=True
-                )
+                logger.exception("Exception raised during asynchronous fetch.")
                 self._error = e
                 if self.producer and not self.producer.is_stopped:
                     await self.producer.stop()
@@ -866,7 +834,7 @@
     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)
+        return fsspec_asyn.sync(self.loop, self.afetch, start, end)
 
     async def aclose(self):
         """Safely shuts down the prefetcher from an asynchronous context."""
@@ -893,7 +861,7 @@
                 timeout=timeout,
                 description="BackgroundPrefetcher teardown",
             )
-        except fsspec.asyn.FSTimeoutError:
+        except fsspec_asyn.FSTimeoutError:
             logger.warning(
                 "BackgroundPrefetcher teardown did not complete within %ss; "
                 "it will keep running in the background.",
--- /tmp/test_prefetcher.gcsfs.pTvwWW.py        2026-09-13 00:00:46.368236184 +0000
+++ /tmp/test_prefetcher.fsspec.yvn5qu.py       2026-09-13 00:00:46.626236127 +0000
@@ -4,10 +4,10 @@
 import threading
 from unittest import mock
 
-import fsspec.asyn
 import pytest
 
-from gcsfs.prefetcher import BackgroundPrefetcher, RunningAverageTracker, _fast_slice
+import fsspec.asyn
+from fsspec.prefetcher import BackgroundPrefetcher, RunningAverageTracker, _fast_slice
 
 
 @pytest.fixture
@@ -371,7 +371,8 @@
     error_object = ValueError("Producer crash")
 
     with mock.patch(
-        "gcsfs.prefetcher.RunningAverageTracker.average", new_callable=mock.PropertyMock
+        "fsspec.prefetcher.RunningAverageTracker.average",
+        new_callable=mock.PropertyMock,
     ) as mocked_avg:
         mocked_avg.side_effect = error_object
         with pytest.raises(ValueError, match="Producer crash"):
@@ -433,6 +434,7 @@
 
 def test_read_runtime_error_on_stopped_empty(prefetcher_factory):
     bp = prefetcher_factory(fetcher=MockFetcher(b"X"), size=100, concurrency=4)
+    fsspec.asyn.sync(bp.loop, bp.producer.stop)
     bp.is_stopped = True
     bp.producer.is_stopped = True
 
@@ -711,7 +713,7 @@
     assert fetcher.call_count > calls_after_next
 
 
-@mock.patch("gcsfs.prefetcher.HAS_CPYTHON_API", False)
+@mock.patch("fsspec.prefetcher.HAS_CPYTHON_API", False)
 def test_fast_slice_pypy_fallback():
     """
     Tests that when HAS_CPYTHON_API is False (e.g., on PyPy), _fast_slice
@@ -778,13 +780,15 @@
         bp = BackgroundPrefetcher(
             fetcher=MockFetcher(b"X" * 10), size=10, concurrency=1, loop=loop
         )
+        real_stop = bp.producer.stop
 
         async def failing_close():
+            await real_stop()
             raise RuntimeError("reentrant close failed")
 
         bp._async_close = failing_close
 
-        with caplog.at_level(logging.ERROR, logger="gcsfs"):
+        with caplog.at_level(logging.ERROR, logger="fsspec.asyn"):
             bp.close()
             await asyncio.sleep(0.05)
 
@@ -903,7 +907,7 @@
 
     bp.producer.stop = boom
 
-    with caplog.at_level(logging.WARNING, logger="gcsfs.prefetcher"):
+    with caplog.at_level(logging.WARNING, logger="fsspec.prefetcher"):
         bp.close()
 
     assert bp.is_stopped is True
@@ -1014,3 +1018,4 @@
         assert bp2.is_stopped
 
     fsspec.asyn.sync(loop, run_aclose_twice)
+

[martindurant]

Great to see this! I hope someone can give it a go on real workloads.

[raj-prince]

Hi Martin, shortly I'll make it ready for review, aligned internally to go with prefetcher as a new cache-type not as engine.

[raj-prince]

@martindurant, CI/CD requires approval, could you please help me with that?

[ankitaluthra1]

should we add specific exception here ?

[raj-prince]

I don't see much value of keeping specific exception other than best practices to avoid Exception in general.

We need to catch multiple specific exceptions AttributeError, TypeError, OSError and not sure if that will be sufficient. So, I would refrain doing this until any significant downside of using Exception.