feat(core.utils): add SQLiteProgramCache backend

Persistent program cache backed by a single sqlite3 file in WAL mode. Keys
are arbitrary bytes (str keys are UTF-8 encoded); values are ObjectCode
instances serialised with pickle. Corrupt entries are treated as cache
misses and pruned on read. A max_size_bytes cap, when supplied, triggers
LRU eviction on writes -- the tracking infrastructure is in place even
though dedicated eviction tests land in a follow-up commit.

Part of issue #178.

Author: cpcloud

cuda_core/cuda/core/utils/__init__.py

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

cuda_core/cuda/core/utils/_program_cache.py

@@ -19,7 +19,12 @@
 
 import abc
 import hashlib
-from typing import Hashable, Sequence
+import os
+import pickle
+import sqlite3
+import time
+from pathlib import Path
+from typing import Hashable, Iterable, Optional, Sequence
 
 from cuda.core._module import ObjectCode
 from cuda.core._program import ProgramOptions
@@ -31,10 +36,30 @@
 
 __all__ = [
     "ProgramCacheResource",
+    "SQLiteProgramCache",
     "make_program_cache_key",
 ]
 
 
+_PICKLE_PROTOCOL = pickle.HIGHEST_PROTOCOL
+
+
+def _require_object_code(value: object) -> ObjectCode:
+    if not isinstance(value, ObjectCode):
+        raise TypeError(
+            f"cache values must be ObjectCode instances, got {type(value).__name__}"
+        )
+    return value
+
+
+def _as_key_bytes(key: object) -> bytes:
+    if isinstance(key, (bytes, bytearray)):
+        return bytes(key)
+    if isinstance(key, str):
+        return key.encode("utf-8")
+    raise TypeError(f"cache keys must be bytes or str, got {type(key).__name__}")
+
+
 # ---------------------------------------------------------------------------
 # Abstract base class
 # ---------------------------------------------------------------------------
@@ -224,3 +249,185 @@ def _update(label: str, payload: bytes) -> None:
         _update("name", n.encode("utf-8"))
 
     return hasher.digest()
+
+
+# ---------------------------------------------------------------------------
+# SQLite backend
+# ---------------------------------------------------------------------------
+
+
+_SQLITE_SCHEMA_VERSION = "1"
+
+
+class SQLiteProgramCache(ProgramCacheResource):
+    """Persistent program cache backed by a single sqlite3 database file.
+
+    Suitable for single-process workflows. Multiple processes *can* share the
+    file (sqlite3 WAL mode serialises writes), but
+    :class:`FileStreamProgramCache` is the recommended choice for concurrent
+    workers.
+
+    Parameters
+    ----------
+    path:
+        Filesystem path to the sqlite3 database. The parent directory is
+        created if missing.
+    max_size_bytes:
+        Optional size cap in bytes. When the sum of stored payload sizes
+        exceeds the cap, the least-recently-used entries are evicted until
+        the total is at or below the cap. ``None`` means unbounded.
+    """
+
+    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._path = Path(path)
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        self._max_size_bytes = max_size_bytes
+        self._conn: Optional[sqlite3.Connection] = None
+        self._open()
+
+    # -- lifecycle -----------------------------------------------------------
+
+    def _open(self) -> None:
+        # ``isolation_level=None`` puts the connection in autocommit mode so
+        # each statement is its own transaction; ``check_same_thread=False``
+        # lets a cache be created in one thread and used from another (writes
+        # are still serialised by sqlite's own lock).
+        self._conn = sqlite3.connect(
+            self._path,
+            isolation_level=None,
+            check_same_thread=False,
+            timeout=5.0,
+        )
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self._conn.execute("PRAGMA synchronous=NORMAL")
+        self._conn.execute("PRAGMA foreign_keys=ON")
+        self._conn.execute("PRAGMA busy_timeout=5000")
+        self._conn.executescript(
+            """
+            CREATE TABLE IF NOT EXISTS schema_meta (
+                key TEXT PRIMARY KEY,
+                value TEXT NOT NULL
+            );
+            CREATE TABLE IF NOT EXISTS entries (
+                key BLOB PRIMARY KEY,
+                payload BLOB NOT NULL,
+                size_bytes INTEGER NOT NULL,
+                created_at REAL NOT NULL,
+                accessed_at REAL NOT NULL
+            );
+            CREATE INDEX IF NOT EXISTS idx_accessed_at
+                ON entries(accessed_at);
+            """
+        )
+        self._conn.execute(
+            "INSERT OR IGNORE INTO schema_meta(key, value) VALUES (?, ?)",
+            ("schema_version", _SQLITE_SCHEMA_VERSION),
+        )
+
+    def close(self) -> None:
+        if self._conn is not None:
+            try:
+                self._conn.close()
+            finally:
+                self._conn = None
+
+    def _require_open(self) -> sqlite3.Connection:
+        if self._conn is None:
+            raise RuntimeError("SQLiteProgramCache is closed")
+        return self._conn
+
+    # -- mapping API ---------------------------------------------------------
+
+    def __contains__(self, key: object) -> bool:
+        k = _as_key_bytes(key)
+        row = self._require_open().execute(
+            "SELECT 1 FROM entries WHERE key = ?", (k,)
+        ).fetchone()
+        return row is not None
+
+    def __getitem__(self, key: object) -> ObjectCode:
+        k = _as_key_bytes(key)
+        conn = self._require_open()
+        row = conn.execute(
+            "SELECT payload FROM entries WHERE key = ?", (k,)
+        ).fetchone()
+        if row is None:
+            raise KeyError(key)
+        payload = row[0]
+        try:
+            value = pickle.loads(payload)
+        except Exception:
+            # Corrupt entry -- delete and treat as a miss.
+            conn.execute("DELETE FROM entries WHERE key = ?", (k,))
+            raise KeyError(key)
+        if not isinstance(value, ObjectCode):
+            conn.execute("DELETE FROM entries WHERE key = ?", (k,))
+            raise KeyError(key)
+        conn.execute(
+            "UPDATE entries SET accessed_at = ? WHERE key = ?",
+            (time.time(), k),
+        )
+        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)
+        now = time.time()
+        conn = self._require_open()
+        conn.execute(
+            """
+            INSERT INTO entries(key, payload, size_bytes, created_at, accessed_at)
+            VALUES (?, ?, ?, ?, ?)
+            ON CONFLICT(key) DO UPDATE SET
+                payload = excluded.payload,
+                size_bytes = excluded.size_bytes,
+                accessed_at = excluded.accessed_at
+            """,
+            (k, payload, len(payload), now, now),
+        )
+        self._enforce_size_cap()
+
+    def __delitem__(self, key: object) -> None:
+        k = _as_key_bytes(key)
+        conn = self._require_open()
+        cur = conn.execute("DELETE FROM entries WHERE key = ?", (k,))
+        if cur.rowcount == 0:
+            raise KeyError(key)
+
+    def __len__(self) -> int:
+        (n,) = self._require_open().execute(
+            "SELECT COUNT(*) FROM entries"
+        ).fetchone()
+        return int(n)
+
+    def clear(self) -> None:
+        self._require_open().execute("DELETE FROM entries")
+
+    # -- eviction ------------------------------------------------------------
+
+    def _enforce_size_cap(self) -> None:
+        if self._max_size_bytes is None:
+            return
+        conn = self._require_open()
+        (total,) = conn.execute(
+            "SELECT COALESCE(SUM(size_bytes), 0) FROM entries"
+        ).fetchone()
+        if total <= self._max_size_bytes:
+            return
+        # Delete oldest (least-recently-used) until at or under the cap.
+        rows: Iterable[tuple[bytes, int]] = conn.execute(
+            "SELECT key, size_bytes FROM entries ORDER BY accessed_at ASC"
+        ).fetchall()
+        for k, sz in rows:
+            if total <= self._max_size_bytes:
+                return
+            conn.execute("DELETE FROM entries WHERE key = ?", (k,))
+            total -= sz

cuda_core/tests/test_program_cache.py

@@ -251,3 +251,128 @@ def test_make_program_cache_key_rejects_non_str_bytes_code():
         make_program_cache_key(
             code=12345, code_type="c++", options=_opts(), target_type="cubin"
         )
+
+
+# ---------------------------------------------------------------------------
+# SQLiteProgramCache -- basic CRUD
+# ---------------------------------------------------------------------------
+
+
+def _fake_object_code(payload: bytes = b"fake-cubin", name: str = "unit"):
+    """Build an ObjectCode without touching the driver."""
+    from cuda.core._module import ObjectCode
+
+    return ObjectCode._init(payload, "cubin", name=name)
+
+
+def test_sqlite_cache_empty_on_create(tmp_path):
+    from cuda.core.utils import SQLiteProgramCache
+
+    db = tmp_path / "cache.db"
+    with SQLiteProgramCache(db) as cache:
+        assert len(cache) == 0
+        assert b"nope" not in cache
+        with pytest.raises(KeyError):
+            cache[b"nope"]
+        assert cache.get(b"nope") is None
+
+
+def test_sqlite_cache_set_get_roundtrip(tmp_path):
+    from cuda.core.utils import SQLiteProgramCache
+
+    db = tmp_path / "cache.db"
+    with SQLiteProgramCache(db) as cache:
+        key = b"k1"
+        cache[key] = _fake_object_code(b"bytes-1", name="a")
+
+        assert key in cache
+        assert len(cache) == 1
+        got = cache[key]
+        assert bytes(got._module) == b"bytes-1"
+        assert got._name == "a"
+        assert got._code_type == "cubin"
+
+
+def test_sqlite_cache_overwrite_same_key(tmp_path):
+    from cuda.core.utils import SQLiteProgramCache
+
+    db = tmp_path / "cache.db"
+    with SQLiteProgramCache(db) as cache:
+        cache[b"k"] = _fake_object_code(b"v1")
+        cache[b"k"] = _fake_object_code(b"v2")
+        assert len(cache) == 1
+        assert bytes(cache[b"k"]._module) == b"v2"
+
+
+def test_sqlite_cache_delete(tmp_path):
+    from cuda.core.utils import SQLiteProgramCache
+
+    db = tmp_path / "cache.db"
+    with SQLiteProgramCache(db) as cache:
+        cache[b"k"] = _fake_object_code()
+        del cache[b"k"]
+        assert b"k" not in cache
+        assert len(cache) == 0
+        with pytest.raises(KeyError):
+            del cache[b"k"]
+
+
+def test_sqlite_cache_clear(tmp_path):
+    from cuda.core.utils import SQLiteProgramCache
+
+    db = tmp_path / "cache.db"
+    with SQLiteProgramCache(db) as cache:
+        cache[b"a"] = _fake_object_code(b"1")
+        cache[b"b"] = _fake_object_code(b"2")
+        cache.clear()
+        assert len(cache) == 0
+
+
+def test_sqlite_cache_persists_across_open(tmp_path):
+    from cuda.core.utils import SQLiteProgramCache
+
+    db = tmp_path / "cache.db"
+    with SQLiteProgramCache(db) as cache:
+        cache[b"k"] = _fake_object_code(b"persisted")
+    with SQLiteProgramCache(db) as cache:
+        assert bytes(cache[b"k"]._module) == b"persisted"
+
+
+def test_sqlite_cache_corruption_is_reported_as_miss(tmp_path):
+    import sqlite3
+
+    from cuda.core.utils import SQLiteProgramCache
+
+    db = tmp_path / "cache.db"
+    with SQLiteProgramCache(db) as cache:
+        cache[b"k"] = _fake_object_code(b"ok")
+    # Overwrite the payload with garbage directly in the DB.
+    with sqlite3.connect(db) as conn:
+        conn.execute(
+            "UPDATE entries SET payload = ? WHERE key = ?",
+            (b"\x00\x01garbage", b"k"),
+        )
+        conn.commit()
+    with SQLiteProgramCache(db) as cache:
+        with pytest.raises(KeyError):
+            cache[b"k"]
+        assert b"k" not in cache  # corrupt entry was pruned
+
+
+def test_sqlite_cache_rejects_non_object_code(tmp_path):
+    from cuda.core.utils import SQLiteProgramCache
+
+    with SQLiteProgramCache(tmp_path / "cache.db") as cache:
+        with pytest.raises(TypeError, match="ObjectCode"):
+            cache[b"k"] = b"not an ObjectCode"
+
+
+def test_sqlite_cache_accepts_str_keys(tmp_path):
+    from cuda.core.utils import SQLiteProgramCache
+
+    db = tmp_path / "cache.db"
+    with SQLiteProgramCache(db) as cache:
+        cache["str-key"] = _fake_object_code(b"v")
+        assert "str-key" in cache
+        # Same bytes representation so the corresponding bytes key also hits.
+        assert b"str-key" in cache