diff --git a/pyproject.toml b/pyproject.toml index 47430ed6..4106ff04 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -279,4 +279,25 @@ log_cli = false # https://docs.pytest.org/en/stable/reference/reference.html#confval-console_output_style console_output_style = 'progress' + +# `pexpect` (used only by the `devx` debugger REPL tests) allocates +# its child pty via stdlib `pty.py:os.forkpty()`, which CPython +# 3.12+ flags as unsafe in a multi-threaded process. This is NOT +# ours to fix at source — we never call `forkpty()`; pexpect does. +# UPSTREAM to actually resolve: pexpect would need to avoid +# `forkpty()` under multithreading (or perform the pty spawn before +# any thread starts); until then this single third-party warning is +# the one we filter rather than fix. +filterwarnings = [ + 'ignore:This process \(pid=\d+\) is multi-threaded, use of forkpty\(\):DeprecationWarning', + + # `tests/test_legacy_one_way_streaming.py` exists *specifically* + # to exercise the DEPRECATED `@tractor.stream` async-gen API, so + # its decoration-time `DeprecationWarning` is expected — and since + # it fires at import (module-level decorators) it can't be caught + # by a per-test mark, hence this test-wide ini filter. Real + # library users still see the warning; drop this once the legacy + # `@tractor.stream` / `Portal.open_stream_from()` API is removed. + 'ignore:`@tractor.stream decorated funcs:DeprecationWarning', +] # ------ tool.pytest ------ diff --git a/scripts/hung-dump.xsh b/scripts/hung-dump.xsh new file mode 100644 index 00000000..564f8467 --- /dev/null +++ b/scripts/hung-dump.xsh @@ -0,0 +1,89 @@ +""" +`hung-dump`: snapshot diagnostic state for a hung +`pytest`/`tractor` process tree. + +For each pid (and all `pgrep -P` descendants), prints +- `ps` forest header, +- `/proc//wchan` + `/proc//stack` (kernel-side + blocked-on syscall), +- `sudo py-spy dump` (python-side stack). + +Usage (xonsh): + source-foreign xonsh ./scripts/hung-dump.xsh + hung-dump [ ...] + +Or just: + source ./scripts/hung-dump.xsh + +Tip — pipe to a paste buffer: + hung-dump 1336765 |t /tmp/hung.log +""" + +def _hung_dump(args): + import subprocess as sp + from pathlib import Path + + if not args: + print('usage: hung-dump [ ...]') + return 1 + + def kids(pid): + try: + out = sp.check_output( + ['pgrep', '-P', str(pid)], + text=True, + ) + except sp.CalledProcessError: + return [] + return [int(p) for p in out.split() if p] + + def tree(pid): + out = [pid] + for k in kids(pid): + out.extend(tree(k)) + return out + + pids = [ + p + for r in (int(a) for a in args) + for p in tree(r) + ] + + print(f'# tree: {pids}') + print('\n## ps forest') + $[ps -o pid,ppid,pgid,stat,cmd -p @(','.join(map(str, pids)))] + + for p in pids: + print(f'\n## pid {p}') + for f in ('wchan', 'stack'): + path = Path(f'/proc/{p}/{f}') + try: + txt = path.read_text().rstrip() + print(f'-- /proc/{p}/{f} --\n{txt}') + except PermissionError: + # `stack` requires CAP_SYS_PTRACE; retry + # via sudo -n so creds-cached users don't + # block on prompt. + try: + txt = sp.check_output( + ['sudo', '-n', 'cat', str(path)], + text=True, + stderr=sp.DEVNULL, + ).rstrip() + print(f'-- /proc/{p}/{f} (via sudo) --\n{txt}') + except sp.CalledProcessError: + print( + f'-- /proc/{p}/{f}: ' + 'PermissionError (try `sudo true` first) --' + ) + except FileNotFoundError: + print(f'-- /proc/{p}/{f}: proc gone --') + + print(f'-- py-spy {p} --') + try: + $[sudo -n py-spy dump --pid @(p)] + except Exception as e: + print(f' (py-spy failed: {e})') + + +aliases['hung-dump'] = _hung_dump diff --git a/tests/devx/test_tooling.py b/tests/devx/test_tooling.py index 4a0e1d5a..4c64b988 100644 --- a/tests/devx/test_tooling.py +++ b/tests/devx/test_tooling.py @@ -92,7 +92,7 @@ def test_shield_pause( expect( child, # end-of-tree delimiter - "end-of-\('root'", + r"end-of-\('root'", ) _before: str = assert_before( child, @@ -157,7 +157,7 @@ def test_shield_pause( expect( child, # end-of-subactor's-tree delimiter - "end-of-\('hanger'", + r"end-of-\('hanger'", ) _before: str = assert_before( child, diff --git a/tests/discovery/test_multi_program.py b/tests/discovery/test_multi_program.py index 47e50e53..1030de6e 100644 --- a/tests/discovery/test_multi_program.py +++ b/tests/discovery/test_multi_program.py @@ -122,7 +122,13 @@ def test_register_duplicate_name( 'test_register_duplicate_name: ' '`wait_for_actor` returned' ) - assert portal.channel.uid in (p2.channel.uid, p1.channel.uid) + assert ( + portal.channel.aid.uid + in ( + p2.channel.aid.uid, + p1.channel.aid.uid, + ) + ) log.cancel( 'test_register_duplicate_name: ' @@ -242,8 +248,14 @@ def test_dup_name_cancel_cascade_escalates_to_hard_kill( # name; doesn't matter which one (registrar will # have last-wins semantics under same-name). async with tractor.wait_for_actor('doggy') as portal: - expected_uids = {p.channel.uid for p in portals} - assert portal.channel.uid in expected_uids + expected_uids = { + p.channel.aid.uid + for p in portals + } + assert ( + portal.channel.aid.uid + in expected_uids + ) # critical section: this MUST return within # `fail_after_s` even when one or more cancel-RPC diff --git a/tests/msg/test_ext_types_msgspec.py b/tests/msg/test_ext_types_msgspec.py index f07d375c..f22e0f18 100644 --- a/tests/msg/test_ext_types_msgspec.py +++ b/tests/msg/test_ext_types_msgspec.py @@ -62,7 +62,10 @@ def enc_nsp(obj: Any) -> Any: actor: Actor = tractor.current_actor( err_on_no_runtime=False, ) - uid: tuple[str, str]|None = None if not actor else actor.uid + uid: tuple[str, str]|None = ( + None if not actor + else actor.aid.uid + ) print(f'{uid} ENC HOOK') match obj: @@ -95,7 +98,10 @@ def dec_nsp( actor: Actor = tractor.current_actor( err_on_no_runtime=False, ) - uid: tuple[str, str]|None = None if not actor else actor.uid + uid: tuple[str, str]|None = ( + None if not actor + else actor.aid.uid + ) print( f'{uid}\n' 'CUSTOM DECODE\n' @@ -419,7 +425,8 @@ async def send_back_values( and ensure we can round trip a func ref with our parent. ''' - uid: tuple = tractor.current_actor().uid + _aid = tractor.current_actor().aid + uid: tuple = _aid.uid # init state in sub-actor should be default chk_codec_applied( diff --git a/tests/test_advanced_streaming.py b/tests/test_advanced_streaming.py index 9b1a476f..c04bf130 100644 --- a/tests/test_advanced_streaming.py +++ b/tests/test_advanced_streaming.py @@ -69,7 +69,7 @@ async def subscribe( new_subs = set(new_subs) remove = new_subs - _registry.keys() - print(f'setting sub to {new_subs} for {ctx.chan.uid}') + print(f'setting sub to {new_subs} for {ctx.chan.aid.uid}') # remove old subs for sub in remove: @@ -84,7 +84,8 @@ async def consumer( subs: list[str], ) -> None: - uid = tractor.current_actor().uid + _aid = tractor.current_actor().aid + uid = _aid.uid async with tractor.wait_for_actor('publisher') as portal: async with portal.open_context(subscribe) as (ctx, first): diff --git a/tests/test_cancellation.py b/tests/test_cancellation.py index 5566691a..ab26c03d 100644 --- a/tests/test_cancellation.py +++ b/tests/test_cancellation.py @@ -351,7 +351,7 @@ async def test_cancel_infinite_streamer( # we support trio's cancellation system assert cancel_scope.cancelled_caught - assert n.cancelled + assert n.cancel_called @pytest.mark.parametrize( @@ -465,7 +465,7 @@ async def test_some_cancels_all( elif isinstance(err, tractor.RemoteActorError): assert err.boxed_type == err_type - assert an.cancelled is True + assert an.cancel_called is True assert not an._children else: pytest.fail("Should have gotten a remote assertion error?") @@ -780,21 +780,28 @@ def test_cancel_via_SIGINT_other_task( async def main(): # should never timeout since SIGINT should cancel the current program with trio.fail_after(timeout): - async with ( - - # XXX ?TODO? why no work!? - # tractor.trionics.collapse_eg(), - trio.open_nursery( - strict_exception_groups=False, - ) as tn, - ): + async with trio.open_nursery() as tn: await tn.start(spawn_and_sleep_forever) if 'mp' in spawn_backend: time.sleep(0.1) os.kill(pid, signal.SIGINT) - with pytest.raises(KeyboardInterrupt): + # SIGINT -> `KeyboardInterrupt`; under `trio>=0.25`'s strict + # exception-groups the KI surfaces wrapped in a (cancel-padded) + # `BaseExceptionGroup` rather than bare — so accept either form + # (replaces the now-deprecated `strict_exception_groups=False`, + # and `collapse_eg()` can't help since the group is multi-exc: + # the KI rides alongside the child-task `Cancelled`s). + with pytest.raises(BaseException) as excinfo: trio.run(main) + exc = excinfo.value + assert ( + isinstance(exc, KeyboardInterrupt) + or ( + isinstance(exc, BaseExceptionGroup) + and exc.subgroup(KeyboardInterrupt) is not None + ) + ) async def spin_for(period=3): diff --git a/tests/test_context_stream_semantics.py b/tests/test_context_stream_semantics.py index 4e23dcae..8ef85426 100644 --- a/tests/test_context_stream_semantics.py +++ b/tests/test_context_stream_semantics.py @@ -311,7 +311,7 @@ def test_parent_cancels( ctx: Context, ) -> None: actor: Actor = current_actor() - uid: tuple = actor.uid + uid: tuple = actor.aid.uid _ctxc: ContextCancelled|None = None if ( @@ -712,9 +712,9 @@ async def test_parent_exits_ctx_after_child_enters_stream( assert ( ctxc.canceller == - current_actor().uid + current_actor().aid.uid == - root.uid + root.aid.uid ) # channel should still be up @@ -1219,7 +1219,7 @@ def test_maybe_allow_overruns_stream( if cancel_ctx: assert isinstance(res, ContextCancelled) - assert tuple(res.canceller) == current_actor().uid + assert tuple(res.canceller) == current_actor().aid.uid else: print(f'RX ROOT SIDE RESULT {res}') diff --git a/tests/test_infected_asyncio.py b/tests/test_infected_asyncio.py index a751e010..bdb1b852 100644 --- a/tests/test_infected_asyncio.py +++ b/tests/test_infected_asyncio.py @@ -956,7 +956,7 @@ async def manage_file( ''' tmp_path: Path = Path(tmp_path_str) - tmp_file: Path = tmp_path / f'{" ".join(ctx._actor.uid)}.file' + tmp_file: Path = tmp_path / f'{" ".join(ctx._actor.aid.uid)}.file' # create a the tmp file and tell the parent where it's at assert not tmp_file.is_file() @@ -1180,7 +1180,7 @@ def test_sigint_closes_lifetime_stack( ): await ctx.wait_for_result() except tractor.ContextCancelled as ctxc: - assert ctxc.canceller == ctx.chan.uid + assert ctxc.canceller == ctx.chan.aid.uid raise except trio.TooSlowError: @@ -1300,7 +1300,7 @@ async def caching_ep( }, # lock around current actor task access - key=tractor.current_actor().uid, + key=tractor.current_actor().aid.uid, ) as (cache_hit, (clients, chan)), ): diff --git a/tests/test_local.py b/tests/test_local.py index 765ff796..b585651c 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -34,7 +34,7 @@ async def test_self_is_registered(reg_addr): assert actor.is_registrar with trio.fail_after(0.2): async with tractor.wait_for_actor('root') as portal: - assert portal.channel.uid[0] == 'root' + assert portal.channel.aid.name == 'root' @tractor_test diff --git a/tests/test_spawning.py b/tests/test_spawning.py index d12c9b28..b6128b96 100644 --- a/tests/test_spawning.py +++ b/tests/test_spawning.py @@ -65,7 +65,7 @@ async def spawn( assert len(an._children) == 1 assert ( - portal.channel.uid + portal.channel.aid.uid in tractor.current_actor().ipc_server._peers ) diff --git a/tractor/_exceptions.py b/tractor/_exceptions.py index 3b087184..ab3043c5 100644 --- a/tractor/_exceptions.py +++ b/tractor/_exceptions.py @@ -1223,7 +1223,7 @@ def pack_error( str, str | tuple[str, str] ] = {} - our_uid: tuple = current_actor().uid + our_uid: tuple[str, str] = current_actor().aid.uid if ( isinstance(exc, RemoteActorError) diff --git a/tractor/_streaming.py b/tractor/_streaming.py index 421ca6c2..adc77dcd 100644 --- a/tractor/_streaming.py +++ b/tractor/_streaming.py @@ -418,7 +418,7 @@ class MsgStream(trio.abc.Channel): # it can't traverse the transport. log.warning( f'Stream was already destroyed?\n' - f'actor: {ctx.chan.uid}\n' + f'actor: {ctx.chan.aid.uid}\n' f'ctx id: {ctx.cid}' ) drained.append(re) @@ -700,7 +700,7 @@ async def open_stream_from_ctx( task: str = trio.lowlevel.current_task().name raise RuntimeError( 'Stream opened after `Context.cancel()` called..?\n' - f'task: {actor.uid[0]}:{task}\n' + f'task: {actor.aid.name}:{task}\n' f'{ctx}' ) @@ -823,7 +823,7 @@ async def open_stream_from_ctx( except KeyError: log.warning( f'Stream was already destroyed?\n' - f'actor: {ctx.chan.uid}\n' + f'actor: {ctx.chan.aid.uid}\n' f'ctx id: {ctx.cid}' ) diff --git a/tractor/_testing/addr.py b/tractor/_testing/addr.py index 6927db77..db95889d 100644 --- a/tractor/_testing/addr.py +++ b/tractor/_testing/addr.py @@ -72,7 +72,12 @@ def get_rando_addr( # `test_tpt_bind_addrs::bind-subset-reg`), # - across parallel pytest sessions, the pid bias # makes coincident port choices unlikely. - port: int = 1000 + ( + # NB. floor at 1024, NOT 1000: ports <1024 are PRIVILEGED + # on linux (`net.ipv4.ip_unprivileged_port_start = 1024`), + # so a non-root `.bind()` on one raises `PermissionError`. + # The old `1000 +` floor could roll 1000-1023 -> flaky + # `[Errno 13] Permission denied` registry-listener binds. + port: int = 1024 + ( random.randint(0, 8999) + os.getpid() ) % 9000 testrun_reg_addr = ( diff --git a/tractor/_testing/pytest.py b/tractor/_testing/pytest.py index 0b2cc6f6..90b0ae67 100644 --- a/tractor/_testing/pytest.py +++ b/tractor/_testing/pytest.py @@ -491,6 +491,16 @@ def pytest_configure( 'cases (e.g. the `subint` GIL-starvation class documented ' 'in `ai/conc-anal/subint_sigint_starvation_issue.md`).' ) + config.addinivalue_line( + 'markers', + 'has_nested_actors: test spawns nested (>1-level) subactor ' + 'trees.' + ) + config.addinivalue_line( + 'markers', + 'trio: legacy mark for tests meant to run under the `trio` ' + 'spawn backend (e.g. `test_local.py`).' + ) # `--enable-stackscope`: install SIGUSR1 → trio task-tree # dump in pytest itself + propagate to every subactor via diff --git a/tractor/devx/debug/_sync.py b/tractor/devx/debug/_sync.py index 21850076..9e1a6395 100644 --- a/tractor/devx/debug/_sync.py +++ b/tractor/devx/debug/_sync.py @@ -92,11 +92,11 @@ async def maybe_wait_for_debugger( # tearing down. ctx_in_debug: Context|None = Lock.ctx_in_debug in_debug: tuple[str, str]|None = ( - ctx_in_debug.chan.uid + ctx_in_debug.chan.aid.uid if ctx_in_debug else None ) - if in_debug == current_actor().uid: + if in_debug == current_actor().aid.uid: log.debug( msg + diff --git a/tractor/experimental/_pubsub.py b/tractor/experimental/_pubsub.py index 97e16e24..c9bf13d6 100644 --- a/tractor/experimental/_pubsub.py +++ b/tractor/experimental/_pubsub.py @@ -38,6 +38,7 @@ import wrapt from ..log import get_logger from .._context import Context +from ..msg import Yield __all__ = ['pub'] @@ -88,7 +89,18 @@ async def fan_out_to_ctxs( if ctx_payloads: for ctx, payload in ctx_payloads: try: - await ctx.send_yield(payload) + # NOTE: inline the (now-deprecated) + # `Context.send_yield()` body to drop its + # `DeprecationWarning` without a behaviour change. + # The "proper" fix is migrating BOTH sides to + # `open_context()`/`open_stream()`, but the + # subscriber still uses the legacy + # `Portal.open_stream_from()` (no `started()` + # handshake) so `Context.open_stream()` can't be + # used here yet — see this module's rework TODO. + await ctx.chan.send( + Yield(cid=ctx.cid, pld=payload) + ) except ( # That's right, anything you can think of... trio.ClosedResourceError, ConnectionResetError, @@ -117,7 +129,7 @@ def modify_subs( Effectively a symbol subscription api. """ - log.info(f"{ctx.chan.uid} changed subscription to {topics}") + log.info(f"{ctx.chan.aid.uid} changed subscription to {topics}") # update map from each symbol to requesting client's chan for topic in topics: diff --git a/tractor/msg/_ops.py b/tractor/msg/_ops.py index 3b4eaa84..a134306f 100644 --- a/tractor/msg/_ops.py +++ b/tractor/msg/_ops.py @@ -345,9 +345,9 @@ class PldRx(Struct): exc=mte, cid=msg.cid, src_uid=( - ipc.chan.uid + ipc.chan.aid.uid if not is_started_send_side - else ipc._actor.uid + else ipc._actor.aid.uid ), ) mte._ipc_msg = err_msg @@ -711,7 +711,7 @@ async def drain_to_final_msg( 'Cancelling `MsgStream` drain since ' f'{reason}\n' f'\n' - f'<= {ctx.chan.uid}\n' + f'<= {ctx.chan.aid.uid}\n' f' |_{ctx._nsf}()\n' f'\n' f'=> {ctx._task}\n' @@ -726,7 +726,7 @@ async def drain_to_final_msg( else: report: str = ( 'Ignoring "yield" msg during `ctx.result()` drain..\n' - f'<= {ctx.chan.uid}\n' + f'<= {ctx.chan.aid.uid}\n' f' |_{ctx._nsf}()\n\n' f'=> {ctx._task}\n' f' |_{ctx._stream}\n\n'