Compare commits

..

No commits in common. "d55c971644c3d187e712cb514a1dc028e13932c0" and "d95cd94e20f319509c1ccf82ad7fb9b1c150ed33" have entirely different histories.

17 changed files with 41 additions and 188 deletions

View File

@ -279,25 +279,4 @@ 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 ------

View File

@ -1,89 +0,0 @@
"""
`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/<pid>/wchan` + `/proc/<pid>/stack` (kernel-side
blocked-on syscall),
- `sudo py-spy dump` (python-side stack).
Usage (xonsh):
source-foreign xonsh ./scripts/hung-dump.xsh
hung-dump <pid> [<pid> ...]
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 <pid> [<pid> ...]')
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

View File

@ -92,7 +92,7 @@ def test_shield_pause(
expect(
child,
# end-of-tree delimiter
r"end-of-\('root'",
"end-of-\('root'",
)
_before: str = assert_before(
child,
@ -157,7 +157,7 @@ def test_shield_pause(
expect(
child,
# end-of-subactor's-tree delimiter
r"end-of-\('hanger'",
"end-of-\('hanger'",
)
_before: str = assert_before(
child,

View File

@ -122,13 +122,7 @@ def test_register_duplicate_name(
'test_register_duplicate_name: '
'`wait_for_actor` returned'
)
assert (
(portal.channel.aid.name, portal.channel.aid.uuid)
in (
(p2.channel.aid.name, p2.channel.aid.uuid),
(p1.channel.aid.name, p1.channel.aid.uuid),
)
)
assert portal.channel.uid in (p2.channel.uid, p1.channel.uid)
log.cancel(
'test_register_duplicate_name: '
@ -248,14 +242,8 @@ 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.aid.name, p.channel.aid.uuid)
for p in portals
}
assert (
(portal.channel.aid.name, portal.channel.aid.uuid)
in expected_uids
)
expected_uids = {p.channel.uid for p in portals}
assert portal.channel.uid in expected_uids
# critical section: this MUST return within
# `fail_after_s` even when one or more cancel-RPC

View File

@ -62,10 +62,7 @@ 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.aid.name, actor.aid.uuid)
)
uid: tuple[str, str]|None = None if not actor else actor.uid
print(f'{uid} ENC HOOK')
match obj:
@ -98,10 +95,7 @@ def dec_nsp(
actor: Actor = tractor.current_actor(
err_on_no_runtime=False,
)
uid: tuple[str, str]|None = (
None if not actor
else (actor.aid.name, actor.aid.uuid)
)
uid: tuple[str, str]|None = None if not actor else actor.uid
print(
f'{uid}\n'
'CUSTOM DECODE\n'
@ -425,8 +419,7 @@ async def send_back_values(
and ensure we can round trip a func ref with our parent.
'''
_aid = tractor.current_actor().aid
uid: tuple = (_aid.name, _aid.uuid)
uid: tuple = tractor.current_actor().uid
# init state in sub-actor should be default
chk_codec_applied(

View File

@ -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.aid.name, ctx.chan.aid.uuid)}')
print(f'setting sub to {new_subs} for {ctx.chan.uid}')
# remove old subs
for sub in remove:
@ -84,8 +84,7 @@ async def consumer(
subs: list[str],
) -> None:
_aid = tractor.current_actor().aid
uid = (_aid.name, _aid.uuid)
uid = tractor.current_actor().uid
async with tractor.wait_for_actor('publisher') as portal:
async with portal.open_context(subscribe) as (ctx, first):

View File

@ -351,7 +351,7 @@ async def test_cancel_infinite_streamer(
# we support trio's cancellation system
assert cancel_scope.cancelled_caught
assert n.cancel_called
assert n.cancelled
@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.cancel_called is True
assert an.cancelled is True
assert not an._children
else:
pytest.fail("Should have gotten a remote assertion error?")
@ -780,28 +780,21 @@ 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 trio.open_nursery() as tn:
async with (
# XXX ?TODO? why no work!?
# tractor.trionics.collapse_eg(),
trio.open_nursery(
strict_exception_groups=False,
) as tn,
):
await tn.start(spawn_and_sleep_forever)
if 'mp' in spawn_backend:
time.sleep(0.1)
os.kill(pid, signal.SIGINT)
# 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:
with pytest.raises(KeyboardInterrupt):
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):

View File

@ -311,7 +311,7 @@ def test_parent_cancels(
ctx: Context,
) -> None:
actor: Actor = current_actor()
uid: tuple = (actor.aid.name, actor.aid.uuid)
uid: tuple = actor.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().aid.name, current_actor().aid.uuid)
current_actor().uid
==
(root.aid.name, root.aid.uuid)
root.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().aid.name, current_actor().aid.uuid)
assert tuple(res.canceller) == current_actor().uid
else:
print(f'RX ROOT SIDE RESULT {res}')

View File

@ -956,7 +956,7 @@ async def manage_file(
'''
tmp_path: Path = Path(tmp_path_str)
tmp_file: Path = tmp_path / f'{" ".join((ctx._actor.aid.name, ctx._actor.aid.uuid))}.file'
tmp_file: Path = tmp_path / f'{" ".join(ctx._actor.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.aid.name, ctx.chan.aid.uuid)
assert ctxc.canceller == ctx.chan.uid
raise
except trio.TooSlowError:
@ -1300,7 +1300,7 @@ async def caching_ep(
},
# lock around current actor task access
key=(tractor.current_actor().aid.name, tractor.current_actor().aid.uuid),
key=tractor.current_actor().uid,
) as (cache_hit, (clients, chan)),
):

View File

@ -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.aid.name == 'root'
assert portal.channel.uid[0] == 'root'
@tractor_test

View File

@ -65,7 +65,7 @@ async def spawn(
assert len(an._children) == 1
assert (
(portal.channel.aid.name, portal.channel.aid.uuid)
portal.channel.uid
in
tractor.current_actor().ipc_server._peers
)

View File

@ -1223,7 +1223,7 @@ def pack_error(
str,
str | tuple[str, str]
] = {}
our_uid: tuple = (current_actor().aid.name, current_actor().aid.uuid)
our_uid: tuple = current_actor().uid
if (
isinstance(exc, RemoteActorError)

View File

@ -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.aid.name, ctx.chan.aid.uuid)}\n'
f'actor: {ctx.chan.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.aid.name}:{task}\n'
f'task: {actor.uid[0]}:{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.aid.name, ctx.chan.aid.uuid)}\n'
f'actor: {ctx.chan.uid}\n'
f'ctx id: {ctx.cid}'
)

View File

@ -491,16 +491,6 @@ 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

View File

@ -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.aid.name, ctx_in_debug.chan.aid.uuid)
ctx_in_debug.chan.uid
if ctx_in_debug
else None
)
if in_debug == (current_actor().aid.name, current_actor().aid.uuid):
if in_debug == current_actor().uid:
log.debug(
msg
+

View File

@ -117,7 +117,7 @@ def modify_subs(
Effectively a symbol subscription api.
"""
log.info(f"{(ctx.chan.aid.name, ctx.chan.aid.uuid)} changed subscription to {topics}")
log.info(f"{ctx.chan.uid} changed subscription to {topics}")
# update map from each symbol to requesting client's chan
for topic in topics:

View File

@ -345,9 +345,9 @@ class PldRx(Struct):
exc=mte,
cid=msg.cid,
src_uid=(
(ipc.chan.aid.name, ipc.chan.aid.uuid)
ipc.chan.uid
if not is_started_send_side
else (ipc._actor.aid.name, ipc._actor.aid.uuid)
else ipc._actor.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.aid.name, ctx.chan.aid.uuid)}\n'
f'<= {ctx.chan.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.aid.name, ctx.chan.aid.uuid)}\n'
f'<= {ctx.chan.uid}\n'
f' |_{ctx._nsf}()\n\n'
f'=> {ctx._task}\n'
f' |_{ctx._stream}\n\n'