From 06f6ce176b562292e41de107dc9b690d6f58107b Mon Sep 17 00:00:00 2001 From: Maarten Breddels Date: Wed, 30 Sep 2026 09:20:32 +0200 Subject: [PATCH] Render requests from other threads never wait and are never lost A state change on a thread that is not rendering used to call render() directly, or only set a flag when another thread was rendering. That gave three kinds of bugs when a background thread (a task that reports progress) changes state while the kernel thread renders: a change that is never rendered, a false "Recursive render detected" raised from the user's own setter, and "Too many renders" counting other threads' updates. And a setter that waited for the render lock could deadlock with a renderer that needs a lock the setter holds. Now a request sets a flag and only tries the lock; every holder checks the flag again after it released the lock, so the request is rendered by the current holder or by the requester. A thread that keeps writing hands its requests to a helper thread after a budget, so the rendering thread is not held up forever. Only a render's own re-renders count towards the limit. Element.create_lock was held while a widget was constructed, and construction opens a comm: in a server that send can wait for a slow client, and one client that stopped reading stalled widget creation for every user. The widget recording it protected is now per thread, so the lock is no longer held during construction. close() sets _closing before it waits, returns for a concurrent second close(), raises instead of hanging when called from its own render, and can be retried after an interrupted attempt. Co-Authored-By: Claude Opus 5.5 (1M context) --- reacton/_threads_test_utils.py | 202 ++++++++ reacton/core.py | 697 +++++++++++++++++----------- reacton/threads_close_test.py | 374 +++++++++++++++ reacton/threads_create_lock_test.py | 110 +++++ reacton/threads_handoff_test.py | 381 +++++++++++++++ reacton/threads_limit_test.py | 233 ++++++++++ reacton/threads_robustness_test.py | 283 +++++++++++ 7 files changed, 2009 insertions(+), 271 deletions(-) create mode 100644 reacton/_threads_test_utils.py create mode 100644 reacton/threads_close_test.py create mode 100644 reacton/threads_create_lock_test.py create mode 100644 reacton/threads_handoff_test.py create mode 100644 reacton/threads_limit_test.py create mode 100644 reacton/threads_robustness_test.py diff --git a/reacton/_threads_test_utils.py b/reacton/_threads_test_utils.py new file mode 100644 index 0000000..5aad2fb --- /dev/null +++ b/reacton/_threads_test_utils.py @@ -0,0 +1,202 @@ +"""Shared helpers for the render hand-off tests. + +Every test forces its interleaving: a thread is paused at a fixed point with an Event (a +component body, an effect, a widget constructor, one of reacton's own logger.info calls, or one +settrace pause at a source line). Only the stress tests depend on the scheduler. +Calls that can hang on a broken tree run on daemon Worker threads with a timeout, so a failing +test fails instead of hanging the run. +""" + +import inspect +import logging +import sys +import threading +import time +import traceback + +import pytest + +from reacton import core + +TIMEOUT = 5.0 +SHORT = 1.0 # "did this call block?" (it returns in microseconds when it does not) + + +class LogHook(logging.Handler): + """Call a callback at one of reacton's logger.info calls, on a given thread, once. + + handle() is overridden: logging.Handler.handle() takes the handler lock, and a thread + paused in emit() would then block every other thread's log call. + """ + + def __init__(self): + super().__init__(level=logging.DEBUG) + self.rules = [] + + def on(self, thread, prefix, callback): + self.rules.append([thread, prefix, callback, False]) + + def handle(self, record): + self.emit(record) + return True + + def emit(self, record): + msg = record.msg if isinstance(record.msg, str) else str(record.msg) + for rule in self.rules: + thread, prefix, callback, fired = rule + if not fired and threading.current_thread() is thread and msg.startswith(prefix): + rule[3] = True + callback() + + +class EveryTimeHook(logging.Handler): + """Call callback() at every reacton log record that starts with one of prefixes, on any + thread except the one named `skip`.""" + + def __init__(self, prefixes, callback, skip): + super().__init__(level=logging.DEBUG) + self.prefixes, self.callback, self.skip = tuple(prefixes), callback, skip + + def handle(self, record): + self.emit(record) + return True + + def emit(self, record): + msg = record.msg if isinstance(record.msg, str) else str(record.msg) + if msg.startswith(self.prefixes) and threading.current_thread().name != self.skip: + self.callback() + + +def install(handler): + logger = logging.getLogger("reacton") + saved = logger.level, logger.propagate + logger.addHandler(handler) + logger.setLevel(logging.INFO) + logger.propagate = False + + def uninstall(): + logger.removeHandler(handler) + logger.setLevel(saved[0]) + logger.propagate = saved[1] + + return uninstall + + +@pytest.fixture +def log_hook(): + hook = LogHook() + uninstall = install(hook) + yield hook + uninstall() + + +@pytest.fixture +def helper_errors(): + """Exceptions raised on other threads (a helper thread renders with handle_error=False).""" + errors = [] + if hasattr(threading, "excepthook"): + old = threading.excepthook + threading.excepthook = lambda args: errors.append(args.exc_value) + try: + yield errors + finally: + threading.excepthook = old + return + + old_run = threading.Thread.run + + def run(thread): + try: + old_run(thread) + except BaseException as e: + errors.append(e) + raise + + threading.Thread.run = run # type: ignore[method-assign] + try: + yield errors + finally: + threading.Thread.run = old_run # type: ignore[method-assign] + + +class Worker(threading.Thread): + def __init__(self, target, name): + super().__init__(name=name, daemon=True) + self._target_f = target + self.error = None + self.error_tb = "" + self.result = None + + def run(self): + try: + self.result = self._target_f() + except BaseException as e: # noqa + self.error = e + self.error_tb = traceback.format_exc() + + +def call_in_thread(f, name, timeout=SHORT): + """Run f on a new thread; return (finished within timeout, worker).""" + t = Worker(f, name) + t.start() + t.join(timeout) + return not t.is_alive(), t + + +def stacks(): + frames = sys._current_frames() + return "\n".join(f"--- {t.name}\n" + "".join(traceback.format_stack(frames[t.ident])[-6:]) for t in threading.enumerate() if t.ident in frames) + + +def text(box): + return box.children[0].description + + +def wait_until(predicate, timeout): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return predicate() + + +class Injected(Exception): + pass + + +def render_loop_function(): + """The function that holds the render loop (RC.render, or _render_locked when present).""" + RC = core._RenderContext + return getattr(RC, "_render_locked", None) or RC.render + + +def source_line(fn, needle, after=None): + lines, first = inspect.getsourcelines(fn) + start = 0 if after is None else next(i for i, line in enumerate(lines) if after in line) + return first + next(i for i, line in enumerate(lines) if i >= start and needle in line) + + +def pause_at_line(fn, lineno, on_pause, after_line=False): + """A sys.settrace function that calls on_pause() once, when fn executes line `lineno` + (after_line=True: at the first line event after `lineno` ran).""" + code = fn.__code__ + state = {"seen": False, "fired": False} + + def local_trace(frame, event, arg): + if event == "line" and not state["fired"]: + if after_line: + if state["seen"] and frame.f_lineno != lineno: + state["fired"] = True + on_pause() + elif frame.f_lineno == lineno: + state["seen"] = True + elif frame.f_lineno == lineno: + state["fired"] = True + on_pause() + return local_trace + + def global_trace(frame, event, arg): + return local_trace if frame.f_code is code else None + + return global_trace diff --git a/reacton/core.py b/reacton/core.py index d4952ce..8b17c40 100644 --- a/reacton/core.py +++ b/reacton/core.py @@ -88,34 +88,47 @@ def get(): # via the widget constructed hook: diffing the global widgets dict per widget # creation is O(total widgets), and solara replaces that dict with a context # aware mapping we should not depend on. -_construction_recording: Optional[List["widgets.Widget"]] = None +# The hook runs on the thread that constructs the widget, so the recording is per +# thread (local.construction_recording): no lock is held while a widget is +# constructed. Construction opens a comm, and in a server that send can wait for a +# slow client; a process wide lock there stalled every other user (solara P3). _chained_construction_callback: Optional[Callable] = None def _record_constructed_widget(widget: "widgets.Widget"): - if _construction_recording is not None: - _construction_recording.append(widget) + recording = getattr(local, "construction_recording", None) + if recording is not None: + recording.append(widget) if _chained_construction_callback is not None: _chained_construction_callback(widget) -def _start_recording_constructed(recording: List["widgets.Widget"]): - global _construction_recording, _chained_construction_callback - current = getattr(widgets.Widget, "_widget_construction_callback", None) - if current is not _record_constructed_widget: - # first time, or someone else registered a callback after us: chain it - _chained_construction_callback = current - widgets.Widget.on_widget_constructed(_record_constructed_widget) - _construction_recording = recording +def _start_recording_constructed(recording: List["widgets.Widget"]) -> Optional[List["widgets.Widget"]]: + """Record the widgets constructed on this thread. Returns the recording it replaces + (a widget whose constructor renders reacton nests one _create_widget in another).""" + global _chained_construction_callback + with Element.create_lock: # a leaf lock now: only guards installing the hook + current = getattr(widgets.Widget, "_widget_construction_callback", None) + if current is not _record_constructed_widget: + # first time, or someone else registered a callback after us: chain it + _chained_construction_callback = current + widgets.Widget.on_widget_constructed(_record_constructed_widget) + previous = getattr(local, "construction_recording", None) + local.construction_recording = recording + return previous -def _stop_recording_constructed(): - global _construction_recording - _construction_recording = None +def _stop_recording_constructed(previous: Optional[List["widgets.Widget"]] = None): + local.construction_recording = previous _last_rc = None # used for testing local = threading.local() +# One call (a state change, a batch exit, render()) takes the render lock at most this many +# times for render requests of other threads; then a helper thread renders the rest (see +# _drain_render_requests). +_MAX_FOREIGN_RENDERS = 50 +_NOT_TAKEN = object() # _hold_and_render: another thread holds the render lock T = TypeVar("T") U = TypeVar("U") W = TypeVar("W") # used for widgets @@ -334,6 +347,7 @@ class Element(Generic[W]): # to make every unique on_value callback to a unique wrapper # so that we can remove the listeners _callback_wrappers: Dict[Tuple[str, str, Callable], Callable] = {} + # a leaf lock: only guards installing the widget construction hook (not held while a widget is constructed) create_lock: ContextManager = threading.Lock() _shared = False @@ -499,32 +513,32 @@ def _create_widget(self, kwargs): # we can't use our own kwarg, since that contains elements, not widgets kwargs, listeners = self._split_kwargs(kwargs) assert isinstance(self.component, ComponentWidget) - # The recording is global state, so we need a lock. - with self.create_lock: - rc = get_render_context(required=True) - recorded: List[widgets.Widget] = [] - _start_recording_constructed(recorded) + # No lock around the construction: the recording is per thread (see + # _record_constructed_widget), and the construction may block in a send. + rc = get_render_context(required=True) + recorded: List[widgets.Widget] = [] + previous_recording = _start_recording_constructed(recorded) + try: try: - try: - widget = self.component.widget(**kwargs) - hold_trait_notifications = widget.hold_trait_notifications - - @contextlib.contextmanager - def hold_trait_notifications_extra(*args, **kwargs): - with rc, hold_trait_notifications(*args, **kwargs): - yield - - widget.hold_trait_notifications = hold_trait_notifications_extra - - if self._meta: - widget._react_meta = dict(self._meta) - except Exception as e: - raise RuntimeError(f"Could not create widget {self.component.widget} with {kwargs}") from e - for name, callback in listeners.items(): - if callback is not None: - self._add_widget_event_listener(widget, name, callback) - finally: - _stop_recording_constructed() + widget = self.component.widget(**kwargs) + hold_trait_notifications = widget.hold_trait_notifications + + @contextlib.contextmanager + def hold_trait_notifications_extra(*args, **kwargs): + with rc, hold_trait_notifications(*args, **kwargs): + yield + + widget.hold_trait_notifications = hold_trait_notifications_extra + + if self._meta: + widget._react_meta = dict(self._meta) + except Exception as e: + raise RuntimeError(f"Could not create widget {self.component.widget} with {kwargs}") from e + for name, callback in listeners.items(): + if callback is not None: + self._add_widget_event_listener(widget, name, callback) + finally: + _stop_recording_constructed(previous_recording) widgets_dict = _get_widgets_dict() orphans = {w.model_id for w in recorded if w is not widget and w.comm is not None and w.model_id in widgets_dict} return widget, orphans @@ -1344,7 +1358,16 @@ def __init__(self, element: Element, container: widgets.Widget = None, children_ self._lock_thread = cast(Optional[threading.Thread], None) self.last_root_widget: widgets.Widget = None self._is_rendering = False + # a re-render asked for by the thread that holds thread_lock (its own setters, effects, + # exception paths); only these count towards the "too many renders" limit self._rerender_needed = False + # a render asked for by a thread that does not hold thread_lock. It never waits for the + # lock: the holder renders it, because every holder checks this again after it released + # the lock (see _request_render and _drain_render_requests) + self._render_requested = False + # the root element to render: update() writes it on any thread, then requests a render; + # every pass reads it into self.element after it cleared the marks + self._element_requested: Optional[Element] = element # the reasons are only read for the "too many renders" error message, and a reason # holds the previous and next state value: keeping all of them kept every old state # value alive until close(). REACTON_RERENDER_REASONS keeps more, for debugging. @@ -1352,6 +1375,8 @@ def __init__(self, element: Element, container: widgets.Widget = None, children_ self._rerender_needed_reasons: Deque[RerenderReason] = collections.deque(maxlen=max_reasons) self.thread_lock = threading.Lock() self._closing = False + self._close_claim = threading.Lock() + self._teardown_done = False self.tracebacks: List[TracebackType] = [] self.handle_error = handle_error self.reconsolidating = False @@ -1388,9 +1413,8 @@ def __enter__(self): def __exit__(self, exc_type, exc_value, traceback): counter = self._batch_counter.decrement() if counter == 0: - logger.info("finishing batch render (%s)", "needs rerender" if self._rerender_needed else "no rerender needed") - if self._rerender_needed: - self._possible_rerender() + logger.info("finishing batch render (%s)", "needs rerender" if self._render_requested else "no rerender needed") + self._drain_render_requests() # requests made during the batch were only marked def find(self, cls: Type[W] = ipywidgets.Widget, **matches): from .find import finder @@ -1400,50 +1424,66 @@ def find(self, cls: Type[W] = ipywidgets.Widget, **matches): _find = find # for backward compatibility def close(self): - with self.thread_lock: + if self._teardown_done: + return + if self._lock_thread is threading.current_thread(): + # thread_lock is not reentrant: waiting for it here would wait for ourselves + raise RuntimeError("close() called from a render of the same render context (avoided deadlock)") + if not self._close_claim.acquire(blocking=False): + return # another close() is running, possibly from one of its cleanups + try: + # Before we wait for the lock: a render on another thread stops re-rendering and + # requests are dropped from now on, so we wait for at most the rest of the pass in flight. self._closing = True - # snapshot the component contexts before _remove_element detaches them from - # their parents: detached contexts would escape the teardown below while the - # setter/handler closures in their state still reference them and us - all_contexts: List[ComponentContext] = [] - - def collect(context: ComponentContext): - all_contexts.append(context) - for child in list(context.children.values()) + list(context.children_next.values()): - collect(child) - - collect(self.context_root) - logger.info("Removing elements...") - self._remove_element(self.element, default_key="/", parent_key=ROOT_KEY) - logger.info("Removing elements done.") - assert self.context is self.context_root - # everything below used to run outside the lock: a render() that was - # blocked on the lock could then interleave with this teardown and - # find self.context None mid-render (AssertionError in - # Element.__exit__, seen in production behind a prompt kernel close) - if self.container: - self.container.close() - if isinstance(self.container, widgets.DOMWidget) and self.container.layout is not None: - self.container.layout.close() - if self._shared_elements: - raise RuntimeError(f"Element not cleaned up: {self._shared_elements}") - if self._orphans: - orphan_widgets = set([_get_widgets_dict()[k] for k in self._orphans]) - raise RuntimeError(f"Orphan widgets not cleaned up for widgets: {orphan_widgets}") - exceptions = [*self.context.exceptions_children, *self.context_root.exceptions_self] - # break the reference cycles through the tree (see _teardown_component_context); - # _closing stays True, making stray setters and event handlers no-ops - for context in all_contexts: - _teardown_component_context(context) - self.context = None - self.context_root = None # type: ignore - # the root element, container and root widget keep the widget tree alive, and - # widgets reference their elements, whose kwargs hold user callbacks - which - # capture use_state setters and therefore this render context - self.element = None # type: ignore - self.container = None - self.last_root_widget = None - self._old_element_ids.clear() + with self.thread_lock: + if self._teardown_done: + return # two close() calls raced past the check above + # snapshot the component contexts before _remove_element detaches them from + # their parents: detached contexts would escape the teardown below while the + # setter/handler closures in their state still reference them and us + all_contexts: List[ComponentContext] = [] + + def collect(context: ComponentContext): + all_contexts.append(context) + for child in list(context.children.values()) + list(context.children_next.values()): + collect(child) + + collect(self.context_root) + logger.info("Removing elements...") + self._remove_element(self.element, default_key="/", parent_key=ROOT_KEY) + logger.info("Removing elements done.") + assert self.context is self.context_root + # everything below used to run outside the lock: a render() that was + # blocked on the lock could then interleave with this teardown and + # find self.context None mid-render (AssertionError in + # Element.__exit__, seen in production behind a prompt kernel close) + if self.container: + self.container.close() + if isinstance(self.container, widgets.DOMWidget) and self.container.layout is not None: + self.container.layout.close() + if self._shared_elements: + raise RuntimeError(f"Element not cleaned up: {self._shared_elements}") + if self._orphans: + orphan_widgets = set([_get_widgets_dict()[k] for k in self._orphans]) + raise RuntimeError(f"Orphan widgets not cleaned up for widgets: {orphan_widgets}") + exceptions = [*self.context.exceptions_children, *self.context_root.exceptions_self] + # break the reference cycles through the tree (see _teardown_component_context); + # _closing stays True, making stray setters and event handlers no-ops + for context in all_contexts: + _teardown_component_context(context) + self.context = None + self.context_root = None # type: ignore + # the root element, container and root widget keep the widget tree alive, and + # widgets reference their elements, whose kwargs hold user callbacks - which + # capture use_state setters and therefore this render context + self.element = None # type: ignore + self.container = None + self.last_root_widget = None + self._old_element_ids.clear() + self._element_requested = None + self._teardown_done = True + finally: + self._close_claim.release() if exceptions: raise exceptions[0] @@ -1579,8 +1619,7 @@ def set_(value): ) else: self._rerender_needed_reasons.append(RerenderReason(reason=f"state changed with key {key}", prev_value=prev_value, next_value=value)) - self._rerender_needed = True - self._possible_rerender() + self._request_render() return set_ @@ -1590,8 +1629,9 @@ def force_update(self): return # a forced update re-walks the whole tree, no subtree skipping self._walk_all = True - if not self._is_rendering: - self.render(self.element, self.container) + if self._is_rendering and self._lock_thread is threading.current_thread(): + return # from our own render pass: unchanged, that pass walks the tree + self._request_render() # before: dropped while another thread rendered def use_effect(self, effect: EffectCallable, dependencies=None): assert self.context is not None @@ -1613,19 +1653,65 @@ def use_effect(self, effect: EffectCallable, dependencies=None): self.context.effect_index += 1 def update(self, element: Element): + if self._closing: + return self._walk_all = True - if self._is_rendering: - self.element = element - self._rerender_needed_reasons.append(RerenderReason(reason="root element changed")) - self._rerender_needed = True - else: - self.render(element, self.container) + # not self.element: the thread that holds the lock may be in the middle of a pass + self._element_requested = element + self._rerender_needed_reasons.append(RerenderReason(reason="root element changed")) + self._request_render() - def _possible_rerender(self): - if not self._is_rendering and self._batch_counter.current() == 0: - self.render(self.element, self.container) - else: - logger.info("No render phase triggered, already rendering") + def _request_render(self): + """Ask for a render. Never waits for thread_lock, and the request is never lost.""" + if self._closing: + return + if self._lock_thread is threading.current_thread(): + if self._is_rendering: + self._rerender_needed = True # our own render loop picks it up (and counts it) + else: + self._render_requested = True # our own drain, after the release, picks it up + return + self._render_requested = True # must come before the try-acquire in the drain + self._drain_render_requests() + + def _drain_render_requests(self, budget: Optional[int] = _MAX_FOREIGN_RENDERS): + """Render while a render is requested, unless another thread holds thread_lock. + + Returning when the lock is taken is safe: every thread that held the lock calls this + after it released the lock, and a requester sets _render_requested before it tries the + lock, so one of the two sees the other. Inside a batch (or while a thread waits in + render(), which counts as a batch) we only mark: that thread renders when it is done. + + budget: how many more times this thread may take the lock for requests before it hands + the rest to a helper thread, so that its own caller (say, an event handler) is not held + up by a thread that keeps changing state. None: no limit. One hold renders the requests + made before it started, and its own re-renders; a request made during the hold ends it + (see _more_passes), and this loop takes the lock again for it. + """ + while self._render_requested and not self._closing and self._batch_counter.current() == 0: + try: + if budget is not None and budget <= 0: + if self._start_render_helper(): + return + budget = None # no threads (pyodide) or interpreter shutdown: render here + if self._hold_and_render(None, self.container, wait=False) is _NOT_TAKEN: + logger.info("No render phase triggered, another thread holds the render lock and will pick it up") + return + if budget is not None: + budget -= 1 + except BaseException: + if self._render_requested and not self._closing: + self._start_render_helper() + raise + + def _start_render_helper(self) -> bool: + logger.info("a helper thread renders the pending render requests") + try: + # no budget: a helper has no caller to hold up + threading.Thread(target=self._drain_render_requests, kwargs={"budget": None}, name="reacton-render", daemon=True).start() + except RuntimeError: + return False + return True def _discard_aborted_pass(self): """Forget everything a render pass staged when it raised before reconciliation. @@ -1657,189 +1743,243 @@ def _discard_aborted_pass(self): self.context = self.context_root def render(self, element: Element, container: widgets.Widget = None): - # render + consolidate - widget = None + """Render element now, and return the root widget. + + The only render entry that waits for thread_lock: it waits for a render (or close) in + progress on another thread. State changes, update() and force_update() go through + _request_render instead, and never wait. + """ if container is None: container = self.container - was_locked = False - if self.thread_lock.locked(): - if self._lock_thread == threading.current_thread(): - raise RuntimeError("Recursive render detected (avoided deadlock), current thread: %r" % threading.current_thread()) - logger.info( - "Render phase still in progress, waiting for mutex to release (locked obtained by %r, we are in thread %r)", - self._lock_thread, - threading.current_thread(), - ) - was_locked = True - with self.thread_lock: - self._lock_thread = threading.current_thread() - if was_locked: - logger.info("Mutex released, continuing render phase") - if self._closing or self.context is None: - # close() won the race for the lock (a disconnect can close the - # kernel while an update was waiting to render): the tree is - # torn down, there is nothing to render into anymore - logger.info("Render requested on a closing/closed render context, ignoring") - return container - prev_rc = getattr(local, "rc", None) - # an exception that escapes while this is True aborted a render pass (see the except below) - in_render_phase = True - try: - local.rc = self - self.element = element - del element - main_render_phase = not self._is_rendering - render_count = self.render_count # make a copy - self._rerender_needed = False - logger.info("Render phase: %r %r of %r", self.render_count, "main" if main_render_phase else "(nested)", self.element) - self.render_count += 1 - self._is_rendering = True - # if we got called recursively, self.context is not the root context - context_prev = self.context - self.context = self.context_root - self.context.exception_handler = False - self.context.exceptions_children = [] - self.context.exceptions_self = [] - self.context.root_element_next = self.element - assert self.context is not None + if self._closing: + logger.info("Render requested on a closing/closed render context, ignoring") + return container + if self._lock_thread is threading.current_thread(): + raise RuntimeError("Recursive render detected (avoided deadlock), current thread: %r" % threading.current_thread()) + try: + widget = self._hold_and_render(element, container, wait=True) + self._drain_render_requests() # requests that came after our last check + except BaseException: + if self._render_requested and not self._closing: + self._start_render_helper() + raise + return widget + def _hold_and_render(self, element: Optional[Element], container: Optional[widgets.Widget], wait: bool): + """Take thread_lock, render until nothing of our own is pending, and release the lock. + + element None: render the root element (the one from update(), if any). Returns + _NOT_TAKEN when wait is False and another thread holds the lock, else the root widget. + The caller must drain after this returns (a request may have come during the hold). + """ + acquired = False + try: + # the acquire is inside the try: an exception right after it (a KeyboardInterrupt, or + # an exception raised by a trace function) must not leak the lock + acquired = self.thread_lock.acquire(blocking=False) + if not acquired and wait: + logger.info( + "Render phase still in progress, waiting for mutex to release (locked obtained by %r, we are in thread %r)", + self._lock_thread, + threading.current_thread(), + ) + # A batch while we wait: other threads leave their render requests to us. + counter = None try: - self._shared_elements_next = set() - self._render(self.element, "/", parent_key=ROOT_KEY) - self.first_render = False - self._walk_all = False - except BaseException: - self._is_rendering = False - raise + counter = self._batch_counter.increment() + if counter == 1: + logger.info("entering batch render") + acquired = self.thread_lock.acquire() + finally: + if counter is not None: + counter = self._batch_counter.decrement() + if counter == 0: + logger.info("finishing batch render (%s)", "needs rerender" if self._render_requested else "no rerender needed") + self._drain_render_requests() + if not acquired: + return _NOT_TAKEN + self._lock_thread = threading.current_thread() + return self._render_locked(element, container) + finally: + if acquired: + self._lock_thread = None # before the release, so the recursion guards see the holder only + self.thread_lock.release() + + def _render_locked(self, element: Optional[Element], container: Optional[widgets.Widget]): + # render + consolidate, with thread_lock held by this thread (see _hold_and_render) + widget = None + if self._closing or self.context is None: + # close() won the race for the lock (a disconnect can close the + # kernel while an update was waiting to render): the tree is + # torn down, there is nothing to render into anymore + logger.info("Render requested on a closing/closed render context, ignoring") + return container + prev_rc = getattr(local, "rc", None) + # an exception that escapes while this is True aborted a render pass (see the except below) + in_render_phase = True + try: + local.rc = self + # consume the requests before this pass reads any state; the root element is state too. + # Only here, at the start of a hold: a request that comes later ends the hold (see + # _more_passes), and the drain after the release takes the lock again for it. + # (Clearing it anywhere else would hide that request from _more_passes, and the + # limit would count it as our own render loop.) + self._render_requested = False + self._rerender_needed = False + if element is not None: + self._element_requested = element # an update() from before this render() is superseded + self.element = self._element_requested # type: ignore # None only after close(), checked above + del element + main_render_phase = not self._is_rendering + render_count = self.render_count # make a copy + logger.info("Render phase: %r %r of %r", self.render_count, "main" if main_render_phase else "(nested)", self.element) + self.render_count += 1 + self._is_rendering = True + # if we got called recursively, self.context is not the root context + context_prev = self.context + self.context = self.context_root + self.context.exception_handler = False + self.context.exceptions_children = [] + self.context.exceptions_self = [] + self.context.root_element_next = self.element + assert self.context is not None - if main_render_phase: - stable = False - render_counts = 0 - while not stable and not self.context_root.exceptions_children: - # we started the rendering loop (main_render_phase is True), so we keep going - # but if an exception bubbled up, we should stop - while self._rerender_needed and not self.context_root.exceptions_children: - if render_counts > 50: - - def format(reason: RerenderReason): - f = f"Reason: {reason.reason}\nValue changed from {reason.prev_value} to {reason.next_value}\n" - if reason.created_stack: - f += f"Created at: {''.join(reason.created_stack)}\n" - if reason.trigger_stack: - f += f"Triggered at: {''.join(reason.trigger_stack)}\n" - return f - - msg = f"Too many renders triggered, your render loop does not stop\nLast reason: {format(self._rerender_needed_reasons[-1])}\n" - if len(self._rerender_needed_reasons) >= 2: - previous = reversed(list(self._rerender_needed_reasons)[:-1]) - msg += f"Previous reasons: {''.join(format(reason) for reason in previous)}\n" - raise RuntimeError(msg) - logger.info("Entering nested render phase: %r", self._rerender_needed_reasons[-1]) - self._rerender_needed = False - self._shared_elements_next = set() - self.context.exception_handler = False - self.context.exceptions_children = [] - self.context.exceptions_self = [] - - self._render(self.element, "/", parent_key=ROOT_KEY) - self._walk_all = False - logger.info("Render done: %r %r", self._rerender_needed, self._rerender_needed_reasons[-1]) - assert self.context is self.context_root - render_counts += 1 - logger.debug("Render phase resulted in (next) elements:") - for el in self._shared_elements_next: - logger.debug("\t%r %x", el, id(el)) - - logger.debug("Current elements:") - for el in self._shared_elements: - logger.debug("\t %r %x", el, id(el)) - if self.context_root.exceptions_children: - # an exception bubbled up render - break - - logger.info("Render reconsolidate...") + self._shared_elements_next = set() + self._render(self.element, "/", parent_key=ROOT_KEY) + self.first_render = False + self._walk_all = False + + if main_render_phase: + stable = False + render_counts = 0 + # decided once and then acted on: other threads can make _more_passes() False + # (a request, close()), and a reconcile needs a render pass before it + more = self._more_passes(in_render_phase=True) + while not stable and not self.context_root.exceptions_children: + # we started the rendering loop (main_render_phase is True), so we keep going + # but if an exception bubbled up, we should stop + while more and not self.context_root.exceptions_children: + if render_counts > 50: + + def format(reason: RerenderReason): + f = f"Reason: {reason.reason}\nValue changed from {reason.prev_value} to {reason.next_value}\n" + if reason.created_stack: + f += f"Created at: {''.join(reason.created_stack)}\n" + if reason.trigger_stack: + f += f"Triggered at: {''.join(reason.trigger_stack)}\n" + return f + + msg = f"Too many renders triggered, your render loop does not stop\nLast reason: {format(self._rerender_needed_reasons[-1])}\n" + if len(self._rerender_needed_reasons) >= 2: + previous = reversed(list(self._rerender_needed_reasons)[:-1]) + msg += f"Previous reasons: {''.join(format(reason) for reason in previous)}\n" + raise RuntimeError(msg) + logger.info("Entering nested render phase: %r", self._rerender_needed_reasons[-1]) + self._rerender_needed = False + if self._element_requested is not self.element: # update() from our own effect + self.element = self.context.root_element_next = self._element_requested # type: ignore # not closing + self._shared_elements_next = set() + self.context.exception_handler = False + self.context.exceptions_children = [] + self.context.exceptions_self = [] + + self._render(self.element, "/", parent_key=ROOT_KEY) + self._walk_all = False + logger.info("Render done: %r %r", self._rerender_needed, self._rerender_needed_reasons[-1]) + assert self.context is self.context_root + render_counts += 1 + more = self._more_passes(in_render_phase=render_counts <= 50) + logger.debug("Render phase resulted in (next) elements:") + for el in self._shared_elements_next: + logger.debug("\t%r %x", el, id(el)) + + logger.debug("Current elements:") + for el in self._shared_elements: + logger.debug("\t %r %x", el, id(el)) + if self.context_root.exceptions_children: + # an exception bubbled up render + break + + logger.info("Render reconsolidate...") + try: self.reconsolidating = True in_render_phase = False - try: - widget = self._reconsolidate(self.element, default_key="/", parent_key=ROOT_KEY) - finally: - self.reconsolidating = False - in_render_phase = True - logger.info("Render reconsolidate done") - self.context.root_element = self.context.root_element_next - self.context.root_element_next = None - - # remove stale elements of the root context itself - # (child contexts are swept during their reconciliation) - self._remove_stale_root_elements(ROOT_KEY) - - if self._shared_elements_next: - raise RuntimeError(f"Element not reconsolidated: {self._shared_elements_next}") - logger.debug("Reconsolidate phase resulted in elements:") - for el in self._shared_elements: - logger.debug("\t%r %x", el, id(el)) - # RESET - assert self.context is self.context_root - if self.element.is_shared: - assert widget in self._shared_widgets.values() - else: - assert widget in self.context_root.widgets.values() - if self.last_root_widget is None: - self.last_root_widget = widget + widget = self._reconsolidate(self.element, default_key="/", parent_key=ROOT_KEY) + finally: + self.reconsolidating = False + in_render_phase = True + logger.info("Render reconsolidate done") + self.context.root_element = self.context.root_element_next + self.context.root_element_next = None + + # remove stale elements of the root context itself + # (child contexts are swept during their reconciliation) + self._remove_stale_root_elements(ROOT_KEY) + + if self._shared_elements_next: + raise RuntimeError(f"Element not reconsolidated: {self._shared_elements_next}") + logger.debug("Reconsolidate phase resulted in elements:") + for el in self._shared_elements: + logger.debug("\t%r %x", el, id(el)) + # RESET + assert self.context is self.context_root + if self.element.is_shared: + assert widget in self._shared_widgets.values() + else: + assert widget in self.context_root.widgets.values() + if self.last_root_widget is None: + self.last_root_widget = widget + else: + if container is None: + if self.last_root_widget != widget: + raise ValueError( + "You are not using a container, and the root component returned a new widget," + "make sure your root component always returns the same component type" + ) + if container: + if widget is None: + # Exception occurred, and we cannot render the widget + container.children = [] else: - if container is None: - if self.last_root_widget != widget: - raise ValueError( - "You are not using a container, and the root component returned a new widget," - "make sure your root component always returns the same component type" - ) - if container: - if widget is None: - # Exception occurred, and we cannot render the widget - container.children = [] - else: - container.children = [widget] + container.children = [widget] - if self.context_root.exceptions_children or self.context_root.exceptions_self: - # an exception bubbled up during reconsolidate - break + if self.context_root.exceptions_children or self.context_root.exceptions_self: + # an exception bubbled up during reconsolidate + break - if self._rerender_needed: - logger.info("Need rerender after reconsolidation: %r", self._rerender_needed_reasons[-1]) - stable = False - else: - stable = True + more = self._more_passes(in_render_phase=False) + if more: + logger.info("Need rerender after reconsolidation: %r", self._rerender_needed_reasons[-1]) + stable = False + else: + stable = True - self._is_rendering = False - self.context = context_prev - logger.info("Done with render phase: %r", render_count) - except BaseException as e: - # Exceptions raised by components are collected in exceptions_self, so an - # exception here comes from the render machinery itself (duplicate key, - # hook misuse, ...) or is a cancellation, and aborted a pass halfway. Drop - # what that pass staged: otherwise the next render reconciles the last - # committed elements, but runs the effects the aborted pass chained, closed - # over elements that were never reconciled (get_widget then fails with - # "found in a previous render"). - if in_render_phase: - self._discard_aborted_pass() - if DEBUG: - # construct a fake traceback (showing how the elements were constructed) - if not self.tracebacks: - raise - e = _with_tracebacks(e, self.tracebacks) - raise e - else: + self._is_rendering = False + self.context = context_prev + logger.info("Done with render phase: %r", render_count) + except BaseException as e: + # Exceptions raised by components are collected in exceptions_self, so an + # exception here comes from the render machinery itself (duplicate key, + # hook misuse, ...) or is a cancellation, and aborted a pass halfway. Drop + # what that pass staged: otherwise the next render reconciles the last + # committed elements, but runs the effects the aborted pass chained, closed + # over elements that were never reconciled (get_widget then fails with + # "found in a previous render"). + if in_render_phase: + self._discard_aborted_pass() + if DEBUG: + # construct a fake traceback (showing how the elements were constructed) + if not self.tracebacks: raise + e = _with_tracebacks(e, self.tracebacks) + raise e + else: + raise - finally: - local.rc = prev_rc # type: ignore - self._is_rendering = False - # clear before the lock is released: a stale _lock_thread makes the - # recursion guard above fire for a thread that merely rendered last, - # while a *different* thread holds the lock (false "Recursive render") - self._lock_thread = None - assert self.context is self.context_root + finally: + local.rc = prev_rc # type: ignore + self._is_rendering = False + assert self.context is self.context_root exceptions = [*self.context.exceptions_children, *self.context_root.exceptions_self] if exceptions: @@ -1858,11 +1998,26 @@ def format(reason: RerenderReason): value = html.escape(error) from . import ipywidgets as w - return self.render(w.HTML(value="
" + value + "
", layout=w.Layout(overflow="auto")), self.container) + # in the same hold of the lock: no other render can come in between + return self._render_locked(w.HTML(value="
" + value + "
", layout=w.Layout(overflow="auto")), self.container) else: raise exc return widget + def _more_passes(self, in_render_phase: bool) -> bool: + """Asked by the thread that renders, between passes: render another pass in this hold? + + Only for our own re-renders (these count towards the limit). Inside the render phase we + go on, as before: a reconcile only commits a settled render. After a reconcile (and at + the limit) a request of another thread ends the hold instead: the drain after the + release renders it in a new hold, which respects batches, a waiting render() and the + budget (see _drain_render_requests). At the limit that also means: not a render loop of + ours, other threads keep changing state; commit what we have. + """ + if self._closing or not self._rerender_needed: + return False + return in_render_phase or not self._render_requested + def _render(self, element: Element, default_key: str, parent_key: str): if not isinstance(element, Element): raise TypeError(f"Expected element, not {element}") diff --git a/reacton/threads_close_test.py b/reacton/threads_close_test.py new file mode 100644 index 0000000..38ecb81 --- /dev/null +++ b/reacton/threads_close_test.py @@ -0,0 +1,374 @@ +"""close(): not starved, no render after it, no self-deadlock, requests during it dropped.""" + +# ruff: noqa: F811 the fixture names shadow the pytest fixture parameters of the same names + +import sys +import threading +from typing import List + +import pytest + +import reacton +import reacton.ipywidgets as w +from reacton import core + +from ._threads_test_utils import TIMEOUT, Injected, Worker, call_in_thread, helper_errors, log_hook, stacks, wait_until # noqa: F401 + + +def test_close_with_batch_in_cleanup_does_not_deadlock(): + """A cleanup enters and leaves a batch (solara's store fire() does `with rc:`) while a render + request is still marked. Before: the batch exit called render() on close()'s own thread, + which waited for its own lock forever. Fails on 40a90a8.""" + setters = {} + + @reacton.component + def Test(): + fail, setters["fail"] = reacton.use_state(False) + rc = core.get_render_context() + + def effect(): + def cleanup(): + with rc: + pass + + return cleanup + + reacton.use_effect(effect, []) + if fail: + raise ValueError("component fails") + return w.Button() + + _box, rc = reacton.render(Test(), handle_error=False) + with pytest.raises(ValueError): + setters["fail"](True) # leaves a render marked (handle_error=False) + finished, _closer = call_in_thread(rc.close, "C", timeout=TIMEOUT) + assert finished, "close() deadlocked on its own render lock\n" + stacks() + + +def test_close_is_not_starved_and_nothing_renders_after_close(): + """K renders, T changes state during every pass (a fast progress reporter), C closes. + close() must not wait for T to stop, and no render pass may start after close() began + (the pass in flight may finish). Fails on 40a90a8.""" + setters = {} + stop = threading.Event() + renders_after_close = [] + closing = threading.Event() + go, done = threading.Semaphore(0), threading.Semaphore(0) + armed = threading.Event() + + @reacton.component + def Test(): + trigger, setters["trigger"] = reacton.use_state(0) + progress, setters["progress"] = reacton.use_state(0) + if closing.is_set(): + renders_after_close.append(progress) + if armed.is_set() and threading.current_thread().name != "T" and not stop.is_set(): + go.release() + done.acquire(timeout=TIMEOUT) + return w.Button(description=f"{trigger} {progress}") + + _box, rc = reacton.render(Test(), handle_error=False) + running = threading.Event() + + def t_target(): + i = 0 + while not stop.is_set(): + if not go.acquire(timeout=0.1): + continue + i += 1 + setters["progress"](i) + if i == 3: + running.set() + done.release() + + t = Worker(t_target, "T") + t.start() + armed.set() + k = Worker(lambda: setters["trigger"](1), "K") + k.start() + assert running.wait(TIMEOUT) + + def close(): + closing.set() + rc.close() + + finished, c = call_in_thread(close, "C", timeout=TIMEOUT) + stop.set() + k.join(TIMEOUT) + t.join(TIMEOUT) + assert finished, "close() waited for the progress reporter to stop\n" + stacks() + assert c.error is None, c.error + assert len(renders_after_close) <= 1, renders_after_close + assert k.error is None, k.error + + +def test_close_from_own_render_raises_instead_of_hanging(): + """Fails on 40a90a8 (hangs).""" + errors = [] + + @reacton.component + def Test(): + rc = core.get_render_context() + + def effect(): + try: + rc.close() + except RuntimeError as e: + errors.append(e) + + reacton.use_effect(effect, []) + return w.Button() + + finished, _t = call_in_thread(lambda: reacton.render(Test(), handle_error=False), "K", timeout=TIMEOUT) + assert finished, "close() inside a render waited for its own lock\n" + stacks() + assert errors + + +@pytest.mark.parametrize("what", ["close", "render"]) +def test_close_or_render_from_a_cleanup_during_close_does_not_hang(what): + """Fails on 40a90a8 (hangs).""" + rcs: dict = {} + + @reacton.component + def Test(): + def effect(): + def cleanup(): + rc = rcs["rc"] + if what == "close": + rc.close() + else: + rc.render(Test()) + + return cleanup + + reacton.use_effect(effect, []) + return w.Button(description="x") + + _box, rc = reacton.render(Test(), handle_error=False) + rcs["rc"] = rc + finished, c = call_in_thread(rc.close, "C", timeout=TIMEOUT) + assert finished, f"close() hung when a cleanup called rc.{what}()\n" + stacks() + assert c.error is None, repr(c.error) + + +def test_close_between_the_decision_to_render_again_and_the_pass(log_hook): + """K reconciled, an effect set state, K decided to render again; close() sets _closing + right then. K must not reconcile without a render pass (KeyError '/'), and close() must + finish. Fails on 40a90a8 (close() sets _closing only after it got the lock, and the + re-decision to render is not guarded).""" + setters = {} + + @reacton.component + def Test(): + trigger, setters["trigger"] = reacton.use_state(0) + derived, set_derived = reacton.use_state(0) + reacton.use_effect(lambda: set_derived(trigger), [trigger]) + return w.Button(description=f"{trigger} {derived}") + + _box, rc = reacton.render(Test(), handle_error=False) + k_paused, resume = threading.Event(), threading.Event() + + def k_target(): + log_hook.on(threading.current_thread(), "Need rerender after reconsolidation", lambda: (k_paused.set(), resume.wait(TIMEOUT * 3))) # type: ignore[func-returns-value] + setters["trigger"](1) + + k = Worker(k_target, "K") + k.start() + assert k_paused.wait(TIMEOUT) + c = Worker(rc.close, "C") + c.start() + assert wait_until(lambda: rc._closing, TIMEOUT) + resume.set() + k.join(TIMEOUT) + c.join(TIMEOUT) + assert not k.is_alive() and not c.is_alive(), stacks() + assert k.error is None, k.error_tb or repr(k.error) + assert c.error is None, repr(c.error) + + +def test_interrupted_close_can_be_retried(): + """An interrupt after close() marks closing must not make every later close() return.""" + setters = {} + k_paused, k_resume = threading.Event(), threading.Event() + cleanups = [] + + @reacton.component + def Test(): + value, setters["value"] = reacton.use_state(0) + + def effect(): + return lambda: cleanups.append("cleanup") + + reacton.use_effect(effect, []) + if value == 1 and threading.current_thread().name == "K": + k_paused.set() + k_resume.wait(TIMEOUT) + return w.Button(description=str(value)) + + _box, rc = reacton.render(Test(), handle_error=False) + k = Worker(lambda: setters["value"](1), "K") + k.start() + assert k_paused.wait(TIMEOUT) + close_code = core._RenderContext.close.__code__ + fired: List[str] = [] + + def trace(frame, event, arg): + if not fired and frame.f_code is close_code and rc._closing and rc.thread_lock.locked(): + fired.append(event) + raise Injected() + return trace + + def interrupted_close(): + sys.settrace(trace) + try: + rc.close() + finally: + sys.settrace(None) + + finished, c = call_in_thread(interrupted_close, "C", timeout=TIMEOUT) + assert finished, "the interrupted close() waited for the render lock\n" + stacks() + assert isinstance(c.error, Injected), repr(c.error) + assert fired, "the probe did not fire" + assert cleanups == [] + k_resume.set() + k.join(TIMEOUT) + assert k.error is None, repr(k.error) + finished, c2 = call_in_thread(rc.close, "C2", timeout=TIMEOUT) + assert finished and c2.error is None, (c2.error, stacks()) + assert cleanups == ["cleanup"] + finished, c3 = call_in_thread(rc.close, "C3", timeout=TIMEOUT) + assert finished and c3.error is None, (c3.error, stacks()) + assert cleanups == ["cleanup"] + + +def test_batched_set_from_another_thread_while_close_runs_cleanups(log_hook): + """The close ABBA. T is inside a setter (past the _closing check) when K starts close(). + K's cleanup waits for T's batch to end (in solara: a store lock). Before: T's batch exit + waited for the render lock that close() holds: deadlock. Fails on 40a90a8.""" + setters = {} + t_batch_done, t_paused, t_go = threading.Event(), threading.Event(), threading.Event() + cleanup_waited = [] + + @reacton.component + def Test(): + value, setters["value"] = reacton.use_state(0) + + def effect(): + def cleanup(): + t_go.set() + cleanup_waited.append(t_batch_done.wait(TIMEOUT)) + + return cleanup + + reacton.use_effect(effect, []) + return w.Button(description=str(value)) + + _box, rc = reacton.render(Test(), handle_error=False) + + def t_target(): + log_hook.on(threading.current_thread(), "Set state = ", lambda: (t_paused.set(), t_go.wait(TIMEOUT))) # type: ignore[func-returns-value] + with rc: # like a solara listener batch + setters["value"](1) + t_batch_done.set() + + def k_target(): + assert t_paused.wait(TIMEOUT) + rc.close() + + t, k = Worker(t_target, "T"), Worker(k_target, "K") + t.start() + k.start() + k.join(TIMEOUT * 2) + t.join(TIMEOUT * 2) + assert not k.is_alive() and not t.is_alive(), "deadlock between close() and a batched set\n" + stacks() + assert k.error is None and t.error is None, (k.error, t.error) + assert cleanup_waited == [True], "the cleanup waited for T's batch, and T's batch exit waited for close()" + + +def test_second_concurrent_close_does_not_deadlock_against_cleanup_lock(): + ext = threading.RLock() + in_cleanup, c2_has_ext = threading.Event(), threading.Event() + + @reacton.component + def Test(): + def effect(): + def cleanup(): + in_cleanup.set() + assert c2_has_ext.wait(TIMEOUT) + with ext: + pass + + return cleanup + + reacton.use_effect(effect, []) + return w.Button(description="x") + + _box, rc = reacton.render(Test(), handle_error=False) + + def c2_target(): + assert in_cleanup.wait(TIMEOUT) + with ext: + c2_has_ext.set() + rc.close() + + c1, c2 = Worker(rc.close, "C1"), Worker(c2_target, "C2") + c1.start() + c2.start() + c1.join(TIMEOUT) + c2.join(TIMEOUT) + assert not c1.is_alive() and not c2.is_alive(), "two close() calls deadlocked across a cleanup lock\n" + stacks() + assert c1.error is None and c2.error is None, (c1.error, c2.error) + + +def test_requests_during_close_are_dropped_and_render_after_close_is_ignored(log_hook): + """Fails on 40a90a8 (T's requests wait for close()).""" + setters = {} + rendered = [] + + @reacton.component + def Test(): + value, setters["value"] = reacton.use_state(0) + rendered.append(value) + return w.Button(description=str(value)) + + box, rc = reacton.render(Test(), handle_error=False) + k_paused, k_go = threading.Event(), threading.Event() + + def k_target(): + log_hook.on(threading.current_thread(), "Removing elements...", lambda: (k_paused.set(), k_go.wait(TIMEOUT * 3))) # type: ignore[func-returns-value] + rc.close() + + def t_target(): + assert k_paused.wait(TIMEOUT) + setters["value"](1) # must return at once and never render + rc.force_update() + rc.update(Test()) + + k = Worker(k_target, "K") + k.start() + finished, t = call_in_thread(t_target, "T", timeout=TIMEOUT) + k_go.set() + k.join(TIMEOUT) + t.join(TIMEOUT) + assert finished, "a request on a closing context waited for close()\n" + stacks() + assert t.error is None and k.error is None, (t.error, k.error) + assert rendered == [0] + finished, r = call_in_thread(lambda: rc.render(Test(), box), "R", timeout=TIMEOUT) + assert finished and r.error is None, (r.error, stacks()) + assert rendered == [0] + + +def test_recursive_render_is_still_detected(): + """A guard (passes on 40a90a8).""" + + @reacton.component + def Page(): + rc = core.get_render_context() + rc.render(rc.element, rc.container) + return w.Button(description="never") + + finished, k = call_in_thread(lambda: reacton.render(Page(), handle_error=True), "K", timeout=TIMEOUT) + assert finished, stacks() + box, rc = k.result + assert "Recursive render detected" in box.children[0].value + rc.close() diff --git a/reacton/threads_create_lock_test.py b/reacton/threads_create_lock_test.py new file mode 100644 index 0000000..c0cffa0 --- /dev/null +++ b/reacton/threads_create_lock_test.py @@ -0,0 +1,110 @@ +"""Widget construction does not hold a shared lock: it does not block other render contexts or +other threads, and a widget that renders in its constructor does not hang. Fail on 40a90a8.""" + +import threading + +import ipywidgets +import pytest + +import reacton +import reacton.ipywidgets as w + +from ._threads_test_utils import TIMEOUT, Worker, call_in_thread, helper_errors, log_hook, stacks # noqa: F401 + + +class BlockingButton(ipywidgets.Button): + """A widget whose construction blocks on thread "A", like comm_open to a slow client.""" + + entered = threading.Event() + release = threading.Event() + + def __init__(self, **kwargs): + super().__init__(**kwargs) + if threading.current_thread().name == "A": + BlockingButton.entered.set() + assert BlockingButton.release.wait(TIMEOUT) + + +@pytest.fixture +def blocking_button(): + BlockingButton.entered.clear() + BlockingButton.release.clear() + yield BlockingButton + BlockingButton.release.set() + + +def test_widget_construction_does_not_block_other_render_contexts(blocking_button): + """Before: one process-wide lock around every widget construction; a construction that + blocks (a send to a slow client) stops every other render context. Side-effect widgets + (Layout, Style) are still attributed to their own render context.""" + BlockingButtonElement = reacton.core.ComponentWidget(blocking_button) + + @reacton.component + def Slow(): + return BlockingButtonElement(description="slow") + + @reacton.component + def Fast(): + return w.Button(description="fast") + + a = Worker(lambda: reacton.render(Slow(), handle_error=False), "A") + a.start() + assert blocking_button.entered.wait(TIMEOUT) + finished, b = call_in_thread(lambda: reacton.render(Fast(), handle_error=False), "B") + blocking_button.release.set() + a.join(TIMEOUT) + b.join(TIMEOUT) + assert finished, "widget creation waited for another render context's widget construction\n" + stacks() + assert a.error is None and b.error is None, (a.error, b.error) + (box_a, rc_a), (box_b, rc_b) = a.result, b.result # type: ignore[misc] + button_a, button_b = box_a.children[0], box_b.children[0] + assert rc_a._orphans[button_a.model_id] == {button_a.layout.model_id, button_a.style.model_id} + assert rc_b._orphans[button_b.model_id] == {button_b.layout.model_id, button_b.style.model_id} + rc_a.close() + rc_b.close() + + +def test_widget_created_by_another_thread_is_not_an_orphan(blocking_button): + """Before: the construction recording was global, so a widget that another thread creates + while a render constructs a widget became an orphan of that widget, and was closed with it.""" + BlockingButtonElement = reacton.core.ComponentWidget(blocking_button) + + @reacton.component + def Slow(): + return BlockingButtonElement(description="slow") + + a = Worker(lambda: reacton.render(Slow(), handle_error=False), "A") + a.start() + assert blocking_button.entered.wait(TIMEOUT) + finished, u = call_in_thread(ipywidgets.IntSlider, "U") # user code on another thread + blocking_button.release.set() + a.join(TIMEOUT) + assert finished, "creating a plain widget waited for another thread's widget construction\n" + stacks() + assert a.error is None, a.error + _box, rc = a.result # type: ignore[misc] + rc.close() + unrelated = u.result + assert unrelated.comm is not None, "a widget of another thread was closed as an orphan" + unrelated.close() + + +def test_widget_that_renders_in_its_constructor(): + """Before: a widget class whose constructor renders reacton (ComponentFunction.widget_class) + used as an element took the non-reentrant create_lock twice on one thread: hang.""" + + @reacton.component + def Inner(label: str = "x"): + return w.Button(description=label) + + InnerWidget = Inner.widget_class() # type: ignore[attr-defined] + + @reacton.component + def Outer(): + return InnerWidget.element(label="nested") + + finished, t = call_in_thread(lambda: reacton.render(Outer(), handle_error=False), "K", timeout=TIMEOUT) + assert finished, "nested widget construction deadlocked\n" + stacks() + assert t.error is None, t.error + box, rc = t.result + assert box.children[0].children[0].description == "nested" + rc.close() diff --git a/reacton/threads_handoff_test.py b/reacton/threads_handoff_test.py new file mode 100644 index 0000000..23cdc3e --- /dev/null +++ b/reacton/threads_handoff_test.py @@ -0,0 +1,381 @@ +"""A render request (state change, update(), force_update(), batch exit) never waits for the +render lock, and is never lost. Fails on 40a90a8 unless noted.""" + +# ruff: noqa: F811 the fixture names shadow the pytest fixture parameters of the same names + +import sys +import threading + +import reacton +import reacton.ipywidgets as w +from reacton import core + +from ._threads_test_utils import ( # noqa: F401 + SHORT, + TIMEOUT, + Injected, + Worker, + call_in_thread, + helper_errors, + log_hook, + pause_at_line, + render_loop_function, + source_line, + stacks, + text, + wait_until, +) + + +def make_ab(): + setters = {} + + @reacton.component + def Test(label=""): + a, setters["a"] = reacton.use_state(0) + b, setters["b"] = reacton.use_state(0) + return w.Button(description=f"{label}{a} {b}") + + return Test, setters + + +def test_state_change_does_not_wait_for_a_render_on_another_thread_prologue(log_hook): + _test_state_change_does_not_wait(log_hook, "Render phase: ") + + +def test_state_change_does_not_wait_for_a_render_on_another_thread_epilogue(log_hook): + _test_state_change_does_not_wait(log_hook, "Done with render phase") + + +def _test_state_change_does_not_wait(log_hook, window): + """K holds the render lock outside its render loop (prologue or epilogue). + Before: T's setter waits for the render lock for the rest of K's render. + After: T's setter returns at once, and K renders T's change after it released the lock.""" + Test, setters = make_ab() + box, rc = reacton.render(Test(), handle_error=False) + k_paused, t_done = threading.Event(), threading.Event() + + def k_target(): + log_hook.on(threading.current_thread(), window, lambda: (k_paused.set(), t_done.wait(TIMEOUT))) # type: ignore[func-returns-value] + setters["a"](1) + + k = Worker(k_target, "K") + k.start() + assert k_paused.wait(TIMEOUT) + finished, t = call_in_thread(lambda: setters["b"](1), "T") + t_done.set() + k.join(TIMEOUT) + t.join(TIMEOUT) + assert finished, "the state change waited for another thread's render\n" + stacks() + assert k.error is None and t.error is None, (k.error, t.error) + assert text(box) == "1 1", "the state change from T was not rendered" + rc.close() + + +def _pause_after_stable(on_pause): + fn = render_loop_function() + return pause_at_line(fn, source_line(fn, "self._is_rendering = False", after="stable = True"), on_pause) + + +def test_state_change_after_the_last_check_is_not_lost(): + """Lost wakeup. K has decided its render is stable and still holds the lock; T sets state + then. Before: T only marks, K releases, nobody renders T's change. + After: K checks again after it released the lock.""" + Test, setters = make_ab() + box, rc = reacton.render(Test(), handle_error=False) + k_paused, t_done = threading.Event(), threading.Event() + tracer = _pause_after_stable(lambda: (k_paused.set(), t_done.wait(TIMEOUT))) # type: ignore[func-returns-value] + + def k_target(): + sys.settrace(tracer) + try: + setters["a"](1) + finally: + sys.settrace(None) + + k = Worker(k_target, "K") + k.start() + assert k_paused.wait(TIMEOUT) + finished, t = call_in_thread(lambda: setters["b"](1), "T") + t_done.set() + k.join(TIMEOUT) + t.join(TIMEOUT) + assert finished and k.error is None and t.error is None, (k.error, t.error, stacks()) + assert text(box) == "1 1", "the state change from T was lost" + rc.close() + + +def test_state_change_after_the_last_check_of_an_explicit_render_is_not_lost(): + """The same window, in an explicit render(): render() must check again after its release.""" + Test, setters = make_ab() + box, rc = reacton.render(Test(), handle_error=False) + w_paused, t_done = threading.Event(), threading.Event() + tracer = _pause_after_stable(lambda: (w_paused.set(), t_done.wait(TIMEOUT))) # type: ignore[func-returns-value] + + def w_target(): + sys.settrace(tracer) + try: + rc.render(Test(label="b")) + finally: + sys.settrace(None) + + wt = Worker(w_target, "W") + wt.start() + assert w_paused.wait(TIMEOUT) + finished, t = call_in_thread(lambda: setters["b"](1), "T") + t_done.set() + wt.join(TIMEOUT) + t.join(TIMEOUT) + assert finished and wt.error is None and t.error is None, (wt.error, t.error, stacks()) + assert wait_until(lambda: text(box) == "b0 1", 2), f"the state change from T was lost: {text(box)!r}" + rc.close() + + +def test_lock_held_around_a_state_change_does_not_deadlock_with_an_effect(log_hook): + """A user lock (a solara store lock, init lock or context.lock behave the same). + T holds L while it sets state; K's render runs an effect that takes L. + Before: T waits for the render lock while it holds L, K waits for L: deadlock.""" + user_lock = threading.RLock() + setters = {} + + @reacton.component + def Test(): + a, setters["a"] = reacton.use_state(0) + b, setters["b"] = reacton.use_state(0) + + def effect(): + if a == 1: + with user_lock: + pass + + reacton.use_effect(effect, [a]) + return w.Button(description=f"{a} {b}") + + box, rc = reacton.render(Test(), handle_error=False) + k_paused, never = threading.Event(), threading.Event() + + def k_target(): + # pause in the render prologue: lock held, not yet rendering + log_hook.on(threading.current_thread(), "Render phase: ", lambda: (k_paused.set(), never.wait(SHORT))) # type: ignore[func-returns-value] + setters["a"](1) + + def t_target(): + with user_lock: + setters["b"](1) + + k = Worker(k_target, "K") + k.start() + assert k_paused.wait(TIMEOUT) + t = Worker(t_target, "T") + t.start() + k.join(TIMEOUT) + t.join(TIMEOUT) + assert not k.is_alive() and not t.is_alive(), "deadlock\n" + stacks() + assert k.error is None and t.error is None, (k.error, t.error) + assert text(box) == "1 1" + rc.close() + + +def _root_element_read(): + """(function, line): the line where the first pass of a hold reads the root element.""" + RC = core._RenderContext + if hasattr(RC, "_render_locked"): + fn = RC._render_locked + for needle in ("self.element = self._element_requested", "self.element = element"): + try: + return fn, source_line(fn, needle) + except StopIteration: + pass + return RC.render, source_line(RC.render, "self.element = element") + + +def test_update_from_another_thread_at_the_start_of_a_pass_is_not_lost(): + """K has just read the root element for its pass; T calls rc.update() with a new root. + T's element must be rendered (the pass start clears the request mark before it reads the + root element). Passes on 40a90a8 (T waits for the lock).""" + setters = {} + + @reacton.component + def Test(label="old"): + a, setters["a"] = reacton.use_state(0) + return w.Button(description=f"{label} {a}") + + box, rc = reacton.render(Test(), handle_error=False) + fn, lineno = _root_element_read() + k_paused, t_done = threading.Event(), threading.Event() + tracer = pause_at_line(fn, lineno, lambda: (k_paused.set(), t_done.wait(SHORT)), after_line=True) # type: ignore[func-returns-value] + + def k_target(): + sys.settrace(tracer) + try: + setters["a"](1) + finally: + sys.settrace(None) + + k = Worker(k_target, "K") + k.start() + assert k_paused.wait(TIMEOUT) + finished, t = call_in_thread(lambda: rc.update(Test(label="new")), "T", timeout=SHORT) + t_done.set() + k.join(TIMEOUT) + t.join(TIMEOUT) + assert k.error is None and t.error is None, (k.error, t.error) + assert wait_until(lambda: text(box) == "new 1", 2), f"the update() from T was lost: {text(box)!r} (T returned at once: {finished})" + rc.close() + + +def test_update_from_an_effect_of_the_render_is_rendered(): + """rc.update() from an effect of the render in progress (same thread): the next pass of + that render must use the new root element. Passes on 40a90a8.""" + calls: list = [] + + @reacton.component + def Test(label="old"): + rc = core.get_render_context() + + def effect(): + if label == "old" and not calls: + calls.append(1) + rc.update(Test(label="new")) + + reacton.use_effect(effect, [label]) + return w.Button(description=label) + + finished, k = call_in_thread(lambda: reacton.render(Test(), handle_error=False), "K", timeout=TIMEOUT) + assert finished and k.error is None, (k.error, stacks()) + box, rc = k.result + assert text(box) == "new" + rc.close() + + +def test_a_reconcile_commits_a_settled_render_only(): + """A component derives state while it renders (`if b != a: set_b(a)`), so the pass after a + change of `a` is not settled. T asks for a render during that pass. The holder must still + finish its render phase before it reconciles: no effect may run on (a=1, b=0). Passes on + 40a90a8.""" + setters = {} + seen = [] + k_in_pass, k_go = threading.Event(), threading.Event() + + @reacton.component + def Test(): + a, setters["a"] = reacton.use_state(0) + b, set_b = reacton.use_state(0) + x, setters["x"] = reacton.use_state(0) + if b != a: + if threading.current_thread().name == "K" and not k_in_pass.is_set(): + k_in_pass.set() + k_go.wait(TIMEOUT) + set_b(a) + reacton.use_effect(lambda: seen.append((a, b)), [a, b]) + return w.Button(description=f"{a} {b} {x}") + + box, rc = reacton.render(Test(), handle_error=False) + k = Worker(lambda: setters["a"](1), "K") + k.start() + assert k_in_pass.wait(TIMEOUT) + _finished, t = call_in_thread(lambda: setters["x"](1), "T", timeout=SHORT * 2) + k_go.set() + k.join(TIMEOUT) + t.join(TIMEOUT) + assert k.error is None and t.error is None, (k.error, t.error) + assert wait_until(lambda: text(box) == "1 1 1", 3), text(box) + assert (1, 0) not in seen, f"an effect ran on a render that was not settled: {seen}" + rc.close() + + +def test_request_during_a_batch_on_another_thread_renders_at_batch_exit(): + """While any thread is inside `with rc:`, a state change only marks, and the thread that + leaves the last batch renders. Passes on 40a90a8 (a guard).""" + Test, setters = make_ab() + box, rc = reacton.render(Test(), handle_error=False) + rc.__enter__() + finished, t = call_in_thread(lambda: setters["a"](1), "T") + assert finished and t.error is None + assert text(box) == "0 0", "rendered inside a batch" + rc.__exit__(None, None, None) + assert text(box) == "1 0" + rc.close() + + +def test_exception_injected_right_after_the_lock_is_taken_does_not_leak_it(): + """Models a KeyboardInterrupt, or solara's cancel_guard (a trace function that raises on a + traced line while the context is not rendering): it fires once, at the first traced event + in reacton/core.py after the thread took the render lock. The lock must not leak. + Passes on 40a90a8.""" + Test, setters = make_ab() + _box, rc = reacton.render(Test(), handle_error=False) + corefile = core.__file__ + fired: list = [] + + def trace(frame, event, arg): + if not fired and frame.f_code.co_filename == corefile and rc.thread_lock.locked() and not rc._is_rendering: + fired.append((frame.f_code.co_name, frame.f_lineno, event)) + raise Injected() + return trace + + def t_target(): + sys.settrace(trace) + try: + setters["a"](1) + finally: + sys.settrace(None) + + finished, t = call_in_thread(t_target, "T", timeout=TIMEOUT) + assert finished + assert isinstance(t.error, Injected), (t.error, fired) + assert not rc.thread_lock.locked(), f"the render lock leaked after an exception at {fired}" + finished, _c = call_in_thread(rc.close, "C", timeout=TIMEOUT) + assert finished, stacks() + + +def test_concurrent_writers_never_hang_raise_or_lose_a_render(): + """Stress, no pauses: K changes `trigger` (its effect writes `b` under a user lock), T writes + `a` under the same lock. No deadlock, no exception, the last state is shown (a helper may + render it a moment after the writers returned).""" + lock = threading.RLock() + setters = {} + + @reacton.component + def Test(): + trigger, setters["trigger"] = reacton.use_state(0) + a, setters["a"] = reacton.use_state(0) + b, set_b = reacton.use_state(0) + + def effect(): + with lock: # like a solara store listener that holds the store lock + set_b(trigger) + + reacton.use_effect(effect, [trigger]) + return w.Button(description=f"{trigger} {a} {b}") + + box, rc = reacton.render(Test(), handle_error=False) + n = 2000 + errors = [] + + def run(f): + try: + for i in range(1, n + 1): + f(i) + except BaseException as e: # noqa + errors.append(e) + + def write_a(i): + with lock: + setters["a"](i) + + old = sys.getswitchinterval() + sys.setswitchinterval(1e-6) + try: + k = Worker(lambda: run(setters["trigger"]), "K") + t = Worker(lambda: run(write_a), "T") + k.start() + t.start() + k.join(60) + t.join(60) + finally: + sys.setswitchinterval(old) + assert not k.is_alive() and not t.is_alive(), "deadlock\n" + stacks() + assert not errors, errors + assert wait_until(lambda: text(box) == f"{n} {n} {n}", 10), text(box) + rc.close() diff --git a/reacton/threads_limit_test.py b/reacton/threads_limit_test.py new file mode 100644 index 0000000..a0107d1 --- /dev/null +++ b/reacton/threads_limit_test.py @@ -0,0 +1,233 @@ +"""The "too many renders" limit counts only a thread's own render loop; a caller is not held +by other threads' updates (the budget and the helper); an explicit render() is not starved.""" + +# ruff: noqa: F811 the fixture names shadow the pytest fixture parameters of the same names + +import threading +import time + +import pytest + +import reacton +import reacton.ipywidgets as w +from reacton import core + +from ._threads_test_utils import TIMEOUT, Worker, call_in_thread, helper_errors, log_hook, stacks, text, wait_until # noqa: F401 + + +@pytest.mark.parametrize("derive", ["none", "effect", "render"]) +def test_updates_from_another_thread_do_not_count_as_a_render_loop(derive): + """T changes state once during each of K's render passes, 60 times. + Before: K raises "Too many renders triggered" after about 50 passes. + derive="effect": an effect derives state from T's value, so every pass for T is followed by + an own pass. derive="render": the component derives it while it renders, so the render + phase does not settle while T keeps changing state; at the limit the holder commits what it + has instead of raising (T's requests are pending). Fails on 40a90a8.""" + n = 60 + setters = {} + remaining = [n] + go, done = threading.Semaphore(0), threading.Semaphore(0) + armed = threading.Event() + + @reacton.component + def Test(): + trigger, setters["trigger"] = reacton.use_state(0) + progress, setters["progress"] = reacton.use_state(0) + derived, set_derived = reacton.use_state(0) + if derive == "effect": + reacton.use_effect(lambda: set_derived(progress), [progress]) + if derive == "render" and derived != progress: + set_derived(progress) + if armed.is_set() and threading.current_thread().name != "T" and remaining[0] > 0: # K, or the helper + go.release() # T reports progress once during this pass + assert done.acquire(timeout=TIMEOUT) + return w.Button(description=f"{trigger} {progress} {derived}") + + box, rc = reacton.render(Test(), handle_error=False) + + def t_target(): + for i in range(n): + assert go.acquire(timeout=TIMEOUT) + setters["progress"](i + 1) + remaining[0] -= 1 + done.release() + + def k_target(): + armed.set() + setters["trigger"](1) + + t = Worker(t_target, "T") + t.start() + finished, k = call_in_thread(k_target, "K", timeout=TIMEOUT * 2) + t.join(TIMEOUT) + assert finished and not t.is_alive(), stacks() + assert k.error is None, repr(k.error)[:300] + assert t.error is None, repr(t.error)[:300] + expected = f"1 {n} {n if derive != 'none' else 0}" + assert wait_until(lambda: text(box) == expected, 5), text(box) + rc.close() + + +@pytest.mark.parametrize("n", [80, 200]) +def test_foreign_updates_seen_by_an_effect_are_not_a_render_loop(n, helper_errors): + """T changes state inside every reconcile of the renderer (inside an effect), n times; + the effect derives own state from T's value. Not a render loop: no "Too many renders", also + not on the helper that renders after K's budget (n=200). Fails on 40a90a8.""" + setters = {} + go, done = threading.Semaphore(0), threading.Semaphore(0) + armed = threading.Event() + remaining = [n] + + @reacton.component + def Test(): + trigger, setters["trigger"] = reacton.use_state(0) + progress, setters["progress"] = reacton.use_state(0) + derived, set_derived = reacton.use_state(0) + + def effect(): + set_derived(progress) + if armed.is_set() and threading.current_thread().name != "T" and remaining[0] > 0: + go.release() + done.acquire(timeout=TIMEOUT) + + reacton.use_effect(effect, [progress, trigger]) + return w.Button(description=f"{trigger} {progress} {derived}") + + box, rc = reacton.render(Test(), handle_error=False) + + def t_target(): + for i in range(n): + assert go.acquire(timeout=TIMEOUT) + setters["progress"](i + 1) + remaining[0] -= 1 + done.release() + + def k_target(): + armed.set() + setters["trigger"](1) + + t = Worker(t_target, "T") + t.start() + finished, k = call_in_thread(k_target, "K", timeout=30) + t.join(30) + assert finished, stacks() + assert k.error is None, repr(k.error)[:300] + assert t.error is None, repr(t.error)[:300] + ok = wait_until(lambda: text(box) == f"1 {n} {n}", 10) + assert not helper_errors, repr(helper_errors[0])[:300] + assert ok, text(box) + rc.close() + + +def test_own_render_loop_is_still_detected(): + """A component that changes its own state on every render still raises. Passes on + 40a90a8 (a guard).""" + + @reacton.component + def Infinite(): + state, set_state = reacton.use_state(0) + set_state(state + 1) + return w.Button(description=str(state)) + + rc = core._RenderContext(Infinite(), handle_error=False) + finished, k = call_in_thread(lambda: rc.render(rc.element), "K", timeout=TIMEOUT) + assert finished, stacks() + assert isinstance(k.error, RuntimeError) and "Too many renders triggered" in str(k.error), repr(k.error) + finished, _c = call_in_thread(rc.close, "C", timeout=TIMEOUT) + assert finished, stacks() + + +def test_renderer_caller_is_not_held_by_a_stream_of_foreign_updates(): + """K renders because of its own state change. T changes state once during every render + pass, until K's call returned (a task that reports progress until the user clicks cancel, + a click that K must process). K's call must return without an error, and T's last value + (made after K returned) must be rendered. Fails on 40a90a8 ("Too many renders").""" + setters = {} + streaming, k_returned = threading.Event(), threading.Event() + go, done = threading.Semaphore(0), threading.Semaphore(0) + + @reacton.component + def Test(): + trigger, setters["trigger"] = reacton.use_state(0) + progress, setters["progress"] = reacton.use_state(0) + if streaming.is_set() and not k_returned.is_set() and threading.current_thread().name != "T": + go.release() + done.acquire(timeout=1) # T reports progress during this pass + return w.Button(description=f"{trigger} {progress}") + + box, rc = reacton.render(Test(), handle_error=False) + count = [0] + + def t_target(): + deadline = time.monotonic() + 40 + while not k_returned.is_set() and time.monotonic() < deadline: + if go.acquire(timeout=0.1): + count[0] += 1 + setters["progress"](count[0]) + done.release() + count[0] += 1 + setters["progress"](count[0]) # after K returned: must be rendered too + + def k_target(): + streaming.set() + setters["trigger"](1) + + t = Worker(t_target, "T") + t.start() + finished, k = call_in_thread(k_target, "K", timeout=10) + k_returned.set() + k.join(45) + t.join(45) + assert finished, f"K's set() did not return while T kept changing state ({count[0]} updates so far)\n" + stacks() + assert k.error is None, repr(k.error) + assert t.error is None, repr(t.error) + assert wait_until(lambda: text(box) == f"1 {count[0]}", 10), f"lost: shows {text(box)!r}, expected '1 {count[0]}'" + rc.close() + + +def test_explicit_render_is_not_starved_by_a_stream(): + """A holder renders a stream of T's updates (one per pass, until W's render() returned). + W calls rc.render() (solara: app.py on a hot reload). It must return. + On 40a90a8 K raises "Too many renders" first. With a real scheduler the blocked waiter + usually wins the lock when the holder releases it between holds.""" + setters = {} + streaming, w_returned = threading.Event(), threading.Event() + go, done = threading.Semaphore(0), threading.Semaphore(0) + + @reacton.component + def Test(label="old"): + trigger, setters["trigger"] = reacton.use_state(0) + progress, setters["progress"] = reacton.use_state(0) + if streaming.is_set() and not w_returned.is_set() and threading.current_thread().name != "T": + go.release() + done.acquire(timeout=1) + return w.Button(description=f"{label} {trigger} {progress}") + + box, rc = reacton.render(Test(), handle_error=False) + count = [0] + + def t_target(): + deadline = time.monotonic() + 40 + while not w_returned.is_set() and time.monotonic() < deadline: + if go.acquire(timeout=0.1): + count[0] += 1 + setters["progress"](count[0]) + done.release() + + def k_target(): + streaming.set() + setters["trigger"](1) + + t = Worker(t_target, "T") + t.start() + k = Worker(k_target, "K") + k.start() + assert wait_until(lambda: count[0] >= 60, 20), "the stream did not start" # past the hand-over to a helper + finished, wt = call_in_thread(lambda: rc.render(Test(label="new")), "W", timeout=10) + w_returned.set() + wt.join(45) + k.join(45) + t.join(45) + assert finished, f"W's render() waited while another thread kept rendering T's updates ({count[0]})\n" + stacks() + assert wait_until(lambda: text(box).startswith("new 1"), 5), text(box) + rc.close() diff --git a/reacton/threads_robustness_test.py b/reacton/threads_robustness_test.py new file mode 100644 index 0000000..1e31a67 --- /dev/null +++ b/reacton/threads_robustness_test.py @@ -0,0 +1,283 @@ +"""Robustness of the render hand-off: an exception in a waiter's batch exit or in a failing +hold must not leak the render lock or lose a request, an explicit render() is not starved by a +stream of derived own-state re-renders, and a foreign request consumed by an own pass is not +a render loop.""" + +# ruff: noqa: F811 the fixture names shadow the pytest fixture parameters of the same names + +import sys +import threading +import time + +import reacton +import reacton.ipywidgets as w +from reacton import core + +from ._threads_test_utils import ( # noqa: F401 + SHORT, + TIMEOUT, + EveryTimeHook, + Injected, + Worker, + call_in_thread, + helper_errors, + install, + log_hook, + source_line, + stacks, + text, + wait_until, +) + + +def test_exception_in_the_waiters_batch_exit_does_not_leak_the_lock(): + """W waits in render() while K holds the lock (K is paused in a component body). When K + releases, W gets the lock and then leaves its waiting batch. An exception there (a + KeyboardInterrupt, a trace function) must not leak the lock or the batch count.""" + setters = {} + k_in_body, k_resume = threading.Event(), threading.Event() + + @reacton.component + def Test(label="a"): + a, setters["a"] = reacton.use_state(0) + if a == 1 and threading.current_thread().name == "K": + k_in_body.set() + k_resume.wait(TIMEOUT) + return w.Button(description=f"{label} {a}") + + box, rc = reacton.render(Test(), handle_error=False) + k = Worker(lambda: setters["a"](1), "K") + k.start() + assert k_in_body.wait(TIMEOUT) + exit_code = core._RenderContext.__exit__.__code__ + hold_code = core._RenderContext._hold_and_render.__code__ + fired: list = [] + + def trace(frame, event, arg): + if not fired and frame.f_code is exit_code and rc.thread_lock.locked() and not rc._is_rendering and k_resume.is_set(): + fired.append(event) + raise Injected() + if ( + not fired + and frame.f_code is hold_code + and rc.thread_lock.locked() + and rc._batch_counter.current() == 0 + and not rc._is_rendering + and k_resume.is_set() + ): + fired.append(event) + raise Injected() + return trace + + def w_target(): + sys.settrace(trace) + try: + rc.render(Test(label="b")) + finally: + sys.settrace(None) + + wt = Worker(w_target, "W") + wt.start() + assert wait_until(lambda: rc._batch_counter.current() == 1, TIMEOUT), "W did not start waiting" + k_resume.set() + k.join(TIMEOUT) + wt.join(TIMEOUT) + assert not k.is_alive() and not wt.is_alive(), stacks() + assert fired, "the probe did not fire" + assert isinstance(wt.error, Injected), repr(wt.error) + assert not rc.thread_lock.locked(), "the render lock leaked after an exception in the waiter's batch exit" + assert rc._batch_counter.current() == 0, "the waiter's batch count leaked" + setters["a"](2) + assert wait_until(lambda: text(box).endswith(" 2"), TIMEOUT), text(box) + finished, _c = call_in_thread(rc.close, "C", timeout=TIMEOUT) + assert finished, "close() hangs on the leaked lock\n" + stacks() + + +def test_request_during_a_failing_hold_is_rendered(log_hook, helper_errors): + """handle_error=False. A's pass fails (a component raises); A pauses after its render loop, + still holding the lock, before it raises. B asks for a recovery render (B's try fails, B + returns). A raises and releases. B's request must still be rendered after the holder raises.""" + setters = {} + + @reacton.component + def Test(): + fail, setters["fail"] = reacton.use_state(False) + if fail: + raise ValueError("fail") + return w.Button(description=f"ok {fail}") + + box, rc = reacton.render(Test(), handle_error=False) + a_paused, a_resume = threading.Event(), threading.Event() + + def a_target(): + log_hook.on(threading.current_thread(), "Done with render phase", lambda: (a_paused.set(), a_resume.wait(TIMEOUT * 3))) # type: ignore[func-returns-value] + setters["fail"](True) + + a = Worker(a_target, "A") + a.start() + assert a_paused.wait(TIMEOUT) + finished, b = call_in_thread(lambda: setters["fail"](False), "B", timeout=SHORT * 2) + a_resume.set() + assert finished and b.error is None, ("B's state change waited for A's render", b.error, stacks()) + a.join(TIMEOUT) + assert isinstance(a.error, ValueError), repr(a.error) + assert wait_until( + lambda: not rc.thread_lock.locked() and not getattr(rc, "_render_requested", False) and text(box) == "ok False", 3 + ), f"B's recovery request is still pending: _render_requested={getattr(rc, '_render_requested', None)}, shows {text(box)!r}" + assert not helper_errors, helper_errors + rc.close() + + +def test_request_after_explicit_render_hold_survives_interrupt(helper_errors): + setters = {} + arm, b_done = threading.Event(), threading.Event() + + @reacton.component + def Test(label="old"): + value, setters["value"] = reacton.use_state(0) + + def effect(): + if arm.is_set(): + arm.clear() + + def b_target(): + setters["value"](1) + b_done.set() + + threading.Thread(target=b_target, name="B", daemon=True).start() + b_done.wait(TIMEOUT) + + reacton.use_effect(effect) + return w.Button(description=f"{label} {value}") + + box, rc = reacton.render(Test(), handle_error=False) + drain_line = source_line(core._RenderContext.render, "self._drain_render_requests()") + + def trace(frame, event, arg): + if event == "line" and frame.f_code is core._RenderContext.render.__code__ and frame.f_lineno == drain_line: + raise Injected() + return trace + + def a_target(): + arm.set() + sys.settrace(trace) + try: + rc.render(Test(label="new")) + finally: + sys.settrace(None) + + a = Worker(a_target, "A") + a.start() + a.join(TIMEOUT) + assert not a.is_alive(), stacks() + assert isinstance(a.error, Injected), repr(a.error) + assert b_done.is_set(), "the request during A's hold did not run" + assert wait_until( + lambda: not getattr(rc, "_render_requested", False) and text(box) == "new 1", TIMEOUT + ), f"the request stayed pending: _render_requested={getattr(rc, '_render_requested', None)}, shows {text(box)!r}" + assert not helper_errors, helper_errors + rc.close() + + +def test_explicit_render_is_not_starved_by_derived_own_state(): + """An effect derives own state from the streamed value, so every pass is followed by an own + re-render. Once the render budget hands the stream over to a helper, W's render() must + still return.""" + setters = {} + streaming, w_returned = threading.Event(), threading.Event() + go, done = threading.Semaphore(0), threading.Semaphore(0) + + @reacton.component + def Test(label="old"): + trigger, setters["trigger"] = reacton.use_state(0) + progress, setters["progress"] = reacton.use_state(0) + _derived, set_derived = reacton.use_state(0) + reacton.use_effect(lambda: set_derived(progress), [progress]) + if streaming.is_set() and not w_returned.is_set() and threading.current_thread().name != "T": + go.release() + done.acquire(timeout=1) + return w.Button(description=f"{label} {trigger} {progress}") + + box, rc = reacton.render(Test(), handle_error=False) + count = [0] + + def t_target(): + deadline = time.monotonic() + 40 + while not w_returned.is_set() and time.monotonic() < deadline: + if go.acquire(timeout=0.1): + count[0] += 1 + setters["progress"](count[0]) + done.release() + + def k_target(): + streaming.set() + setters["trigger"](1) + + t = Worker(t_target, "T") + t.start() + k = Worker(k_target, "K") + k.start() + assert wait_until(lambda: count[0] >= 150, 30), "the stream did not start" # past the hand-over to a helper + finished, wt = call_in_thread(lambda: rc.render(Test(label="new")), "W", timeout=10) + w_returned.set() + wt.join(45) + k.join(45) + t.join(45) + assert finished, f"W's render() waited while a helper kept rendering ({count[0]} updates)\n" + stacks() + assert wait_until(lambda: text(box).startswith("new 1"), 5), text(box) + rc.close() + + +def test_foreign_request_consumed_by_an_own_pass_is_not_a_render_loop(helper_errors): + """An effect derives own state from T's value, so the renderer renders own passes. T changes + state at the start of every pass (at the log calls "Render phase: " and "Entering nested + render phase"). 60 times. That is not a render loop: no "Too many renders", and T's last + value is rendered.""" + n = 60 + setters = {} + go, done = threading.Semaphore(0), threading.Semaphore(0) + remaining = [n] + armed = threading.Event() + + @reacton.component + def Test(): + trigger, setters["trigger"] = reacton.use_state(0) + progress, setters["progress"] = reacton.use_state(0) + derived, set_derived = reacton.use_state(0) + reacton.use_effect(lambda: set_derived(progress + trigger), [progress, trigger]) + return w.Button(description=f"{trigger} {progress} {derived}") + + box, rc = reacton.render(Test(), handle_error=False) + + def at_pass_start(): + if armed.is_set() and remaining[0] > 0: + go.release() + done.acquire(timeout=TIMEOUT) + + uninstall = install(EveryTimeHook(["Entering nested render phase", "Render phase: "], at_pass_start, skip="T")) + + def t_target(): + for i in range(n): + assert go.acquire(timeout=TIMEOUT) + setters["progress"](i + 1) + remaining[0] -= 1 + done.release() + + def k_target(): + armed.set() + setters["trigger"](1) + + try: + t = Worker(t_target, "T") + t.start() + finished, k = call_in_thread(k_target, "K", timeout=60) + t.join(60) + assert finished, stacks() + assert k.error is None, repr(k.error)[:300] + assert t.error is None, repr(t.error)[:300] + ok = wait_until(lambda: text(box) == f"1 {n} {n + 1}", 10) + assert not helper_errors, repr(helper_errors[0])[:300] + assert ok, text(box) + finally: + uninstall() + rc.close()