Compare commits
No commits in common. "a69bc00593145eb68d5a179661ffd0bc13aa8001" and "f7469442e35171e77cd04b7d41224fb769e7ed0d" have entirely different histories.
a69bc00593
...
f7469442e3
|
|
@ -2,10 +2,7 @@ import asyncio
|
|||
|
||||
import trio
|
||||
import tractor
|
||||
from tractor import (
|
||||
to_asyncio,
|
||||
Portal,
|
||||
)
|
||||
from tractor import to_asyncio
|
||||
|
||||
|
||||
async def aio_sleep_forever():
|
||||
|
|
@ -46,7 +43,7 @@ async def bp_then_error(
|
|||
@tractor.context
|
||||
async def trio_ctx(
|
||||
ctx: tractor.Context,
|
||||
bp_before_started: bool = True,
|
||||
bp_before_started: bool = False,
|
||||
):
|
||||
|
||||
# this will block until the ``asyncio`` task sends a "first"
|
||||
|
|
@ -60,6 +57,7 @@ async def trio_ctx(
|
|||
|
||||
trio.open_nursery() as n,
|
||||
):
|
||||
|
||||
assert first == 'start'
|
||||
|
||||
if bp_before_started:
|
||||
|
|
@ -75,18 +73,15 @@ async def trio_ctx(
|
|||
|
||||
|
||||
async def main(
|
||||
bps_all_over: bool = True,
|
||||
bps_all_over: bool = False,
|
||||
|
||||
) -> None:
|
||||
|
||||
async with tractor.open_nursery(
|
||||
debug_mode=True,
|
||||
maybe_enable_greenback=True,
|
||||
# loglevel='devx',
|
||||
# loglevel='runtime',
|
||||
# debug_mode=True,
|
||||
) as n:
|
||||
|
||||
ptl: Portal = await n.start_actor(
|
||||
p = await n.start_actor(
|
||||
'aio_daemon',
|
||||
enable_modules=[__name__],
|
||||
infect_asyncio=True,
|
||||
|
|
@ -94,7 +89,7 @@ async def main(
|
|||
loglevel='cancel',
|
||||
)
|
||||
|
||||
async with ptl.open_context(
|
||||
async with p.open_context(
|
||||
trio_ctx,
|
||||
bp_before_started=bps_all_over,
|
||||
) as (ctx, first):
|
||||
|
|
@ -110,7 +105,7 @@ async def main(
|
|||
|
||||
# TODO: case where we cancel from trio-side while asyncio task
|
||||
# has debugger lock?
|
||||
# await ptl.cancel_actor()
|
||||
# await p.cancel_actor()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -25,8 +25,7 @@ async def main():
|
|||
"""
|
||||
async with tractor.open_nursery(
|
||||
debug_mode=True,
|
||||
# loglevel='cancel',
|
||||
# loglevel='devx',
|
||||
loglevel='cancel',
|
||||
) as n:
|
||||
|
||||
p0 = await n.start_actor('bp_forever', enable_modules=[__name__])
|
||||
|
|
|
|||
|
|
@ -10,7 +10,6 @@ import pytest
|
|||
from pexpect.exceptions import (
|
||||
TIMEOUT,
|
||||
)
|
||||
from pexpect.spawnbase import SpawnBase
|
||||
from tractor._testing import (
|
||||
mk_cmd,
|
||||
)
|
||||
|
|
@ -108,7 +107,7 @@ def expect(
|
|||
|
||||
|
||||
def in_prompt_msg(
|
||||
child: SpawnBase,
|
||||
prompt: str,
|
||||
parts: list[str],
|
||||
|
||||
pause_on_false: bool = False,
|
||||
|
|
@ -126,20 +125,18 @@ def in_prompt_msg(
|
|||
'''
|
||||
__tracebackhide__: bool = False
|
||||
|
||||
before: str = str(child.before.decode())
|
||||
for part in parts:
|
||||
if part not in before:
|
||||
if part not in prompt:
|
||||
if pause_on_false:
|
||||
import pdbp
|
||||
pdbp.set_trace()
|
||||
|
||||
if print_prompt_on_false:
|
||||
print(before)
|
||||
print(prompt)
|
||||
|
||||
if err_on_false:
|
||||
raise ValueError(
|
||||
f'Could not find pattern in `before` output?\n'
|
||||
f'part: {part!r}\n'
|
||||
f'Could not find pattern: {part!r} in `before` output?'
|
||||
)
|
||||
return False
|
||||
|
||||
|
|
@ -150,7 +147,7 @@ def in_prompt_msg(
|
|||
# against call stack frame output from the the 'll' command the like!
|
||||
# -[ ] SO answer for stipping ANSI codes: https://stackoverflow.com/a/14693789
|
||||
def assert_before(
|
||||
child: SpawnBase,
|
||||
child,
|
||||
patts: list[str],
|
||||
|
||||
**kwargs,
|
||||
|
|
@ -158,8 +155,10 @@ def assert_before(
|
|||
) -> None:
|
||||
__tracebackhide__: bool = False
|
||||
|
||||
# as in before the prompt end
|
||||
before: str = str(child.before.decode())
|
||||
assert in_prompt_msg(
|
||||
child=child,
|
||||
prompt=before,
|
||||
parts=patts,
|
||||
|
||||
# since this is an "assert" helper ;)
|
||||
|
|
|
|||
|
|
@ -96,15 +96,14 @@ def test_root_actor_error(
|
|||
# scan for the prompt
|
||||
expect(child, PROMPT)
|
||||
|
||||
before = str(child.before.decode())
|
||||
|
||||
# make sure expected logging and error arrives
|
||||
assert in_prompt_msg(
|
||||
child,
|
||||
[
|
||||
_crash_msg,
|
||||
"('root'",
|
||||
'AssertionError',
|
||||
]
|
||||
before,
|
||||
[_crash_msg, "('root'"]
|
||||
)
|
||||
assert 'AssertionError' in before
|
||||
|
||||
# send user command
|
||||
child.sendline(user_input)
|
||||
|
|
@ -244,12 +243,10 @@ def test_subactor_error(
|
|||
# scan for the prompt
|
||||
child.expect(PROMPT)
|
||||
|
||||
before = str(child.before.decode())
|
||||
assert in_prompt_msg(
|
||||
child,
|
||||
[
|
||||
_crash_msg,
|
||||
"('name_error'",
|
||||
]
|
||||
before,
|
||||
[_crash_msg, "('name_error'"]
|
||||
)
|
||||
|
||||
if do_next:
|
||||
|
|
@ -268,15 +265,17 @@ def test_subactor_error(
|
|||
child.sendline('continue')
|
||||
|
||||
child.expect(PROMPT)
|
||||
assert in_prompt_msg(
|
||||
child,
|
||||
[
|
||||
_crash_msg,
|
||||
before = str(child.before.decode())
|
||||
|
||||
# root actor gets debugger engaged
|
||||
"('root'",
|
||||
assert in_prompt_msg(
|
||||
before,
|
||||
[_crash_msg, "('root'"]
|
||||
)
|
||||
# error is a remote error propagated from the subactor
|
||||
"('name_error'",
|
||||
]
|
||||
assert in_prompt_msg(
|
||||
before,
|
||||
[_crash_msg, "('name_error'"]
|
||||
)
|
||||
|
||||
# another round
|
||||
|
|
@ -297,11 +296,14 @@ def test_subactor_breakpoint(
|
|||
"Single subactor with an infinite breakpoint loop"
|
||||
|
||||
child = spawn('subactor_breakpoint')
|
||||
|
||||
# scan for the prompt
|
||||
child.expect(PROMPT)
|
||||
|
||||
before = str(child.before.decode())
|
||||
assert in_prompt_msg(
|
||||
child,
|
||||
[_pause_msg,
|
||||
"('breakpoint_forever'",]
|
||||
before,
|
||||
[_pause_msg, "('breakpoint_forever'"]
|
||||
)
|
||||
|
||||
# do some "next" commands to demonstrate recurrent breakpoint
|
||||
|
|
@ -317,8 +319,9 @@ def test_subactor_breakpoint(
|
|||
for _ in range(5):
|
||||
child.sendline('continue')
|
||||
child.expect(PROMPT)
|
||||
before = str(child.before.decode())
|
||||
assert in_prompt_msg(
|
||||
child,
|
||||
before,
|
||||
[_pause_msg, "('breakpoint_forever'"]
|
||||
)
|
||||
|
||||
|
|
@ -331,8 +334,9 @@ def test_subactor_breakpoint(
|
|||
# child process should exit but parent will capture pdb.BdbQuit
|
||||
child.expect(PROMPT)
|
||||
|
||||
before = str(child.before.decode())
|
||||
assert in_prompt_msg(
|
||||
child,
|
||||
before,
|
||||
['RemoteActorError:',
|
||||
"('breakpoint_forever'",
|
||||
'bdb.BdbQuit',]
|
||||
|
|
@ -347,8 +351,9 @@ def test_subactor_breakpoint(
|
|||
# process should exit
|
||||
child.expect(EOF)
|
||||
|
||||
before = str(child.before.decode())
|
||||
assert in_prompt_msg(
|
||||
child,
|
||||
before,
|
||||
['RemoteActorError:',
|
||||
"('breakpoint_forever'",
|
||||
'bdb.BdbQuit',]
|
||||
|
|
@ -372,7 +377,7 @@ def test_multi_subactors(
|
|||
|
||||
before = str(child.before.decode())
|
||||
assert in_prompt_msg(
|
||||
child,
|
||||
before,
|
||||
[_pause_msg, "('breakpoint_forever'"]
|
||||
)
|
||||
|
||||
|
|
@ -393,14 +398,12 @@ def test_multi_subactors(
|
|||
|
||||
# first name_error failure
|
||||
child.expect(PROMPT)
|
||||
before = str(child.before.decode())
|
||||
assert in_prompt_msg(
|
||||
child,
|
||||
[
|
||||
_crash_msg,
|
||||
"('name_error'",
|
||||
"NameError",
|
||||
]
|
||||
before,
|
||||
[_crash_msg, "('name_error'"]
|
||||
)
|
||||
assert "NameError" in before
|
||||
|
||||
if ctlc:
|
||||
do_ctlc(child)
|
||||
|
|
@ -424,8 +427,9 @@ def test_multi_subactors(
|
|||
# breakpoint loop should re-engage
|
||||
child.sendline('c')
|
||||
child.expect(PROMPT)
|
||||
before = str(child.before.decode())
|
||||
assert in_prompt_msg(
|
||||
child,
|
||||
before,
|
||||
[_pause_msg, "('breakpoint_forever'"]
|
||||
)
|
||||
|
||||
|
|
@ -518,28 +522,25 @@ def test_multi_daemon_subactors(
|
|||
# the root's tty lock first so anticipate either crash
|
||||
# message on the first entry.
|
||||
|
||||
bp_forev_parts = [
|
||||
_pause_msg,
|
||||
"('bp_forever'",
|
||||
]
|
||||
bp_forev_parts = [_pause_msg, "('bp_forever'"]
|
||||
bp_forev_in_msg = partial(
|
||||
in_prompt_msg,
|
||||
parts=bp_forev_parts,
|
||||
)
|
||||
|
||||
name_error_msg: str = "NameError: name 'doggypants' is not defined"
|
||||
name_error_parts: list[str] = [name_error_msg]
|
||||
name_error_msg = "NameError: name 'doggypants' is not defined"
|
||||
name_error_parts = [name_error_msg]
|
||||
|
||||
before = str(child.before.decode())
|
||||
|
||||
if bp_forev_in_msg(child=child):
|
||||
if bp_forev_in_msg(prompt=before):
|
||||
next_parts = name_error_parts
|
||||
|
||||
elif name_error_msg in before:
|
||||
next_parts = bp_forev_parts
|
||||
|
||||
else:
|
||||
raise ValueError('Neither log msg was found !?')
|
||||
raise ValueError("Neither log msg was found !?")
|
||||
|
||||
if ctlc:
|
||||
do_ctlc(child)
|
||||
|
|
@ -608,12 +609,14 @@ def test_multi_daemon_subactors(
|
|||
# wait for final error in root
|
||||
# where it crashs with boxed error
|
||||
while True:
|
||||
try:
|
||||
child.sendline('c')
|
||||
child.expect(PROMPT)
|
||||
if not in_prompt_msg(
|
||||
assert_before(
|
||||
child,
|
||||
bp_forev_parts
|
||||
):
|
||||
)
|
||||
except AssertionError:
|
||||
break
|
||||
|
||||
assert_before(
|
||||
|
|
@ -794,13 +797,10 @@ def test_root_nursery_cancels_before_child_releases_tty_lock(
|
|||
child = spawn('root_cancelled_but_child_is_in_tty_lock')
|
||||
|
||||
child.expect(PROMPT)
|
||||
assert_before(
|
||||
child,
|
||||
[
|
||||
"NameError: name 'doggypants' is not defined",
|
||||
"tractor._exceptions.RemoteActorError: ('name_error'",
|
||||
],
|
||||
)
|
||||
|
||||
before = str(child.before.decode())
|
||||
assert "NameError: name 'doggypants' is not defined" in before
|
||||
assert "tractor._exceptions.RemoteActorError: ('name_error'" not in before
|
||||
time.sleep(0.5)
|
||||
|
||||
if ctlc:
|
||||
|
|
@ -891,8 +891,9 @@ def test_different_debug_mode_per_actor(
|
|||
child.expect(PROMPT)
|
||||
|
||||
# only one actor should enter the debugger
|
||||
before = str(child.before.decode())
|
||||
assert in_prompt_msg(
|
||||
child,
|
||||
before,
|
||||
[_crash_msg, "('debugged_boi'", "RuntimeError"],
|
||||
)
|
||||
|
||||
|
|
@ -902,6 +903,8 @@ def test_different_debug_mode_per_actor(
|
|||
child.sendline('c')
|
||||
child.expect(EOF)
|
||||
|
||||
before = str(child.before.decode())
|
||||
|
||||
# NOTE: this debugged actor error currently WON'T show up since the
|
||||
# root will actually cancel and terminate the nursery before the error
|
||||
# msg reported back from the debug mode actor is processed.
|
||||
|
|
@ -953,8 +956,9 @@ def test_pause_from_sync(
|
|||
child.expect(PROMPT)
|
||||
|
||||
# XXX shouldn't see gb loaded message with PDB loglevel!
|
||||
before = str(child.before.decode())
|
||||
assert not in_prompt_msg(
|
||||
child,
|
||||
before,
|
||||
['`greenback` portal opened!'],
|
||||
)
|
||||
# should be same root task
|
||||
|
|
@ -1035,7 +1039,7 @@ def test_pause_from_sync(
|
|||
# at the same time as the one that was detected above.
|
||||
for key, other_patts in attach_patts.copy().items():
|
||||
assert not in_prompt_msg(
|
||||
child,
|
||||
before,
|
||||
other_patts,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -291,7 +291,7 @@ def test_context_spawns_aio_task_that_errors(
|
|||
|
||||
err = excinfo.value
|
||||
assert isinstance(err, expect)
|
||||
assert err.boxed_type is AssertionError
|
||||
assert err.boxed_type == AssertionError
|
||||
|
||||
|
||||
async def aio_cancel():
|
||||
|
|
@ -497,7 +497,7 @@ def test_trio_error_cancels_intertask_chan(reg_addr):
|
|||
trio.run(main)
|
||||
|
||||
# ensure boxed error type
|
||||
excinfo.value.boxed_type is Exception
|
||||
excinfo.value.boxed_type == Exception
|
||||
|
||||
|
||||
def test_trio_closes_early_and_channel_exits(reg_addr):
|
||||
|
|
@ -533,7 +533,7 @@ def test_aio_errors_and_channel_propagates_and_closes(reg_addr):
|
|||
) as excinfo:
|
||||
trio.run(main)
|
||||
|
||||
excinfo.value.boxed_type is Exception
|
||||
excinfo.value.boxed_type == Exception
|
||||
|
||||
|
||||
@tractor.context
|
||||
|
|
|
|||
|
|
@ -20,7 +20,6 @@ Multi-core debugging for da peeps!
|
|||
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import asyncio
|
||||
import bdb
|
||||
from contextlib import (
|
||||
asynccontextmanager as acm,
|
||||
|
|
@ -68,7 +67,6 @@ from trio import (
|
|||
TaskStatus,
|
||||
)
|
||||
import tractor
|
||||
from tractor.to_asyncio import run_trio_task_in_future
|
||||
from tractor.log import get_logger
|
||||
from tractor._context import Context
|
||||
from tractor import _state
|
||||
|
|
@ -298,7 +296,7 @@ class Lock:
|
|||
)
|
||||
|
||||
@classmethod
|
||||
# @pdbp.hideframe
|
||||
@pdbp.hideframe
|
||||
def release(
|
||||
cls,
|
||||
raise_on_thread: bool = True,
|
||||
|
|
@ -312,8 +310,6 @@ class Lock:
|
|||
we_released: bool = False
|
||||
ctx_in_debug: Context|None = cls.ctx_in_debug
|
||||
repl_task: Task|Thread|None = DebugStatus.repl_task
|
||||
|
||||
try:
|
||||
if not DebugStatus.is_main_trio_thread():
|
||||
thread: threading.Thread = threading.current_thread()
|
||||
message: str = (
|
||||
|
|
@ -346,6 +342,7 @@ class Lock:
|
|||
f'|_{task}\n'
|
||||
)
|
||||
|
||||
try:
|
||||
lock: trio.StrictFIFOLock = cls._debug_lock
|
||||
owner: Task = lock.statistics().owner
|
||||
if (
|
||||
|
|
@ -791,14 +788,7 @@ class DebugStatus:
|
|||
# in which case schedule the SIGINT shielding override
|
||||
# to in the main thread.
|
||||
# https://docs.python.org/3/library/signal.html#signals-and-threads
|
||||
if (
|
||||
not cls.is_main_trio_thread()
|
||||
and
|
||||
not _state._runtime_vars.get(
|
||||
'_is_infected_aio',
|
||||
False,
|
||||
)
|
||||
):
|
||||
if not cls.is_main_trio_thread():
|
||||
cls._orig_sigint_handler: Callable = trio.from_thread.run_sync(
|
||||
signal.signal,
|
||||
signal.SIGINT,
|
||||
|
|
@ -823,16 +813,7 @@ class DebugStatus:
|
|||
# always restore ``trio``'s sigint handler. see notes below in
|
||||
# the pdb factory about the nightmare that is that code swapping
|
||||
# out the handler when the repl activates...
|
||||
# if not cls.is_main_trio_thread():
|
||||
if (
|
||||
not cls.is_main_trio_thread()
|
||||
and
|
||||
# not _state._runtime_vars.get(
|
||||
# '_is_infected_aio',
|
||||
# False,
|
||||
# )
|
||||
not current_actor().is_infected_aio()
|
||||
):
|
||||
if not cls.is_main_trio_thread():
|
||||
trio.from_thread.run_sync(
|
||||
signal.signal,
|
||||
signal.SIGINT,
|
||||
|
|
@ -890,7 +871,7 @@ class DebugStatus:
|
|||
return False
|
||||
|
||||
@classmethod
|
||||
# @pdbp.hideframe
|
||||
@pdbp.hideframe
|
||||
def release(
|
||||
cls,
|
||||
cancel_req_task: bool = False,
|
||||
|
|
@ -899,21 +880,11 @@ class DebugStatus:
|
|||
try:
|
||||
# sometimes the task might already be terminated in
|
||||
# which case this call will raise an RTE?
|
||||
if repl_release is not None:
|
||||
if (
|
||||
repl_release is not None
|
||||
):
|
||||
if cls.is_main_trio_thread():
|
||||
repl_release.set()
|
||||
|
||||
elif current_actor().is_infected_aio():
|
||||
|
||||
async def _set_repl_release():
|
||||
repl_release.set()
|
||||
|
||||
fute: asyncio.Future = run_trio_task_in_future(
|
||||
_set_repl_release
|
||||
)
|
||||
if not fute.done():
|
||||
log.warning('REPL release state unknown..?')
|
||||
|
||||
else:
|
||||
# XXX NOTE ONLY used for bg root-actor sync
|
||||
# threads, see `.pause_from_sync()`.
|
||||
|
|
@ -1687,24 +1658,18 @@ async def _pause(
|
|||
try:
|
||||
task: Task = current_task()
|
||||
except RuntimeError as rte:
|
||||
# NOTE, 2 cases we might get here:
|
||||
#
|
||||
# - ACTUALLY not a `trio.lowlevel.Task` nor runtime caller,
|
||||
# |_ error out as normal
|
||||
#
|
||||
# - an infected `asycio` actor calls it from an actual
|
||||
# `asyncio.Task`
|
||||
# |_ in this case we DO NOT want to RTE!
|
||||
__tracebackhide__: bool = False
|
||||
if actor.is_infected_aio():
|
||||
log.exception(
|
||||
'Failed to get current `trio`-task?'
|
||||
)
|
||||
raise RuntimeError(
|
||||
'An `asyncio` task should not be calling this!?'
|
||||
) from rte
|
||||
else:
|
||||
task = asyncio.current_task()
|
||||
# if actor.is_infected_aio():
|
||||
# mk_pdb().set_trace()
|
||||
# raise RuntimeError(
|
||||
# '`tractor.pause[_from_sync]()` not yet supported '
|
||||
# 'directly (infected) `asyncio` tasks!'
|
||||
# ) from rte
|
||||
|
||||
raise rte
|
||||
|
||||
if debug_func is not None:
|
||||
debug_func = partial(debug_func)
|
||||
|
|
@ -2095,7 +2060,6 @@ async def _pause(
|
|||
f'on behalf of {repl_task} ??\n'
|
||||
)
|
||||
|
||||
if not actor.is_infected_aio():
|
||||
DebugStatus.release(cancel_req_task=True)
|
||||
|
||||
# sanity checks for ^ on request/status teardown
|
||||
|
|
@ -2149,9 +2113,7 @@ def _set_trace(
|
|||
log.pdb(
|
||||
f'{_pause_msg}\n'
|
||||
f'>(\n'
|
||||
f'|_{actor.uid}\n'
|
||||
f' |_{task}\n' # @ {actor.uid}\n'
|
||||
# f'|_{task}\n'
|
||||
f'|_ {task} @ {actor.uid}\n'
|
||||
# ^-TODO-^ more compact pformating?
|
||||
# -[ ] make an `Actor.__repr()__`
|
||||
# -[ ] should we use `log.pformat_task_uid()`?
|
||||
|
|
@ -2428,6 +2390,9 @@ def pause_from_sync(
|
|||
actor: tractor.Actor = current_actor(
|
||||
err_on_no_runtime=False,
|
||||
)
|
||||
message: str = (
|
||||
f'{actor.uid} task called `tractor.pause_from_sync()`\n'
|
||||
)
|
||||
if not actor:
|
||||
raise RuntimeError(
|
||||
'Not inside the `tractor`-runtime?\n'
|
||||
|
|
@ -2435,9 +2400,6 @@ def pause_from_sync(
|
|||
'- `async with tractor.open_nursery()` or,\n'
|
||||
'- `async with tractor.open_root_actor()`\n'
|
||||
)
|
||||
message: str = (
|
||||
f'{actor.uid} task called `tractor.pause_from_sync()`\n'
|
||||
)
|
||||
|
||||
# TODO: once supported, remove this AND the one
|
||||
# inside `._pause()`!
|
||||
|
|
@ -2447,17 +2409,16 @@ def pause_from_sync(
|
|||
# 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!'
|
||||
# )
|
||||
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_root: bool = is_root_process()
|
||||
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
|
||||
|
|
@ -2470,11 +2431,8 @@ def pause_from_sync(
|
|||
# when called from a (bg) thread, run an async task in a new
|
||||
# thread which will call `._pause()` manually with special
|
||||
# handling for root-actor caller usage.
|
||||
if (
|
||||
not DebugStatus.is_main_trio_thread()
|
||||
and
|
||||
not is_aio # see below for this usage
|
||||
):
|
||||
if not DebugStatus.is_main_trio_thread():
|
||||
|
||||
# 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..
|
||||
|
|
@ -2482,21 +2440,10 @@ def pause_from_sync(
|
|||
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'
|
||||
)
|
||||
elif is_aio:
|
||||
message += (
|
||||
f'-> called from a `asyncio`-task bg {thread}\n'
|
||||
)
|
||||
message += (
|
||||
'-> scheduling `._pause_from_bg_root_thread()`..\n'
|
||||
f'-> scheduling `._pause_from_bg_root_thread()`..\n'
|
||||
)
|
||||
# XXX SUBTLE BADNESS XXX that should really change!
|
||||
# don't over-write the `repl` here since when
|
||||
|
|
@ -2515,8 +2462,7 @@ def pause_from_sync(
|
|||
hide_tb=hide_tb,
|
||||
**_pause_kwargs,
|
||||
),
|
||||
),
|
||||
trio_token=trio.lowlevel.current_trio_token(),
|
||||
)
|
||||
)
|
||||
DebugStatus.shield_sigint()
|
||||
message += (
|
||||
|
|
@ -2549,29 +2495,6 @@ def pause_from_sync(
|
|||
DebugStatus.shield_sigint()
|
||||
assert bg_task is not DebugStatus.repl_task
|
||||
|
||||
elif is_aio:
|
||||
greenback: ModuleType = maybe_import_greenback()
|
||||
repl_owner: Task = asyncio.current_task()
|
||||
fute: asyncio.Future = run_trio_task_in_future(
|
||||
partial(
|
||||
_pause,
|
||||
debug_func=None,
|
||||
repl=repl,
|
||||
hide_tb=hide_tb,
|
||||
|
||||
# XXX to prevent `._pause()` for setting
|
||||
# `DebugStatus.repl_task` to the gb task!
|
||||
called_from_sync=True,
|
||||
called_from_bg_thread=True,
|
||||
|
||||
**_pause_kwargs
|
||||
)
|
||||
)
|
||||
# TODO: for async version -> `.pause_from_aio()`?
|
||||
# bg_task, _ = await fute
|
||||
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
|
||||
greenback: ModuleType = maybe_import_greenback()
|
||||
|
|
@ -2586,8 +2509,8 @@ def pause_from_sync(
|
|||
# NOTE XXX seems to need to be set BEFORE the `_pause()`
|
||||
# invoke using gb below?
|
||||
DebugStatus.shield_sigint()
|
||||
repl_owner: Task = current_task()
|
||||
|
||||
repl_owner: Task = current_task()
|
||||
message += '-> calling `greenback.await_(_pause(debug_func=None))` from sync caller..\n'
|
||||
try:
|
||||
out = greenback.await_(
|
||||
|
|
@ -2649,10 +2572,6 @@ def pause_from_sync(
|
|||
# -[ ] tried to use `@pdbp.hideframe` decoration but
|
||||
# still doesn't work
|
||||
except BaseException as err:
|
||||
log.exception(
|
||||
'Failed to sync-pause from\n\n'
|
||||
f'{repl_owner}\n'
|
||||
)
|
||||
__tracebackhide__: bool = False
|
||||
raise err
|
||||
|
||||
|
|
|
|||
|
|
@ -562,100 +562,6 @@ class AsyncioRuntimeTranslationError(RuntimeError):
|
|||
'''
|
||||
|
||||
|
||||
def run_trio_task_in_future(
|
||||
async_fn,
|
||||
*args,
|
||||
) -> asyncio.Future:
|
||||
'''
|
||||
Run an async-func as a `trio` task from an `asyncio.Task` wrapped
|
||||
in a `asyncio.Future` which is returned to the caller.
|
||||
|
||||
Another astounding feat by the great @oremanj !!
|
||||
|
||||
Bo
|
||||
|
||||
'''
|
||||
result_future = asyncio.Future()
|
||||
cancel_scope = trio.CancelScope()
|
||||
finished: bool = False
|
||||
|
||||
# Monkeypatch the returned future's cancel() method to forward
|
||||
# cancellation to the Trio task
|
||||
cancel_message = None
|
||||
orig_cancel = result_future.cancel
|
||||
|
||||
def wrapped_cancel(msg=None):
|
||||
nonlocal cancel_message
|
||||
if finished:
|
||||
# We're being called back after the task completed
|
||||
if msg is not None:
|
||||
return orig_cancel(msg)
|
||||
elif cancel_message is not None:
|
||||
return orig_cancel(cancel_message)
|
||||
else:
|
||||
return orig_cancel()
|
||||
|
||||
if result_future.done():
|
||||
return False
|
||||
|
||||
# Forward cancellation to the Trio task, don't mark
|
||||
# future as cancelled until it completes
|
||||
cancel_message = msg
|
||||
cancel_scope.cancel()
|
||||
return True
|
||||
|
||||
result_future.cancel = wrapped_cancel
|
||||
# End of monkeypatching
|
||||
|
||||
async def trio_task() -> None:
|
||||
nonlocal finished
|
||||
try:
|
||||
with cancel_scope:
|
||||
try:
|
||||
# TODO: type this with new tech in 3.13
|
||||
result: Any = await async_fn(*args)
|
||||
finally:
|
||||
finished = True
|
||||
|
||||
# Propagate result or cancellation to the Future
|
||||
if cancel_scope.cancelled_caught:
|
||||
result_future.cancel()
|
||||
|
||||
elif not result_future.cancelled():
|
||||
result_future.set_result(result)
|
||||
|
||||
except BaseException as exc:
|
||||
# The result future gets all the non-Cancelled
|
||||
# exceptions. Any Cancelled need to keep propagating
|
||||
# out of this stack frame in order to reach the cancel
|
||||
# scope for which they're intended.
|
||||
cancelled: BaseException|None
|
||||
rest: BaseException|None
|
||||
if isinstance(exc, BaseExceptionGroup):
|
||||
cancelled, rest = exc.split(trio.Cancelled)
|
||||
|
||||
elif isinstance(exc, trio.Cancelled):
|
||||
cancelled, rest = exc, None
|
||||
|
||||
else:
|
||||
cancelled, rest = None, exc
|
||||
|
||||
if not result_future.cancelled():
|
||||
if rest:
|
||||
result_future.set_exception(rest)
|
||||
else:
|
||||
result_future.cancel()
|
||||
|
||||
if cancelled:
|
||||
raise cancelled
|
||||
|
||||
trio.lowlevel.spawn_system_task(
|
||||
trio_task,
|
||||
name=async_fn,
|
||||
)
|
||||
return result_future
|
||||
|
||||
|
||||
def run_as_asyncio_guest(
|
||||
trio_main: Callable,
|
||||
# ^-NOTE-^ when spawned with `infected_aio=True` this func is
|
||||
|
|
|
|||
Loading…
Reference in New Issue