Compare commits
No commits in common. "a1d75625e46bf19452365ce5f4ec5be960130fbf" and "fb04f746057f634061f443602c37e291e305ad21" have entirely different histories.
a1d75625e4
...
fb04f74605
|
|
@ -150,18 +150,6 @@ def pytest_generate_tests(metafunc):
|
|||
metafunc.parametrize("start_method", [spawn_backend], scope='module')
|
||||
|
||||
|
||||
# TODO: a way to let test scripts (like from `examples/`)
|
||||
# guarantee they won't registry addr collide!
|
||||
# @pytest.fixture
|
||||
# def open_test_runtime(
|
||||
# reg_addr: tuple,
|
||||
# ) -> AsyncContextManager:
|
||||
# return partial(
|
||||
# tractor.open_nursery,
|
||||
# registry_addrs=[reg_addr],
|
||||
# )
|
||||
|
||||
|
||||
def sig_prog(proc, sig):
|
||||
"Kill the actor-process with ``sig``."
|
||||
proc.send_signal(sig)
|
||||
|
|
|
|||
|
|
@ -218,9 +218,10 @@ def expect_any_of(
|
|||
)
|
||||
|
||||
return expected_patts
|
||||
# yield child
|
||||
|
||||
|
||||
def test_sync_pause_from_aio_task(
|
||||
def test_pause_from_asyncio_task(
|
||||
spawn,
|
||||
ctlc: bool
|
||||
# ^TODO, fix for `asyncio`!!
|
||||
|
|
@ -326,25 +327,3 @@ def test_sync_pause_from_aio_task(
|
|||
|
||||
child.sendline('c')
|
||||
child.expect(EOF)
|
||||
|
||||
|
||||
def test_sync_pause_from_non_greenbacked_aio_task():
|
||||
'''
|
||||
Where the `breakpoint()` caller task is NOT spawned by
|
||||
`tractor.to_asyncio` and thus never activates
|
||||
a `greenback.ensure_portal()` beforehand, presumably bc the task
|
||||
was started by some lib/dep as in often seen in the field.
|
||||
|
||||
Ensure sync pausing works when the pause is in,
|
||||
|
||||
- the root actor running in infected-mode?
|
||||
|_ since we don't need any IPC to acquire the debug lock?
|
||||
|_ is there some way to handle this like the non-main-thread case?
|
||||
|
||||
All other cases need to error out appropriately right?
|
||||
|
||||
- for any subactor we can't avoid needing the repl lock..
|
||||
|_ is there a way to hook into `asyncio.ensure_future(obj)`?
|
||||
|
||||
'''
|
||||
pass
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@ The hipster way to force SC onto the stdlib's "async": 'infection mode'.
|
|||
import asyncio
|
||||
import builtins
|
||||
from contextlib import ExitStack
|
||||
# from functools import partial
|
||||
import itertools
|
||||
import importlib
|
||||
import os
|
||||
|
|
@ -109,9 +108,7 @@ async def asyncio_actor(
|
|||
|
||||
except BaseException as err:
|
||||
if expect_err:
|
||||
assert isinstance(err, error_type), (
|
||||
f'{type(err)} is not {error_type}?'
|
||||
)
|
||||
assert isinstance(err, error_type)
|
||||
|
||||
raise
|
||||
|
||||
|
|
@ -183,8 +180,8 @@ def test_trio_cancels_aio(reg_addr):
|
|||
with trio.move_on_after(1):
|
||||
# cancel the nursery shortly after boot
|
||||
|
||||
async with tractor.open_nursery() as tn:
|
||||
await tn.run_in_actor(
|
||||
async with tractor.open_nursery() as n:
|
||||
await n.run_in_actor(
|
||||
asyncio_actor,
|
||||
target='aio_sleep_forever',
|
||||
expect_err='trio.Cancelled',
|
||||
|
|
@ -204,12 +201,9 @@ async def trio_ctx(
|
|||
# this will block until the ``asyncio`` task sends a "first"
|
||||
# message.
|
||||
with trio.fail_after(2):
|
||||
try:
|
||||
async with (
|
||||
trio.open_nursery(
|
||||
# TODO, for new `trio` / py3.13
|
||||
# strict_exception_groups=False,
|
||||
) as tn,
|
||||
trio.open_nursery() as n,
|
||||
|
||||
tractor.to_asyncio.open_channel_from(
|
||||
sleep_and_err,
|
||||
) as (first, chan),
|
||||
|
|
@ -218,20 +212,12 @@ async def trio_ctx(
|
|||
assert first == 'start'
|
||||
|
||||
# spawn another asyncio task for the cuck of it.
|
||||
tn.start_soon(
|
||||
n.start_soon(
|
||||
tractor.to_asyncio.run_task,
|
||||
aio_sleep_forever,
|
||||
)
|
||||
await trio.sleep_forever()
|
||||
|
||||
# TODO, factor this into a `trionics.collapse()`?
|
||||
except* BaseException as beg:
|
||||
# await tractor.pause(shield=True)
|
||||
if len(excs := beg.exceptions) == 1:
|
||||
raise excs[0]
|
||||
else:
|
||||
raise
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'parent_cancels',
|
||||
|
|
@ -249,6 +235,7 @@ def test_context_spawns_aio_task_that_errors(
|
|||
|
||||
'''
|
||||
async def main():
|
||||
|
||||
with trio.fail_after(2):
|
||||
async with tractor.open_nursery() as n:
|
||||
p = await n.start_actor(
|
||||
|
|
@ -320,9 +307,7 @@ async def aio_cancel():
|
|||
await aio_sleep_forever()
|
||||
|
||||
|
||||
def test_aio_cancelled_from_aio_causes_trio_cancelled(
|
||||
reg_addr: tuple,
|
||||
):
|
||||
def test_aio_cancelled_from_aio_causes_trio_cancelled(reg_addr):
|
||||
'''
|
||||
When the `asyncio.Task` cancels itself the `trio` side cshould
|
||||
also cancel and teardown and relay the cancellation cross-process
|
||||
|
|
@ -419,7 +404,6 @@ async def stream_from_aio(
|
|||
sequence=seq,
|
||||
expect_cancel=raise_err or exit_early,
|
||||
fail_early=aio_raise_err,
|
||||
|
||||
) as (first, chan):
|
||||
|
||||
assert first is True
|
||||
|
|
@ -438,15 +422,10 @@ async def stream_from_aio(
|
|||
if raise_err:
|
||||
raise Exception
|
||||
elif exit_early:
|
||||
print('`consume()` breaking early!\n')
|
||||
break
|
||||
|
||||
print('returning from `consume()`..\n')
|
||||
|
||||
# run 2 tasks each pulling from
|
||||
# the inter-task-channel with the 2nd
|
||||
# using a fan-out `BroadcastReceiver`.
|
||||
if fan_out:
|
||||
# start second task that get's the same stream value set.
|
||||
async with (
|
||||
|
||||
# NOTE: this has to come first to avoid
|
||||
|
|
@ -456,19 +435,11 @@ async def stream_from_aio(
|
|||
|
||||
trio.open_nursery() as n,
|
||||
):
|
||||
# start 2nd task that get's broadcast the same
|
||||
# value set.
|
||||
n.start_soon(consume, br)
|
||||
await consume(chan)
|
||||
|
||||
else:
|
||||
await consume(chan)
|
||||
except BaseException as err:
|
||||
import logging
|
||||
log = logging.getLogger()
|
||||
log.exception('aio-subactor errored!\n')
|
||||
raise err
|
||||
|
||||
finally:
|
||||
|
||||
if (
|
||||
|
|
@ -489,8 +460,7 @@ async def stream_from_aio(
|
|||
assert not fan_out
|
||||
assert pulled == expect[:51]
|
||||
|
||||
print('trio guest-mode task completed!')
|
||||
assert chan._aio_task.done()
|
||||
print('trio guest mode task completed!')
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -530,37 +500,19 @@ def test_trio_error_cancels_intertask_chan(reg_addr):
|
|||
excinfo.value.boxed_type is Exception
|
||||
|
||||
|
||||
def test_trio_closes_early_and_channel_exits(
|
||||
reg_addr: tuple[str, int],
|
||||
):
|
||||
'''
|
||||
Check that if the `trio`-task "exits early" on `async for`ing the
|
||||
inter-task-channel (via a `break`) we exit silently from the
|
||||
`open_channel_from()` block and get a final `Return[None]` msg.
|
||||
|
||||
'''
|
||||
def test_trio_closes_early_and_channel_exits(reg_addr):
|
||||
async def main():
|
||||
with trio.fail_after(2):
|
||||
async with tractor.open_nursery(
|
||||
# debug_mode=True,
|
||||
# enable_stack_on_sig=True,
|
||||
) as n:
|
||||
async with tractor.open_nursery() as n:
|
||||
portal = await n.run_in_actor(
|
||||
stream_from_aio,
|
||||
exit_early=True,
|
||||
infect_asyncio=True,
|
||||
)
|
||||
# should raise RAE diectly
|
||||
print('waiting on final infected subactor result..')
|
||||
res: None = await portal.wait_for_result()
|
||||
assert res is None
|
||||
print('infected subactor returned result: {res!r}\n')
|
||||
await portal.result()
|
||||
|
||||
# should be a quiet exit on a simple channel exit
|
||||
trio.run(
|
||||
main,
|
||||
# strict_exception_groups=False,
|
||||
)
|
||||
trio.run(main)
|
||||
|
||||
|
||||
def test_aio_errors_and_channel_propagates_and_closes(reg_addr):
|
||||
|
|
@ -584,10 +536,15 @@ def test_aio_errors_and_channel_propagates_and_closes(reg_addr):
|
|||
excinfo.value.boxed_type is Exception
|
||||
|
||||
|
||||
async def aio_echo_server(
|
||||
@tractor.context
|
||||
async def trio_to_aio_echo_server(
|
||||
ctx: tractor.Context,
|
||||
):
|
||||
|
||||
async def aio_echo_server(
|
||||
to_trio: trio.MemorySendChannel,
|
||||
from_trio: asyncio.Queue,
|
||||
) -> None:
|
||||
) -> None:
|
||||
|
||||
to_trio.send_nowait('start')
|
||||
|
||||
|
|
@ -605,19 +562,15 @@ async def aio_echo_server(
|
|||
|
||||
print('exiting asyncio task')
|
||||
|
||||
|
||||
@tractor.context
|
||||
async def trio_to_aio_echo_server(
|
||||
ctx: tractor.Context|None,
|
||||
):
|
||||
async with to_asyncio.open_channel_from(
|
||||
aio_echo_server,
|
||||
) as (first, chan):
|
||||
assert first == 'start'
|
||||
|
||||
assert first == 'start'
|
||||
await ctx.started(first)
|
||||
|
||||
async with ctx.open_stream() as stream:
|
||||
|
||||
async for msg in stream:
|
||||
print(f'asyncio echoing {msg}')
|
||||
await chan.send(msg)
|
||||
|
|
@ -696,6 +649,7 @@ def test_echoserver_detailed_mechanics(
|
|||
trio.run(main)
|
||||
|
||||
|
||||
|
||||
@tractor.context
|
||||
async def manage_file(
|
||||
ctx: tractor.Context,
|
||||
|
|
|
|||
|
|
@ -1,244 +0,0 @@
|
|||
'''
|
||||
Special attention cases for using "infect `asyncio`" mode from a root
|
||||
actor; i.e. not using a std `trio.run()` bootstrap.
|
||||
|
||||
'''
|
||||
import asyncio
|
||||
from functools import partial
|
||||
|
||||
import pytest
|
||||
import trio
|
||||
import tractor
|
||||
from tractor import (
|
||||
to_asyncio,
|
||||
)
|
||||
from tests.test_infected_asyncio import (
|
||||
aio_echo_server,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'raise_error_mid_stream',
|
||||
[
|
||||
False,
|
||||
Exception,
|
||||
KeyboardInterrupt,
|
||||
],
|
||||
ids='raise_error={}'.format,
|
||||
)
|
||||
def test_infected_root_actor(
|
||||
raise_error_mid_stream: bool|Exception,
|
||||
|
||||
# conftest wide
|
||||
loglevel: str,
|
||||
debug_mode: bool,
|
||||
):
|
||||
'''
|
||||
Verify you can run the `tractor` runtime with `Actor.is_infected_aio() == True`
|
||||
in the root actor.
|
||||
|
||||
'''
|
||||
async def _trio_main():
|
||||
with trio.fail_after(2):
|
||||
first: str
|
||||
chan: to_asyncio.LinkedTaskChannel
|
||||
async with (
|
||||
tractor.open_root_actor(
|
||||
debug_mode=debug_mode,
|
||||
loglevel=loglevel,
|
||||
),
|
||||
to_asyncio.open_channel_from(
|
||||
aio_echo_server,
|
||||
) as (first, chan),
|
||||
):
|
||||
assert first == 'start'
|
||||
|
||||
for i in range(1000):
|
||||
await chan.send(i)
|
||||
out = await chan.receive()
|
||||
assert out == i
|
||||
print(f'asyncio echoing {i}')
|
||||
|
||||
if raise_error_mid_stream and i == 500:
|
||||
raise raise_error_mid_stream
|
||||
|
||||
if out is None:
|
||||
try:
|
||||
out = await chan.receive()
|
||||
except trio.EndOfChannel:
|
||||
break
|
||||
else:
|
||||
raise RuntimeError(
|
||||
'aio channel never stopped?'
|
||||
)
|
||||
|
||||
if raise_error_mid_stream:
|
||||
with pytest.raises(raise_error_mid_stream):
|
||||
tractor.to_asyncio.run_as_asyncio_guest(
|
||||
trio_main=_trio_main,
|
||||
)
|
||||
else:
|
||||
tractor.to_asyncio.run_as_asyncio_guest(
|
||||
trio_main=_trio_main,
|
||||
)
|
||||
|
||||
|
||||
|
||||
async def sync_and_err(
|
||||
# just signature placeholders for compat with
|
||||
# ``to_asyncio.open_channel_from()``
|
||||
to_trio: trio.MemorySendChannel,
|
||||
from_trio: asyncio.Queue,
|
||||
ev: asyncio.Event,
|
||||
|
||||
):
|
||||
if to_trio:
|
||||
to_trio.send_nowait('start')
|
||||
|
||||
await ev.wait()
|
||||
raise RuntimeError('asyncio-side')
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'aio_err_trigger',
|
||||
[
|
||||
'before_start_point',
|
||||
'after_trio_task_starts',
|
||||
'after_start_point',
|
||||
],
|
||||
ids='aio_err_triggered={}'.format
|
||||
)
|
||||
def test_trio_prestarted_task_bubbles(
|
||||
aio_err_trigger: str,
|
||||
|
||||
# conftest wide
|
||||
loglevel: str,
|
||||
debug_mode: bool,
|
||||
):
|
||||
async def pre_started_err(
|
||||
raise_err: bool = False,
|
||||
pre_sleep: float|None = None,
|
||||
aio_trigger: asyncio.Event|None = None,
|
||||
task_status=trio.TASK_STATUS_IGNORED,
|
||||
):
|
||||
'''
|
||||
Maybe pre-started error then sleep.
|
||||
|
||||
'''
|
||||
if pre_sleep is not None:
|
||||
print(f'Sleeping from trio for {pre_sleep!r}s !')
|
||||
await trio.sleep(pre_sleep)
|
||||
|
||||
# signal aio-task to raise JUST AFTER this task
|
||||
# starts but has not yet `.started()`
|
||||
if aio_trigger:
|
||||
print('Signalling aio-task to raise from `trio`!!')
|
||||
aio_trigger.set()
|
||||
|
||||
if raise_err:
|
||||
print('Raising from trio!')
|
||||
raise TypeError('trio-side')
|
||||
|
||||
task_status.started()
|
||||
await trio.sleep_forever()
|
||||
|
||||
async def _trio_main():
|
||||
# with trio.fail_after(2):
|
||||
with trio.fail_after(999):
|
||||
first: str
|
||||
chan: to_asyncio.LinkedTaskChannel
|
||||
aio_ev = asyncio.Event()
|
||||
|
||||
async with (
|
||||
tractor.open_root_actor(
|
||||
debug_mode=False,
|
||||
loglevel=loglevel,
|
||||
),
|
||||
):
|
||||
# TODO, tests for this with 3.13 egs?
|
||||
# from tractor.devx import open_crash_handler
|
||||
# with open_crash_handler():
|
||||
async with (
|
||||
# where we'll start a sub-task that errors BEFORE
|
||||
# calling `.started()` such that the error should
|
||||
# bubble before the guest run terminates!
|
||||
trio.open_nursery() as tn,
|
||||
|
||||
# THEN start an infect task which should error just
|
||||
# after the trio-side's task does.
|
||||
to_asyncio.open_channel_from(
|
||||
partial(
|
||||
sync_and_err,
|
||||
ev=aio_ev,
|
||||
)
|
||||
) as (first, chan),
|
||||
):
|
||||
|
||||
for i in range(5):
|
||||
pre_sleep: float|None = None
|
||||
last_iter: bool = (i == 4)
|
||||
|
||||
# TODO, missing cases?
|
||||
# -[ ] error as well on
|
||||
# 'after_start_point' case as well for
|
||||
# another case?
|
||||
raise_err: bool = False
|
||||
|
||||
if last_iter:
|
||||
raise_err: bool = True
|
||||
|
||||
# trigger aio task to error on next loop
|
||||
# tick/checkpoint
|
||||
if aio_err_trigger == 'before_start_point':
|
||||
aio_ev.set()
|
||||
|
||||
pre_sleep: float = 0
|
||||
|
||||
await tn.start(
|
||||
pre_started_err,
|
||||
raise_err,
|
||||
pre_sleep,
|
||||
(aio_ev if (
|
||||
aio_err_trigger == 'after_trio_task_starts'
|
||||
and
|
||||
last_iter
|
||||
) else None
|
||||
),
|
||||
)
|
||||
|
||||
if (
|
||||
aio_err_trigger == 'after_start_point'
|
||||
and
|
||||
last_iter
|
||||
):
|
||||
aio_ev.set()
|
||||
|
||||
with pytest.raises(
|
||||
expected_exception=ExceptionGroup,
|
||||
) as excinfo:
|
||||
tractor.to_asyncio.run_as_asyncio_guest(
|
||||
trio_main=_trio_main,
|
||||
)
|
||||
|
||||
eg = excinfo.value
|
||||
rte_eg, rest_eg = eg.split(RuntimeError)
|
||||
|
||||
# ensure the trio-task's error bubbled despite the aio-side
|
||||
# having (maybe) errored first.
|
||||
if aio_err_trigger in (
|
||||
'after_trio_task_starts',
|
||||
'after_start_point',
|
||||
):
|
||||
assert len(errs := rest_eg.exceptions) == 1
|
||||
typerr = errs[0]
|
||||
assert (
|
||||
type(typerr) is TypeError
|
||||
and
|
||||
'trio-side' in typerr.args
|
||||
)
|
||||
|
||||
# when aio errors BEFORE (last) trio task is scheduled, we should
|
||||
# never see anythinb but the aio-side.
|
||||
else:
|
||||
assert len(rtes := rte_eg.exceptions) == 1
|
||||
assert 'asyncio-side' in rtes[0].args[0]
|
||||
|
|
@ -3,10 +3,6 @@ Reminders for oddities in `trio` that we need to stay aware of and/or
|
|||
want to see changed.
|
||||
|
||||
'''
|
||||
from contextlib import (
|
||||
asynccontextmanager as acm,
|
||||
)
|
||||
|
||||
import pytest
|
||||
import trio
|
||||
from trio import TaskStatus
|
||||
|
|
@ -84,115 +80,3 @@ def test_stashed_child_nursery(use_start_soon):
|
|||
|
||||
with pytest.raises(NameError):
|
||||
trio.run(main)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
('unmask_from_canc', 'canc_from_finally'),
|
||||
[
|
||||
(True, False),
|
||||
(True, True),
|
||||
pytest.param(False, True,
|
||||
marks=pytest.mark.xfail(reason="never raises!")
|
||||
),
|
||||
],
|
||||
# TODO, ask ronny how to impl this .. XD
|
||||
# ids='unmask_from_canc={0}, canc_from_finally={1}',#.format,
|
||||
)
|
||||
def test_acm_embedded_nursery_propagates_enter_err(
|
||||
canc_from_finally: bool,
|
||||
unmask_from_canc: bool,
|
||||
):
|
||||
'''
|
||||
Demo how a masking `trio.Cancelled` could be handled by unmasking from the
|
||||
`.__context__` field when a user (by accident) re-raises from a `finally:`.
|
||||
|
||||
'''
|
||||
import tractor
|
||||
|
||||
@acm
|
||||
async def maybe_raise_from_masking_exc(
|
||||
tn: trio.Nursery,
|
||||
unmask_from: BaseException|None = trio.Cancelled
|
||||
|
||||
# TODO, maybe offer a collection?
|
||||
# unmask_from: set[BaseException] = {
|
||||
# trio.Cancelled,
|
||||
# },
|
||||
):
|
||||
if not unmask_from:
|
||||
yield
|
||||
return
|
||||
|
||||
try:
|
||||
yield
|
||||
except* unmask_from as be_eg:
|
||||
|
||||
# TODO, if we offer `unmask_from: set`
|
||||
# for masker_exc_type in unmask_from:
|
||||
|
||||
matches, rest = be_eg.split(unmask_from)
|
||||
if not matches:
|
||||
raise
|
||||
|
||||
for exc_match in be_eg.exceptions:
|
||||
if (
|
||||
(exc_ctx := exc_match.__context__)
|
||||
and
|
||||
type(exc_ctx) not in {
|
||||
# trio.Cancelled, # always by default?
|
||||
unmask_from,
|
||||
}
|
||||
):
|
||||
exc_ctx.add_note(
|
||||
f'\n'
|
||||
f'WARNING: the above error was masked by a {unmask_from!r} !?!\n'
|
||||
f'Are you always cancelling? Say from a `finally:` ?\n\n'
|
||||
|
||||
f'{tn!r}'
|
||||
)
|
||||
raise exc_ctx from exc_match
|
||||
|
||||
|
||||
@acm
|
||||
async def wraps_tn_that_always_cancels():
|
||||
async with (
|
||||
trio.open_nursery() as tn,
|
||||
maybe_raise_from_masking_exc(
|
||||
tn=tn,
|
||||
unmask_from=(
|
||||
trio.Cancelled
|
||||
if unmask_from_canc
|
||||
else None
|
||||
),
|
||||
)
|
||||
):
|
||||
try:
|
||||
yield tn
|
||||
finally:
|
||||
if canc_from_finally:
|
||||
tn.cancel_scope.cancel()
|
||||
await trio.lowlevel.checkpoint()
|
||||
|
||||
async def _main():
|
||||
with tractor.devx.open_crash_handler() as bxerr:
|
||||
assert not bxerr.value
|
||||
|
||||
async with (
|
||||
wraps_tn_that_always_cancels() as tn,
|
||||
):
|
||||
assert not tn.cancel_scope.cancel_called
|
||||
assert 0
|
||||
|
||||
assert (
|
||||
(err := bxerr.value)
|
||||
and
|
||||
type(err) is AssertionError
|
||||
)
|
||||
|
||||
with pytest.raises(ExceptionGroup) as excinfo:
|
||||
trio.run(_main)
|
||||
|
||||
eg: ExceptionGroup = excinfo.value
|
||||
assert_eg, rest_eg = eg.split(AssertionError)
|
||||
|
||||
assert len(assert_eg.exceptions) == 1
|
||||
|
|
|
|||
|
|
@ -1146,51 +1146,19 @@ def unpack_error(
|
|||
|
||||
|
||||
def is_multi_cancelled(
|
||||
exc: BaseException|BaseExceptionGroup,
|
||||
|
||||
ignore_nested: set[BaseException] = set(),
|
||||
|
||||
) -> bool|BaseExceptionGroup:
|
||||
exc: BaseException|BaseExceptionGroup
|
||||
) -> bool:
|
||||
'''
|
||||
Predicate to determine if an `BaseExceptionGroup` only contains
|
||||
some (maybe nested) set of sub-grouped exceptions (like only
|
||||
`trio.Cancelled`s which get swallowed silently by default) and is
|
||||
thus the result of "gracefully cancelling" a collection of
|
||||
sub-tasks (or other conc primitives) and receiving a "cancelled
|
||||
ACK" from each after termination.
|
||||
|
||||
Docs:
|
||||
----
|
||||
- https://docs.python.org/3/library/exceptions.html#exception-groups
|
||||
- https://docs.python.org/3/library/exceptions.html#BaseExceptionGroup.subgroup
|
||||
Predicate to determine if a possible ``BaseExceptionGroup`` contains
|
||||
only ``trio.Cancelled`` sub-exceptions (and is likely the result of
|
||||
cancelling a collection of subtasks.
|
||||
|
||||
'''
|
||||
|
||||
if (
|
||||
not ignore_nested
|
||||
or
|
||||
trio.Cancelled in ignore_nested
|
||||
# XXX always count-in `trio`'s native signal
|
||||
):
|
||||
ignore_nested |= {trio.Cancelled}
|
||||
|
||||
if isinstance(exc, BaseExceptionGroup):
|
||||
matched_exc: BaseExceptionGroup|None = exc.subgroup(
|
||||
tuple(ignore_nested),
|
||||
return exc.subgroup(
|
||||
lambda exc: isinstance(exc, trio.Cancelled)
|
||||
) is not None
|
||||
|
||||
# TODO, complain about why not allowed XD
|
||||
# condition=tuple(ignore_nested),
|
||||
)
|
||||
if matched_exc is not None:
|
||||
return matched_exc
|
||||
|
||||
# NOTE, IFF no excs types match (throughout the error-tree)
|
||||
# -> return `False`, OW return the matched sub-eg.
|
||||
#
|
||||
# IOW, for the inverse of ^ for the purpose of
|
||||
# maybe-enter-REPL--logic: "only debug when the err-tree contains
|
||||
# at least one exc-type NOT in `ignore_nested`" ; i.e. the case where
|
||||
# we fallthrough and return `False` here.
|
||||
return False
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -95,13 +95,6 @@ async def open_root_actor(
|
|||
|
||||
hide_tb: bool = True,
|
||||
|
||||
# XXX, proxied directly to `.devx._debug._maybe_enter_pm()`
|
||||
# for REPL-entry logic.
|
||||
debug_filter: Callable[
|
||||
[BaseException|BaseExceptionGroup],
|
||||
bool,
|
||||
] = lambda err: not is_multi_cancelled(err),
|
||||
|
||||
# TODO, a way for actors to augment passing derived
|
||||
# read-only state to sublayers?
|
||||
# extra_rt_vars: dict|None = None,
|
||||
|
|
@ -341,10 +334,6 @@ async def open_root_actor(
|
|||
loglevel=loglevel,
|
||||
enable_modules=enable_modules,
|
||||
)
|
||||
# XXX, in case the root actor runtime was actually run from
|
||||
# `tractor.to_asyncio.run_as_asyncio_guest()` and NOt
|
||||
# `.trio.run()`.
|
||||
actor._infected_aio = _state._runtime_vars['_is_infected_aio']
|
||||
|
||||
# Start up main task set via core actor-runtime nurseries.
|
||||
try:
|
||||
|
|
@ -386,7 +375,6 @@ async def open_root_actor(
|
|||
Exception,
|
||||
BaseExceptionGroup,
|
||||
) as err:
|
||||
|
||||
# XXX NOTE XXX see equiv note inside
|
||||
# `._runtime.Actor._stream_handler()` where in the
|
||||
# non-root or root-that-opened-this-mahually case we
|
||||
|
|
@ -395,15 +383,11 @@ async def open_root_actor(
|
|||
entered: bool = await _debug._maybe_enter_pm(
|
||||
err,
|
||||
api_frame=inspect.currentframe(),
|
||||
debug_filter=debug_filter,
|
||||
)
|
||||
|
||||
if (
|
||||
not entered
|
||||
and
|
||||
not is_multi_cancelled(
|
||||
err,
|
||||
)
|
||||
not is_multi_cancelled(err)
|
||||
):
|
||||
logger.exception('Root actor crashed\n')
|
||||
|
||||
|
|
|
|||
|
|
@ -75,7 +75,6 @@ from tractor import _state
|
|||
from tractor._exceptions import (
|
||||
InternalError,
|
||||
NoRuntime,
|
||||
is_multi_cancelled,
|
||||
)
|
||||
from tractor._state import (
|
||||
current_actor,
|
||||
|
|
@ -317,7 +316,6 @@ class Lock:
|
|||
we_released: bool = False
|
||||
ctx_in_debug: Context|None = cls.ctx_in_debug
|
||||
repl_task: Task|Thread|None = DebugStatus.repl_task
|
||||
message: str = ''
|
||||
|
||||
try:
|
||||
if not DebugStatus.is_main_trio_thread():
|
||||
|
|
@ -445,10 +443,7 @@ class Lock:
|
|||
f'|_{repl_task}\n'
|
||||
)
|
||||
|
||||
if message:
|
||||
log.devx(message)
|
||||
else:
|
||||
import pdbp; pdbp.set_trace()
|
||||
|
||||
return we_released
|
||||
|
||||
|
|
@ -1748,7 +1743,7 @@ async def _pause(
|
|||
] = trio.TASK_STATUS_IGNORED,
|
||||
**debug_func_kwargs,
|
||||
|
||||
) -> tuple[Task, PdbREPL]|None:
|
||||
) -> tuple[PdbREPL, Task]|None:
|
||||
'''
|
||||
Inner impl for `pause()` to avoid the `trio.CancelScope.__exit__()`
|
||||
stack frame when not shielded (since apparently i can't figure out
|
||||
|
|
@ -1934,7 +1929,7 @@ async def _pause(
|
|||
)
|
||||
with trio.CancelScope(shield=shield):
|
||||
await trio.lowlevel.checkpoint()
|
||||
return (repl, task)
|
||||
return repl, task
|
||||
|
||||
# elif repl_task:
|
||||
# log.warning(
|
||||
|
|
@ -2535,17 +2530,26 @@ def pause_from_sync(
|
|||
f'{actor.uid} task called `tractor.pause_from_sync()`\n'
|
||||
)
|
||||
|
||||
# TODO: once supported, remove this AND the one
|
||||
# inside `._pause()`!
|
||||
# outstanding impl fixes:
|
||||
# -[ ] need to make `.shield_sigint()` below work here!
|
||||
# -[ ] how to handle `asyncio`'s new SIGINT-handler
|
||||
# injection?
|
||||
# -[ ] should `breakpoint()` work and what does it normally
|
||||
# do in `asyncio` ctxs?
|
||||
# if actor.is_infected_aio():
|
||||
# raise RuntimeError(
|
||||
# '`tractor.pause[_from_sync]()` not yet supported '
|
||||
# 'for infected `asyncio` mode!'
|
||||
# )
|
||||
|
||||
repl: PdbREPL = mk_pdb()
|
||||
|
||||
# message += f'-> created local REPL {repl}\n'
|
||||
is_trio_thread: bool = DebugStatus.is_main_trio_thread()
|
||||
is_root: bool = is_root_process()
|
||||
is_infected_aio: bool = actor.is_infected_aio()
|
||||
thread: Thread = threading.current_thread()
|
||||
|
||||
asyncio_task: asyncio.Task|None = None
|
||||
if is_infected_aio:
|
||||
asyncio_task = asyncio.current_task()
|
||||
is_aio: bool = actor.is_infected_aio()
|
||||
|
||||
# TODO: we could also check for a non-`.to_thread` context
|
||||
# using `trio.from_thread.check_cancelled()` (says
|
||||
|
|
@ -2561,14 +2565,20 @@ def pause_from_sync(
|
|||
if (
|
||||
not is_trio_thread
|
||||
and
|
||||
not asyncio_task
|
||||
not is_aio # see below for this usage
|
||||
):
|
||||
# TODO: `threading.Lock()` this so we don't get races in
|
||||
# multi-thr cases where they're acquiring/releasing the
|
||||
# REPL and setting request/`Lock` state, etc..
|
||||
repl_owner: Thread = thread
|
||||
thread: threading.Thread = threading.current_thread()
|
||||
repl_owner = thread
|
||||
|
||||
# TODO: make root-actor bg thread usage work!
|
||||
if (
|
||||
is_root
|
||||
# or
|
||||
# is_aio
|
||||
):
|
||||
if is_root:
|
||||
message += (
|
||||
f'-> called from a root-actor bg {thread}\n'
|
||||
|
|
@ -2627,32 +2637,13 @@ def pause_from_sync(
|
|||
DebugStatus.shield_sigint()
|
||||
assert bg_task is not DebugStatus.repl_task
|
||||
|
||||
# TODO: once supported, remove this AND the one
|
||||
# inside `._pause()`!
|
||||
# outstanding impl fixes:
|
||||
# -[ ] need to make `.shield_sigint()` below work here!
|
||||
# -[ ] how to handle `asyncio`'s new SIGINT-handler
|
||||
# injection?
|
||||
# -[ ] should `breakpoint()` work and what does it normally
|
||||
# do in `asyncio` ctxs?
|
||||
# if actor.is_infected_aio():
|
||||
# raise RuntimeError(
|
||||
# '`tractor.pause[_from_sync]()` not yet supported '
|
||||
# 'for infected `asyncio` mode!'
|
||||
# )
|
||||
elif (
|
||||
not is_trio_thread
|
||||
and
|
||||
is_infected_aio # as in, the special actor-runtime mode
|
||||
# ^NOTE XXX, that doesn't mean the caller is necessarily
|
||||
# an `asyncio.Task` just that `trio` has been embedded on
|
||||
# the `asyncio` event loop!
|
||||
and
|
||||
asyncio_task # transitive caller is an actual `asyncio.Task`
|
||||
is_aio
|
||||
):
|
||||
greenback: ModuleType = maybe_import_greenback()
|
||||
|
||||
if greenback.has_portal():
|
||||
repl_owner: Task = asyncio.current_task()
|
||||
DebugStatus.shield_sigint()
|
||||
fute: asyncio.Future = run_trio_task_in_future(
|
||||
partial(
|
||||
|
|
@ -2669,53 +2660,11 @@ def pause_from_sync(
|
|||
**_pause_kwargs
|
||||
)
|
||||
)
|
||||
repl_owner = asyncio_task
|
||||
bg_task, _ = greenback.await_(fute)
|
||||
# TODO: ASYNC version -> `.pause_from_aio()`?
|
||||
|
||||
# TODO: for async version -> `.pause_from_aio()`?
|
||||
# bg_task, _ = await fute
|
||||
|
||||
# handle the case where an `asyncio` task has been
|
||||
# spawned WITHOUT enabling a `greenback` portal..
|
||||
# => can often happen in 3rd party libs.
|
||||
else:
|
||||
bg_task = repl_owner
|
||||
|
||||
# TODO, ostensibly we can just acquire the
|
||||
# debug lock directly presuming we're the
|
||||
# root actor running in infected asyncio
|
||||
# mode?
|
||||
#
|
||||
# TODO, this would be a special case where
|
||||
# a `_pause_from_root()` would come in very
|
||||
# handy!
|
||||
# if is_root:
|
||||
# import pdbp; pdbp.set_trace()
|
||||
# log.warning(
|
||||
# 'Allowing `asyncio` task to acquire debug-lock in root-actor..\n'
|
||||
# 'This is not fully implemented yet; there may be teardown hangs!\n\n'
|
||||
# )
|
||||
# else:
|
||||
|
||||
# simply unsupported, since there exists no hack (i
|
||||
# can think of) to workaround this in a subactor
|
||||
# which needs to lock the root's REPL ow we're sure
|
||||
# to get prompt stdstreams clobbering..
|
||||
cf_repr: str = ''
|
||||
if api_frame:
|
||||
caller_frame: FrameType = api_frame.f_back
|
||||
cf_repr: str = f'caller_frame: {caller_frame!r}\n'
|
||||
|
||||
raise RuntimeError(
|
||||
f"CAN'T USE `greenback._await()` without a portal !?\n\n"
|
||||
f'Likely this task was NOT spawned via the `tractor.to_asyncio` API..\n'
|
||||
f'{asyncio_task}\n'
|
||||
f'{cf_repr}\n'
|
||||
|
||||
f'Prolly the task was started out-of-band (from some lib?)\n'
|
||||
f'AND one of the below was never called ??\n'
|
||||
f'- greenback.ensure_portal()\n'
|
||||
f'- greenback.bestow_portal(<task>)\n'
|
||||
)
|
||||
bg_task, _ = greenback.await_(fute)
|
||||
bg_task: asyncio.Task = asyncio.current_task()
|
||||
|
||||
else: # we are presumably the `trio.run()` + main thread
|
||||
# raises on not-found by default
|
||||
|
|
@ -2966,14 +2915,8 @@ async def _maybe_enter_pm(
|
|||
tb: TracebackType|None = None,
|
||||
api_frame: FrameType|None = None,
|
||||
hide_tb: bool = False,
|
||||
|
||||
# only enter debugger REPL when returns `True`
|
||||
debug_filter: Callable[
|
||||
[BaseException|BaseExceptionGroup],
|
||||
bool,
|
||||
] = lambda err: not is_multi_cancelled(err),
|
||||
|
||||
):
|
||||
from tractor._exceptions import is_multi_cancelled
|
||||
if (
|
||||
debug_mode()
|
||||
|
||||
|
|
@ -2990,8 +2933,7 @@ async def _maybe_enter_pm(
|
|||
|
||||
# Really we just want to mostly avoid catching KBIs here so there
|
||||
# might be a simpler check we can do?
|
||||
and
|
||||
debug_filter(err)
|
||||
and not is_multi_cancelled(err)
|
||||
):
|
||||
api_frame: FrameType = api_frame or inspect.currentframe()
|
||||
tb: TracebackType = tb or sys.exc_info()[2]
|
||||
|
|
@ -3172,7 +3114,7 @@ async def maybe_wait_for_debugger(
|
|||
@cm
|
||||
def open_crash_handler(
|
||||
catch: set[BaseException] = {
|
||||
# Exception,
|
||||
Exception,
|
||||
BaseException,
|
||||
},
|
||||
ignore: set[BaseException] = {
|
||||
|
|
@ -3193,30 +3135,14 @@ def open_crash_handler(
|
|||
'''
|
||||
__tracebackhide__: bool = tb_hide
|
||||
|
||||
class BoxedMaybeException(Struct):
|
||||
value: BaseException|None = None
|
||||
|
||||
# TODO, yield a `outcome.Error`-like boxed type?
|
||||
# -[~] use `outcome.Value/Error` X-> frozen!
|
||||
# -[x] write our own..?
|
||||
# -[ ] consider just wtv is used by `pytest.raises()`?
|
||||
#
|
||||
boxed_maybe_exc = BoxedMaybeException()
|
||||
err: BaseException
|
||||
try:
|
||||
yield boxed_maybe_exc
|
||||
yield
|
||||
except tuple(catch) as err:
|
||||
boxed_maybe_exc.value = err
|
||||
if (
|
||||
type(err) not in ignore
|
||||
and
|
||||
not is_multi_cancelled(
|
||||
err,
|
||||
ignore_nested=ignore
|
||||
)
|
||||
):
|
||||
try:
|
||||
if type(err) not in ignore:
|
||||
|
||||
# use our re-impl-ed version
|
||||
try:
|
||||
_post_mortem(
|
||||
repl=mk_pdb(),
|
||||
tb=sys.exc_info()[2],
|
||||
|
|
@ -3224,13 +3150,13 @@ def open_crash_handler(
|
|||
)
|
||||
except bdb.BdbQuit:
|
||||
__tracebackhide__: bool = False
|
||||
raise err
|
||||
raise
|
||||
|
||||
# XXX NOTE, `pdbp`'s version seems to lose the up-stack
|
||||
# tb-info?
|
||||
# pdbp.xpm()
|
||||
|
||||
raise err
|
||||
raise
|
||||
|
||||
|
||||
@cm
|
||||
|
|
|
|||
|
|
@ -92,7 +92,7 @@ def pformat_boxed_tb(
|
|||
f' ------ {boxer_header} ------\n'
|
||||
f'{tb_body}'
|
||||
f' ------ {boxer_header}- ------\n'
|
||||
f'_|'
|
||||
f'_|\n'
|
||||
)
|
||||
tb_box_indent: str = (
|
||||
tb_box_indent
|
||||
|
|
|
|||
|
|
@ -258,28 +258,20 @@ class ActorContextInfo(Mapping):
|
|||
|
||||
|
||||
def get_logger(
|
||||
name: str|None = None,
|
||||
|
||||
name: str | None = None,
|
||||
_root_name: str = _proj_name,
|
||||
|
||||
logger: Logger|None = None,
|
||||
|
||||
# TODO, using `.config.dictConfig()` api?
|
||||
# -[ ] SO answer with docs links
|
||||
# |_https://stackoverflow.com/questions/7507825/where-is-a-complete-example-of-logging-config-dictconfig
|
||||
# |_https://docs.python.org/3/library/logging.config.html#configuration-dictionary-schema
|
||||
subsys_spec: str|None = None,
|
||||
|
||||
) -> StackLevelAdapter:
|
||||
'''Return the package log or a sub-logger for ``name`` if provided.
|
||||
|
||||
'''
|
||||
log: Logger
|
||||
log = rlog = logger or logging.getLogger(_root_name)
|
||||
log = rlog = logging.getLogger(_root_name)
|
||||
|
||||
if (
|
||||
name
|
||||
and
|
||||
name != _proj_name
|
||||
and name != _proj_name
|
||||
):
|
||||
|
||||
# NOTE: for handling for modules that use ``get_logger(__name__)``
|
||||
|
|
@ -291,7 +283,7 @@ def get_logger(
|
|||
# since in python the {filename} is always this same
|
||||
# module-file.
|
||||
|
||||
sub_name: None|str = None
|
||||
sub_name: None | str = None
|
||||
rname, _, sub_name = name.partition('.')
|
||||
pkgpath, _, modfilename = sub_name.rpartition('.')
|
||||
|
||||
|
|
@ -314,10 +306,7 @@ def get_logger(
|
|||
|
||||
# add our actor-task aware adapter which will dynamically look up
|
||||
# the actor and task names at each log emit
|
||||
logger = StackLevelAdapter(
|
||||
log,
|
||||
ActorContextInfo(),
|
||||
)
|
||||
logger = StackLevelAdapter(log, ActorContextInfo())
|
||||
|
||||
# additional levels
|
||||
for name, val in CUSTOM_LEVELS.items():
|
||||
|
|
@ -330,25 +319,15 @@ def get_logger(
|
|||
|
||||
|
||||
def get_console_log(
|
||||
level: str|None = None,
|
||||
logger: Logger|None = None,
|
||||
level: str | None = None,
|
||||
**kwargs,
|
||||
|
||||
) -> LoggerAdapter:
|
||||
'''
|
||||
Get a `tractor`-style logging instance: a `Logger` wrapped in
|
||||
a `StackLevelAdapter` which injects various concurrency-primitive
|
||||
(process, thread, task) fields and enables a `StreamHandler` that
|
||||
writes on stderr using `colorlog` formatting.
|
||||
|
||||
Yeah yeah, i know we can use `logging.config.dictConfig()`. You do it.
|
||||
'''Get the package logger and enable a handler which writes to stderr.
|
||||
|
||||
Yeah yeah, i know we can use ``DictConfig``. You do it.
|
||||
'''
|
||||
log = get_logger(
|
||||
logger=logger,
|
||||
**kwargs
|
||||
) # set a root logger
|
||||
logger: Logger = log.logger
|
||||
log = get_logger(**kwargs) # our root logger
|
||||
logger = log.logger
|
||||
|
||||
if not level:
|
||||
return log
|
||||
|
|
@ -367,13 +346,9 @@ def get_console_log(
|
|||
None,
|
||||
)
|
||||
):
|
||||
fmt = LOG_FORMAT
|
||||
# if logger:
|
||||
# fmt = None
|
||||
|
||||
handler = StreamHandler()
|
||||
formatter = colorlog.ColoredFormatter(
|
||||
fmt=fmt,
|
||||
LOG_FORMAT,
|
||||
datefmt=DATE_FORMAT,
|
||||
log_colors=STD_PALETTE,
|
||||
secondary_log_colors=BOLD_PALETTE,
|
||||
|
|
@ -390,7 +365,7 @@ def get_loglevel() -> str:
|
|||
|
||||
|
||||
# global module logger for tractor itself
|
||||
log: StackLevelAdapter = get_logger('tractor')
|
||||
log = get_logger('tractor')
|
||||
|
||||
|
||||
def at_least_level(
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue