From 828152a759aac39d91756be488f80b391a37395f Mon Sep 17 00:00:00 2001 From: Yao Yao Date: Thu, 6 Aug 2026 06:48:49 +0000 Subject: [PATCH 1/6] [None][fix] reject nested control_action() instead of corrupting its handshake control_request_barrier and control_action_done are single-slot events shared by every control action. A control action whose body opened another one would have the inner exit clear the barrier and set done, releasing the executor loop while the outer body was still running -- the drain it asked for silently no longer holding. Nothing nests today: update_weights() reaches the reuse state through the undecorated self.engine.reset_prefix_cache(), the Ray worker's sleep()/wakeup() call no decorated sibling, and the base-worker control blocks only touch CUDA and memory. The guard is there so that stays true, and so a future caller that wraps one @control_action_decorator method in another fails at the point of the mistake rather than at whatever breaks later. Raise before enqueue_control_request() so a refused nesting attempt leaves no orphaned control request, and clear the flag in the existing finally so a body that raises does not wedge the executor. Register the new test file in l0_cpu.yml -- test-db lists files individually, so an unregistered unit test never runs in CI. Signed-off-by: Yao Yao --- tensorrt_llm/_torch/pyexecutor/py_executor.py | 20 ++++ .../integration/test_lists/test-db/l0_cpu.yml | 1 + .../test_control_action_reentrancy.py | 109 ++++++++++++++++++ 3 files changed, 130 insertions(+) create mode 100644 tests/unittest/executor/test_control_action_reentrancy.py diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py index 6cf98b759f02..d57e379a3854 100644 --- a/tensorrt_llm/_torch/pyexecutor/py_executor.py +++ b/tensorrt_llm/_torch/pyexecutor/py_executor.py @@ -897,6 +897,11 @@ def on_detected(): self.control_request_barrier = threading.Event() self.control_action_done = threading.Event() + # Guards against a control_action() body entering control_action() + # again. The two events above are shared, single-slot state, so a + # nested entry would clear the outer action's barrier and let the + # executor loop resume while the outer body is still running. + self._control_action_in_progress = False self._active_control_id: Optional[str] = None self._sleep_wakeup_pending_aborts: Dict[str, str] = {} self._sleep_wakeup_pending_abort_lock = threading.Lock() @@ -4456,17 +4461,32 @@ def control_action(self, In-flight requests keep their KV caches across the action; same-batch requests fetched after the sentinel are parked until the ``with`` block exits. + + Not re-entrant: ``control_request_barrier`` and ``control_action_done`` + are single-slot events, so a nested entry would clear the outer + action's barrier and release the executor loop while the outer body is + still running. Nesting is rejected rather than allowed to corrupt that + handshake. """ + if self._control_action_in_progress: + raise RuntimeError( + "control_action() is already in progress; it is not re-entrant. " + "A control action must not invoke another control action - call " + "the undecorated operation instead (e.g. self.engine.() " + "rather than the @control_action_decorator wrapper).") + if self.dist.rank == 0: self.executor_request_queue.enqueue_control_request( drain=drain, control_id=control_id) self.control_request_barrier.wait() + self._control_action_in_progress = True try: yield self finally: + self._control_action_in_progress = False self.control_action_done.set() self.control_request_barrier.clear() diff --git a/tests/integration/test_lists/test-db/l0_cpu.yml b/tests/integration/test_lists/test-db/l0_cpu.yml index bf67c89236da..e5d359bca39c 100644 --- a/tests/integration/test_lists/test-db/l0_cpu.yml +++ b/tests/integration/test_lists/test-db/l0_cpu.yml @@ -28,6 +28,7 @@ l0_cpu: - unittest/bindings - unittest/disaggregated - unittest/executor/test_base_worker.py ISOLATION + - unittest/executor/test_control_action_reentrancy.py - unittest/executor/test_fatal_error_health_check.py - unittest/executor/test_ipc.py - unittest/executor/test_rpc.py diff --git a/tests/unittest/executor/test_control_action_reentrancy.py b/tests/unittest/executor/test_control_action_reentrancy.py new file mode 100644 index 000000000000..ea8f23553b5d --- /dev/null +++ b/tests/unittest/executor/test_control_action_reentrancy.py @@ -0,0 +1,109 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Guard tests for PyExecutor.control_action() re-entrancy. + +No GPU, MPI or model weights required: the executor is built with +object.__new__ and only the handful of attributes control_action() touches. +""" + +import threading +from types import SimpleNamespace + +import pytest + +pytestmark = pytest.mark.cpu_only + + +def _make_executor(): + """Build a PyExecutor shell exercising only control_action()'s state. + + rank is 1 so the rank-0 enqueue path is skipped, and the barrier starts + set so wait() returns immediately instead of blocking on a real executor + loop. + """ + from tensorrt_llm._torch.pyexecutor.py_executor import PyExecutor + + ex = object.__new__(PyExecutor) + ex.dist = SimpleNamespace(rank=1) + ex.executor_request_queue = None # unreachable while rank != 0 + ex.control_request_barrier = threading.Event() + ex.control_request_barrier.set() + ex.control_action_done = threading.Event() + ex._control_action_in_progress = False + return ex + + +def test_nested_control_action_is_rejected(): + """A control action must not open another control action.""" + ex = _make_executor() + + with ex.control_action(): + assert ex._control_action_in_progress + with pytest.raises(RuntimeError, match="not re-entrant"): + with ex.control_action(): + pytest.fail("nested control_action() should not have yielded") + + +def test_rejected_nesting_leaves_the_outer_handshake_intact(): + """The rejection must not touch the events the outer action still owns. + + control_request_barrier / control_action_done are single-slot state: if a + refused nested entry cleared the barrier or set done, the executor loop + would resume while the outer body is still running -- the very corruption + the guard exists to prevent. + """ + ex = _make_executor() + + with ex.control_action(): + with pytest.raises(RuntimeError): + with ex.control_action(): + pass + # Still mid-outer-action: barrier held, completion not signalled. + assert ex.control_request_barrier.is_set() + assert not ex.control_action_done.is_set() + assert ex._control_action_in_progress + + # Outer exit performs the handshake exactly once. + assert ex.control_action_done.is_set() + assert not ex.control_request_barrier.is_set() + assert not ex._control_action_in_progress + + +def test_sequential_control_actions_are_allowed(): + """The guard rejects nesting, not repeated use.""" + ex = _make_executor() + + for _ in range(3): + ex.control_request_barrier.set() + ex.control_action_done.clear() + with ex.control_action(): + assert ex._control_action_in_progress + assert not ex._control_action_in_progress + + +def test_flag_is_cleared_when_the_body_raises(): + """An exception inside the body must not leave the executor wedged.""" + ex = _make_executor() + + with pytest.raises(ValueError): + with ex.control_action(): + raise ValueError("boom") + + assert not ex._control_action_in_progress + + # A later control action still works. + ex.control_request_barrier.set() + with ex.control_action(): + assert ex._control_action_in_progress From bad27767ec9e1fbfbdfeaebf9f8c8f227ad0db75 Mon Sep 17 00:00:00 2001 From: Yao Yao Date: Thu, 6 Aug 2026 14:20:37 +0000 Subject: [PATCH 2/6] [None][fix] serialise control_action() with a lock, not a bool Review of the first cut pointed out the flag was not atomic, and BowenFu found the evidence that settles it: base_worker.py:715 already documents this exact race and works around it by hand with _sleep_wakeup_lock -- "control_action() uses an Event-based barrier, not a mutex, so two concurrent callers can both pass the barrier". That workaround covers three call sites; the @control_action_decorator path takes no lock at all, so the five decorated methods can race each other or a sleep()/wakeup(). An Event is a broadcast, not a token: set() releases every waiter. So two threads could both clear control_request_barrier and both set control_action_done, releasing the executor loop while one body still ran -- the corruption the guard is supposed to prevent, reachable by the other of its two routes. Replace the bool with a real mutex, which is the primitive the handshake was missing. Concurrent callers serialise rather than fail: that is what the existing _sleep_wakeup_lock sites already rely on, and refusing them would report contention as re-entrancy. Only a nested call from the thread already holding the lock raises, since blocking there would deadlock; it is distinguished by thread ident rather than by taking the lock again. _sleep_wakeup_lock is now redundant but left in place -- removing it touches multi-rank sleep/wakeup, which deserves its own change. Signed-off-by: Yao Yao --- tensorrt_llm/_torch/pyexecutor/py_executor.py | 59 +++++--- .../test_control_action_reentrancy.py | 143 +++++++++++++++--- 2 files changed, 159 insertions(+), 43 deletions(-) diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py index d57e379a3854..f1e4595fb5a6 100644 --- a/tensorrt_llm/_torch/pyexecutor/py_executor.py +++ b/tensorrt_llm/_torch/pyexecutor/py_executor.py @@ -897,11 +897,12 @@ def on_detected(): self.control_request_barrier = threading.Event() self.control_action_done = threading.Event() - # Guards against a control_action() body entering control_action() - # again. The two events above are shared, single-slot state, so a - # nested entry would clear the outer action's barrier and let the - # executor loop resume while the outer body is still running. - self._control_action_in_progress = False + # The two events above are a broadcast handshake, not a mutex: set() + # releases every waiter, so callers are serialised here instead. + self._control_action_lock = threading.Lock() + # Holder's thread ident, so a nested call can be rejected rather than + # deadlock on the non-reentrant lock above. + self._control_action_owner: Optional[int] = None self._active_control_id: Optional[str] = None self._sleep_wakeup_pending_aborts: Dict[str, str] = {} self._sleep_wakeup_pending_abort_lock = threading.Lock() @@ -4462,33 +4463,41 @@ def control_action(self, action; same-batch requests fetched after the sentinel are parked until the ``with`` block exits. - Not re-entrant: ``control_request_barrier`` and ``control_action_done`` - are single-slot events, so a nested entry would clear the outer - action's barrier and release the executor loop while the outer body is - still running. Nesting is rejected rather than allowed to corrupt that - handshake. + Mutually exclusive and not re-entrant: the two events are a broadcast + handshake rather than a mutex, so concurrent callers would both pass + the barrier and both clear it, releasing the executor loop while one + body still runs. Callers therefore serialise on + ``_control_action_lock``; a nested call from the holding thread is + rejected rather than blocked, which would deadlock. """ - if self._control_action_in_progress: + # Unsynchronised read is sound: only the owning thread writes its own + # ident, and clears it before releasing. So this can match only for a + # thread that really holds the lock; any other value falls through to + # the lock, which does the actual exclusion. + if self._control_action_owner == threading.get_ident(): raise RuntimeError( - "control_action() is already in progress; it is not re-entrant. " - "A control action must not invoke another control action - call " - "the undecorated operation instead (e.g. self.engine.() " + "control_action() is not re-entrant: this thread already holds " + "one. A control action must not invoke another control action - " + "call the undecorated operation instead (e.g. self.engine.() " "rather than the @control_action_decorator wrapper).") - if self.dist.rank == 0: - self.executor_request_queue.enqueue_control_request( - drain=drain, control_id=control_id) + with self._control_action_lock: + self._control_action_owner = threading.get_ident() + try: + if self.dist.rank == 0: + self.executor_request_queue.enqueue_control_request( + drain=drain, control_id=control_id) - self.control_request_barrier.wait() + self.control_request_barrier.wait() - self._control_action_in_progress = True - try: - yield self - finally: - self._control_action_in_progress = False - self.control_action_done.set() - self.control_request_barrier.clear() + try: + yield self + finally: + self.control_action_done.set() + self.control_request_barrier.clear() + finally: + self._control_action_owner = None def _wait_for_model_engine_input_copy(self): wait_for_input_copy = getattr(self.model_engine, "wait_for_input_copy", diff --git a/tests/unittest/executor/test_control_action_reentrancy.py b/tests/unittest/executor/test_control_action_reentrancy.py index ea8f23553b5d..3382b5efc283 100644 --- a/tests/unittest/executor/test_control_action_reentrancy.py +++ b/tests/unittest/executor/test_control_action_reentrancy.py @@ -12,10 +12,15 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -"""Guard tests for PyExecutor.control_action() re-entrancy. +"""Guard tests for PyExecutor.control_action() admission. + +control_request_barrier / control_action_done are a broadcast handshake, not a +mutex: set() releases every waiter. So the method has to provide the exclusion +itself -- serialising callers from different threads, and rejecting a nested +call from the thread that already holds it (blocking there would deadlock). No GPU, MPI or model weights required: the executor is built with -object.__new__ and only the handful of attributes control_action() touches. +object.__new__ and only the attributes control_action() touches. """ import threading @@ -25,6 +30,10 @@ pytestmark = pytest.mark.cpu_only +# Long enough that a wrongly-blocked thread is unambiguous, short enough that a +# regression does not stall the suite. +_TIMEOUT = 10.0 + def _make_executor(): """Build a PyExecutor shell exercising only control_action()'s state. @@ -41,16 +50,21 @@ def _make_executor(): ex.control_request_barrier = threading.Event() ex.control_request_barrier.set() ex.control_action_done = threading.Event() - ex._control_action_in_progress = False + ex._control_action_lock = threading.Lock() + ex._control_action_owner = None return ex +# --------------------------------------------------------------------------- +# Nesting (same thread) -- must raise, because blocking would deadlock +# --------------------------------------------------------------------------- + + def test_nested_control_action_is_rejected(): - """A control action must not open another control action.""" ex = _make_executor() with ex.control_action(): - assert ex._control_action_in_progress + assert ex._control_action_owner == threading.get_ident() with pytest.raises(RuntimeError, match="not re-entrant"): with ex.control_action(): pytest.fail("nested control_action() should not have yielded") @@ -59,9 +73,8 @@ def test_nested_control_action_is_rejected(): def test_rejected_nesting_leaves_the_outer_handshake_intact(): """The rejection must not touch the events the outer action still owns. - control_request_barrier / control_action_done are single-slot state: if a - refused nested entry cleared the barrier or set done, the executor loop - would resume while the outer body is still running -- the very corruption + If a refused nested entry cleared the barrier or set done, the executor + loop would resume while the outer body is still running -- the corruption the guard exists to prevent. """ ex = _make_executor() @@ -70,15 +83,109 @@ def test_rejected_nesting_leaves_the_outer_handshake_intact(): with pytest.raises(RuntimeError): with ex.control_action(): pass - # Still mid-outer-action: barrier held, completion not signalled. assert ex.control_request_barrier.is_set() assert not ex.control_action_done.is_set() - assert ex._control_action_in_progress + assert ex._control_action_owner == threading.get_ident() - # Outer exit performs the handshake exactly once. assert ex.control_action_done.is_set() assert not ex.control_request_barrier.is_set() - assert not ex._control_action_in_progress + assert ex._control_action_owner is None + assert not ex._control_action_lock.locked() + + +# --------------------------------------------------------------------------- +# Concurrency (different threads) -- must serialise, NOT raise +# --------------------------------------------------------------------------- + + +def test_concurrent_callers_serialise_and_do_not_raise(): + """A second thread waits its turn rather than being refused. + + Rejecting it would be wrong twice over: contention is not re-entrancy, and + the existing callers (base_worker's _sleep_wakeup_lock sites) rely on + concurrent control actions serialising. + """ + ex = _make_executor() + events = [] + errors = [] + first_inside = threading.Event() + release_first = threading.Event() + second_inside = threading.Event() + + def first(): + try: + with ex.control_action(): + events.append("first-enter") + first_inside.set() + release_first.wait(timeout=_TIMEOUT) + events.append("first-exit") + except BaseException as exc: # noqa: BLE001 - surfaced via assert below + errors.append(("first", exc)) + + def second(): + try: + with ex.control_action(): + events.append("second-enter") + second_inside.set() + except BaseException as exc: # noqa: BLE001 + errors.append(("second", exc)) + + t1 = threading.Thread(target=first) + t1.start() + assert first_inside.wait(timeout=_TIMEOUT), "first thread never entered" + + t2 = threading.Thread(target=second) + t2.start() + + # While the first action holds the lock the second must be blocked, not + # raising and not running the body. + assert not second_inside.wait(timeout=0.5), "second thread entered concurrently" + assert errors == [], f"a concurrent caller was refused: {errors}" + + release_first.set() + t1.join(timeout=_TIMEOUT) + + # The first action's cleanup cleared the barrier; the real executor loop + # re-arms it for the next control request, so do that here. + ex.control_request_barrier.set() + assert second_inside.wait(timeout=_TIMEOUT), "second thread never got its turn" + t2.join(timeout=_TIMEOUT) + + assert errors == [] + assert events == ["first-enter", "first-exit", "second-enter"] + assert ex._control_action_owner is None + assert not ex._control_action_lock.locked() + + +def test_owner_is_per_thread_so_a_sibling_thread_is_not_mistaken_for_nesting(): + """The nesting check keys on thread ident, not a global flag.""" + ex = _make_executor() + seen = {} + first_inside = threading.Event() + release_first = threading.Event() + + def first(): + with ex.control_action(): + first_inside.set() + release_first.wait(timeout=_TIMEOUT) + + t1 = threading.Thread(target=first) + t1.start() + assert first_inside.wait(timeout=_TIMEOUT) + + # From the main thread the owner is someone else, so this is contention, + # not nesting -- the pre-check must not fire. + seen["owner"] = ex._control_action_owner + assert seen["owner"] not in (None, threading.get_ident()) + + release_first.set() + t1.join(timeout=_TIMEOUT) + assert ex._control_action_owner is None + + +# --------------------------------------------------------------------------- +# Cleanup +# --------------------------------------------------------------------------- def test_sequential_control_actions_are_allowed(): @@ -89,11 +196,11 @@ def test_sequential_control_actions_are_allowed(): ex.control_request_barrier.set() ex.control_action_done.clear() with ex.control_action(): - assert ex._control_action_in_progress - assert not ex._control_action_in_progress + assert ex._control_action_owner == threading.get_ident() + assert ex._control_action_owner is None -def test_flag_is_cleared_when_the_body_raises(): +def test_lock_and_owner_are_released_when_the_body_raises(): """An exception inside the body must not leave the executor wedged.""" ex = _make_executor() @@ -101,9 +208,9 @@ def test_flag_is_cleared_when_the_body_raises(): with ex.control_action(): raise ValueError("boom") - assert not ex._control_action_in_progress + assert ex._control_action_owner is None + assert not ex._control_action_lock.locked() - # A later control action still works. ex.control_request_barrier.set() with ex.control_action(): - assert ex._control_action_in_progress + assert ex._control_action_owner == threading.get_ident() From 9d8139cd4e6edadc8b387d02e86a21b2fd59ec29 Mon Sep 17 00:00:00 2001 From: Yao Yao Date: Thu, 6 Aug 2026 14:33:59 +0000 Subject: [PATCH 3/6] [None][test] annotate control_action() guard test functions CODING_GUIDELINES.md requires a return annotation on every function. Add -> None to the tests and their worker closures, and a TYPE_CHECKING-quoted PyExecutor return type to _make_executor() so the deliberately deferred runtime import stays inside the function body. Signed-off-by: Yao Yao --- .../test_control_action_reentrancy.py | 24 +++++++++++-------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/tests/unittest/executor/test_control_action_reentrancy.py b/tests/unittest/executor/test_control_action_reentrancy.py index 3382b5efc283..dd25013302df 100644 --- a/tests/unittest/executor/test_control_action_reentrancy.py +++ b/tests/unittest/executor/test_control_action_reentrancy.py @@ -25,9 +25,13 @@ import threading from types import SimpleNamespace +from typing import TYPE_CHECKING import pytest +if TYPE_CHECKING: + from tensorrt_llm._torch.pyexecutor.py_executor import PyExecutor + pytestmark = pytest.mark.cpu_only # Long enough that a wrongly-blocked thread is unambiguous, short enough that a @@ -35,7 +39,7 @@ _TIMEOUT = 10.0 -def _make_executor(): +def _make_executor() -> "PyExecutor": """Build a PyExecutor shell exercising only control_action()'s state. rank is 1 so the rank-0 enqueue path is skipped, and the barrier starts @@ -60,7 +64,7 @@ def _make_executor(): # --------------------------------------------------------------------------- -def test_nested_control_action_is_rejected(): +def test_nested_control_action_is_rejected() -> None: ex = _make_executor() with ex.control_action(): @@ -70,7 +74,7 @@ def test_nested_control_action_is_rejected(): pytest.fail("nested control_action() should not have yielded") -def test_rejected_nesting_leaves_the_outer_handshake_intact(): +def test_rejected_nesting_leaves_the_outer_handshake_intact() -> None: """The rejection must not touch the events the outer action still owns. If a refused nested entry cleared the barrier or set done, the executor @@ -98,7 +102,7 @@ def test_rejected_nesting_leaves_the_outer_handshake_intact(): # --------------------------------------------------------------------------- -def test_concurrent_callers_serialise_and_do_not_raise(): +def test_concurrent_callers_serialise_and_do_not_raise() -> None: """A second thread waits its turn rather than being refused. Rejecting it would be wrong twice over: contention is not re-entrancy, and @@ -112,7 +116,7 @@ def test_concurrent_callers_serialise_and_do_not_raise(): release_first = threading.Event() second_inside = threading.Event() - def first(): + def first() -> None: try: with ex.control_action(): events.append("first-enter") @@ -122,7 +126,7 @@ def first(): except BaseException as exc: # noqa: BLE001 - surfaced via assert below errors.append(("first", exc)) - def second(): + def second() -> None: try: with ex.control_action(): events.append("second-enter") @@ -157,14 +161,14 @@ def second(): assert not ex._control_action_lock.locked() -def test_owner_is_per_thread_so_a_sibling_thread_is_not_mistaken_for_nesting(): +def test_owner_is_per_thread_so_a_sibling_thread_is_not_mistaken_for_nesting() -> None: """The nesting check keys on thread ident, not a global flag.""" ex = _make_executor() seen = {} first_inside = threading.Event() release_first = threading.Event() - def first(): + def first() -> None: with ex.control_action(): first_inside.set() release_first.wait(timeout=_TIMEOUT) @@ -188,7 +192,7 @@ def first(): # --------------------------------------------------------------------------- -def test_sequential_control_actions_are_allowed(): +def test_sequential_control_actions_are_allowed() -> None: """The guard rejects nesting, not repeated use.""" ex = _make_executor() @@ -200,7 +204,7 @@ def test_sequential_control_actions_are_allowed(): assert ex._control_action_owner is None -def test_lock_and_owner_are_released_when_the_body_raises(): +def test_lock_and_owner_are_released_when_the_body_raises() -> None: """An exception inside the body must not leave the executor wedged.""" ex = _make_executor() From b5fd58aef5d98f2439a36bcfd67b6c893a97b58d Mon Sep 17 00:00:00 2001 From: Yao Yao Date: Fri, 7 Aug 2026 02:53:08 +0000 Subject: [PATCH 4/6] [None][fix] bound control_action()'s barrier wait and close review gaps Review follow-ups from brnguyen2 on #17346. The lock is held across control_request_barrier.wait(), which was unbounded. _handle_control_request pulses set(); clear() on the aborted-control-request path without waiting for control_action_done, so a caller that has not reached the wait yet misses the edge. Previously that stalled one caller; with the lock it wedged every later control action silently. Wait in bounded slices instead: shutdown or a dead executor loop fails within one poll interval, and a lost edge fails after _CONTROL_BARRIER_TIMEOUT_S. The timeout is deliberately generous because drain=True is unbounded by design - it is a wedge-breaker, not a latency budget. Also note in the docstring that the lock is local mutual exclusion only and does not order control actions across ranks, and add a rank=0 test with a recording queue asserting a rejected nested call enqueues nothing - the previous tests all ran at rank=1, where the enqueue is skipped, so moving the guard below it went undetected. Signed-off-by: Yao Yao --- tensorrt_llm/_torch/pyexecutor/py_executor.py | 52 ++++++++- .../test_control_action_reentrancy.py | 101 ++++++++++++++++-- 2 files changed, 146 insertions(+), 7 deletions(-) diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py index f1e4595fb5a6..166b816d517d 100644 --- a/tensorrt_llm/_torch/pyexecutor/py_executor.py +++ b/tensorrt_llm/_torch/pyexecutor/py_executor.py @@ -155,6 +155,16 @@ class _SleepWakeupAction(StrEnum): _SLEEP_WAKEUP_ACK_TIMEOUT_S = 30.0 _SLEEP_WAKEUP_ACK_POLL_INTERVAL_S = 0.01 +# Backstop for control_action()'s barrier wait. Deliberately generous: with +# drain=True the sentinel is parked until active_requests and waiting_queue +# empty, so a legitimate wait is bounded only by how long in-flight requests +# take. This is not a latency budget, it is a last resort that turns a lost +# barrier edge into a loud failure instead of a permanently wedged lock. +# Shutdown and a dead executor loop are detected by the liveness poll below, +# so they fail in _CONTROL_BARRIER_POLL_INTERVAL_S rather than waiting this out. +_CONTROL_BARRIER_TIMEOUT_S = 1800.0 +_CONTROL_BARRIER_POLL_INTERVAL_S = 0.5 + def _sleep_wakeup_ack_ready(comm, source: int, tag: _SleepWakeupTag) -> bool: """Return whether an ACK is ready without blocking on recv.""" @@ -4469,6 +4479,14 @@ def control_action(self, body still runs. Callers therefore serialise on ``_control_action_lock``; a nested call from the holding thread is rejected rather than blocked, which would deadlock. + + The lock provides *local* mutual exclusion only. It does not order + control actions across ranks: only rank 0 enqueues, and the order in + which callers on rank N acquire the lock is not tied to rank 0's + enqueue order, so concurrent control actions could still pair + different bodies with the same barrier cycle on different ranks. + Cross-rank correctness relies on rank 0 being the single enqueue + point - do not issue concurrent collective control actions. """ # Unsynchronised read is sound: only the owning thread writes its own @@ -4489,7 +4507,7 @@ def control_action(self, self.executor_request_queue.enqueue_control_request( drain=drain, control_id=control_id) - self.control_request_barrier.wait() + self._wait_for_control_barrier(control_id) try: yield self @@ -4499,6 +4517,38 @@ def control_action(self, finally: self._control_action_owner = None + def _wait_for_control_barrier(self, control_id: Optional[str]) -> None: + """Wait for the executor loop to fire our control request. + + Bounded rather than a bare ``wait()``, because the barrier edge can be + lost: ``_handle_control_request`` pulses ``set()`` then ``clear()`` on + the aborted-control-request path without waiting for + ``control_action_done``. A caller that has not reached the wait yet + when that happens would otherwise block forever *while holding* + ``_control_action_lock``, wedging every later control action with no + diagnostic. Failing loudly here blames the caller that actually lost + the edge. + """ + deadline = time.monotonic() + _CONTROL_BARRIER_TIMEOUT_S + while not self.control_request_barrier.wait( + timeout=_CONTROL_BARRIER_POLL_INTERVAL_S): + # The loop that would set the barrier is gone; waiting out the full + # timeout would serve no purpose. + worker = getattr(self, "worker_thread", None) + if self.shutdown_event.is_set() or (worker is not None + and not worker.is_alive()): + raise RuntimeError( + "control_action() barrier never fired: the executor loop " + f"is shut down (control_id={control_id}). The control " + "request cannot be serviced.") + if time.monotonic() >= deadline: + raise RuntimeError( + "control_action() timed out after " + f"{_CONTROL_BARRIER_TIMEOUT_S}s waiting for the control " + f"request barrier (control_id={control_id}). The barrier " + "edge was likely lost - e.g. the request was aborted " + "between enqueue and this wait.") + def _wait_for_model_engine_input_copy(self): wait_for_input_copy = getattr(self.model_engine, "wait_for_input_copy", None) diff --git a/tests/unittest/executor/test_control_action_reentrancy.py b/tests/unittest/executor/test_control_action_reentrancy.py index dd25013302df..1aa0747edba6 100644 --- a/tests/unittest/executor/test_control_action_reentrancy.py +++ b/tests/unittest/executor/test_control_action_reentrancy.py @@ -24,6 +24,7 @@ """ import threading +import time from types import SimpleNamespace from typing import TYPE_CHECKING @@ -39,26 +40,38 @@ _TIMEOUT = 10.0 -def _make_executor() -> "PyExecutor": +def _make_executor(rank: int = 1, queue: object = None) -> "PyExecutor": """Build a PyExecutor shell exercising only control_action()'s state. - rank is 1 so the rank-0 enqueue path is skipped, and the barrier starts - set so wait() returns immediately instead of blocking on a real executor - loop. + rank defaults to 1 so the rank-0 enqueue path is skipped, and the barrier + starts set so the wait returns immediately instead of blocking on a real + executor loop. Pass rank=0 with a recording queue to exercise enqueue + ordering. """ from tensorrt_llm._torch.pyexecutor.py_executor import PyExecutor ex = object.__new__(PyExecutor) - ex.dist = SimpleNamespace(rank=1) - ex.executor_request_queue = None # unreachable while rank != 0 + ex.dist = SimpleNamespace(rank=rank) + ex.executor_request_queue = queue # unreachable while rank != 0 ex.control_request_barrier = threading.Event() ex.control_request_barrier.set() ex.control_action_done = threading.Event() + ex.shutdown_event = threading.Event() ex._control_action_lock = threading.Lock() ex._control_action_owner = None return ex +class _RecordingQueue: + """Records enqueue_control_request() calls so ordering can be asserted.""" + + def __init__(self) -> None: + self.calls: list = [] + + def enqueue_control_request(self, **kwargs: object) -> None: + self.calls.append(kwargs) + + # --------------------------------------------------------------------------- # Nesting (same thread) -- must raise, because blocking would deadlock # --------------------------------------------------------------------------- @@ -97,6 +110,31 @@ def test_rejected_nesting_leaves_the_outer_handshake_intact() -> None: assert not ex._control_action_lock.locked() +def test_rejected_nesting_enqueues_no_control_request() -> None: + """On rank 0 the guard must fire BEFORE enqueue_control_request(). + + This is the property the change exists for: a refused nesting attempt must + not leave an orphaned sentinel in the queue for the executor loop to fire + at a body that never runs. Needs rank=0 -- with rank=1 the enqueue is + skipped entirely, so moving the guard below it would go unnoticed. + """ + queue = _RecordingQueue() + ex = _make_executor(rank=0, queue=queue) + + with ex.control_action(control_id="outer"): + assert len(queue.calls) == 1, "outer action should enqueue exactly once" + with pytest.raises(RuntimeError, match="not re-entrant"): + with ex.control_action(control_id="nested"): + pytest.fail("nested control_action() should not have yielded") + # The decisive assertion: still one call, so the rejected nested entry + # enqueued nothing. + assert len(queue.calls) == 1, ( + f"rejected nesting left an orphaned control request: {queue.calls}" + ) + + assert [c.get("control_id") for c in queue.calls] == ["outer"] + + # --------------------------------------------------------------------------- # Concurrency (different threads) -- must serialise, NOT raise # --------------------------------------------------------------------------- @@ -204,6 +242,57 @@ def test_sequential_control_actions_are_allowed() -> None: assert ex._control_action_owner is None +# --------------------------------------------------------------------------- +# Bounded barrier wait -- a lost edge must fail loudly, not wedge the lock +# --------------------------------------------------------------------------- + + +def test_lost_barrier_edge_fails_loudly_instead_of_wedging_the_lock() -> None: + """Reproduces the aborted-control-request pulse. + + _handle_control_request does set(); clear() on the abort path without + waiting for control_action_done. A caller arriving after that pulse sees + a cleared barrier and would block forever while holding the lock, wedging + every later control action. It must raise instead, and release the lock. + """ + from tensorrt_llm._torch.pyexecutor import py_executor + + ex = _make_executor() + ex.control_request_barrier.clear() # the edge was pulsed and missed + + with pytest.MonkeyPatch.context() as mp: + mp.setattr(py_executor, "_CONTROL_BARRIER_TIMEOUT_S", 0.3) + mp.setattr(py_executor, "_CONTROL_BARRIER_POLL_INTERVAL_S", 0.05) + with pytest.raises(RuntimeError, match="timed out"): + with ex.control_action(control_id="lost-edge"): + pytest.fail("should not have yielded on a lost barrier edge") + + # The whole point: the next caller is not wedged behind us. + assert not ex._control_action_lock.locked() + assert ex._control_action_owner is None + + +def test_shutdown_fails_fast_without_waiting_out_the_timeout() -> None: + """A dead executor loop must not cost a full timeout to discover.""" + from tensorrt_llm._torch.pyexecutor import py_executor + + ex = _make_executor() + ex.control_request_barrier.clear() + ex.shutdown_event.set() + + with pytest.MonkeyPatch.context() as mp: + # Large timeout: if shutdown were not detected this test would hang. + mp.setattr(py_executor, "_CONTROL_BARRIER_TIMEOUT_S", 300.0) + mp.setattr(py_executor, "_CONTROL_BARRIER_POLL_INTERVAL_S", 0.05) + start = time.monotonic() + with pytest.raises(RuntimeError, match="shut down"): + with ex.control_action(control_id="after-shutdown"): + pytest.fail("should not have yielded after shutdown") + assert time.monotonic() - start < 5.0, "did not fail fast on shutdown" + + assert not ex._control_action_lock.locked() + + def test_lock_and_owner_are_released_when_the_body_raises() -> None: """An exception inside the body must not leave the executor wedged.""" ex = _make_executor() From 1a04578ec09f4306f485837adfc73ef931749bdf Mon Sep 17 00:00:00 2001 From: Yao Yao Date: Fri, 7 Aug 2026 03:11:18 +0000 Subject: [PATCH 5/6] [None][test] cover control_action()'s dead-worker fast-fail branch _wait_for_control_barrier fast-fails on two conditions: shutdown_event being set, or worker_thread having died. Only the first was covered -- _make_executor never assigns worker_thread, so getattr(...) returned None in every test and the liveness half of the disjunct always evaluated False. Removing it entirely kept the suite green. Add a test that stubs worker_thread with is_alive() -> False while shutdown_event stays clear, so the branch is reachable, and assert the call fails well inside the (monkeypatched, 300s) barrier timeout rather than waiting it out. Also assert _control_action_owner is released in the existing shutdown test, matching the other cleanup tests. Signed-off-by: Yao Yao --- .../test_control_action_reentrancy.py | 31 +++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/tests/unittest/executor/test_control_action_reentrancy.py b/tests/unittest/executor/test_control_action_reentrancy.py index 1aa0747edba6..22c2f7ae2346 100644 --- a/tests/unittest/executor/test_control_action_reentrancy.py +++ b/tests/unittest/executor/test_control_action_reentrancy.py @@ -291,6 +291,37 @@ def test_shutdown_fails_fast_without_waiting_out_the_timeout() -> None: assert time.monotonic() - start < 5.0, "did not fail fast on shutdown" assert not ex._control_action_lock.locked() + assert ex._control_action_owner is None + + +def test_dead_worker_thread_fails_fast_without_waiting_out_the_timeout() -> None: + """A crashed executor loop must be caught with ``shutdown_event`` clear. + + ``shutdown_event`` only covers the orderly path. A worker that died + without setting it is the other half of the fast-fail condition, and is + reachable only through the liveness check -- ``_make_executor`` leaves + ``worker_thread`` unset, so every other test evaluates that half as + ``False`` and would not notice it breaking. + """ + from tensorrt_llm._torch.pyexecutor import py_executor + + ex = _make_executor() + ex.control_request_barrier.clear() + # shutdown_event stays CLEAR: only the worker liveness check can catch this. + ex.worker_thread = SimpleNamespace(is_alive=lambda: False) + + with pytest.MonkeyPatch.context() as mp: + # Large timeout: if the dead worker were not detected this would hang. + mp.setattr(py_executor, "_CONTROL_BARRIER_TIMEOUT_S", 300.0) + mp.setattr(py_executor, "_CONTROL_BARRIER_POLL_INTERVAL_S", 0.05) + start = time.monotonic() + with pytest.raises(RuntimeError, match="shut down"): + with ex.control_action(control_id="dead-worker"): + pytest.fail("should not have yielded with a dead worker") + assert time.monotonic() - start < 5.0, "did not fail fast on a dead worker" + + assert not ex._control_action_lock.locked() + assert ex._control_action_owner is None def test_lock_and_owner_are_released_when_the_body_raises() -> None: From a6b2aafe6275eaacbc84e70cd30805e1cdb624b3 Mon Sep 17 00:00:00 2001 From: Yao Yao Date: Fri, 7 Aug 2026 11:01:00 +0000 Subject: [PATCH 6/6] [None][fix] drop control_action()'s wall-clock barrier deadline The deadline stranded the sentinel. Rank 0 enqueues before waiting, so raising while the executor loop is still alive left the loop to pop that sentinel, set the barrier and block in the untimed control_action_done wait with no caller to answer - hanging the executor until the hang detector killed the job. That is worse than the stall the deadline was meant to avoid. Keep polling, but only report a loop that cannot serve the request at all. Both surviving conditions are safe because each implies no consumer is left to strand: shutdown_event is set in _executor_loop_cleanup(), i.e. only once the loop has exited, and a dead worker_thread cannot pop anything either. An orphaned sentinel is inert in both cases. A lost barrier edge therefore still blocks, as it did before this PR. Fixing that needs the abort routed back to the caller, not a timeout; the docstring says so to stop a deadline being reinstated later. Drop the test asserting the removed raise - without the deadline it would hang rather than fail. The two fast-fail tests now run off-thread and join with a timeout so a regression fails instead of hanging CI, and a new test pins the property that matters: with the loop alive the caller keeps waiting instead of stranding the sentinel. Signed-off-by: Yao Yao --- tensorrt_llm/_torch/pyexecutor/py_executor.py | 51 +++++---- .../test_control_action_reentrancy.py | 101 +++++++++++------- 2 files changed, 86 insertions(+), 66 deletions(-) diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py index 166b816d517d..41d738f0f6b5 100644 --- a/tensorrt_llm/_torch/pyexecutor/py_executor.py +++ b/tensorrt_llm/_torch/pyexecutor/py_executor.py @@ -155,14 +155,9 @@ class _SleepWakeupAction(StrEnum): _SLEEP_WAKEUP_ACK_TIMEOUT_S = 30.0 _SLEEP_WAKEUP_ACK_POLL_INTERVAL_S = 0.01 -# Backstop for control_action()'s barrier wait. Deliberately generous: with -# drain=True the sentinel is parked until active_requests and waiting_queue -# empty, so a legitimate wait is bounded only by how long in-flight requests -# take. This is not a latency budget, it is a last resort that turns a lost -# barrier edge into a loud failure instead of a permanently wedged lock. -# Shutdown and a dead executor loop are detected by the liveness poll below, -# so they fail in _CONTROL_BARRIER_POLL_INTERVAL_S rather than waiting this out. -_CONTROL_BARRIER_TIMEOUT_S = 1800.0 +# How often control_action() re-checks executor-loop liveness while waiting for +# the control request barrier. There is deliberately NO wall-clock deadline on +# that wait -- see _wait_for_control_barrier() for why one would be unsafe. _CONTROL_BARRIER_POLL_INTERVAL_S = 0.5 @@ -4520,20 +4515,31 @@ def control_action(self, def _wait_for_control_barrier(self, control_id: Optional[str]) -> None: """Wait for the executor loop to fire our control request. - Bounded rather than a bare ``wait()``, because the barrier edge can be - lost: ``_handle_control_request`` pulses ``set()`` then ``clear()`` on - the aborted-control-request path without waiting for - ``control_action_done``. A caller that has not reached the wait yet - when that happens would otherwise block forever *while holding* - ``_control_action_lock``, wedging every later control action with no - diagnostic. Failing loudly here blames the caller that actually lost - the edge. + Waits indefinitely, but polls so that a *dead* executor loop is + reported instead of hung on. + + There is deliberately no wall-clock deadline. Rank 0 has already + enqueued the sentinel by the time we get here, so bailing out while the + loop is still alive would strand it: the loop would later pop that + sentinel, ``set()`` the barrier and block in the untimed + ``control_action_done.wait()`` with no caller left to answer, hanging + the executor until the hang detector kills the job. That is strictly + worse than the stall a deadline would have avoided. + + The two conditions below are safe precisely because each one implies + there is no consumer left to strand: ``shutdown_event`` is set in + ``_executor_loop_cleanup()``, i.e. only once the loop has exited, and a + dead ``worker_thread`` cannot pop anything either. So an orphaned + sentinel is inert in both cases. + + Consequence: a lost barrier edge (``_handle_control_request`` pulses + ``set(); clear()`` on the aborted path without waiting for + ``control_action_done``) still blocks here while holding + ``_control_action_lock``. That is pre-existing behaviour; fixing it + needs the abort to be routed back to this caller, not a timeout. """ - deadline = time.monotonic() + _CONTROL_BARRIER_TIMEOUT_S while not self.control_request_barrier.wait( timeout=_CONTROL_BARRIER_POLL_INTERVAL_S): - # The loop that would set the barrier is gone; waiting out the full - # timeout would serve no purpose. worker = getattr(self, "worker_thread", None) if self.shutdown_event.is_set() or (worker is not None and not worker.is_alive()): @@ -4541,13 +4547,6 @@ def _wait_for_control_barrier(self, control_id: Optional[str]) -> None: "control_action() barrier never fired: the executor loop " f"is shut down (control_id={control_id}). The control " "request cannot be serviced.") - if time.monotonic() >= deadline: - raise RuntimeError( - "control_action() timed out after " - f"{_CONTROL_BARRIER_TIMEOUT_S}s waiting for the control " - f"request barrier (control_id={control_id}). The barrier " - "edge was likely lost - e.g. the request was aborted " - "between enqueue and this wait.") def _wait_for_model_engine_input_copy(self): wait_for_input_copy = getattr(self.model_engine, "wait_for_input_copy", diff --git a/tests/unittest/executor/test_control_action_reentrancy.py b/tests/unittest/executor/test_control_action_reentrancy.py index 22c2f7ae2346..09cec5b66d66 100644 --- a/tests/unittest/executor/test_control_action_reentrancy.py +++ b/tests/unittest/executor/test_control_action_reentrancy.py @@ -24,7 +24,6 @@ """ import threading -import time from types import SimpleNamespace from typing import TYPE_CHECKING @@ -243,37 +242,36 @@ def test_sequential_control_actions_are_allowed() -> None: # --------------------------------------------------------------------------- -# Bounded barrier wait -- a lost edge must fail loudly, not wedge the lock +# Dead executor loop -- must be reported, not waited on forever +# +# The wait is deliberately unbounded (a deadline would strand the already +# enqueued sentinel and hang the loop), so these are the ONLY two escapes. +# Each runs the call on a worker thread and joins with a timeout, so a +# regression surfaces as a test failure rather than a hung CI stage. # --------------------------------------------------------------------------- -def test_lost_barrier_edge_fails_loudly_instead_of_wedging_the_lock() -> None: - """Reproduces the aborted-control-request pulse. - - _handle_control_request does set(); clear() on the abort path without - waiting for control_action_done. A caller arriving after that pulse sees - a cleared barrier and would block forever while holding the lock, wedging - every later control action. It must raise instead, and release the lock. - """ - from tensorrt_llm._torch.pyexecutor import py_executor +def _run_expecting_shutdown_error(ex: "PyExecutor", control_id: str) -> str: + """Call control_action() off-thread; return "raised"/"yielded"/"hung".""" + outcome = [] - ex = _make_executor() - ex.control_request_barrier.clear() # the edge was pulsed and missed - - with pytest.MonkeyPatch.context() as mp: - mp.setattr(py_executor, "_CONTROL_BARRIER_TIMEOUT_S", 0.3) - mp.setattr(py_executor, "_CONTROL_BARRIER_POLL_INTERVAL_S", 0.05) - with pytest.raises(RuntimeError, match="timed out"): - with ex.control_action(control_id="lost-edge"): - pytest.fail("should not have yielded on a lost barrier edge") + def body() -> None: + try: + with ex.control_action(control_id=control_id): + outcome.append("yielded") + except RuntimeError as exc: + outcome.append("raised" if "shut down" in str(exc) else str(exc)) - # The whole point: the next caller is not wedged behind us. - assert not ex._control_action_lock.locked() - assert ex._control_action_owner is None + t = threading.Thread(target=body, daemon=True) + t.start() + t.join(timeout=_TIMEOUT) + if t.is_alive(): + return "hung" + return outcome[0] if outcome else "no-outcome" -def test_shutdown_fails_fast_without_waiting_out_the_timeout() -> None: - """A dead executor loop must not cost a full timeout to discover.""" +def test_shutdown_is_reported_instead_of_waited_on() -> None: + """shutdown_event set => the loop already exited, so nothing will fire.""" from tensorrt_llm._torch.pyexecutor import py_executor ex = _make_executor() @@ -281,20 +279,14 @@ def test_shutdown_fails_fast_without_waiting_out_the_timeout() -> None: ex.shutdown_event.set() with pytest.MonkeyPatch.context() as mp: - # Large timeout: if shutdown were not detected this test would hang. - mp.setattr(py_executor, "_CONTROL_BARRIER_TIMEOUT_S", 300.0) mp.setattr(py_executor, "_CONTROL_BARRIER_POLL_INTERVAL_S", 0.05) - start = time.monotonic() - with pytest.raises(RuntimeError, match="shut down"): - with ex.control_action(control_id="after-shutdown"): - pytest.fail("should not have yielded after shutdown") - assert time.monotonic() - start < 5.0, "did not fail fast on shutdown" + assert _run_expecting_shutdown_error(ex, "after-shutdown") == "raised" assert not ex._control_action_lock.locked() assert ex._control_action_owner is None -def test_dead_worker_thread_fails_fast_without_waiting_out_the_timeout() -> None: +def test_dead_worker_thread_is_reported_instead_of_waited_on() -> None: """A crashed executor loop must be caught with ``shutdown_event`` clear. ``shutdown_event`` only covers the orderly path. A worker that died @@ -311,14 +303,43 @@ def test_dead_worker_thread_fails_fast_without_waiting_out_the_timeout() -> None ex.worker_thread = SimpleNamespace(is_alive=lambda: False) with pytest.MonkeyPatch.context() as mp: - # Large timeout: if the dead worker were not detected this would hang. - mp.setattr(py_executor, "_CONTROL_BARRIER_TIMEOUT_S", 300.0) mp.setattr(py_executor, "_CONTROL_BARRIER_POLL_INTERVAL_S", 0.05) - start = time.monotonic() - with pytest.raises(RuntimeError, match="shut down"): - with ex.control_action(control_id="dead-worker"): - pytest.fail("should not have yielded with a dead worker") - assert time.monotonic() - start < 5.0, "did not fail fast on a dead worker" + assert _run_expecting_shutdown_error(ex, "dead-worker") == "raised" + + assert not ex._control_action_lock.locked() + assert ex._control_action_owner is None + + +def test_live_loop_keeps_waiting_rather_than_stranding_the_sentinel() -> None: + """With the loop alive, a missing barrier edge must NOT raise. + + Rank 0 has already enqueued the sentinel by this point. Bailing out would + leave the loop to pop it, set the barrier and block forever in the untimed + control_action_done.wait() with no caller left -- hanging the executor. + So the caller has to keep waiting, and resume once the edge arrives. + """ + from tensorrt_llm._torch.pyexecutor import py_executor + + ex = _make_executor() + ex.control_request_barrier.clear() # edge not fired yet + ex.worker_thread = SimpleNamespace(is_alive=lambda: True) # loop is healthy + entered = threading.Event() + + def body() -> None: + with ex.control_action(control_id="slow-drain"): + entered.set() + + with pytest.MonkeyPatch.context() as mp: + mp.setattr(py_executor, "_CONTROL_BARRIER_POLL_INTERVAL_S", 0.05) + t = threading.Thread(target=body, daemon=True) + t.start() + # Must still be waiting, not raising, well past several poll intervals. + assert not entered.wait(timeout=1.0), "caller bailed out on a live loop" + assert t.is_alive(), "caller must not have raised while the loop is alive" + + ex.control_request_barrier.set() # the loop finally fires it + assert entered.wait(timeout=_TIMEOUT), "caller never resumed" + t.join(timeout=_TIMEOUT) assert not ex._control_action_lock.locked() assert ex._control_action_owner is None