From c32aaf02e537e08c4e03a71a3fc05db38eb54584 Mon Sep 17 00:00:00 2001 From: Chuck Date: Sun, 12 Jul 2026 19:59:29 -0400 Subject: [PATCH] perf(plugins): run scheduled updates off the render thread MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit plugin update() executed inline in the render loop — execute_update's internal thread.join(timeout=30) blocked it, so one slow plugin HTTP fetch froze scrolling for the whole fetch (up to 30s; DNS-retry storms made this a regular occurrence on flaky networks). Scheduling stays on the render thread and keeps every existing gate (enabled, circuit breaker, can_execute, interval); due updates are now enqueued to a single background worker (serialized — same one-at-a-time execution as before, no thundering herd). RUNNING is set at enqueue so can_execute blocks re-entry alongside the pending-set dedup. Per-plugin locks make the old implicit update/display no-overlap guarantee explicit: the worker holds the plugin's lock through its update; the display side try-locks and, when the plugin is mid-update, holds the last frame for that iteration — reported as success so a mid-update skip never advances the rotation. Unlike before, the guarantee now also holds across the post-timeout window (previously the lingering update thread overlapped display()). Deadlock-free by construction: the worker takes one lock; display never blocks. Timeout semantics unchanged (lingering daemon thread documented). Kill switch: plugin_system.synchronous_updates: true restores the inline path. 8 new concurrency tests (non-blocking scheduler, overlap assertion under a hammering display loop, lock release on failure/timeout paths, dedup, kill switch); 4-min devpi soak clean (updates completing, rotation advancing, no stuck RUNNING states). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam --- src/display_controller.py | 96 ++++++++++---- src/plugin_system/plugin_manager.py | 190 +++++++++++++++++++++------ test/test_async_plugin_updates.py | 197 ++++++++++++++++++++++++++++ test/test_plugin_system.py | 8 ++ 4 files changed, 422 insertions(+), 69 deletions(-) create mode 100644 test/test_async_plugin_updates.py diff --git a/src/display_controller.py b/src/display_controller.py index 8ae64f34..79680291 100644 --- a/src/display_controller.py +++ b/src/display_controller.py @@ -23,6 +23,7 @@ import time import os import json +from contextlib import contextmanager from pathlib import Path from typing import Dict, Any, List, Optional, Callable from datetime import datetime @@ -839,6 +840,30 @@ def _tick_plugin_updates(self): except Exception: # pylint: disable=broad-except logger.exception("Error running scheduled plugin updates") + @contextmanager + def _display_lock_or_skip(self, plugin_id): + """Try-lock guard keeping a plugin's display() off its in-flight update(). + + Yields True when display may run (lock held, released on exit) or + when no lock support exists (older plugin manager). Yields False when + the plugin's update() is currently executing on the background + worker — the caller should treat the frame as displayed (the panel + holds the last pushed frame) rather than as a plugin failure, so a + mid-update skip never advances the rotation. + """ + pm = self.plugin_manager + if not pm or not hasattr(pm, 'get_plugin_lock') or not plugin_id: + yield True + return + lock = pm.get_plugin_lock(plugin_id) + if not lock.acquire(blocking=False): + yield False + return + try: + yield True + finally: + lock.release() + _FOLLOWER_SEND_INTERVAL = 1.0 / 90 # raw bytes are cheap; 90fps > follower render rate def _follower_rebuild_scroll_image(self) -> None: @@ -1836,25 +1861,30 @@ def run(self): ) _accepts_display_mode = self._plugin_accepts_display_mode[_cache_key] - # Use PluginExecutor for safe execution with timeout - if self.plugin_manager and hasattr(self.plugin_manager, 'plugin_executor'): - result = self.plugin_manager.plugin_executor.execute_display( - manager_to_display, - plugin_id, - force_clear=self.force_change, - display_mode=active_mode if _accepts_display_mode else None - ) - # execute_display returns bool, convert to expected format - if result: - result = True # Success - else: - result = False # Failed - else: - # Fallback to direct call if executor not available - if _accepts_display_mode: - result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change) + with self._display_lock_or_skip(plugin_id) as can_display: + if not can_display: + # update() in flight on the worker — hold + # the last frame; not a plugin failure + result = True + elif self.plugin_manager and hasattr(self.plugin_manager, 'plugin_executor'): + # Use PluginExecutor for safe execution with timeout + result = self.plugin_manager.plugin_executor.execute_display( + manager_to_display, + plugin_id, + force_clear=self.force_change, + display_mode=active_mode if _accepts_display_mode else None + ) + # execute_display returns bool, convert to expected format + if result: + result = True # Success + else: + result = False # Failed else: - result = manager_to_display.display(force_clear=self.force_change) + # Fallback to direct call if executor not available + if _accepts_display_mode: + result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change) + else: + result = manager_to_display.display(force_clear=self.force_change) logger.debug(f"display() returned: {result} (type: {type(result)})") # Check if display() returned a boolean (new behavior) @@ -2139,11 +2169,16 @@ def _should_exit_dynamic(elapsed_time: float) -> bool: while True: try: - # Pass display_mode to maintain sticky manager state - if _accepts_display_mode: - result = manager_to_display.display(display_mode=active_mode, force_clear=False) - else: - result = manager_to_display.display(force_clear=False) + with self._display_lock_or_skip(plugin_id) as can_display: + if can_display: + # Pass display_mode to maintain sticky manager state + if _accepts_display_mode: + result = manager_to_display.display(display_mode=active_mode, force_clear=False) + else: + result = manager_to_display.display(force_clear=False) + else: + # update() in flight — hold the last frame + result = True if isinstance(result, bool) and not result: logger.debug("Display returned False, breaking early") break @@ -2203,11 +2238,16 @@ def _should_exit_dynamic(elapsed_time: float) -> bool: break try: - # Pass display_mode to maintain sticky manager state - if _accepts_display_mode: - result = manager_to_display.display(display_mode=active_mode, force_clear=False) - else: - result = manager_to_display.display(force_clear=False) + with self._display_lock_or_skip(plugin_id) as can_display: + if can_display: + # Pass display_mode to maintain sticky manager state + if _accepts_display_mode: + result = manager_to_display.display(display_mode=active_mode, force_clear=False) + else: + result = manager_to_display.display(force_clear=False) + else: + # update() in flight — hold the last frame + result = True if isinstance(result, bool) and not result: # For dynamic duration plugins, don't exit on False - keep looping # until cycle is complete or max duration is reached diff --git a/src/plugin_system/plugin_manager.py b/src/plugin_system/plugin_manager.py index 0c6c558b..f6b68bfb 100644 --- a/src/plugin_system/plugin_manager.py +++ b/src/plugin_system/plugin_manager.py @@ -8,12 +8,13 @@ """ import json +import queue import sys import time import threading import types from pathlib import Path -from typing import Dict, List, Optional, Any +from typing import Dict, List, Optional, Any, Tuple import logging from src.exceptions import PluginError, ConfigError from src.logging_config import get_logger @@ -91,6 +92,32 @@ def __init__(self, plugins_dir: str = "plugins", # Health tracking (optional, set by display_controller if available) self.health_tracker = None self.resource_monitor = None + + # --- Asynchronous plugin updates ------------------------------- + # update() used to run inline in the render loop (execute_update's + # internal thread.join(timeout=30) blocked it), so one slow plugin + # HTTP fetch froze scrolling for the whole fetch. Scheduling still + # happens on the render thread (run_scheduled_updates), but + # execution moves to this single background worker. Per-plugin + # locks keep a plugin's update() and display() mutually exclusive — + # today's implicit guarantee, now explicit (and, unlike today, + # also held across the post-timeout window). + # Kill switch: plugin_system.synchronous_updates: true restores the + # inline path. + self._update_queue: "queue.Queue[Optional[Tuple[str, float]]]" = queue.Queue() + self._pending_updates: set = set() + self._pending_lock = threading.Lock() + self._plugin_locks: Dict[str, threading.Lock] = {} + self._plugin_locks_guard = threading.Lock() + self._update_worker: Optional[threading.Thread] = None + self._synchronous_updates = False + try: + if self.config_manager is not None: + cfg = self.config_manager.get_config() or {} + self._synchronous_updates = bool( + cfg.get('plugin_system', {}).get('synchronous_updates', False)) + except Exception: + self._synchronous_updates = False # Ensure plugins directory exists with proper permissions try: @@ -734,46 +761,127 @@ def run_scheduled_updates(self, current_time: Optional[float] = None) -> None: last_update = self.plugin_last_update.get(plugin_id, 0.0) if last_update == 0.0 or (current_time - last_update) >= interval: - # Update state to RUNNING - self.state_manager.set_state(plugin_id, PluginState.RUNNING) - - try: - # Use PluginExecutor for safe execution - success = False - if self.resource_monitor: - # If resource monitor exists, wrap the call - def monitored_update(): - self.resource_monitor.monitor_call(plugin_id, plugin_instance.update) - # SimpleNamespace stores `update` as an *instance* - # attribute, so attribute lookup returns the plain - # function object as-is. A dynamically-built class - # (`type(..., {'update': monitored_update})`) instead - # stores it as a *class* attribute, which the - # descriptor protocol turns into a bound method on - # access -- silently prepending the instance as an - # implicit first argument to a function that takes - # none, raising "monitored_update() takes 0 - # positional arguments but 1 was given" on every call. - success = self.plugin_executor.execute_update( - types.SimpleNamespace(update=monitored_update), - plugin_id - ) - else: - success = self.plugin_executor.execute_update(plugin_instance, plugin_id) - - if success: - self.plugin_last_update[plugin_id] = current_time - self.state_manager.record_update(plugin_id) - # Update state back to ENABLED - self.state_manager.set_state(plugin_id, PluginState.ENABLED) - # Record success - if self.health_tracker: - self.health_tracker.record_success(plugin_id) - else: - self._record_update_failure(plugin_id) - except Exception as exc: # pylint: disable=broad-except - self.logger.exception("Error updating plugin %s: %s", plugin_id, exc) - self._record_update_failure(plugin_id, exc=exc) + if self._synchronous_updates: + # Kill-switch path: the original inline execution + # (blocks the caller until update() completes/times out) + self.state_manager.set_state(plugin_id, PluginState.RUNNING) + self._execute_update_now(plugin_id, plugin_instance, current_time) + else: + self._enqueue_update(plugin_id, current_time) + + def get_plugin_lock(self, plugin_id: str) -> threading.Lock: + """Per-plugin lock keeping update() and display() mutually exclusive. + + The update worker holds it for the duration of a plugin's update(); + the display side acquires it non-blocking and skips that frame's + display() call when the plugin is mid-update. + """ + with self._plugin_locks_guard: + lock = self._plugin_locks.get(plugin_id) + if lock is None: + lock = threading.Lock() + self._plugin_locks[plugin_id] = lock + return lock + + def _enqueue_update(self, plugin_id: str, scheduled_time: float) -> None: + """Queue a due update for the background worker (dedup while pending).""" + with self._pending_lock: + if plugin_id in self._pending_updates: + return + self._pending_updates.add(plugin_id) + # RUNNING is set at enqueue time so can_execute() blocks re-entry and + # the web UI shows the truthful state while the item waits its turn. + self.state_manager.set_state(plugin_id, PluginState.RUNNING) + self._ensure_update_worker() + self._update_queue.put((plugin_id, scheduled_time)) + + def _ensure_update_worker(self) -> None: + if self._update_worker is not None and self._update_worker.is_alive(): + return + self._update_worker = threading.Thread( + target=self._update_worker_loop, name='plugin-update-worker', + daemon=True) + self._update_worker.start() + + def _update_worker_loop(self) -> None: + """Single worker: serializes all plugin updates (matching the old + inline behavior — no thundering herd of concurrent fetches), off the + render thread.""" + while True: + item = self._update_queue.get() + if item is None: # shutdown sentinel + return + plugin_id, scheduled_time = item + try: + plugin_instance = self.plugins.get(plugin_id) + if plugin_instance is None: # unloaded while queued + self.state_manager.set_state(plugin_id, PluginState.ENABLED) + continue + lock = self.get_plugin_lock(plugin_id) + with lock: + self._execute_update_now(plugin_id, plugin_instance, + scheduled_time) + except Exception: # pylint: disable=broad-except + self.logger.exception("update worker: unexpected error for %s", + plugin_id) + finally: + with self._pending_lock: + self._pending_updates.discard(plugin_id) + + def stop_update_worker(self, timeout: float = 5.0) -> None: + """Signal the worker to exit (used by cleanup; thread is a daemon).""" + if self._update_worker is not None and self._update_worker.is_alive(): + self._update_queue.put(None) + self._update_worker.join(timeout=timeout) + + def _execute_update_now(self, plugin_id: str, plugin_instance: Any, + scheduled_time: float) -> None: + """The (unchanged) update execution block: executor + bookkeeping. + + Caller is responsible for having set RUNNING state and, on the async + path, for holding the plugin's lock. plugin_executor's internal + thread.join(timeout) blocks only the calling thread; on timeout the + lingering daemon update-thread keeps running unkillable — exactly the + pre-async behavior, documented here so nobody "fixes" it into a + thread leak hunt. + """ + try: + # Use PluginExecutor for safe execution + success = False + if self.resource_monitor: + # If resource monitor exists, wrap the call + def monitored_update(): + self.resource_monitor.monitor_call(plugin_id, plugin_instance.update) + # SimpleNamespace stores `update` as an *instance* + # attribute, so attribute lookup returns the plain + # function object as-is. A dynamically-built class + # (`type(..., {'update': monitored_update})`) instead + # stores it as a *class* attribute, which the + # descriptor protocol turns into a bound method on + # access -- silently prepending the instance as an + # implicit first argument to a function that takes + # none, raising "monitored_update() takes 0 + # positional arguments but 1 was given" on every call. + success = self.plugin_executor.execute_update( + types.SimpleNamespace(update=monitored_update), + plugin_id + ) + else: + success = self.plugin_executor.execute_update(plugin_instance, plugin_id) + + if success: + self.plugin_last_update[plugin_id] = scheduled_time + self.state_manager.record_update(plugin_id) + # Update state back to ENABLED + self.state_manager.set_state(plugin_id, PluginState.ENABLED) + # Record success + if self.health_tracker: + self.health_tracker.record_success(plugin_id) + else: + self._record_update_failure(plugin_id) + except Exception as exc: # pylint: disable=broad-except + self.logger.exception("Error updating plugin %s: %s", plugin_id, exc) + self._record_update_failure(plugin_id, exc=exc) def update_all_plugins(self) -> None: """ diff --git a/test/test_async_plugin_updates.py b/test/test_async_plugin_updates.py new file mode 100644 index 00000000..0d3ebfc5 --- /dev/null +++ b/test/test_async_plugin_updates.py @@ -0,0 +1,197 @@ +"""Tests for asynchronous plugin updates (plugin_manager background worker). + +The invariants that keep this change safe: +1. run_scheduled_updates returns immediately — a slow update() can never + again freeze the render loop (the original defect: 30s scroll freezes). +2. A plugin's update() and display() are NEVER concurrent — the per-plugin + lock makes the old implicit no-overlap guarantee explicit (and, unlike + before, holds it across the post-timeout window too). +3. Failure/timeout bookkeeping is unchanged (same executor, same + _record_update_failure path, same last-update stamping). +4. The kill switch (plugin_system.synchronous_updates) restores the + inline path exactly. +""" + +import os +import sys +import threading +import time + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from src.plugin_system.plugin_manager import PluginManager # noqa: E402 +from src.plugin_system.plugin_state import PluginState # noqa: E402 + + +class SlowPlugin: + """Fake plugin whose update() sleeps and records overlap violations.""" + + def __init__(self, update_seconds=0.5): + self.enabled = True + self.update_seconds = update_seconds + self.update_calls = 0 + self.display_calls = 0 + self.in_update = False + self.overlap_detected = False + + def update(self): + self.in_update = True + self.update_calls += 1 + time.sleep(self.update_seconds) + self.in_update = False + return True + + def display(self, force_clear=False): + if self.in_update: + self.overlap_detected = True + self.display_calls += 1 + return True + + +@pytest.fixture +def pm(tmp_path): + manager = PluginManager(plugins_dir=str(tmp_path), config_manager=None, + display_manager=None, cache_manager=None) + yield manager + manager.stop_update_worker() + + +def _install(pm, plugin, plugin_id="slow-plugin"): + pm.plugins[plugin_id] = plugin + pm._update_interval_cache[plugin_id] = 0.01 # always due + # load_plugin normally registers state; can_execute() gates on it + pm.state_manager.set_state(plugin_id, PluginState.ENABLED) + return plugin_id + + +class TestSchedulerNonBlocking: + def test_run_scheduled_updates_returns_immediately(self, pm): + plugin_id = _install(pm, SlowPlugin(update_seconds=2.0)) + start = time.monotonic() + pm.run_scheduled_updates() + elapsed = time.monotonic() - start + assert elapsed < 0.1, f"scheduler blocked for {elapsed:.2f}s" + # the update actually runs in the background + deadline = time.monotonic() + 5 + while pm.plugins[plugin_id].update_calls == 0 and time.monotonic() < deadline: + time.sleep(0.05) + assert pm.plugins[plugin_id].update_calls == 1 + + def test_no_double_enqueue_while_pending(self, pm): + plugin_id = _install(pm, SlowPlugin(update_seconds=0.8)) + for _ in range(20): + pm.run_scheduled_updates() + time.sleep(0.01) + time.sleep(1.5) # let the single queued update finish + assert pm.plugins[plugin_id].update_calls == 1 + + +class TestUpdateDisplayExclusion: + def test_display_lock_held_during_update(self, pm): + plugin_id = _install(pm, SlowPlugin(update_seconds=0.6)) + pm.run_scheduled_updates() + # give the worker a moment to take the lock and enter update() + deadline = time.monotonic() + 2 + while not pm.plugins[plugin_id].in_update and time.monotonic() < deadline: + time.sleep(0.01) + lock = pm.get_plugin_lock(plugin_id) + assert lock.acquire(blocking=False) is False, \ + "lock must be held while update() runs" + # and released afterwards + deadline = time.monotonic() + 3 + while pm.plugins[plugin_id].in_update and time.monotonic() < deadline: + time.sleep(0.05) + time.sleep(0.1) + assert lock.acquire(blocking=False) is True + lock.release() + + def test_no_overlap_under_hammering_display_loop(self, pm): + """Simulate the render loop's try-lock display pattern at high rate + while updates fire — the plugin itself asserts no overlap.""" + plugin = SlowPlugin(update_seconds=0.15) + plugin_id = _install(pm, plugin) + stop = threading.Event() + + def render_loop(): + while not stop.is_set(): + lock = pm.get_plugin_lock(plugin_id) + if lock.acquire(blocking=False): + try: + plugin.display() + finally: + lock.release() + time.sleep(0.002) + + renderer = threading.Thread(target=render_loop, daemon=True) + renderer.start() + try: + for _ in range(6): + pm.plugin_last_update.pop(plugin_id, None) # force due + pm.run_scheduled_updates() + time.sleep(0.3) + finally: + stop.set() + renderer.join(timeout=2) + assert plugin.update_calls >= 3 + assert plugin.display_calls > 10 + assert plugin.overlap_detected is False + + def test_state_returns_to_enabled_after_update(self, pm): + """RUNNING is set at enqueue (blocks re-entry via can_execute) and + must return to an executable state once the update finishes.""" + plugin_id = _install(pm, SlowPlugin(update_seconds=0.1)) + pm.run_scheduled_updates() + # while queued/running, re-entry is blocked + assert pm.state_manager.can_execute(plugin_id) is False + deadline = time.monotonic() + 3 + while time.monotonic() < deadline: + if (pm.plugins[plugin_id].update_calls + and pm.state_manager.can_execute(plugin_id)): + break + time.sleep(0.05) + assert pm.plugins[plugin_id].update_calls == 1 + assert pm.state_manager.can_execute(plugin_id) is True + + +class TestFailurePaths: + def test_update_failure_routes_through_failure_bookkeeping(self, pm): + class FailingPlugin(SlowPlugin): + def update(self): + self.update_calls += 1 + raise RuntimeError("boom") + + plugin_id = _install(pm, FailingPlugin()) + pm.run_scheduled_updates() + deadline = time.monotonic() + 3 + while pm.plugins[plugin_id].update_calls == 0 and time.monotonic() < deadline: + time.sleep(0.05) + time.sleep(0.2) + # failure stamped so the interval gate holds (no hot retry loop) + assert pm.plugin_last_update.get(plugin_id, 0) > 0 + # lock released after failure + assert pm.get_plugin_lock(plugin_id).acquire(blocking=False) is True + pm.get_plugin_lock(plugin_id).release() + + def test_unloaded_while_queued_is_harmless(self, pm): + plugin_id = _install(pm, SlowPlugin(update_seconds=0.1)) + pm._enqueue_update(plugin_id, time.time()) + del pm.plugins[plugin_id] + time.sleep(0.3) # worker drains the item without crashing + assert pm._update_worker is None or pm._update_worker.is_alive() + + +class TestKillSwitch: + def test_synchronous_mode_blocks_like_before(self, pm): + pm._synchronous_updates = True + plugin_id = _install(pm, SlowPlugin(update_seconds=0.4)) + start = time.monotonic() + pm.run_scheduled_updates() + elapsed = time.monotonic() - start + assert elapsed >= 0.4, "synchronous mode must run inline" + assert pm.plugins[plugin_id].update_calls == 1 + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-v"])) diff --git a/test/test_plugin_system.py b/test/test_plugin_system.py index cfc3f95d..fce98c0f 100644 --- a/test/test_plugin_system.py +++ b/test/test_plugin_system.py @@ -123,6 +123,14 @@ def test_run_scheduled_updates_calls_update_with_resource_monitor( pm.run_scheduled_updates(current_time=time.time()) + # Updates now execute on the background worker (the scheduler + # returns immediately) — wait for completion before asserting. + deadline = time.time() + 5 + while (plugin_instance.update.call_count == 0 + and time.time() < deadline): + time.sleep(0.02) + pm.stop_update_worker() + plugin_instance.update.assert_called_once() assert "test_plugin" in pm.plugin_last_update assert pm.state_manager.get_state("test_plugin") == PluginState.ENABLED