feat(core.utils): add FileStreamProgramCache backend

Persistent program cache backed by a directory of entry files, one per
key hash. Writes stage into a tmp/ subdirectory and promote via os.replace
so concurrent readers never observe a torn file. Corrupt entries are
treated as cache misses and pruned. The max_size_bytes cap is enforced
opportunistically on writes by oldest mtime; this is deliberately
best-effort for multi-process use. Use SQLiteProgramCache for strict
LRU semantics within a single process.

Part of issue #178.

Author: cpcloud

cuda_core/cuda/core/utils/__init__.py

@@ -7,6 +7,7 @@
     args_viewable_as_strided_memory,  # noqa: F401
 )
 from cuda.core.utils._program_cache import (
+    FileStreamProgramCache,  # noqa: F401
     ProgramCacheResource,  # noqa: F401
     SQLiteProgramCache,  # noqa: F401
     make_program_cache_key,  # noqa: F401

cuda_core/cuda/core/utils/_program_cache.py

@@ -22,6 +22,7 @@
 import os
 import pickle
 import sqlite3
+import tempfile
 import time
 from pathlib import Path
 from typing import Hashable, Iterable, Optional, Sequence
@@ -35,6 +36,7 @@
 )
 
 __all__ = [
+    "FileStreamProgramCache",
     "ProgramCacheResource",
     "SQLiteProgramCache",
     "make_program_cache_key",
@@ -431,3 +433,187 @@ def _enforce_size_cap(self) -> None:
                 return
             conn.execute("DELETE FROM entries WHERE key = ?", (k,))
             total -= sz
+
+
+# ---------------------------------------------------------------------------
+# FileStream backend
+# ---------------------------------------------------------------------------
+
+
+_FILESTREAM_SCHEMA_VERSION = 1
+_ENTRIES_SUBDIR = "entries"
+_TMP_SUBDIR = "tmp"
+_SCHEMA_FILE = "SCHEMA_VERSION"
+
+
+class FileStreamProgramCache(ProgramCacheResource):
+    """Persistent program cache backed by a directory of atomic files.
+
+    Designed for multi-process use: writes stage a temporary file and then
+    :func:`os.replace` it into place, so concurrent readers never observe a
+    partially-written entry. There is no cross-process LRU tracking; size
+    enforcement is best-effort by file mtime.
+
+    Parameters
+    ----------
+    path:
+        Directory that owns the cache. Created if missing.
+    max_size_bytes:
+        Optional soft cap on total on-disk size. Enforced opportunistically
+        on writes; concurrent writers may briefly exceed it.
+    """
+
+    def __init__(
+        self,
+        path: str | os.PathLike,
+        *,
+        max_size_bytes: Optional[int] = None,
+    ) -> None:
+        if max_size_bytes is not None and max_size_bytes < 0:
+            raise ValueError("max_size_bytes must be non-negative or None")
+        self._root = Path(path)
+        self._entries = self._root / _ENTRIES_SUBDIR
+        self._tmp = self._root / _TMP_SUBDIR
+        self._schema_path = self._root / _SCHEMA_FILE
+        self._max_size_bytes = max_size_bytes
+        self._root.mkdir(parents=True, exist_ok=True)
+        self._entries.mkdir(exist_ok=True)
+        self._tmp.mkdir(exist_ok=True)
+        if not self._schema_path.exists():
+            self._schema_path.write_text(str(_FILESTREAM_SCHEMA_VERSION))
+
+    # -- key-to-path helpers -------------------------------------------------
+
+    def _path_for_key(self, key: object) -> Path:
+        k = _as_key_bytes(key)
+        hex_ = k.hex() if k else "empty"
+        if len(hex_) < 3:
+            hex_ = hex_.rjust(3, "0")
+        return self._entries / hex_[:2] / hex_[2:]
+
+    # -- mapping API ---------------------------------------------------------
+
+    def __contains__(self, key: object) -> bool:
+        return self._path_for_key(key).exists()
+
+    def __getitem__(self, key: object) -> ObjectCode:
+        path = self._path_for_key(key)
+        try:
+            data = path.read_bytes()
+        except FileNotFoundError:
+            raise KeyError(key)
+        k = _as_key_bytes(key)
+        try:
+            record = pickle.loads(data)
+            schema, stored_key, payload, _created_at = record
+            if schema != _FILESTREAM_SCHEMA_VERSION:
+                raise ValueError(f"unknown schema {schema}")
+            if stored_key != k:
+                raise ValueError("key mismatch")
+            value = pickle.loads(payload)
+        except Exception:
+            # Corrupt entry -- delete and treat as a miss.
+            try:
+                path.unlink()
+            except FileNotFoundError:
+                pass
+            raise KeyError(key)
+        if not isinstance(value, ObjectCode):
+            try:
+                path.unlink()
+            except FileNotFoundError:
+                pass
+            raise KeyError(key)
+        return value
+
+    def __setitem__(self, key: object, value: object) -> None:
+        obj = _require_object_code(value)
+        k = _as_key_bytes(key)
+        payload = pickle.dumps(obj, protocol=_PICKLE_PROTOCOL)
+        record = pickle.dumps(
+            (_FILESTREAM_SCHEMA_VERSION, k, payload, time.time()),
+            protocol=_PICKLE_PROTOCOL,
+        )
+
+        target = self._path_for_key(key)
+        target.parent.mkdir(parents=True, exist_ok=True)
+
+        fd, tmp_name = tempfile.mkstemp(prefix="entry-", dir=self._tmp)
+        tmp_path = Path(tmp_name)
+        try:
+            with os.fdopen(fd, "wb") as fh:
+                fh.write(record)
+                fh.flush()
+                os.fsync(fh.fileno())
+            os.replace(tmp_path, target)
+        except BaseException:
+            try:
+                tmp_path.unlink()
+            except FileNotFoundError:
+                pass
+            raise
+        self._enforce_size_cap()
+
+    def __delitem__(self, key: object) -> None:
+        path = self._path_for_key(key)
+        try:
+            path.unlink()
+        except FileNotFoundError:
+            raise KeyError(key)
+
+    def __len__(self) -> int:
+        count = 0
+        for _ in self._iter_entry_paths():
+            count += 1
+        return count
+
+    def clear(self) -> None:
+        for path in list(self._iter_entry_paths()):
+            try:
+                path.unlink()
+            except FileNotFoundError:
+                pass
+        # Remove empty subdirs (best-effort; concurrent writers may re-create).
+        if self._entries.exists():
+            for sub in sorted(self._entries.iterdir(), reverse=True):
+                if sub.is_dir():
+                    try:
+                        sub.rmdir()
+                    except OSError:
+                        pass
+
+    # -- internals -----------------------------------------------------------
+
+    def _iter_entry_paths(self) -> Iterable[Path]:
+        if not self._entries.exists():
+            return
+        for sub in self._entries.iterdir():
+            if not sub.is_dir():
+                continue
+            for entry in sub.iterdir():
+                if entry.is_file():
+                    yield entry
+
+    def _enforce_size_cap(self) -> None:
+        if self._max_size_bytes is None:
+            return
+        entries = []
+        total = 0
+        for path in self._iter_entry_paths():
+            try:
+                st = path.stat()
+            except FileNotFoundError:
+                continue
+            entries.append((st.st_mtime, st.st_size, path))
+            total += st.st_size
+        if total <= self._max_size_bytes:
+            return
+        entries.sort()  # oldest mtime first
+        for _mtime, size, path in entries:
+            if total <= self._max_size_bytes:
+                return
+            try:
+                path.unlink()
+                total -= size
+            except FileNotFoundError:
+                pass

cuda_core/tests/test_program_cache.py

@@ -434,3 +434,128 @@ def test_sqlite_cache_unbounded_by_default(tmp_path):
         for i in range(25):
             cache[f"k{i}".encode()] = _fake_object_code(b"X" * 1024, name=f"n{i}")
         assert len(cache) == 25
+
+
+# ---------------------------------------------------------------------------
+# FileStreamProgramCache -- single-process CRUD
+# ---------------------------------------------------------------------------
+
+
+def test_filestream_cache_empty_on_create(tmp_path):
+    from cuda.core.utils import FileStreamProgramCache
+
+    with FileStreamProgramCache(tmp_path / "fc") as cache:
+        assert len(cache) == 0
+        assert b"nope" not in cache
+        with pytest.raises(KeyError):
+            cache[b"nope"]
+
+
+def test_filestream_cache_roundtrip(tmp_path):
+    from cuda.core.utils import FileStreamProgramCache
+
+    with FileStreamProgramCache(tmp_path / "fc") as cache:
+        cache[b"k1"] = _fake_object_code(b"v1", name="x")
+        assert b"k1" in cache
+        got = cache[b"k1"]
+        assert bytes(got._module) == b"v1"
+        assert got._name == "x"
+        assert got._code_type == "cubin"
+
+
+def test_filestream_cache_delete(tmp_path):
+    from cuda.core.utils import FileStreamProgramCache
+
+    with FileStreamProgramCache(tmp_path / "fc") as cache:
+        cache[b"k"] = _fake_object_code()
+        del cache[b"k"]
+        assert b"k" not in cache
+        with pytest.raises(KeyError):
+            del cache[b"k"]
+
+
+def test_filestream_cache_len_counts_all(tmp_path):
+    from cuda.core.utils import FileStreamProgramCache
+
+    with FileStreamProgramCache(tmp_path / "fc") as cache:
+        cache[b"a"] = _fake_object_code(b"1")
+        cache[b"b"] = _fake_object_code(b"2")
+        cache[b"c"] = _fake_object_code(b"3")
+        assert len(cache) == 3
+
+
+def test_filestream_cache_clear(tmp_path):
+    from cuda.core.utils import FileStreamProgramCache
+
+    root = tmp_path / "fc"
+    with FileStreamProgramCache(root) as cache:
+        cache[b"a"] = _fake_object_code()
+        cache.clear()
+        assert len(cache) == 0
+
+
+def test_filestream_cache_persists_across_reopen(tmp_path):
+    from cuda.core.utils import FileStreamProgramCache
+
+    root = tmp_path / "fc"
+    with FileStreamProgramCache(root) as cache:
+        cache[b"k"] = _fake_object_code(b"persisted")
+    with FileStreamProgramCache(root) as cache:
+        assert bytes(cache[b"k"]._module) == b"persisted"
+
+
+def test_filestream_cache_atomic_no_half_written_file(tmp_path, monkeypatch):
+    # Simulate a crash during write: patch os.replace to raise.
+    import os as _os
+
+    from cuda.core.utils import FileStreamProgramCache
+
+    with FileStreamProgramCache(tmp_path / "fc") as cache:
+        def _boom(src, dst):
+            raise RuntimeError("crash during replace")
+
+        monkeypatch.setattr(_os, "replace", _boom)
+        with pytest.raises(RuntimeError, match="crash"):
+            cache[b"k"] = _fake_object_code(b"v")
+        monkeypatch.undo()
+        assert b"k" not in cache
+
+
+def test_filestream_cache_corruption_is_reported_as_miss(tmp_path):
+    from cuda.core.utils import FileStreamProgramCache
+
+    root = tmp_path / "fc"
+    with FileStreamProgramCache(root) as cache:
+        cache[b"k"] = _fake_object_code(b"ok")
+        path = cache._path_for_key(b"k")
+
+    # Corrupt the file on disk.
+    path.write_bytes(b"\x00not-a-pickle")
+    with FileStreamProgramCache(root) as cache:
+        with pytest.raises(KeyError):
+            cache[b"k"]
+        assert b"k" not in cache
+
+
+def test_filestream_cache_rejects_non_object_code(tmp_path):
+    from cuda.core.utils import FileStreamProgramCache
+
+    with FileStreamProgramCache(tmp_path / "fc") as cache:
+        with pytest.raises(TypeError, match="ObjectCode"):
+            cache[b"k"] = b"not an ObjectCode"
+
+
+def test_filestream_cache_rejects_negative_size_cap(tmp_path):
+    from cuda.core.utils import FileStreamProgramCache
+
+    with pytest.raises(ValueError, match="non-negative"):
+        FileStreamProgramCache(tmp_path / "fc", max_size_bytes=-1)
+
+
+def test_filestream_cache_accepts_str_keys(tmp_path):
+    from cuda.core.utils import FileStreamProgramCache
+
+    with FileStreamProgramCache(tmp_path / "fc") as cache:
+        cache["my-key"] = _fake_object_code(b"v")
+        assert "my-key" in cache
+        assert b"my-key" in cache