Compare commits
No commits in common. "d55c971644c3d187e712cb514a1dc028e13932c0" and "d95cd94e20f319509c1ccf82ad7fb9b1c150ed33" have entirely different histories.
d55c971644
...
d95cd94e20
|
|
@ -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 ------
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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}')
|
||||
|
|
|
|||
|
|
@ -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)),
|
||||
):
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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}'
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
+
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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'
|
||||
|
|
|
|||
Loading…
Reference in New Issue