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
|
# https://docs.pytest.org/en/stable/reference/reference.html#confval-console_output_style
|
||||||
console_output_style = 'progress'
|
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 ------
|
# ------ 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(
|
expect(
|
||||||
child,
|
child,
|
||||||
# end-of-tree delimiter
|
# end-of-tree delimiter
|
||||||
r"end-of-\('root'",
|
"end-of-\('root'",
|
||||||
)
|
)
|
||||||
_before: str = assert_before(
|
_before: str = assert_before(
|
||||||
child,
|
child,
|
||||||
|
|
@ -157,7 +157,7 @@ def test_shield_pause(
|
||||||
expect(
|
expect(
|
||||||
child,
|
child,
|
||||||
# end-of-subactor's-tree delimiter
|
# end-of-subactor's-tree delimiter
|
||||||
r"end-of-\('hanger'",
|
"end-of-\('hanger'",
|
||||||
)
|
)
|
||||||
_before: str = assert_before(
|
_before: str = assert_before(
|
||||||
child,
|
child,
|
||||||
|
|
|
||||||
|
|
@ -122,13 +122,7 @@ def test_register_duplicate_name(
|
||||||
'test_register_duplicate_name: '
|
'test_register_duplicate_name: '
|
||||||
'`wait_for_actor` returned'
|
'`wait_for_actor` returned'
|
||||||
)
|
)
|
||||||
assert (
|
assert portal.channel.uid in (p2.channel.uid, p1.channel.uid)
|
||||||
(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),
|
|
||||||
)
|
|
||||||
)
|
|
||||||
|
|
||||||
log.cancel(
|
log.cancel(
|
||||||
'test_register_duplicate_name: '
|
'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
|
# name; doesn't matter which one (registrar will
|
||||||
# have last-wins semantics under same-name).
|
# have last-wins semantics under same-name).
|
||||||
async with tractor.wait_for_actor('doggy') as portal:
|
async with tractor.wait_for_actor('doggy') as portal:
|
||||||
expected_uids = {
|
expected_uids = {p.channel.uid for p in portals}
|
||||||
(p.channel.aid.name, p.channel.aid.uuid)
|
assert portal.channel.uid in expected_uids
|
||||||
for p in portals
|
|
||||||
}
|
|
||||||
assert (
|
|
||||||
(portal.channel.aid.name, portal.channel.aid.uuid)
|
|
||||||
in expected_uids
|
|
||||||
)
|
|
||||||
|
|
||||||
# critical section: this MUST return within
|
# critical section: this MUST return within
|
||||||
# `fail_after_s` even when one or more cancel-RPC
|
# `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(
|
actor: Actor = tractor.current_actor(
|
||||||
err_on_no_runtime=False,
|
err_on_no_runtime=False,
|
||||||
)
|
)
|
||||||
uid: tuple[str, str]|None = (
|
uid: tuple[str, str]|None = None if not actor else actor.uid
|
||||||
None if not actor
|
|
||||||
else (actor.aid.name, actor.aid.uuid)
|
|
||||||
)
|
|
||||||
print(f'{uid} ENC HOOK')
|
print(f'{uid} ENC HOOK')
|
||||||
|
|
||||||
match obj:
|
match obj:
|
||||||
|
|
@ -98,10 +95,7 @@ def dec_nsp(
|
||||||
actor: Actor = tractor.current_actor(
|
actor: Actor = tractor.current_actor(
|
||||||
err_on_no_runtime=False,
|
err_on_no_runtime=False,
|
||||||
)
|
)
|
||||||
uid: tuple[str, str]|None = (
|
uid: tuple[str, str]|None = None if not actor else actor.uid
|
||||||
None if not actor
|
|
||||||
else (actor.aid.name, actor.aid.uuid)
|
|
||||||
)
|
|
||||||
print(
|
print(
|
||||||
f'{uid}\n'
|
f'{uid}\n'
|
||||||
'CUSTOM DECODE\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.
|
and ensure we can round trip a func ref with our parent.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
_aid = tractor.current_actor().aid
|
uid: tuple = tractor.current_actor().uid
|
||||||
uid: tuple = (_aid.name, _aid.uuid)
|
|
||||||
|
|
||||||
# init state in sub-actor should be default
|
# init state in sub-actor should be default
|
||||||
chk_codec_applied(
|
chk_codec_applied(
|
||||||
|
|
|
||||||
|
|
@ -69,7 +69,7 @@ async def subscribe(
|
||||||
new_subs = set(new_subs)
|
new_subs = set(new_subs)
|
||||||
remove = new_subs - _registry.keys()
|
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
|
# remove old subs
|
||||||
for sub in remove:
|
for sub in remove:
|
||||||
|
|
@ -84,8 +84,7 @@ async def consumer(
|
||||||
subs: list[str],
|
subs: list[str],
|
||||||
) -> None:
|
) -> None:
|
||||||
|
|
||||||
_aid = tractor.current_actor().aid
|
uid = tractor.current_actor().uid
|
||||||
uid = (_aid.name, _aid.uuid)
|
|
||||||
|
|
||||||
async with tractor.wait_for_actor('publisher') as portal:
|
async with tractor.wait_for_actor('publisher') as portal:
|
||||||
async with portal.open_context(subscribe) as (ctx, first):
|
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
|
# we support trio's cancellation system
|
||||||
assert cancel_scope.cancelled_caught
|
assert cancel_scope.cancelled_caught
|
||||||
assert n.cancel_called
|
assert n.cancelled
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
|
|
@ -465,7 +465,7 @@ async def test_some_cancels_all(
|
||||||
elif isinstance(err, tractor.RemoteActorError):
|
elif isinstance(err, tractor.RemoteActorError):
|
||||||
assert err.boxed_type == err_type
|
assert err.boxed_type == err_type
|
||||||
|
|
||||||
assert an.cancel_called is True
|
assert an.cancelled is True
|
||||||
assert not an._children
|
assert not an._children
|
||||||
else:
|
else:
|
||||||
pytest.fail("Should have gotten a remote assertion error?")
|
pytest.fail("Should have gotten a remote assertion error?")
|
||||||
|
|
@ -780,28 +780,21 @@ def test_cancel_via_SIGINT_other_task(
|
||||||
async def main():
|
async def main():
|
||||||
# should never timeout since SIGINT should cancel the current program
|
# should never timeout since SIGINT should cancel the current program
|
||||||
with trio.fail_after(timeout):
|
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)
|
await tn.start(spawn_and_sleep_forever)
|
||||||
if 'mp' in spawn_backend:
|
if 'mp' in spawn_backend:
|
||||||
time.sleep(0.1)
|
time.sleep(0.1)
|
||||||
os.kill(pid, signal.SIGINT)
|
os.kill(pid, signal.SIGINT)
|
||||||
|
|
||||||
# SIGINT -> `KeyboardInterrupt`; under `trio>=0.25`'s strict
|
with pytest.raises(KeyboardInterrupt):
|
||||||
# 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)
|
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):
|
async def spin_for(period=3):
|
||||||
|
|
|
||||||
|
|
@ -311,7 +311,7 @@ def test_parent_cancels(
|
||||||
ctx: Context,
|
ctx: Context,
|
||||||
) -> None:
|
) -> None:
|
||||||
actor: Actor = current_actor()
|
actor: Actor = current_actor()
|
||||||
uid: tuple = (actor.aid.name, actor.aid.uuid)
|
uid: tuple = actor.uid
|
||||||
_ctxc: ContextCancelled|None = None
|
_ctxc: ContextCancelled|None = None
|
||||||
|
|
||||||
if (
|
if (
|
||||||
|
|
@ -712,9 +712,9 @@ async def test_parent_exits_ctx_after_child_enters_stream(
|
||||||
assert (
|
assert (
|
||||||
ctxc.canceller
|
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
|
# channel should still be up
|
||||||
|
|
@ -1219,7 +1219,7 @@ def test_maybe_allow_overruns_stream(
|
||||||
|
|
||||||
if cancel_ctx:
|
if cancel_ctx:
|
||||||
assert isinstance(res, ContextCancelled)
|
assert isinstance(res, ContextCancelled)
|
||||||
assert tuple(res.canceller) == (current_actor().aid.name, current_actor().aid.uuid)
|
assert tuple(res.canceller) == current_actor().uid
|
||||||
|
|
||||||
else:
|
else:
|
||||||
print(f'RX ROOT SIDE RESULT {res}')
|
print(f'RX ROOT SIDE RESULT {res}')
|
||||||
|
|
|
||||||
|
|
@ -956,7 +956,7 @@ async def manage_file(
|
||||||
'''
|
'''
|
||||||
|
|
||||||
tmp_path: Path = Path(tmp_path_str)
|
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
|
# create a the tmp file and tell the parent where it's at
|
||||||
assert not tmp_file.is_file()
|
assert not tmp_file.is_file()
|
||||||
|
|
@ -1180,7 +1180,7 @@ def test_sigint_closes_lifetime_stack(
|
||||||
):
|
):
|
||||||
await ctx.wait_for_result()
|
await ctx.wait_for_result()
|
||||||
except tractor.ContextCancelled as ctxc:
|
except tractor.ContextCancelled as ctxc:
|
||||||
assert ctxc.canceller == (ctx.chan.aid.name, ctx.chan.aid.uuid)
|
assert ctxc.canceller == ctx.chan.uid
|
||||||
raise
|
raise
|
||||||
|
|
||||||
except trio.TooSlowError:
|
except trio.TooSlowError:
|
||||||
|
|
@ -1300,7 +1300,7 @@ async def caching_ep(
|
||||||
},
|
},
|
||||||
|
|
||||||
# lock around current actor task access
|
# 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)),
|
) as (cache_hit, (clients, chan)),
|
||||||
):
|
):
|
||||||
|
|
|
||||||
|
|
@ -34,7 +34,7 @@ async def test_self_is_registered(reg_addr):
|
||||||
assert actor.is_registrar
|
assert actor.is_registrar
|
||||||
with trio.fail_after(0.2):
|
with trio.fail_after(0.2):
|
||||||
async with tractor.wait_for_actor('root') as portal:
|
async with tractor.wait_for_actor('root') as portal:
|
||||||
assert portal.channel.aid.name == 'root'
|
assert portal.channel.uid[0] == 'root'
|
||||||
|
|
||||||
|
|
||||||
@tractor_test
|
@tractor_test
|
||||||
|
|
|
||||||
|
|
@ -65,7 +65,7 @@ async def spawn(
|
||||||
|
|
||||||
assert len(an._children) == 1
|
assert len(an._children) == 1
|
||||||
assert (
|
assert (
|
||||||
(portal.channel.aid.name, portal.channel.aid.uuid)
|
portal.channel.uid
|
||||||
in
|
in
|
||||||
tractor.current_actor().ipc_server._peers
|
tractor.current_actor().ipc_server._peers
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -1223,7 +1223,7 @@ def pack_error(
|
||||||
str,
|
str,
|
||||||
str | tuple[str, str]
|
str | tuple[str, str]
|
||||||
] = {}
|
] = {}
|
||||||
our_uid: tuple = (current_actor().aid.name, current_actor().aid.uuid)
|
our_uid: tuple = current_actor().uid
|
||||||
|
|
||||||
if (
|
if (
|
||||||
isinstance(exc, RemoteActorError)
|
isinstance(exc, RemoteActorError)
|
||||||
|
|
|
||||||
|
|
@ -418,7 +418,7 @@ class MsgStream(trio.abc.Channel):
|
||||||
# it can't traverse the transport.
|
# it can't traverse the transport.
|
||||||
log.warning(
|
log.warning(
|
||||||
f'Stream was already destroyed?\n'
|
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}'
|
f'ctx id: {ctx.cid}'
|
||||||
)
|
)
|
||||||
drained.append(re)
|
drained.append(re)
|
||||||
|
|
@ -700,7 +700,7 @@ async def open_stream_from_ctx(
|
||||||
task: str = trio.lowlevel.current_task().name
|
task: str = trio.lowlevel.current_task().name
|
||||||
raise RuntimeError(
|
raise RuntimeError(
|
||||||
'Stream opened after `Context.cancel()` called..?\n'
|
'Stream opened after `Context.cancel()` called..?\n'
|
||||||
f'task: {actor.aid.name}:{task}\n'
|
f'task: {actor.uid[0]}:{task}\n'
|
||||||
f'{ctx}'
|
f'{ctx}'
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -823,7 +823,7 @@ async def open_stream_from_ctx(
|
||||||
except KeyError:
|
except KeyError:
|
||||||
log.warning(
|
log.warning(
|
||||||
f'Stream was already destroyed?\n'
|
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}'
|
f'ctx id: {ctx.cid}'
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -491,16 +491,6 @@ def pytest_configure(
|
||||||
'cases (e.g. the `subint` GIL-starvation class documented '
|
'cases (e.g. the `subint` GIL-starvation class documented '
|
||||||
'in `ai/conc-anal/subint_sigint_starvation_issue.md`).'
|
'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
|
# `--enable-stackscope`: install SIGUSR1 → trio task-tree
|
||||||
# dump in pytest itself + propagate to every subactor via
|
# dump in pytest itself + propagate to every subactor via
|
||||||
|
|
|
||||||
|
|
@ -92,11 +92,11 @@ async def maybe_wait_for_debugger(
|
||||||
# tearing down.
|
# tearing down.
|
||||||
ctx_in_debug: Context|None = Lock.ctx_in_debug
|
ctx_in_debug: Context|None = Lock.ctx_in_debug
|
||||||
in_debug: tuple[str, str]|None = (
|
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
|
if ctx_in_debug
|
||||||
else None
|
else None
|
||||||
)
|
)
|
||||||
if in_debug == (current_actor().aid.name, current_actor().aid.uuid):
|
if in_debug == current_actor().uid:
|
||||||
log.debug(
|
log.debug(
|
||||||
msg
|
msg
|
||||||
+
|
+
|
||||||
|
|
|
||||||
|
|
@ -117,7 +117,7 @@ def modify_subs(
|
||||||
|
|
||||||
Effectively a symbol subscription api.
|
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
|
# update map from each symbol to requesting client's chan
|
||||||
for topic in topics:
|
for topic in topics:
|
||||||
|
|
|
||||||
|
|
@ -345,9 +345,9 @@ class PldRx(Struct):
|
||||||
exc=mte,
|
exc=mte,
|
||||||
cid=msg.cid,
|
cid=msg.cid,
|
||||||
src_uid=(
|
src_uid=(
|
||||||
(ipc.chan.aid.name, ipc.chan.aid.uuid)
|
ipc.chan.uid
|
||||||
if not is_started_send_side
|
if not is_started_send_side
|
||||||
else (ipc._actor.aid.name, ipc._actor.aid.uuid)
|
else ipc._actor.uid
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
mte._ipc_msg = err_msg
|
mte._ipc_msg = err_msg
|
||||||
|
|
@ -711,7 +711,7 @@ async def drain_to_final_msg(
|
||||||
'Cancelling `MsgStream` drain since '
|
'Cancelling `MsgStream` drain since '
|
||||||
f'{reason}\n'
|
f'{reason}\n'
|
||||||
f'\n'
|
f'\n'
|
||||||
f'<= {(ctx.chan.aid.name, ctx.chan.aid.uuid)}\n'
|
f'<= {ctx.chan.uid}\n'
|
||||||
f' |_{ctx._nsf}()\n'
|
f' |_{ctx._nsf}()\n'
|
||||||
f'\n'
|
f'\n'
|
||||||
f'=> {ctx._task}\n'
|
f'=> {ctx._task}\n'
|
||||||
|
|
@ -726,7 +726,7 @@ async def drain_to_final_msg(
|
||||||
else:
|
else:
|
||||||
report: str = (
|
report: str = (
|
||||||
'Ignoring "yield" msg during `ctx.result()` drain..\n'
|
'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._nsf}()\n\n'
|
||||||
f'=> {ctx._task}\n'
|
f'=> {ctx._task}\n'
|
||||||
f' |_{ctx._stream}\n\n'
|
f' |_{ctx._stream}\n\n'
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue