Compare commits
20 Commits
11d3c1ee58
...
3bc6680982
| Author | SHA1 | Date |
|---|---|---|
|
|
3bc6680982 | |
|
|
23580e9cb8 | |
|
|
2a7c46add2 | |
|
|
6764251e6b | |
|
|
02cb789901 | |
|
|
a3f8765264 | |
|
|
e076296cda | |
|
|
7f6782b9be | |
|
|
35414a42a9 | |
|
|
8fcb9eb7d6 | |
|
|
14329ec1d6 | |
|
|
008824d650 | |
|
|
6255c17db9 | |
|
|
6a1149d401 | |
|
|
88cc68edd3 | |
|
|
3727ce9e6f | |
|
|
90c316f90c | |
|
|
6db796ed83 | |
|
|
eb7ef3f0bf | |
|
|
bb0b2ffa3c |
|
|
@ -263,9 +263,18 @@ def test_simple_context(
|
||||||
except error_parent:
|
except error_parent:
|
||||||
pass
|
pass
|
||||||
except BaseExceptionGroup as beg:
|
except BaseExceptionGroup as beg:
|
||||||
# XXX: on windows it seems we may have to expect the group error
|
# XXX: on windows — and under `trio>=0.33`'s strict
|
||||||
|
# exception-groups, where a lone `error_parent` now
|
||||||
|
# arrives wrapped in a 1-exc `ExceptionGroup` rather than
|
||||||
|
# collapsed to the bare exc — accept either the
|
||||||
|
# `error_parent` nested in the group OR a cancel-only
|
||||||
|
# (multi-cancelled) group.
|
||||||
from tractor.trionics import is_multi_cancelled
|
from tractor.trionics import is_multi_cancelled
|
||||||
assert is_multi_cancelled(beg)
|
assert (
|
||||||
|
is_multi_cancelled(beg)
|
||||||
|
or
|
||||||
|
beg.subgroup(error_parent) is not None
|
||||||
|
)
|
||||||
else:
|
else:
|
||||||
trio.run(main)
|
trio.run(main)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -150,7 +150,12 @@ async def test_most_beautiful_word(
|
||||||
The main ``tractor`` routine.
|
The main ``tractor`` routine.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
with trio.fail_after(1):
|
# actor spawn + IPC round-trip is comfortably sub-second on a
|
||||||
|
# warm box, but slow/noisy CI runners (esp. macOS) blow a flat
|
||||||
|
# 1s deadline. Scale for CI/CPU-throttle headroom — `== 1s`
|
||||||
|
# locally where `cpu_scaling_factor()` is `1.0`.
|
||||||
|
from .conftest import cpu_scaling_factor
|
||||||
|
with trio.fail_after(1 * cpu_scaling_factor()):
|
||||||
async with tractor.open_nursery(
|
async with tractor.open_nursery(
|
||||||
debug_mode=debug_mode,
|
debug_mode=debug_mode,
|
||||||
) as an:
|
) as an:
|
||||||
|
|
|
||||||
|
|
@ -14,6 +14,7 @@ Hermetic `trio`-only coverage (no actor-runtime needed):
|
||||||
|
|
||||||
'''
|
'''
|
||||||
from functools import partial
|
from functools import partial
|
||||||
|
import platform
|
||||||
import subprocess
|
import subprocess
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
@ -69,6 +70,10 @@ def test_stdout_relayed_per_line(monkeypatch):
|
||||||
assert any('line=3' in r for r in out_lines)
|
assert any('line=3' in r for r in out_lines)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif(
|
||||||
|
platform.system() != 'Linux',
|
||||||
|
reason='reads `/proc/self/fd` — Linux-only',
|
||||||
|
)
|
||||||
def test_parent_tty_isolated(monkeypatch):
|
def test_parent_tty_isolated(monkeypatch):
|
||||||
records = _capture_relay(monkeypatch)
|
records = _capture_relay(monkeypatch)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -216,18 +216,18 @@ def _read_comm(pid: int) -> str:
|
||||||
# regardless of cwd / venv path / launch context. Used by
|
# regardless of cwd / venv path / launch context. Used by
|
||||||
# `_is_tractor_subactor()` below.
|
# `_is_tractor_subactor()` below.
|
||||||
#
|
#
|
||||||
# - cmdline `tractor[`: matches the `setproctitle`-set form
|
# - cmdline `_subactor[` (`_proctitle._def_prefix`): matches
|
||||||
# (`tractor[<aid.reprol()>]`) — set in
|
# the `setproctitle`-set form (`_subactor[<aid.reprol()>]`)
|
||||||
# `_actor_child_main` for ALL backends, mutates argv via
|
# — set in `_actor_child_main` for ALL backends, mutates
|
||||||
# libc so visible in `/proc/<pid>/cmdline`.
|
# argv via libc so visible in `/proc/<pid>/cmdline`.
|
||||||
# - cmdline `tractor._child`: matches the legacy
|
# - cmdline `tractor._child`: matches the legacy
|
||||||
# `python -m tractor._child --uid (...)` form. Catches
|
# `python -m tractor._child --uid (...)` form. Catches
|
||||||
# procs that died before `_actor_child_main` got to call
|
# procs that died before `_actor_child_main` got to call
|
||||||
# `setproctitle()` — argv from exec is still kernel-
|
# `setproctitle()` — argv from exec is still kernel-
|
||||||
# visible at that point.
|
# visible at that point.
|
||||||
# - comm `tractor[`: same proctitle-set form, but visible
|
# - comm `_subactor[`: same proctitle-set form, but visible
|
||||||
# via `/proc/<pid>/comm` (kernel-truncated to ~15 bytes,
|
# via `/proc/<pid>/comm` (kernel-truncated to ~15 bytes,
|
||||||
# `tractor[doggy:`). Critical for ZOMBIES — kernel
|
# `_subactor[doggy:`). Critical for ZOMBIES — kernel
|
||||||
# preserves `comm` past task-exit until parent reaps,
|
# preserves `comm` past task-exit until parent reaps,
|
||||||
# while `cmdline` for zombies often reads as empty.
|
# while `cmdline` for zombies often reads as empty.
|
||||||
_TRACTOR_PROC_CMDLINE_MARKERS: tuple[str, ...] = (
|
_TRACTOR_PROC_CMDLINE_MARKERS: tuple[str, ...] = (
|
||||||
|
|
@ -253,7 +253,7 @@ def _is_tractor_subactor(pid: int) -> bool:
|
||||||
|
|
||||||
'''
|
'''
|
||||||
# 1. cmdline match — catches both `setproctitle`-set
|
# 1. cmdline match — catches both `setproctitle`-set
|
||||||
# `tractor[<aid>]` (live) AND legacy `python -m
|
# `_subactor[<aid>]` (live) AND legacy `python -m
|
||||||
# tractor._child` (any) form.
|
# tractor._child` (any) form.
|
||||||
cmdline: str = _read_cmdline(pid)
|
cmdline: str = _read_cmdline(pid)
|
||||||
if any(m in cmdline for m in _TRACTOR_PROC_CMDLINE_MARKERS):
|
if any(m in cmdline for m in _TRACTOR_PROC_CMDLINE_MARKERS):
|
||||||
|
|
|
||||||
|
|
@ -281,8 +281,11 @@ def add_log_level(
|
||||||
`StackLevelAdapter` so `log.<name>('msg')` works (and so
|
`StackLevelAdapter` so `log.<name>('msg')` works (and so
|
||||||
`get_logger()`'s per-level method audit passes).
|
`get_logger()`'s per-level method audit passes).
|
||||||
|
|
||||||
Idempotent: re-registering an existing name is a no-op-ish
|
Idempotent: re-registering an existing name refreshes its
|
||||||
refresh (won't clobber an already-bound method).
|
value + color in place. The bound `StackLevelAdapter.<name>`
|
||||||
|
method reads the current `CUSTOM_LEVELS` value at call time, so
|
||||||
|
it tracks the new level across re-registration (the method
|
||||||
|
object is bound once, never re-bound/clobbered).
|
||||||
|
|
||||||
'''
|
'''
|
||||||
name_up: str = name.upper()
|
name_up: str = name.upper()
|
||||||
|
|
@ -294,16 +297,18 @@ def add_log_level(
|
||||||
BOLD_PALETTE['bold'][name_up] = f'bold_{color}'
|
BOLD_PALETTE['bold'][name_up] = f'bold_{color}'
|
||||||
|
|
||||||
if not hasattr(StackLevelAdapter, name_lo):
|
if not hasattr(StackLevelAdapter, name_lo):
|
||||||
# bind via default-arg so `value` is captured (not
|
# read the level from `CUSTOM_LEVELS[name_up]` at CALL time
|
||||||
# late-bound); delegates to `.log()` exactly like the
|
# (NOT captured at bind time) so a later
|
||||||
# hand-written level methods above.
|
# `add_log_level(name, <new-value>)` re-registration stays
|
||||||
|
# consistent with this once-bound method. The closure binds
|
||||||
|
# *this* call's `name_up` local (stable, never reassigned),
|
||||||
|
# so there's no late-binding hazard. Delegates to `.log()`
|
||||||
|
# exactly like the hand-written level methods above.
|
||||||
def _emit(
|
def _emit(
|
||||||
self,
|
self,
|
||||||
msg: str,
|
msg: str,
|
||||||
*,
|
|
||||||
_level: int = value,
|
|
||||||
) -> None:
|
) -> None:
|
||||||
return self.log(_level, msg)
|
return self.log(CUSTOM_LEVELS[name_up], msg)
|
||||||
|
|
||||||
_emit.__name__ = name_lo
|
_emit.__name__ = name_lo
|
||||||
_emit.__qualname__ = f'StackLevelAdapter.{name_lo}'
|
_emit.__qualname__ = f'StackLevelAdapter.{name_lo}'
|
||||||
|
|
|
||||||
|
|
@ -42,6 +42,22 @@ log = get_logger()
|
||||||
_UNSET = object()
|
_UNSET = object()
|
||||||
|
|
||||||
|
|
||||||
|
# cap the stderr bytes retained for a rc!=0 `CalledProcessError`
|
||||||
|
# note: a long-lived (`check=True`) child could otherwise buffer
|
||||||
|
# its ENTIRE stderr history in parent mem — keep only the TAIL
|
||||||
|
# (the bytes nearest the failure, which are what you want to see).
|
||||||
|
_STDERR_NOTE_TAIL_BYTES: int = 16 * 1024
|
||||||
|
|
||||||
|
|
||||||
|
def _decode_line(raw: bytes) -> str:
|
||||||
|
'''
|
||||||
|
Decode one relayed line: replace undecodable bytes and trim a
|
||||||
|
trailing carriage-return (so CRLF-term'd lines read clean).
|
||||||
|
|
||||||
|
'''
|
||||||
|
return raw.decode(errors='replace').rstrip('\r')
|
||||||
|
|
||||||
|
|
||||||
def _add_stderr_note(
|
def _add_stderr_note(
|
||||||
cpe: subprocess.CalledProcessError,
|
cpe: subprocess.CalledProcessError,
|
||||||
stderr_bytes: bytes,
|
stderr_bytes: bytes,
|
||||||
|
|
@ -65,6 +81,7 @@ async def _relay_stream_lines(
|
||||||
emit: Callable[[str], None]|None = None,
|
emit: Callable[[str], None]|None = None,
|
||||||
tag: str = '',
|
tag: str = '',
|
||||||
accum: bytearray|None = None,
|
accum: bytearray|None = None,
|
||||||
|
accum_cap: int|None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
'''
|
'''
|
||||||
Concurrently drain a child subproc's `stdout`/`stderr`
|
Concurrently drain a child subproc's `stdout`/`stderr`
|
||||||
|
|
@ -95,15 +112,21 @@ async def _relay_stream_lines(
|
||||||
async for chunk in stream: # ends at child-exit EOF
|
async for chunk in stream: # ends at child-exit EOF
|
||||||
if accum is not None:
|
if accum is not None:
|
||||||
accum += chunk
|
accum += chunk
|
||||||
|
# bound retained bytes to the TAIL (see caller +
|
||||||
|
# `_STDERR_NOTE_TAIL_BYTES`) so a chatty long-lived
|
||||||
|
# child can't grow parent mem without limit.
|
||||||
|
if (
|
||||||
|
accum_cap is not None
|
||||||
|
and
|
||||||
|
len(accum) > accum_cap
|
||||||
|
):
|
||||||
|
del accum[:len(accum) - accum_cap]
|
||||||
if emit is None:
|
if emit is None:
|
||||||
continue # drain(+accum)-only
|
continue # drain(+accum)-only
|
||||||
buf: bytes = residual + chunk
|
buf: bytes = residual + chunk
|
||||||
*lines, residual = buf.split(b'\n')
|
*lines, residual = buf.split(b'\n')
|
||||||
for raw in lines:
|
for raw in lines:
|
||||||
line: str = raw.decode(
|
emit(f'[{tag}] {_decode_line(raw)}')
|
||||||
errors='replace',
|
|
||||||
).rstrip('\r')
|
|
||||||
emit(f'[{tag}] {line}')
|
|
||||||
|
|
||||||
# flush any trailing partial (un-newline-term'd) line @ EOF
|
# flush any trailing partial (un-newline-term'd) line @ EOF
|
||||||
if (
|
if (
|
||||||
|
|
@ -111,10 +134,7 @@ async def _relay_stream_lines(
|
||||||
and
|
and
|
||||||
residual
|
residual
|
||||||
):
|
):
|
||||||
line: str = residual.decode(
|
emit(f'[{tag}] {_decode_line(residual)}')
|
||||||
errors='replace',
|
|
||||||
).rstrip('\r')
|
|
||||||
emit(f'[{tag}] {line}')
|
|
||||||
|
|
||||||
|
|
||||||
async def supervise_run_process(
|
async def supervise_run_process(
|
||||||
|
|
@ -140,7 +160,7 @@ async def supervise_run_process(
|
||||||
# non-relay `stdout` override; defaults (via `_UNSET`) to
|
# non-relay `stdout` override; defaults (via `_UNSET`) to
|
||||||
# `DEVNULL` so we NEVER inherit (+ thus can't clobber) the
|
# `DEVNULL` so we NEVER inherit (+ thus can't clobber) the
|
||||||
# parent controlling-tty.
|
# parent controlling-tty.
|
||||||
stdout: int = _UNSET,
|
stdout: int|None = _UNSET,
|
||||||
|
|
||||||
task_status: trio.TaskStatus[
|
task_status: trio.TaskStatus[
|
||||||
trio.Process
|
trio.Process
|
||||||
|
|
@ -158,11 +178,14 @@ async def supervise_run_process(
|
||||||
|
|
||||||
- surfaces a rc!=0 `subprocess.CalledProcessError`
|
- surfaces a rc!=0 `subprocess.CalledProcessError`
|
||||||
DETERMINISTICALLY: we pass `check=False` to `trio` and
|
DETERMINISTICALLY: we pass `check=False` to `trio` and
|
||||||
do our OWN post-drain rc-check, (re)building + raising a
|
do our OWN post-drain rc-check, (re)building + raising the
|
||||||
BARE CPE (with a `.stderr` note) from this coro's body
|
CPE (with a `.stderr` note) from this coro's body AFTER the
|
||||||
AFTER the child exits — so there's no nursery-eg-wrapped
|
child exits — so it's never wrapped by `trio.run_process`'s
|
||||||
CPE to catch/`collapse_eg`, and the relay reader is never
|
INTERNAL nursery nor race-cancelled mid-drain. It IS still
|
||||||
race-cancelled mid-drain.
|
raised as a task into the *parent* `tn`, so — like any
|
||||||
|
`tn.start()`-task raise — it surfaces `ExceptionGroup`-
|
||||||
|
wrapped under `trio>=0.33`; unwrap via `except*` or
|
||||||
|
`trionics.collapse_eg()`.
|
||||||
|
|
||||||
- ALWAYS isolates the parent controlling-tty
|
- ALWAYS isolates the parent controlling-tty
|
||||||
(`stdin=DEVNULL`, and `stdout=DEVNULL` unless
|
(`stdin=DEVNULL`, and `stdout=DEVNULL` unless
|
||||||
|
|
@ -190,7 +213,14 @@ async def supervise_run_process(
|
||||||
)
|
)
|
||||||
|
|
||||||
'''
|
'''
|
||||||
emit: Callable[[str], None] = getattr(log, relay_level)
|
# resolve the relay emit-method ONLY when actually relaying so
|
||||||
|
# a bad/typo'd `relay_level` can't crash a NON-relay call (and
|
||||||
|
# we don't bind an unused method on the common silent path).
|
||||||
|
emit: Callable[[str], None]|None = (
|
||||||
|
getattr(log, relay_level)
|
||||||
|
if (relay_stdout or relay_stderr)
|
||||||
|
else None
|
||||||
|
)
|
||||||
tag: str = (
|
tag: str = (
|
||||||
label
|
label
|
||||||
or
|
or
|
||||||
|
|
@ -201,6 +231,20 @@ async def supervise_run_process(
|
||||||
# MANAGED keys below override on conflict.
|
# MANAGED keys below override on conflict.
|
||||||
rp_kwargs: dict = dict(run_process_kwargs)
|
rp_kwargs: dict = dict(run_process_kwargs)
|
||||||
|
|
||||||
|
# XXX reject `trio`'s `capture_stdout`/`capture_stderr`: they
|
||||||
|
# ALIAS our MANAGED `stdout`/`stderr` keys, so forwarding them
|
||||||
|
# makes `trio.run_process` raise an opaque
|
||||||
|
# `ValueError("can't specify both ...")`. Fail LOUD + early
|
||||||
|
# with a pointer to the supported knobs instead.
|
||||||
|
for _cap_kw in ('capture_stdout', 'capture_stderr'):
|
||||||
|
if _cap_kw in rp_kwargs:
|
||||||
|
raise ValueError(
|
||||||
|
f'{_cap_kw!r} is unsupported here; use '
|
||||||
|
f'`relay_stdout=`/`relay_stderr=` for a live relay, '
|
||||||
|
f'or the `stdout=` override / default `check=True` '
|
||||||
|
f'stderr-capture instead.'
|
||||||
|
)
|
||||||
|
|
||||||
# XXX ALWAYS isolate the controlling-tty's stdin.
|
# XXX ALWAYS isolate the controlling-tty's stdin.
|
||||||
rp_kwargs['stdin'] = subprocess.DEVNULL
|
rp_kwargs['stdin'] = subprocess.DEVNULL
|
||||||
|
|
||||||
|
|
@ -210,6 +254,16 @@ async def supervise_run_process(
|
||||||
if relay_stdout:
|
if relay_stdout:
|
||||||
rp_kwargs['stdout'] = subprocess.PIPE
|
rp_kwargs['stdout'] = subprocess.PIPE
|
||||||
elif stdout is not _UNSET:
|
elif stdout is not _UNSET:
|
||||||
|
# XXX a bare `stdout=PIPE` override has NO drain reader
|
||||||
|
# (only `relay_stdout` spins one up), so it would deadlock
|
||||||
|
# once the ~64KiB OS pipe buffer fills — reject it.
|
||||||
|
if stdout is subprocess.PIPE:
|
||||||
|
raise ValueError(
|
||||||
|
'Use `relay_stdout=True` to PIPE *and* drain '
|
||||||
|
'stdout; a bare `stdout=subprocess.PIPE` override '
|
||||||
|
'has no reader and will deadlock once the ~64KiB '
|
||||||
|
'pipe buffer fills.'
|
||||||
|
)
|
||||||
rp_kwargs['stdout'] = stdout
|
rp_kwargs['stdout'] = stdout
|
||||||
else:
|
else:
|
||||||
rp_kwargs['stdout'] = subprocess.DEVNULL
|
rp_kwargs['stdout'] = subprocess.DEVNULL
|
||||||
|
|
@ -262,6 +316,7 @@ async def supervise_run_process(
|
||||||
emit=emit if relay_stderr else None,
|
emit=emit if relay_stderr else None,
|
||||||
tag=f'{tag}:err',
|
tag=f'{tag}:err',
|
||||||
accum=stderr_accum,
|
accum=stderr_accum,
|
||||||
|
accum_cap=_STDERR_NOTE_TAIL_BYTES,
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -287,6 +342,10 @@ async def supervise_run_process(
|
||||||
if stderr_accum is not None
|
if stderr_accum is not None
|
||||||
else b''
|
else b''
|
||||||
)
|
)
|
||||||
|
# NOTE: stdout is NOT captured (DEVNULL unless overridden)
|
||||||
|
# so the CPE carries only (tail-bounded) stderr; a cmd that
|
||||||
|
# logs failures to *stdout* should `relay_stdout=True` to
|
||||||
|
# surface them in the note.
|
||||||
cpe = subprocess.CalledProcessError(
|
cpe = subprocess.CalledProcessError(
|
||||||
returncode=trio_proc.returncode,
|
returncode=trio_proc.returncode,
|
||||||
cmd=trio_proc.args,
|
cmd=trio_proc.args,
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue