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.Hm8K6x.py     2026-08-07 16:50:52.006827951 +0000
+++ /tmp/prefetcher.fsspec.z2jMge.py    2026-08-07 16:50:52.347827859 +0000
@@ -4,15 +4,23 @@
 import weakref
 from collections import deque
 
-import fsspec.asyn
+from . import asyn as fsspec_asyn
 
 logger = logging.getLogger(__name__)
 
-from gcsfs.zb_hns_utils import (
-    HAS_CPYTHON_API,
-    PyBytes_AsString,
-    PyBytes_FromStringAndSize,
-)
+try:
+    PyBytes_FromStringAndSize = ctypes.pythonapi.PyBytes_FromStringAndSize
+    PyBytes_FromStringAndSize.argtypes = (ctypes.c_void_p, ctypes.c_ssize_t)
+    PyBytes_FromStringAndSize.restype = ctypes.py_object
+
+    PyBytes_AsString = ctypes.pythonapi.PyBytes_AsString
+    PyBytes_AsString.argtypes = (ctypes.py_object,)
+    PyBytes_AsString.restype = ctypes.c_void_p
+    HAS_CPYTHON_API = True
+except Exception:
+    PyBytes_FromStringAndSize = None
+    PyBytes_AsString = None
+    HAS_CPYTHON_API = False
 
 
 # Please refer to following discussion to understand why this is required at this point
@@ -229,9 +237,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 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.
@@ -293,10 +299,9 @@
         except asyncio.CancelledError:
             logger.debug("PrefetchProducer loop was cancelled.")
         except Exception as e:
-            logger.error(
+            logger.exception(
                 "PrefetchProducer loop encountered an unexpected error: %s",
                 e,
-                exc_info=True,
             )
             self.is_stopped = True
             self.orchestrator.set_error(e)
@@ -552,7 +557,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: %s", e)
                     self.orchestrator.set_error(e)
                     raise e
 
@@ -708,7 +713,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
@@ -812,9 +817,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: %s", e)
                 self._error = e
                 if self.producer and not self.producer.is_stopped:
                     await self.producer.stop()
@@ -864,7 +867,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."""
@@ -872,4 +875,4 @@
 
     def close(self):
         """Safely shuts down the prefetcher from a synchronous context."""
-        fsspec.asyn.sync(self.loop, self._async_close)
+        fsspec_asyn.sync(self.loop, self._async_close)
--- /tmp/test_prefetcher.gcsfs.p13ZhM.py        2026-08-07 16:50:52.580827796 +0000
+++ /tmp/test_prefetcher.fsspec.ULYtQJ.py       2026-08-07 16:50:52.907827706 +0000
@@ -1,10 +1,10 @@
 import asyncio
 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
@@ -368,7 +368,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"):
@@ -708,7 +709,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

[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?