mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-12 14:18:06 +00:00
fix(cache): one cleanup thread per cache directory, not per manager
The display process ran three cleanup threads over one directory:
14:22:59.954 display_controller (the real manager)
14:22:59.973 startup validation, run 1 (discarded)
14:23:01.055 startup validation, run 2 (discarded)
Two of those managers existed only to read a directory path.
StartupValidator._validate_cache_directory built a whole CacheManager to
call get_cache_dir(), and validation runs twice -- once before the
plugin manager exists and again after. Each construction also probes
writability by writing and deleting .writetest on the card.
The discarded ones never went away. cleanup_loop closes over `self`, so
the thread keeps its manager alive: two objects that could never be
collected, waking every 24 hours to re-scan the same 9,000-file
directory. Nothing stopped them either -- stop_cleanup_thread had no
callers anywhere in the tree.
Two changes. The validator now takes the CacheManager the application
actually uses, which is also the more correct thing to validate; when
no caller supplies one it still builds its own, but stops the thread
afterwards. And CacheManager now tracks which directory it is sweeping,
so the second manager over a directory skips starting a thread at all.
That is the right granularity regardless of call sites: the sweep lists
a directory and deletes from it, so a second thread only duplicates the
scan. Ownership is released on stop, so a survivor can take over rather
than leaving the directory permanently unclaimed by a dead owner.
Measured directly, three managers over one directory: 3 threads before,
1 after.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Udr6MfaFLUPhX5Fgo67Jf5
This commit is contained in:
co-authored by
Claude Opus 5
parent
bb1a1671ec
commit
56dbfe6c8b
+43
-4
@@ -46,7 +46,21 @@ from src.cache.disk_cache import DateTimeEncoder # noqa: F401 - deliberate re-e
|
|||||||
|
|
||||||
class CacheManager:
|
class CacheManager:
|
||||||
"""Manages caching of API responses to reduce API calls."""
|
"""Manages caching of API responses to reduce API calls."""
|
||||||
|
|
||||||
|
# Which cache directories already have a cleanup thread in this process.
|
||||||
|
#
|
||||||
|
# The sweep is directory-scoped work -- it lists a directory and deletes
|
||||||
|
# from it -- so one per directory is the right number no matter how many
|
||||||
|
# managers exist. Nothing enforced that before: every instance started its
|
||||||
|
# own, and because the loop closes over `self`, a discarded manager could
|
||||||
|
# never be collected and its thread woke to re-scan the same directory
|
||||||
|
# every 24 hours for the life of the process. Startup validation runs
|
||||||
|
# twice and built a throwaway manager each time, so a display process
|
||||||
|
# carried three threads for one cache.
|
||||||
|
_cleanup_owners: Dict[str, 'CacheManager'] = {}
|
||||||
|
_cleanup_owners_lock = threading.Lock()
|
||||||
|
|
||||||
|
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
# Initialize logger first
|
# Initialize logger first
|
||||||
self.logger: logging.Logger = get_logger(__name__)
|
self.logger: logging.Logger = get_logger(__name__)
|
||||||
@@ -718,11 +732,29 @@ class CacheManager:
|
|||||||
}
|
}
|
||||||
|
|
||||||
def start_cleanup_thread(self) -> None:
|
def start_cleanup_thread(self) -> None:
|
||||||
"""Start background thread for periodic disk cache cleanup."""
|
"""Start background thread for periodic disk cache cleanup.
|
||||||
|
|
||||||
|
At most one thread per cache directory per process: the sweep is
|
||||||
|
directory-scoped, so a second one only duplicates the scan.
|
||||||
|
"""
|
||||||
if self._cleanup_thread and self._cleanup_thread.is_alive():
|
if self._cleanup_thread and self._cleanup_thread.is_alive():
|
||||||
self.logger.debug("Cleanup thread already running")
|
self.logger.debug("Cleanup thread already running")
|
||||||
return
|
return
|
||||||
|
|
||||||
|
with CacheManager._cleanup_owners_lock:
|
||||||
|
owner = CacheManager._cleanup_owners.get(self.cache_dir)
|
||||||
|
if owner is not None and owner is not self:
|
||||||
|
thread = owner._cleanup_thread
|
||||||
|
if thread is not None and thread.is_alive():
|
||||||
|
self.logger.debug(
|
||||||
|
"Cleanup thread for %s already owned by another cache "
|
||||||
|
"manager in this process; not starting a second",
|
||||||
|
self.cache_dir)
|
||||||
|
return
|
||||||
|
# The owner's thread died or was stopped -- take over.
|
||||||
|
CacheManager._cleanup_owners[self.cache_dir] = self
|
||||||
|
|
||||||
|
|
||||||
def cleanup_loop():
|
def cleanup_loop():
|
||||||
"""Background loop that runs cleanup periodically."""
|
"""Background loop that runs cleanup periodically."""
|
||||||
self.logger.info("Disk cache cleanup thread started (interval: %d hours)",
|
self.logger.info("Disk cache cleanup thread started (interval: %d hours)",
|
||||||
@@ -770,10 +802,17 @@ class CacheManager:
|
|||||||
Signals the thread to stop and waits for it to finish (with timeout).
|
Signals the thread to stop and waits for it to finish (with timeout).
|
||||||
This allows for clean shutdown during testing or application termination.
|
This allows for clean shutdown during testing or application termination.
|
||||||
"""
|
"""
|
||||||
|
# Release ownership first and unconditionally, so a manager that never
|
||||||
|
# started a thread (or whose thread already exited) cannot keep the
|
||||||
|
# directory claimed and block a live manager from sweeping it.
|
||||||
|
with CacheManager._cleanup_owners_lock:
|
||||||
|
if CacheManager._cleanup_owners.get(self.cache_dir) is self:
|
||||||
|
del CacheManager._cleanup_owners[self.cache_dir]
|
||||||
|
|
||||||
if not self._cleanup_thread or not self._cleanup_thread.is_alive():
|
if not self._cleanup_thread or not self._cleanup_thread.is_alive():
|
||||||
self.logger.debug("Cleanup thread not running")
|
self.logger.debug("Cleanup thread not running")
|
||||||
return
|
return
|
||||||
|
|
||||||
self.logger.info("Stopping disk cache cleanup thread...")
|
self.logger.info("Stopping disk cache cleanup thread...")
|
||||||
self._cleanup_stop_event.set() # Signal thread to stop
|
self._cleanup_stop_event.set() # Signal thread to stop
|
||||||
|
|
||||||
|
|||||||
@@ -90,7 +90,8 @@ class DisplayController:
|
|||||||
# Validate startup configuration
|
# Validate startup configuration
|
||||||
try:
|
try:
|
||||||
from src.startup_validator import StartupValidator
|
from src.startup_validator import StartupValidator
|
||||||
validator = StartupValidator(self.config_manager)
|
validator = StartupValidator(self.config_manager,
|
||||||
|
cache_manager=self.cache_manager)
|
||||||
is_valid, errors, warnings = validator.validate_all()
|
is_valid, errors, warnings = validator.validate_all()
|
||||||
|
|
||||||
if warnings:
|
if warnings:
|
||||||
@@ -258,7 +259,8 @@ class DisplayController:
|
|||||||
# Validate plugins after plugin manager is created
|
# Validate plugins after plugin manager is created
|
||||||
try:
|
try:
|
||||||
from src.startup_validator import StartupValidator
|
from src.startup_validator import StartupValidator
|
||||||
validator = StartupValidator(self.config_manager, self.plugin_manager)
|
validator = StartupValidator(self.config_manager, self.plugin_manager,
|
||||||
|
cache_manager=self.cache_manager)
|
||||||
is_valid, errors, warnings = validator.validate_all()
|
is_valid, errors, warnings = validator.validate_all()
|
||||||
|
|
||||||
if warnings:
|
if warnings:
|
||||||
|
|||||||
@@ -15,16 +15,23 @@ from src.logging_config import get_logger
|
|||||||
class StartupValidator:
|
class StartupValidator:
|
||||||
"""Validates system state on startup."""
|
"""Validates system state on startup."""
|
||||||
|
|
||||||
def __init__(self, config_manager: Any, plugin_manager: Optional[Any] = None) -> None:
|
def __init__(self, config_manager: Any, plugin_manager: Optional[Any] = None,
|
||||||
|
cache_manager: Optional[Any] = None) -> None:
|
||||||
"""
|
"""
|
||||||
Initialize the startup validator.
|
Initialize the startup validator.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
config_manager: ConfigManager instance
|
config_manager: ConfigManager instance
|
||||||
plugin_manager: Optional PluginManager instance
|
plugin_manager: Optional PluginManager instance
|
||||||
|
cache_manager: The CacheManager the application will actually use.
|
||||||
|
Pass it. Without one this validator builds its own just to read
|
||||||
|
a directory path, which reports on a cache the app does not
|
||||||
|
use and leaves behind a cleanup thread that nothing stops --
|
||||||
|
validation runs twice per startup, so that was two of them.
|
||||||
"""
|
"""
|
||||||
self.config_manager = config_manager
|
self.config_manager = config_manager
|
||||||
self.plugin_manager = plugin_manager
|
self.plugin_manager = plugin_manager
|
||||||
|
self.cache_manager = cache_manager
|
||||||
self.logger = get_logger(__name__)
|
self.logger = get_logger(__name__)
|
||||||
self.errors: List[str] = []
|
self.errors: List[str] = []
|
||||||
self.warnings: List[str] = []
|
self.warnings: List[str] = []
|
||||||
@@ -91,9 +98,21 @@ class StartupValidator:
|
|||||||
def _validate_cache_directory(self) -> None:
|
def _validate_cache_directory(self) -> None:
|
||||||
"""Validate cache directory permissions."""
|
"""Validate cache directory permissions."""
|
||||||
try:
|
try:
|
||||||
from src.cache_manager import CacheManager
|
cache_manager = self.cache_manager
|
||||||
cache_manager = CacheManager()
|
if cache_manager is None:
|
||||||
cache_dir = cache_manager.get_cache_dir()
|
# No caller supplied one (older embedders, direct use in a
|
||||||
|
# script). Build one, but do not leave its cleanup thread
|
||||||
|
# running behind us -- this instance is discarded on the next
|
||||||
|
# line but the thread is a closure over it, so it would never
|
||||||
|
# be collected.
|
||||||
|
from src.cache_manager import CacheManager
|
||||||
|
cache_manager = CacheManager()
|
||||||
|
try:
|
||||||
|
cache_dir = cache_manager.get_cache_dir()
|
||||||
|
finally:
|
||||||
|
cache_manager.stop_cleanup_thread()
|
||||||
|
else:
|
||||||
|
cache_dir = cache_manager.get_cache_dir()
|
||||||
|
|
||||||
if not cache_dir:
|
if not cache_dir:
|
||||||
self.warnings.append("Cache directory not available - caching will be disabled")
|
self.warnings.append("Cache directory not available - caching will be disabled")
|
||||||
|
|||||||
@@ -0,0 +1,146 @@
|
|||||||
|
"""Tests that one cache directory gets one cleanup thread per process.
|
||||||
|
|
||||||
|
The sweep lists a directory and deletes from it, so a second thread over the
|
||||||
|
same directory only duplicates the scan. Nothing enforced that: every
|
||||||
|
CacheManager started its own, and since the loop closes over `self`, a
|
||||||
|
discarded manager could never be collected -- its thread stayed alive and
|
||||||
|
re-scanned the same directory every 24 hours for the life of the process.
|
||||||
|
|
||||||
|
On the dev rig a display process carried three, for one cache directory:
|
||||||
|
|
||||||
|
14:22:59.954 display_controller (the real one)
|
||||||
|
14:22:59.973 startup validation, run 1 (discarded)
|
||||||
|
14:23:01.055 startup validation, run 2 (discarded)
|
||||||
|
|
||||||
|
Startup validation runs twice and built a throwaway manager each time, purely
|
||||||
|
to read a directory path.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import threading
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from src.cache_manager import CacheManager
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def _clean_registry():
|
||||||
|
CacheManager._cleanup_owners.clear()
|
||||||
|
yield
|
||||||
|
for owner in list(CacheManager._cleanup_owners.values()):
|
||||||
|
owner.stop_cleanup_thread()
|
||||||
|
CacheManager._cleanup_owners.clear()
|
||||||
|
|
||||||
|
|
||||||
|
def _live_cleanup_threads():
|
||||||
|
return [t for t in threading.enumerate()
|
||||||
|
if t.name == 'DiskCacheCleanup' and t.is_alive()]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def manager(tmp_path, monkeypatch):
|
||||||
|
"""A CacheManager pinned to a temp dir, so tests never touch the real one."""
|
||||||
|
monkeypatch.setattr(CacheManager, '_get_writable_cache_dir',
|
||||||
|
lambda self: str(tmp_path))
|
||||||
|
return CacheManager
|
||||||
|
|
||||||
|
|
||||||
|
class TestOneThreadPerDirectory:
|
||||||
|
def test_a_single_manager_starts_one(self, manager):
|
||||||
|
before = len(_live_cleanup_threads())
|
||||||
|
m = manager()
|
||||||
|
try:
|
||||||
|
assert len(_live_cleanup_threads()) == before + 1
|
||||||
|
finally:
|
||||||
|
m.stop_cleanup_thread()
|
||||||
|
|
||||||
|
def test_three_managers_still_start_one(self, manager):
|
||||||
|
# Exactly the rig's shape: the real manager plus two throwaways.
|
||||||
|
before = len(_live_cleanup_threads())
|
||||||
|
managers = [manager() for _ in range(3)]
|
||||||
|
try:
|
||||||
|
assert len(_live_cleanup_threads()) == before + 1
|
||||||
|
finally:
|
||||||
|
for m in managers:
|
||||||
|
m.stop_cleanup_thread()
|
||||||
|
|
||||||
|
def test_the_first_one_owns_it(self, manager):
|
||||||
|
first, second = manager(), manager()
|
||||||
|
try:
|
||||||
|
assert CacheManager._cleanup_owners[first.cache_dir] is first
|
||||||
|
assert second._cleanup_thread is None
|
||||||
|
finally:
|
||||||
|
first.stop_cleanup_thread()
|
||||||
|
second.stop_cleanup_thread()
|
||||||
|
|
||||||
|
def test_the_survivor_can_take_over(self, manager):
|
||||||
|
first = manager()
|
||||||
|
first.stop_cleanup_thread()
|
||||||
|
assert not _live_cleanup_threads()
|
||||||
|
|
||||||
|
second = manager()
|
||||||
|
try:
|
||||||
|
# Ownership was released, so the directory is swept again rather
|
||||||
|
# than being left permanently unclaimed by a dead owner.
|
||||||
|
assert len(_live_cleanup_threads()) == 1
|
||||||
|
assert CacheManager._cleanup_owners[second.cache_dir] is second
|
||||||
|
finally:
|
||||||
|
second.stop_cleanup_thread()
|
||||||
|
|
||||||
|
def test_stopping_a_non_owner_does_not_unclaim_the_directory(self, manager):
|
||||||
|
first, second = manager(), manager()
|
||||||
|
try:
|
||||||
|
second.stop_cleanup_thread() # never owned it
|
||||||
|
assert CacheManager._cleanup_owners[first.cache_dir] is first
|
||||||
|
assert len(_live_cleanup_threads()) == 1
|
||||||
|
finally:
|
||||||
|
first.stop_cleanup_thread()
|
||||||
|
|
||||||
|
def test_separate_directories_get_separate_threads(self, tmp_path, monkeypatch):
|
||||||
|
a, b = tmp_path / 'a', tmp_path / 'b'
|
||||||
|
a.mkdir()
|
||||||
|
b.mkdir()
|
||||||
|
dirs = iter([str(a), str(b)])
|
||||||
|
monkeypatch.setattr(CacheManager, '_get_writable_cache_dir',
|
||||||
|
lambda self: next(dirs))
|
||||||
|
first, second = CacheManager(), CacheManager()
|
||||||
|
try:
|
||||||
|
assert first.cache_dir != second.cache_dir
|
||||||
|
assert len(_live_cleanup_threads()) == 2
|
||||||
|
finally:
|
||||||
|
first.stop_cleanup_thread()
|
||||||
|
second.stop_cleanup_thread()
|
||||||
|
|
||||||
|
def test_no_thread_leaks_across_many_constructions(self, manager):
|
||||||
|
before = len(_live_cleanup_threads())
|
||||||
|
made = [manager() for _ in range(12)]
|
||||||
|
try:
|
||||||
|
assert len(_live_cleanup_threads()) == before + 1
|
||||||
|
finally:
|
||||||
|
for m in made:
|
||||||
|
m.stop_cleanup_thread()
|
||||||
|
assert len(_live_cleanup_threads()) == before
|
||||||
|
|
||||||
|
|
||||||
|
class TestValidatorDoesNotBuildItsOwn:
|
||||||
|
def test_it_uses_the_cache_manager_it_is_given(self, manager):
|
||||||
|
from src.startup_validator import StartupValidator
|
||||||
|
|
||||||
|
shared = manager()
|
||||||
|
try:
|
||||||
|
before = len(_live_cleanup_threads())
|
||||||
|
v = StartupValidator(config_manager=object(), cache_manager=shared)
|
||||||
|
v._validate_cache_directory()
|
||||||
|
assert len(_live_cleanup_threads()) == before, (
|
||||||
|
"validation started another cleanup thread")
|
||||||
|
finally:
|
||||||
|
shared.stop_cleanup_thread()
|
||||||
|
|
||||||
|
def test_without_one_it_cleans_up_after_itself(self, manager):
|
||||||
|
from src.startup_validator import StartupValidator
|
||||||
|
|
||||||
|
before = len(_live_cleanup_threads())
|
||||||
|
v = StartupValidator(config_manager=object())
|
||||||
|
v._validate_cache_directory()
|
||||||
|
assert len(_live_cleanup_threads()) == before, (
|
||||||
|
"the fallback manager left its cleanup thread running")
|
||||||
Reference in New Issue
Block a user