diff --git a/src/cachier/core.py b/src/cachier/core.py index 5a4bc4aa..30ccb15b 100644 --- a/src/cachier/core.py +++ b/src/cachier/core.py @@ -102,10 +102,20 @@ def _set_max_workers(max_workers): _get_executor(True) -def _get_executor(reset=False): - if reset or not hasattr(_get_executor, "executor"): - _get_executor.executor = ThreadPoolExecutor(_max_workers()) - return _get_executor.executor +def _get_executor(reset: bool = False) -> ThreadPoolExecutor: + """Return the worker pool, retiring the previous pool after a reset. + + Already submitted work is allowed to finish without blocking the caller. + + """ + executor_state: Any = _get_executor + if reset or not hasattr(executor_state, "executor"): + executor = ThreadPoolExecutor(_max_workers()) + previous = getattr(executor_state, "executor", None) + executor_state.executor = executor + if previous is not None: + previous.shutdown(wait=False) + return executor_state.executor def _function_thread(core: _BaseCore, key, func, args, kwds): diff --git a/tests/test_executor_reset.py b/tests/test_executor_reset.py new file mode 100644 index 00000000..f221e41e --- /dev/null +++ b/tests/test_executor_reset.py @@ -0,0 +1,69 @@ +"""Regression tests for replacing background worker pools.""" + +import threading +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from cachier.core import MAX_WORKERS_ENVAR_NAME, _get_executor + + +def test_reset_retires_pool_without_dropping_pending_work(monkeypatch: pytest.MonkeyPatch) -> None: + """Keep running and queued work alive while retiring the old pool. + + The replacement must become usable before the old work completes. + + """ + previous = ThreadPoolExecutor(max_workers=1) + monkeypatch.setattr(_get_executor, "executor", previous, raising=False) + monkeypatch.setenv(MAX_WORKERS_ENVAR_NAME, "1") + started = threading.Event() + release = threading.Event() + + def blocked_work() -> str: + """Wait for the test to release the old worker. + + A timeout bounds cleanup if an assertion fails. + + """ + started.set() + release.wait(timeout=5) + return "running" + + running = previous.submit(blocked_work) + reset = threading.Thread(target=_get_executor, kwargs={"reset": True}) + try: + assert started.wait(timeout=2) + queued = previous.submit(str, "queued") + reset.start() + reset.join(timeout=1) + assert not reset.is_alive() + replacement = _get_executor() + assert replacement is not previous + assert replacement.submit(str, "new").result(timeout=2) == "new" + with pytest.raises(RuntimeError, match="shutdown"): + previous.submit(str, "rejected") + release.set() + assert running.result(timeout=2) == "running" + assert queued.result(timeout=2) == "queued" + finally: + release.set() + if reset.ident is not None: + reset.join(timeout=5) + previous.shutdown(wait=True) + _get_executor().shutdown(wait=True) + + +def test_failed_reset_preserves_usable_pool(monkeypatch: pytest.MonkeyPatch) -> None: + """Keep the existing pool available when replacement creation fails. + + Invalid worker settings must not shut down a working executor. + + """ + with ThreadPoolExecutor(max_workers=1) as previous: + monkeypatch.setattr(_get_executor, "executor", previous, raising=False) + monkeypatch.setenv(MAX_WORKERS_ENVAR_NAME, "0") + with pytest.raises(ValueError, match="max_workers"): + _get_executor(reset=True) + assert _get_executor() is previous + assert previous.submit(str, "still usable").result(timeout=2) == "still usable"