diff --git a/python/fi/prompt/cache.py b/python/fi/prompt/cache.py index 295c4b0..58da235 100644 --- a/python/fi/prompt/cache.py +++ b/python/fi/prompt/cache.py @@ -1,11 +1,9 @@ from __future__ import annotations -import atexit import logging import threading import time -from queue import Empty, Queue -from typing import Callable, Dict, Optional, Tuple +from typing import Dict, Optional # We deliberately import via string to avoid circular import at runtime from typing import TYPE_CHECKING @@ -19,7 +17,6 @@ # --------------------------------------------------------------------------- DEFAULT_TTL_SEC = 60 * 5 # 5 minutes -DEFAULT_REFRESH_WORKERS = 2 logger = logging.getLogger("fi.prompt.cache") @@ -42,71 +39,6 @@ def is_stale(self) -> bool: return time.time() >= self.expiry -# --------------------------------------------------------------------------- -# Background refresh infra (Queue + workers) -# --------------------------------------------------------------------------- - - -class _RefreshWorker(threading.Thread): - """Continuously processes refresh callables from the shared queue.""" - - def __init__(self, q: "Queue[Callable[[], None]]", identifier: int): - super().__init__(daemon=True, name=f"PromptCacheWorker-{identifier}") - self._queue = q - self._running = True - - def run(self) -> None: # noqa: D401 – imperative mood fine - while self._running: - try: - task = self._queue.get(timeout=1) - except Empty: - continue # check _running flag again - - try: - task() - except Exception as exc: # pragma: no cover – log + continue - logger.warning("Prompt cache refresh task failed: %s", exc, exc_info=True) - finally: - self._queue.task_done() - - def stop(self) -> None: - self._running = False - - -class _TaskManager: - """Manages background refresh workers and graceful shutdown.""" - - def __init__(self, num_workers: int): - self._queue: "Queue[Callable[[], None]]" = Queue() - self._workers = [_RefreshWorker(self._queue, i) for i in range(num_workers)] - for w in self._workers: - w.start() - - atexit.register(self._shutdown) - - # Public API ----------------------------------------------------------- - - def submit(self, task: Callable[[], None]): - self._queue.put(task) - - # Private -------------------------------------------------------------- - - def _shutdown(self): - logger.debug("Shutting down PromptCache workers …") - for w in self._workers: - w.stop() - # Drain queue quickly - while not self._queue.empty(): - try: - self._queue.get_nowait() - self._queue.task_done() - except Empty: - break - for w in self._workers: - w.join(timeout=1) - logger.debug("PromptCache workers shut down.") - - # --------------------------------------------------------------------------- # Public cache API # --------------------------------------------------------------------------- @@ -115,12 +47,10 @@ def _shutdown(self): class PromptCache: """Thread-safe, stale-while-revalidate cache for `PromptTemplate` objects.""" - def __init__(self, ttl_sec: int = DEFAULT_TTL_SEC, max_workers: int = DEFAULT_REFRESH_WORKERS): + def __init__(self, ttl_sec: int = DEFAULT_TTL_SEC): self._ttl_sec = ttl_sec self._store: Dict[str, _CacheItem] = {} self._lock = threading.Lock() # protects _store mutations - self._refreshing_keys: set[str] = set() - self._tm = _TaskManager(max_workers) # ------------------------------- helpers ---------------------------- @@ -161,26 +91,7 @@ def invalidate(self, key_prefix: str): for k in to_delete: del self._store[k] - # -------------------------- refresh management --------------------- - - def refresh_async(self, key: str, fetch_fn: Callable[[], "PromptTemplate"]): - """Schedule a refresh if one is not already in-flight.""" - with self._lock: - if key in self._refreshing_keys: - return - self._refreshing_keys.add(key) - - def _task(): - try: - tpl = fetch_fn() - self.set(key, tpl) - finally: - with self._lock: - self._refreshing_keys.discard(key) - - self._tm.submit(_task) - # Global singleton -------------------------------------------------------- -prompt_cache = PromptCache() \ No newline at end of file +prompt_cache = PromptCache() diff --git a/python/tests/test_prompt_cache.py b/python/tests/test_prompt_cache.py new file mode 100644 index 0000000..efbb322 --- /dev/null +++ b/python/tests/test_prompt_cache.py @@ -0,0 +1,13 @@ +"""Tests for fi.prompt.cache.""" + +import subprocess +import sys + + +def test_importing_fi_prompt_spawns_no_threads(): + code = "import threading, fi.prompt; print([t.name for t in threading.enumerate()])" + result = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, text=True, check=True, + ) + assert "PromptCacheWorker" not in result.stdout