diff --git a/param/parameterized.py b/param/parameterized.py index 6c3ebded..b5f2ecb0 100644 --- a/param/parameterized.py +++ b/param/parameterized.py @@ -2579,10 +2579,11 @@ def __getattr__(self_, attr: str) -> t.Any: ns_type = type(self_) - # The cached parameters are read from _param__private directly rather - # than via the _cls_parameters property. If that property raised an - # AttributeError we would be called to handle it and recurse, so it is - # only invoked as a descriptor, i.e. without attribute lookup on self_. + # Read the cached parameters from _param__private rather than from the + # _cls_parameters property: an AttributeError raised by that property is + # dispatched to this method, so accessing it as an attribute here would + # recurse. When the cache is empty the property is invoked directly as a + # descriptor instead, which bypasses that dispatch. params = cls._param__private.params if not params: params = _find_descriptor(ns_type, '_cls_parameters').__get__(self_, ns_type) @@ -2742,7 +2743,10 @@ def _sync_refs(self_, *events): if new_val is Skip or new_val is Undefined: continue elif is_async: - async_executor(partial(self_._async_ref, pname, t.cast("t.Awaitable[t.Any]", new_val))) + generation = self_._schedule_async_ref(pname) + async_executor(partial( + self_._async_ref, pname, t.cast("t.Awaitable[t.Any]", new_val), generation + )) continue updates[pname] = new_val @@ -2763,39 +2767,85 @@ def _resolve_ref(self_, pobj: Parameter, value: t.Any): except Skip: value = Undefined if is_async and pobj.name: - async_executor(partial(self_._async_ref, pobj.name, t.cast("t.Awaitable[t.Any]", value))) + generation = self_._schedule_async_ref(pobj.name) + async_executor(partial( + self_._async_ref, pobj.name, t.cast("t.Awaitable[t.Any]", value), generation + )) value = None return ref, deps, value, is_async - async def _async_ref(self_, pname: str, awaitable: t.Awaitable[t.Any]): + def _schedule_async_ref(self_, pname: str) -> int: + """ + Record that an asynchronous reference resolution is about to be + scheduled and return the generation identifying it. + + The generation is bumped synchronously, before the task is handed to + the executor, so that the reference reads as unsettled from the moment + it is superseded rather than only once the task starts running. + """ + if self_.self is None: + return 0 + private = self_.self._param__private + private.async_ref_scheduled[pname] += 1 + return private.async_ref_scheduled[pname] + + def _settle_async_ref(self_, pname: str, generation: int): + """ + Record that the resolution identified by ``generation`` produced a + value, or gave up on producing one. + + A superseded task settling late must not mark the reference settled for + the generation that superseded it, so the generation is recorded rather + than cleared. Generators settle on every value they yield, so a + reference is unsettled only until its next value arrives, not until the + generator is exhausted. + """ + if self_.self is None or not generation: + return + settled = self_.self._param__private.async_ref_settled + settled[pname] = max(settled[pname], generation) + + def _awaiting_ref(self_, pname: str) -> bool: + """Whether an asynchronous reference has not yet produced a value.""" + if self_.self is None: + return False + private = self_.self._param__private + return private.async_ref_scheduled[pname] != private.async_ref_settled[pname] + + async def _async_ref(self_, pname: str, awaitable: t.Awaitable[t.Any], generation: int = 0): if self_.self is None: return if not self_.self._param__private.initialized: - async_executor(partial(self_._async_ref, pname, awaitable)) + async_executor(partial(self_._async_ref, pname, awaitable, generation)) return import asyncio current_task = asyncio.current_task() running_task = self_.self._param__private.async_refs.get(pname) - if running_task is None: + if running_task is not current_task: + if running_task is not None: + running_task.cancel() self_.self._param__private.async_refs[pname] = current_task - elif current_task is not running_task: - self_.self._param__private.async_refs[pname].cancel() try: if isinstance(awaitable, types.AsyncGeneratorType): async for new_obj in awaitable: with _syncing(self_.self, (pname,)): self_.update({pname: new_obj}) + self_._settle_async_ref(pname, generation) else: with _syncing(self_.self, (pname,)): try: self_.update({pname: await awaitable}) except Skip: pass + self_._settle_async_ref(pname, generation) finally: - # Ensure we clean up but only if the task matches the current task - if self_.self._param__private.async_refs.get(pname) is current_task: - del self_.self._param__private.async_refs[pname] + self_._settle_async_ref(pname, generation) + # Ensure we clean up but only if the task matches the current task, + # i.e. only the resolution that still owns the reference clears it. + async_refs = self_.self._param__private.async_refs + if pname in async_refs and async_refs[pname] is current_task: + del async_refs[pname] @classmethod def _changed(cls, event): @@ -5638,6 +5688,8 @@ class _InstancePrivate: 'dynamic_watchers', 'params', 'async_refs', + 'async_ref_scheduled', + 'async_ref_settled', 'refs', 'ref_watchers', 'syncing', @@ -5651,6 +5703,8 @@ class _InstancePrivate: dynamic_watchers: defaultdict[str, list[Watcher]] params: dict[str, Parameter] async_refs: dict[str, t.Any] + async_ref_scheduled: defaultdict[str, int] + async_ref_settled: defaultdict[str, int] refs: dict[str, t.Any] ref_watchers: list[tuple[tuple[str, ...], Watcher]] syncing: set[str] @@ -5681,6 +5735,8 @@ def __init__( } self.ref_watchers = [] self.async_refs = {} + self.async_ref_scheduled = defaultdict(int) + self.async_ref_settled = defaultdict(int) self.parameters_state = parameters_state self.dynamic_watchers = defaultdict(list, dynamic_watchers or ()) self.params = {} if params is None else params diff --git a/tests/testrefs.py b/tests/testrefs.py index 8241e75b..fdf877c0 100644 --- a/tests/testrefs.py +++ b/tests/testrefs.py @@ -304,6 +304,55 @@ async def gen_strings2(): assert task2 is async_task2 assert p._param__private.async_refs['string'] is task2 +async def test_async_ref_cancelled_on_dependency_change(): + started, finished = [], [] + + class Source(param.Parameterized): + x = param.Number(default=0) + + async def slow(i): + started.append(i) + await asyncio.sleep(0.1) + finished.append(i) + return str(i) + + source = Source() + p = Parameters(string=bind(slow, source.param.x)) + for i in (1, 2, 3): + source.x = i + await asyncio.sleep(0.01) + + await wait_for_value(p, 'string', '3', timeout=1) + assert started == [0, 1, 2, 3] + assert finished == [3] + + await asyncio.sleep(0.05) + assert 'string' not in p._param__private.async_refs + +async def test_async_generator_ref_cancelled_on_dependency_change(): + emitted = [] + + class Source(param.Parameterized): + x = param.Number(default=0) + + async def gen(i): + while True: + emitted.append(i) + yield str(i) + await asyncio.sleep(0.01) + + source = Source() + p = Parameters(string=bind(gen, source.param.x)) + await asyncio.sleep(0.05) + source.x = 1 + await asyncio.sleep(0.05) + source.x = 2 + await wait_for_value(p, 'string', '2', timeout=0.5) + + emitted.clear() + await asyncio.sleep(0.05) + assert set(emitted) == {2} + async def test_generator_ref_cancelled(): threads = [] def gen_strings1():