holoviz/panel · #8775
fix: Stop shared ESM watchers when unused
panel/custom.py8 + / 29 −
@@ -1,6 +1,5 @@ from __future__ import annotations -import asyncio import hashlib import importlib import inspect@@ -25,6 +24,7 @@ from .io.resource_spec import resource_spec from .io.resources import component_resource_path from .io.state import state+from .io.watcher import get_path_watcher from .layout.base import Panel from .models import ( AnyWidgetComponent as _BkAnyWidgetComponent,@@ -410,32 +410,9 @@ def _cleanup(self, root: Model | None = None) -> None: child._cleanup(root) super()._cleanup(root) if not self._models and self._watching_esm:- self._watching_esm.set()- if self._watching_esm in state._watch_events:- state._watch_events.remove(self._watching_esm)+ self._watching_esm.unsubscribe(self, '_update_esm') self._watching_esm = None - async def _watch_esm(self):- import watchfiles- path = self._esm_path(compiled=False)- async for changes in watchfiles.awatch(path, stop_event=self._watching_esm):- update = False- for (change, path) in changes:- if change != watchfiles.Change.modified:- continue- # Ensure that the file exists (in case filesystem write is not atomic)- retries = 0- while not os.path.exists(path):- await asyncio.sleep(0.1)- retries += 1- if retries == 5:- break- # Ignore change if file is not written within 0.5 seconds- if retries != 5:- update = True- if update:- self._update_esm()- def _update_esm(self): for ref, (model, _) in self._models.copy().items(): if ref not in state._views:@@ -562,10 +539,12 @@ def _setup_autoreload(self): if not ((config.autoreload or getattr(self, '_debug', False)) and import_available('watchfiles')): return super()._setup_autoreload()- if (self._esm_path(compiled=False) and not self._watching_esm):- self._watching_esm = event = asyncio.Event()- state._watch_events.append(event)- state.execute(self._watch_esm)+ esm_path = self._esm_path(compiled=False)+ if esm_path and not self._watching_esm:+ # Components rendering from the same file, e.g. all components in+ # a shared bundle, subscribe to a single watcher for that path.+ self._watching_esm = watcher = get_path_watcher(esm_path)+ watcher.subscribe(self, '_update_esm') def _get_model( self, doc: Document, root: Model | None = None,panel/io/watcher.pyadded157 + / 0 −
@@ -0,0 +1,157 @@+"""+Shared file watching for the autoreload development loop.++Components that render from the same file, e.g. every component compiled+into a shared ESM bundle, would otherwise each start their own watcher+task. This module hands out one watcher per path and notifies all its+subscribers when the file changes.+"""+from __future__ import annotations++import asyncio+import os+import pathlib+import typing as t+import weakref++from .state import state++_watchers: dict[pathlib.Path, PathWatcher] = {}+++class PathWatcher:+ """+ Watches a single path on behalf of any number of subscribers.++ Subscribers are held weakly, so subscribing does not keep a component+ alive; each is notified by calling the method it subscribed with.+ """++ def __init__(self, path: pathlib.Path):+ self.path = path+ self._subscribers: weakref.WeakKeyDictionary[t.Any, set[str]] = weakref.WeakKeyDictionary()+ self._stop_event: asyncio.Event | None = None+ self._task: asyncio.Task | None = None++ def subscribe(self, obj: t.Any, method: str) -> None:+ """+ Subscribes an object to changes of the watched path.++ Parameters+ ----------+ obj: Any+ The object to notify, held weakly.+ method: str+ Name of the method to invoke on the object.+ """+ self._subscribers.setdefault(obj, set()).add(method)+ self._ensure_running()++ def unsubscribe(self, obj: t.Any, method: str | None = None) -> None:+ """+ Unsubscribes an object, either from a single method or entirely.+ """+ if method is None:+ self._subscribers.pop(obj, None)+ else:+ methods = self._subscribers.get(obj)+ if methods is None:+ return+ methods.discard(method)+ if not methods:+ self._subscribers.pop(obj, None)+ if not self._subscribers:+ if _watchers.get(self.path) is self:+ del _watchers[self.path]+ if self._stop_event is not None:+ self._stop_event.set()++ @property+ def running(self) -> bool:+ return self._task is not None and not self._task.done()++ def _ensure_running(self) -> None:+ if self.running:+ return+ try:+ loop = asyncio.get_running_loop()+ except RuntimeError:+ # Without a running loop there is nothing to schedule the+ # watcher on, e.g. when rendering to a static file.+ return+ self._stop_event = asyncio.Event()+ # Registering the event lets the server stop the watcher on shutdown+ # alongside its own autoreload watchers.+ state._watch_events.append(self._stop_event)+ # Deliberately not scheduled via state.execute, which would bind the+ # task to whichever Document happens to be rendering; the watcher is+ # shared by all sessions and must outlive any one of them.+ self._task = loop.create_task(self._watch())++ async def _watch(self) -> None:+ import watchfiles+ stop_event = self._stop_event+ try:+ async for changes in watchfiles.awatch(self.path, stop_event=stop_event):+ if await self._is_complete_write(changes, watchfiles):+ self._notify()+ finally:+ if stop_event in state._watch_events:+ state._watch_events.remove(stop_event)+ self._stop_event = None+ self._task = None++ async def _is_complete_write(self, changes, watchfiles) -> bool:+ """+ Whether any change represents a finished write. Bundlers write+ non-atomically, so a modification event may arrive while the file+ is still missing or only partially written. A write that replaces+ the file atomically is reported as a deletion followed by an+ addition, so additions count as a new version while a deletion on+ its own does not.+ """+ updated = False+ for change, path in changes:+ if change is watchfiles.Change.deleted:+ continue+ for _ in range(5):+ if os.path.exists(path):+ updated = True+ break+ await asyncio.sleep(0.1)+ return updated++ def _notify(self) -> None:+ from ..config import config+ for obj, methods in list(self._subscribers.items()):+ for method in list(methods):+ try:+ getattr(obj, method)()+ except Exception as e:+ # The watcher is shared, so a failing subscriber must not+ # tear down reloading for all the others.+ if config.exception_handler:+ config.exception_handler(e)+ else:+ state.log(+ f'{type(obj).__name__}.{method} raised {e!r} while '+ f'handling a change to {self.path}.', level='error'+ )+++def get_path_watcher(path: str | os.PathLike) -> PathWatcher:+ """+ Returns the process-wide watcher for the given path, creating it on+ first use.+ """+ path = pathlib.Path(path).absolute()+ if path not in _watchers:+ _watchers[path] = PathWatcher(path)+ return _watchers[path]+++def current_path_watchers() -> dict[pathlib.Path, PathWatcher]:+ """+ Returns the watchers that have been created so far.+ """+ return dict(_watchers)panel/tests/io/test_watcher.pyadded236 + / 0 −
@@ -0,0 +1,236 @@+import asyncio+import gc+import pathlib++import pytest++from panel.io.state import state+from panel.io.watcher import (+ PathWatcher, current_path_watchers, get_path_watcher,+)+from panel.util.checks import import_available+++class Subscriber:++ def __init__(self):+ self.calls = 0+ self.notified = asyncio.Event()++ def _update(self):+ self.calls += 1+ self.notified.set()+++class FailingSubscriber(Subscriber):++ def _update(self):+ super()._update()+ raise ValueError('nope')+++@pytest.fixture(autouse=True)+def clean_watchers(monkeypatch):+ from panel.io import watcher++ # Native filesystem events are not always delivered in CI containers+ # or sandboxes, so poll instead of relying on the platform backend.+ monkeypatch.setenv('WATCHFILES_FORCE_POLLING', '1')+ watcher._watchers.clear()+ yield+ for path_watcher in watcher._watchers.values():+ if path_watcher._stop_event is not None:+ path_watcher._stop_event.set()+ watcher._watchers.clear()+ state._watch_events.clear()+++def test_get_path_watcher_is_shared_per_path(tmp_path):+ path = tmp_path / 'bundle.js'+ watcher = get_path_watcher(path)+ assert get_path_watcher(path) is watcher+ assert get_path_watcher(str(path)) is watcher+ assert get_path_watcher(tmp_path / 'other.js') is not watcher+ assert set(current_path_watchers()) == {path, tmp_path / 'other.js'}+++def test_subscribe_and_unsubscribe(tmp_path):+ path = tmp_path / 'bundle.js'+ watcher = get_path_watcher(path)+ sub = Subscriber()++ watcher.subscribe(sub, '_update')+ watcher.subscribe(sub, '_update')+ assert list(watcher._subscribers) == [sub]+ assert watcher._subscribers[sub] == {'_update'}++ watcher.unsubscribe(sub, '_update')+ assert list(watcher._subscribers) == []+ assert path not in current_path_watchers()+++def test_subscribers_are_held_weakly(tmp_path):+ watcher = get_path_watcher(tmp_path / 'bundle.js')+ sub = Subscriber()+ watcher.subscribe(sub, '_update')++ del sub+ gc.collect()++ assert list(watcher._subscribers) == []+++def test_notify_calls_all_subscribers(tmp_path):+ watcher = get_path_watcher(tmp_path / 'bundle.js')+ subs = [Subscriber() for _ in range(3)]+ for sub in subs:+ watcher.subscribe(sub, '_update')++ watcher._notify()++ assert all(sub.calls == 1 for sub in subs)+++def test_notify_isolates_failing_subscriber(tmp_path):+ watcher = get_path_watcher(tmp_path / 'bundle.js')+ failing, working = FailingSubscriber(), Subscriber()+ watcher.subscribe(failing, '_update')+ watcher.subscribe(working, '_update')++ watcher._notify()++ assert failing.calls == 1+ assert working.calls == 1+++def test_notify_routes_error_to_exception_handler(tmp_path):+ from panel.config import config++ watcher = get_path_watcher(tmp_path / 'bundle.js')+ failing = FailingSubscriber()+ watcher.subscribe(failing, '_update')+ errors = []++ with config.set(exception_handler=errors.append):+ watcher._notify()++ assert len(errors) == 1+ assert isinstance(errors[0], ValueError)+++def test_ensure_running_without_loop_is_noop(tmp_path):+ watcher = get_path_watcher(tmp_path / 'bundle.js')+ watcher.subscribe(Subscriber(), '_update')++ assert not watcher.running+++@pytest.mark.skipif(not import_available('watchfiles'), reason='watchfiles is not installed')+async def test_watcher_notifies_on_file_change(tmp_path):+ path = tmp_path / 'bundle.js'+ path.write_text('export default {}')++ watcher = get_path_watcher(path)+ sub = Subscriber()+ watcher.subscribe(sub, '_update')++ assert watcher.running+ assert watcher._stop_event in state._watch_events++ # watchfiles snapshots the path asynchronously, so a change made before+ # it is watching goes unnoticed; keep writing until it is picked up.+ for version in range(20):+ path.write_text(f'export default {{version: {version}}}')+ try:+ await asyncio.wait_for(sub.notified.wait(), timeout=0.5)+ break+ except TimeoutError:+ continue++ assert sub.calls >= 1+++@pytest.mark.skipif(not import_available('watchfiles'), reason='watchfiles is not installed')+async def test_watcher_task_is_shared(tmp_path):+ path = tmp_path / 'bundle.js'+ path.write_text('export default {}')++ watcher = get_path_watcher(path)+ first, second = Subscriber(), Subscriber()+ watcher.subscribe(first, '_update')+ task = watcher._task++ watcher.subscribe(second, '_update')++ assert watcher._task is task+ assert len([e for e in state._watch_events if e is watcher._stop_event]) == 1+++@pytest.mark.skipif(not import_available('watchfiles'), reason='watchfiles is not installed')+async def test_unsubscribe_stops_idle_watcher(tmp_path):+ path = tmp_path / 'bundle.js'+ path.write_text('export default {}')+ watcher = get_path_watcher(path)+ first, second = Subscriber(), Subscriber()+ watcher.subscribe(first, '_update')+ watcher.subscribe(second, '_update')+ task = watcher._task+ stop_event = watcher._stop_event++ watcher.unsubscribe(first)+ assert watcher.running+ assert get_path_watcher(path) is watcher++ watcher.unsubscribe(second, '_update')+ assert path not in current_path_watchers()+ assert stop_event.is_set()+ await task+ assert stop_event not in state._watch_events+ assert not watcher.running+ assert get_path_watcher(path) is not watcher+++@pytest.mark.skipif(not import_available('watchfiles'), reason='watchfiles is not installed')+async def test_stopping_watcher_deregisters_event(tmp_path):+ path = tmp_path / 'bundle.js'+ path.write_text('export default {}')++ watcher = get_path_watcher(path)+ sub = Subscriber()+ watcher.subscribe(sub, '_update')+ stop_event = watcher._stop_event++ stop_event.set()+ await watcher._task++ assert stop_event not in state._watch_events+ assert not watcher.running+ assert watcher._stop_event is None+++@pytest.mark.skipif(not import_available('watchfiles'), reason='watchfiles is not installed')+def test_esm_components_share_a_watcher(document, comm, tmp_path):+ from panel.config import config+ from panel.custom import JSComponent++ esm = tmp_path / 'CustomShared.js'+ esm.write_text('export function render() {}')++ class CustomShared(JSComponent):+ _esm = pathlib.Path(esm)++ with config.set(autoreload=True):+ first, second = CustomShared(), CustomShared()+ root = first.get_root(document, comm)+ second.get_root(document, comm)++ watcher = get_path_watcher(esm)+ assert isinstance(watcher, PathWatcher)+ assert current_path_watchers() == {esm: watcher}+ assert set(watcher._subscribers) == {first, second}+ assert first._watching_esm is watcher++ first._cleanup(root)++ assert set(watcher._subscribers) == {second}+ assert first._watching_esm is None