Compare commits
No commits in common. "e89fe03da7ae129acadae23b8d28b4b33b920ddb" and "683476cc96a16e951d7247f48fdcea0e0c967189" have entirely different histories.
e89fe03da7
...
683476cc96
|
|
@ -74,22 +74,16 @@ jobs:
|
|||
# run: mypy tractor/ --ignore-missing-imports --show-traceback
|
||||
|
||||
|
||||
testing:
|
||||
name: '${{ matrix.os }} Python${{ matrix.python-version }} - spawn_backend=${{ matrix.spawn_backend }}'
|
||||
timeout-minutes: 16
|
||||
testing-linux:
|
||||
name: '${{ matrix.os }} Python ${{ matrix.python }} - ${{ matrix.spawn_backend }}'
|
||||
timeout-minutes: 10
|
||||
runs-on: ${{ matrix.os }}
|
||||
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
os: [
|
||||
ubuntu-latest,
|
||||
macos-latest,
|
||||
]
|
||||
python-version: [
|
||||
'3.13',
|
||||
# '3.14',
|
||||
]
|
||||
os: [ubuntu-latest]
|
||||
python-version: ['3.13']
|
||||
spawn_backend: [
|
||||
'trio',
|
||||
# 'mp_spawn',
|
||||
|
|
@ -97,6 +91,7 @@ jobs:
|
|||
]
|
||||
|
||||
steps:
|
||||
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: 'Install uv + py-${{ matrix.python-version }}'
|
||||
|
|
|
|||
|
|
@ -641,15 +641,13 @@ Help us push toward the future of distributed `Python`.
|
|||
- Typed capability-based (dialog) protocols ( see `#196
|
||||
<https://github.com/goodboy/tractor/issues/196>`_ with draft work
|
||||
started in `#311 <https://github.com/goodboy/tractor/pull/311>`_)
|
||||
- **macOS is now officially supported** and tested in CI
|
||||
alongside Linux!
|
||||
- We **recently disabled CI-testing on windows** and need
|
||||
help getting it running again! (see `#327
|
||||
<https://github.com/goodboy/tractor/pull/327>`_). **We do
|
||||
have windows support** (and have for quite a while) but
|
||||
since no active hacker exists in the user-base to help
|
||||
test on that OS, for now we're not actively maintaining
|
||||
testing due to the added hassle and general latency..
|
||||
- We **recently disabled CI-testing on windows** and need help getting
|
||||
it running again! (see `#327
|
||||
<https://github.com/goodboy/tractor/pull/327>`_). **We do have windows
|
||||
support** (and have for quite a while) but since no active hacker
|
||||
exists in the user-base to help test on that OS, for now we're not
|
||||
actively maintaining testing due to the added hassle and general
|
||||
latency..
|
||||
|
||||
|
||||
Feel like saying hi?
|
||||
|
|
|
|||
|
|
@ -18,14 +18,15 @@ async def aio_sleep_forever():
|
|||
|
||||
|
||||
async def bp_then_error(
|
||||
chan: to_asyncio.LinkedTaskChannel,
|
||||
to_trio: trio.MemorySendChannel,
|
||||
from_trio: asyncio.Queue,
|
||||
|
||||
raise_after_bp: bool = True,
|
||||
|
||||
) -> None:
|
||||
|
||||
# sync with `trio`-side (caller) task
|
||||
chan.started_nowait('start')
|
||||
to_trio.send_nowait('start')
|
||||
|
||||
# NOTE: what happens here inside the hook needs some refinement..
|
||||
# => seems like it's still `.debug._set_trace()` but
|
||||
|
|
|
|||
|
|
@ -3,7 +3,6 @@ Verify we can dump a `stackscope` tree on a hang.
|
|||
|
||||
'''
|
||||
import os
|
||||
import platform
|
||||
import signal
|
||||
|
||||
import trio
|
||||
|
|
@ -32,26 +31,13 @@ async def main(
|
|||
from_test: bool = False,
|
||||
) -> None:
|
||||
|
||||
if platform.system() != 'Darwin':
|
||||
tpt = 'uds'
|
||||
else:
|
||||
# XXX, precisely we can't use pytest's tmp-path generation
|
||||
# for tests.. apparently because:
|
||||
#
|
||||
# > The OSError: AF_UNIX path too long in macOS Python occurs
|
||||
# > because the path to the Unix domain socket exceeds the
|
||||
# > operating system's maximum path length limit (around 104
|
||||
#
|
||||
# WHICH IS just, wtf hillarious XD
|
||||
tpt = 'tcp'
|
||||
|
||||
async with (
|
||||
tractor.open_nursery(
|
||||
debug_mode=True,
|
||||
enable_stack_on_sig=True,
|
||||
# maybe_enable_greenback=False,
|
||||
loglevel='devx',
|
||||
enable_transports=[tpt],
|
||||
enable_transports=['uds'],
|
||||
) as an,
|
||||
):
|
||||
ptl: tractor.Portal = await an.start_actor(
|
||||
|
|
|
|||
|
|
@ -1,5 +1,3 @@
|
|||
import platform
|
||||
|
||||
import tractor
|
||||
import trio
|
||||
|
||||
|
|
@ -36,22 +34,9 @@ async def just_bp(
|
|||
|
||||
async def main():
|
||||
|
||||
if platform.system() != 'Darwin':
|
||||
tpt = 'uds'
|
||||
else:
|
||||
# XXX, precisely we can't use pytest's tmp-path generation
|
||||
# for tests.. apparently because:
|
||||
#
|
||||
# > The OSError: AF_UNIX path too long in macOS Python occurs
|
||||
# > because the path to the Unix domain socket exceeds the
|
||||
# > operating system's maximum path length limit (around 104
|
||||
#
|
||||
# WHICH IS just, wtf hillarious XD
|
||||
tpt = 'tcp'
|
||||
|
||||
async with tractor.open_nursery(
|
||||
debug_mode=True,
|
||||
enable_transports=[tpt],
|
||||
enable_transports=['uds'],
|
||||
loglevel='devx',
|
||||
) as n:
|
||||
p = await n.start_actor(
|
||||
|
|
|
|||
|
|
@ -90,7 +90,7 @@ async def main() -> list[int]:
|
|||
# yes, a nursery which spawns `trio`-"actors" B)
|
||||
an: ActorNursery
|
||||
async with tractor.open_nursery(
|
||||
loglevel='error',
|
||||
loglevel='cancel',
|
||||
# debug_mode=True,
|
||||
) as an:
|
||||
|
||||
|
|
@ -118,10 +118,8 @@ async def main() -> list[int]:
|
|||
cancelled: bool = await portal.cancel_actor()
|
||||
assert cancelled
|
||||
|
||||
print(
|
||||
f"STREAM TIME = {time.time() - start}\n"
|
||||
f"STREAM + SPAWN TIME = {time.time() - pre_start}\n"
|
||||
)
|
||||
print(f"STREAM TIME = {time.time() - start}")
|
||||
print(f"STREAM + SPAWN TIME = {time.time() - pre_start}")
|
||||
assert result_stream == list(range(seed))
|
||||
return result_stream
|
||||
|
||||
|
|
|
|||
|
|
@ -11,17 +11,21 @@ import tractor
|
|||
|
||||
|
||||
async def aio_echo_server(
|
||||
chan: tractor.to_asyncio.LinkedTaskChannel,
|
||||
to_trio: trio.MemorySendChannel,
|
||||
from_trio: asyncio.Queue,
|
||||
|
||||
) -> None:
|
||||
|
||||
# a first message must be sent **from** this ``asyncio``
|
||||
# task or the ``trio`` side will never unblock from
|
||||
# ``tractor.to_asyncio.open_channel_from():``
|
||||
chan.started_nowait('start')
|
||||
to_trio.send_nowait('start')
|
||||
|
||||
# XXX: this uses an ``from_trio: asyncio.Queue`` currently but we
|
||||
# should probably offer something better.
|
||||
while True:
|
||||
# echo the msg back
|
||||
chan.send_nowait(await chan.get())
|
||||
to_trio.send_nowait(await from_trio.get())
|
||||
await asyncio.sleep(0)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -24,7 +24,6 @@ keywords = [
|
|||
classifiers = [
|
||||
"Development Status :: 3 - Alpha",
|
||||
"Operating System :: POSIX :: Linux",
|
||||
"Operating System :: MacOS",
|
||||
"Framework :: Trio",
|
||||
"License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)",
|
||||
"Programming Language :: Python :: Implementation :: CPython",
|
||||
|
|
@ -49,7 +48,6 @@ dependencies = [
|
|||
"msgspec>=0.19.0",
|
||||
"cffi>=1.17.1",
|
||||
"bidict>=0.23.1",
|
||||
"platformdirs>=4.4.0",
|
||||
]
|
||||
|
||||
# ------ project ------
|
||||
|
|
|
|||
|
|
@ -11,7 +11,6 @@ import platform
|
|||
import time
|
||||
|
||||
import pytest
|
||||
import tractor
|
||||
from tractor._testing import (
|
||||
examples_dir as examples_dir,
|
||||
tractor_test as tractor_test,
|
||||
|
|
@ -23,7 +22,6 @@ pytest_plugins: list[str] = [
|
|||
'tractor._testing.pytest',
|
||||
]
|
||||
|
||||
_non_linux: bool = platform.system() != 'Linux'
|
||||
|
||||
# Sending signal.SIGINT on subprocess fails on windows. Use CTRL_* alternatives
|
||||
if platform.system() == 'Windows':
|
||||
|
|
@ -46,10 +44,6 @@ no_windows = pytest.mark.skipif(
|
|||
platform.system() == "Windows",
|
||||
reason="Test is unsupported on windows",
|
||||
)
|
||||
no_macos = pytest.mark.skipif(
|
||||
platform.system() == "Darwin",
|
||||
reason="Test is unsupported on MacOS",
|
||||
)
|
||||
|
||||
|
||||
def pytest_addoption(
|
||||
|
|
@ -67,7 +61,7 @@ def pytest_addoption(
|
|||
|
||||
|
||||
@pytest.fixture(scope='session', autouse=True)
|
||||
def loglevel(request) -> str:
|
||||
def loglevel(request):
|
||||
import tractor
|
||||
orig = tractor.log._default_loglevel
|
||||
level = tractor.log._default_loglevel = request.config.option.loglevel
|
||||
|
|
@ -75,46 +69,11 @@ def loglevel(request) -> str:
|
|||
level=level,
|
||||
name='tractor', # <- enable root logger
|
||||
)
|
||||
log.info(
|
||||
f'Test-harness set runtime loglevel: {level!r}\n'
|
||||
)
|
||||
log.info(f'Test-harness logging level: {level}\n')
|
||||
yield level
|
||||
tractor.log._default_loglevel = orig
|
||||
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def test_log(
|
||||
request,
|
||||
loglevel: str,
|
||||
) -> tractor.log.StackLevelAdapter:
|
||||
'''
|
||||
Deliver a per test-module-fn logger instance for reporting from
|
||||
within actual test bodies/fixtures.
|
||||
|
||||
For example this can be handy to report certain error cases from
|
||||
exception handlers using `test_log.exception()`.
|
||||
|
||||
'''
|
||||
modname: str = request.function.__module__
|
||||
log = tractor.log.get_logger(
|
||||
name=modname, # <- enable root logger
|
||||
# pkg_name='tests',
|
||||
)
|
||||
_log = tractor.log.get_console_log(
|
||||
level=loglevel,
|
||||
logger=log,
|
||||
name=modname,
|
||||
# pkg_name='tests',
|
||||
)
|
||||
_log.debug(
|
||||
f'In-test-logging requested\n'
|
||||
f'test_log.name: {log.name!r}\n'
|
||||
f'level: {loglevel!r}\n'
|
||||
|
||||
)
|
||||
yield _log
|
||||
|
||||
|
||||
_ci_env: bool = os.environ.get('CI', False)
|
||||
|
||||
|
||||
|
|
@ -151,7 +110,6 @@ def daemon(
|
|||
testdir: pytest.Pytester,
|
||||
reg_addr: tuple[str, int],
|
||||
tpt_proto: str,
|
||||
ci_env: bool,
|
||||
|
||||
) -> subprocess.Popen:
|
||||
'''
|
||||
|
|
@ -189,25 +147,13 @@ def daemon(
|
|||
**kwargs,
|
||||
)
|
||||
|
||||
# TODO! we should poll for the registry socket-bind to take place
|
||||
# and only once that's done yield to the requester!
|
||||
# -[ ] TCP: use the `._root.open_root_actor()`::`ping_tpt_socket()`
|
||||
# closure!
|
||||
# -[ ] UDS: can we do something similar for 'pinging" the
|
||||
# file-socket?
|
||||
#
|
||||
global _PROC_SPAWN_WAIT
|
||||
# UDS sockets are **really** fast to bind()/listen()/connect()
|
||||
# so it's often required that we delay a bit more starting
|
||||
# the first actor-tree..
|
||||
if tpt_proto == 'uds':
|
||||
global _PROC_SPAWN_WAIT
|
||||
_PROC_SPAWN_WAIT = 0.6
|
||||
|
||||
if _non_linux and ci_env:
|
||||
_PROC_SPAWN_WAIT += 1
|
||||
|
||||
# XXX, allow time for the sub-py-proc to boot up.
|
||||
# !TODO, see ping-polling ideas above!
|
||||
time.sleep(_PROC_SPAWN_WAIT)
|
||||
|
||||
assert not proc.returncode
|
||||
|
|
@ -217,30 +163,18 @@ def daemon(
|
|||
# XXX! yeah.. just be reaaal careful with this bc sometimes it
|
||||
# can lock up on the `_io.BufferedReader` and hang..
|
||||
stderr: str = proc.stderr.read().decode()
|
||||
stdout: str = proc.stdout.read().decode()
|
||||
if (
|
||||
stderr
|
||||
or
|
||||
stdout
|
||||
):
|
||||
if stderr:
|
||||
print(
|
||||
f'Daemon actor tree produced output:\n'
|
||||
f'Daemon actor tree produced STDERR:\n'
|
||||
f'{proc.args}\n'
|
||||
f'\n'
|
||||
f'stderr: {stderr!r}\n'
|
||||
f'stdout: {stdout!r}\n'
|
||||
f'{stderr}\n'
|
||||
)
|
||||
|
||||
if (rc := proc.returncode) != -2:
|
||||
msg: str = (
|
||||
f'Daemon actor tree was not cancelled !?\n'
|
||||
f'proc.args: {proc.args!r}\n'
|
||||
f'proc.returncode: {rc!r}\n'
|
||||
if proc.returncode != -2:
|
||||
raise RuntimeError(
|
||||
'Daemon actor tree failed !?\n'
|
||||
f'{proc.args}\n'
|
||||
)
|
||||
if rc < 0:
|
||||
raise RuntimeError(msg)
|
||||
|
||||
log.error(msg)
|
||||
|
||||
|
||||
# @pytest.fixture(autouse=True)
|
||||
|
|
|
|||
|
|
@ -3,9 +3,8 @@
|
|||
|
||||
'''
|
||||
from __future__ import annotations
|
||||
import platform
|
||||
import signal
|
||||
import time
|
||||
import signal
|
||||
from typing import (
|
||||
Callable,
|
||||
TYPE_CHECKING,
|
||||
|
|
@ -34,17 +33,6 @@ if TYPE_CHECKING:
|
|||
from pexpect import pty_spawn
|
||||
|
||||
|
||||
_non_linux: bool = platform.system() != 'Linux'
|
||||
|
||||
|
||||
def pytest_configure(config):
|
||||
# register custom marks to avoid warnings see,
|
||||
# https://docs.pytest.org/en/stable/how-to/writing_plugins.html#registering-custom-markers
|
||||
config.addinivalue_line(
|
||||
'markers',
|
||||
'ctlcs_bish: test will (likely) not behave under SIGINT..'
|
||||
)
|
||||
|
||||
# a fn that sub-instantiates a `pexpect.spawn()`
|
||||
# and returns it.
|
||||
type PexpectSpawner = Callable[
|
||||
|
|
@ -80,10 +68,7 @@ def spawn(
|
|||
|
||||
'''
|
||||
import os
|
||||
# disable colored tbs
|
||||
os.environ['PYTHON_COLORS'] = '0'
|
||||
# disable all ANSI color output
|
||||
# os.environ['NO_COLOR'] = '1'
|
||||
|
||||
spawned: PexpectSpawner|None = None
|
||||
|
||||
|
|
@ -98,10 +83,7 @@ def spawn(
|
|||
cmd,
|
||||
**mkcmd_kwargs,
|
||||
),
|
||||
expect_timeout=(
|
||||
10 if _non_linux and _ci_env
|
||||
else 3
|
||||
),
|
||||
expect_timeout=3,
|
||||
# preexec_fn=unset_colors,
|
||||
# ^TODO? get `pytest` core to expose underlying
|
||||
# `pexpect.spawn()` stuff?
|
||||
|
|
@ -164,8 +146,6 @@ def ctlc(
|
|||
mark.name == 'ctlcs_bish'
|
||||
and
|
||||
use_ctlc
|
||||
and
|
||||
all(mark.args)
|
||||
):
|
||||
pytest.skip(
|
||||
f'Test {node} prolly uses something from the stdlib (namely `asyncio`..)\n'
|
||||
|
|
@ -271,13 +251,12 @@ def assert_before(
|
|||
err_on_false=True,
|
||||
**kwargs
|
||||
)
|
||||
return str(child.before.decode())
|
||||
|
||||
|
||||
def do_ctlc(
|
||||
child,
|
||||
count: int = 3,
|
||||
delay: float|None = None,
|
||||
delay: float = 0.1,
|
||||
patt: str|None = None,
|
||||
|
||||
# expect repl UX to reprint the prompt after every
|
||||
|
|
@ -289,7 +268,6 @@ def do_ctlc(
|
|||
) -> str|None:
|
||||
|
||||
before: str|None = None
|
||||
delay = delay or 0.1
|
||||
|
||||
# make sure ctl-c sends don't do anything but repeat output
|
||||
for _ in range(count):
|
||||
|
|
@ -300,10 +278,7 @@ def do_ctlc(
|
|||
# if you run this test manually it works just fine..
|
||||
if expect_prompt:
|
||||
time.sleep(delay)
|
||||
child.expect(
|
||||
PROMPT,
|
||||
timeout=(child.timeout * 2) if _ci_env else child.timeout,
|
||||
)
|
||||
child.expect(PROMPT)
|
||||
before = str(child.before.decode())
|
||||
time.sleep(delay)
|
||||
|
||||
|
|
|
|||
|
|
@ -37,9 +37,6 @@ from .conftest import (
|
|||
in_prompt_msg,
|
||||
assert_before,
|
||||
)
|
||||
from ..conftest import (
|
||||
_ci_env,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ..conftest import PexpectSpawner
|
||||
|
|
@ -54,14 +51,13 @@ if TYPE_CHECKING:
|
|||
# - recurrent root errors
|
||||
|
||||
|
||||
_non_linux: bool = platform.system() != 'Linux'
|
||||
|
||||
if platform.system() == 'Windows':
|
||||
pytest.skip(
|
||||
'Debugger tests have no windows support (yet)',
|
||||
allow_module_level=True,
|
||||
)
|
||||
|
||||
|
||||
# TODO: was trying to this xfail style but some weird bug i see in CI
|
||||
# that's happening at collect time.. pretty soon gonna dump actions i'm
|
||||
# thinkin...
|
||||
|
|
@ -197,11 +193,6 @@ def test_root_actor_bp_forever(
|
|||
child.expect(EOF)
|
||||
|
||||
|
||||
# skip on non-Linux CI
|
||||
@pytest.mark.ctlcs_bish(
|
||||
_non_linux,
|
||||
_ci_env,
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
'do_next',
|
||||
(True, False),
|
||||
|
|
@ -267,11 +258,6 @@ def test_subactor_error(
|
|||
child.expect(EOF)
|
||||
|
||||
|
||||
# skip on non-Linux CI
|
||||
@pytest.mark.ctlcs_bish(
|
||||
_non_linux,
|
||||
_ci_env,
|
||||
)
|
||||
def test_subactor_breakpoint(
|
||||
spawn,
|
||||
ctlc: bool,
|
||||
|
|
@ -494,24 +480,8 @@ def test_multi_daemon_subactors(
|
|||
stream.
|
||||
|
||||
'''
|
||||
non_linux = _non_linux
|
||||
if non_linux and ctlc:
|
||||
pytest.skip(
|
||||
'Ctl-c + MacOS is too unreliable/racy for this test..\n'
|
||||
)
|
||||
# !TODO, if someone with more patience then i wants to muck
|
||||
# with the timings on this please feel free to see all the
|
||||
# `non_linux` branching logic i added on my first attempt
|
||||
# below!
|
||||
#
|
||||
# my conclusion was that if i were to run the script
|
||||
# manually, and thus as slowly as a human would, the test
|
||||
# would and should pass as described in this test fn, however
|
||||
# after fighting with it for >= 1hr. i decided more then
|
||||
# likely the more extensive `linux` testing should cover most
|
||||
# regressions.
|
||||
|
||||
child = spawn('multi_daemon_subactors')
|
||||
|
||||
child.expect(PROMPT)
|
||||
|
||||
# there can be a race for which subactor will acquire
|
||||
|
|
@ -541,19 +511,8 @@ def test_multi_daemon_subactors(
|
|||
else:
|
||||
raise ValueError('Neither log msg was found !?')
|
||||
|
||||
non_linux_delay: float = 0.3
|
||||
if ctlc:
|
||||
do_ctlc(
|
||||
child,
|
||||
delay=(
|
||||
non_linux_delay
|
||||
if non_linux
|
||||
else None
|
||||
),
|
||||
)
|
||||
|
||||
if non_linux:
|
||||
time.sleep(1)
|
||||
do_ctlc(child)
|
||||
|
||||
# NOTE: previously since we did not have clobber prevention
|
||||
# in the root actor this final resume could result in the debugger
|
||||
|
|
@ -584,66 +543,33 @@ def test_multi_daemon_subactors(
|
|||
# assert "in use by child ('bp_forever'," in before
|
||||
|
||||
if ctlc:
|
||||
do_ctlc(
|
||||
child,
|
||||
delay=(
|
||||
non_linux_delay
|
||||
if non_linux
|
||||
else None
|
||||
),
|
||||
)
|
||||
|
||||
if non_linux:
|
||||
time.sleep(1)
|
||||
do_ctlc(child)
|
||||
|
||||
# expect another breakpoint actor entry
|
||||
child.sendline('c')
|
||||
child.expect(PROMPT)
|
||||
|
||||
try:
|
||||
before: str = assert_before(
|
||||
assert_before(
|
||||
child,
|
||||
bp_forev_parts,
|
||||
)
|
||||
except AssertionError:
|
||||
before: str = assert_before(
|
||||
assert_before(
|
||||
child,
|
||||
name_error_parts,
|
||||
)
|
||||
|
||||
else:
|
||||
if ctlc:
|
||||
before: str = do_ctlc(
|
||||
child,
|
||||
delay=(
|
||||
non_linux_delay
|
||||
if non_linux
|
||||
else None
|
||||
),
|
||||
)
|
||||
|
||||
if non_linux:
|
||||
time.sleep(1)
|
||||
do_ctlc(child)
|
||||
|
||||
# should crash with the 2nd name error (simulates
|
||||
# a retry) and then the root eventually (boxed) errors
|
||||
# after 1 or more further bp actor entries.
|
||||
|
||||
child.sendline('c')
|
||||
try:
|
||||
child.expect(
|
||||
PROMPT,
|
||||
timeout=3,
|
||||
)
|
||||
except EOF:
|
||||
before: str = child.before.decode()
|
||||
print(
|
||||
f'\n'
|
||||
f'??? NEVER RXED `pdb` PROMPT ???\n'
|
||||
f'\n'
|
||||
f'{before}\n'
|
||||
)
|
||||
raise
|
||||
|
||||
child.expect(PROMPT)
|
||||
assert_before(
|
||||
child,
|
||||
name_error_parts,
|
||||
|
|
@ -763,8 +689,7 @@ def test_multi_subactors_root_errors(
|
|||
|
||||
@has_nested_actors
|
||||
def test_multi_nested_subactors_error_through_nurseries(
|
||||
ci_env: bool,
|
||||
spawn: PexpectSpawner,
|
||||
spawn,
|
||||
|
||||
# TODO: address debugger issue for nested tree:
|
||||
# https://github.com/goodboy/tractor/issues/320
|
||||
|
|
@ -787,16 +712,7 @@ def test_multi_nested_subactors_error_through_nurseries(
|
|||
|
||||
for send_char in itertools.cycle(['c', 'q']):
|
||||
try:
|
||||
child.expect(
|
||||
PROMPT,
|
||||
timeout=(
|
||||
6 if (
|
||||
_non_linux
|
||||
and
|
||||
ci_env
|
||||
) else -1
|
||||
),
|
||||
)
|
||||
child.expect(PROMPT)
|
||||
child.sendline(send_char)
|
||||
time.sleep(0.01)
|
||||
|
||||
|
|
@ -973,11 +889,6 @@ def test_different_debug_mode_per_actor(
|
|||
)
|
||||
|
||||
|
||||
# skip on non-Linux CI
|
||||
@pytest.mark.ctlcs_bish(
|
||||
_non_linux,
|
||||
_ci_env,
|
||||
)
|
||||
def test_post_mortem_api(
|
||||
spawn,
|
||||
ctlc: bool,
|
||||
|
|
@ -1222,20 +1133,14 @@ def test_ctxep_pauses_n_maybe_ipc_breaks(
|
|||
# closed so verify we see error reporting as well as
|
||||
# a failed crash-REPL request msg and can CTL-c our way
|
||||
# out.
|
||||
|
||||
# ?TODO, match depending on `tpt_proto(s)`?
|
||||
# - [ ] how can we pass it into the script tho?
|
||||
tpt: str = 'UDS'
|
||||
if _non_linux:
|
||||
tpt: str = 'TCP'
|
||||
|
||||
assert_before(
|
||||
child,
|
||||
['peer IPC channel closed abruptly?',
|
||||
'another task closed this fd',
|
||||
'Debug lock request was CANCELLED?',
|
||||
f"'Msgpack{tpt}Stream' was already closed locally?",
|
||||
f"TransportClosed: 'Msgpack{tpt}Stream' was already closed 'by peer'?",
|
||||
"'MsgpackUDSStream' was already closed locally?",
|
||||
"TransportClosed: 'MsgpackUDSStream' was already closed 'by peer'?",
|
||||
# ?TODO^? match depending on `tpt_proto(s)`?
|
||||
]
|
||||
|
||||
# XXX races on whether these show/hit?
|
||||
|
|
|
|||
|
|
@ -31,9 +31,6 @@ from .conftest import (
|
|||
PROMPT,
|
||||
_pause_msg,
|
||||
)
|
||||
from ..conftest import (
|
||||
no_macos,
|
||||
)
|
||||
|
||||
import pytest
|
||||
from pexpect.exceptions import (
|
||||
|
|
@ -45,7 +42,6 @@ if TYPE_CHECKING:
|
|||
from ..conftest import PexpectSpawner
|
||||
|
||||
|
||||
@no_macos
|
||||
def test_shield_pause(
|
||||
spawn: PexpectSpawner,
|
||||
):
|
||||
|
|
@ -61,7 +57,6 @@ def test_shield_pause(
|
|||
expect(
|
||||
child,
|
||||
'Yo my child hanging..?',
|
||||
timeout=3,
|
||||
)
|
||||
assert_before(
|
||||
child,
|
||||
|
|
|
|||
|
|
@ -1,4 +0,0 @@
|
|||
'''
|
||||
`tractor.msg.*` sub-sys test suite.
|
||||
|
||||
'''
|
||||
|
|
@ -1,4 +0,0 @@
|
|||
'''
|
||||
`tractor.msg.*` test sub-pkg conf.
|
||||
|
||||
'''
|
||||
|
|
@ -1,240 +0,0 @@
|
|||
'''
|
||||
Unit tests for `tractor.msg.pretty_struct`
|
||||
private-field filtering in `pformat()`.
|
||||
|
||||
'''
|
||||
import pytest
|
||||
|
||||
from tractor.msg.pretty_struct import (
|
||||
Struct,
|
||||
pformat,
|
||||
iter_struct_ppfmt_lines,
|
||||
)
|
||||
from tractor.msg._codec import (
|
||||
MsgDec,
|
||||
mk_dec,
|
||||
)
|
||||
|
||||
|
||||
# ------ test struct definitions ------ #
|
||||
|
||||
class PublicOnly(Struct):
|
||||
'''
|
||||
All-public fields for baseline testing.
|
||||
|
||||
'''
|
||||
name: str = 'alice'
|
||||
age: int = 30
|
||||
|
||||
|
||||
class PrivateOnly(Struct):
|
||||
'''
|
||||
Only underscore-prefixed (private) fields.
|
||||
|
||||
'''
|
||||
_secret: str = 'hidden'
|
||||
_internal: int = 99
|
||||
|
||||
|
||||
class MixedFields(Struct):
|
||||
'''
|
||||
Mix of public and private fields.
|
||||
|
||||
'''
|
||||
name: str = 'bob'
|
||||
_hidden: int = 42
|
||||
value: float = 3.14
|
||||
_meta: str = 'internal'
|
||||
|
||||
|
||||
class Inner(
|
||||
Struct,
|
||||
frozen=True,
|
||||
):
|
||||
'''
|
||||
Frozen inner struct with a private field,
|
||||
for nesting tests.
|
||||
|
||||
'''
|
||||
x: int = 1
|
||||
_secret: str = 'nope'
|
||||
|
||||
|
||||
class Outer(Struct):
|
||||
'''
|
||||
Outer struct nesting an `Inner`.
|
||||
|
||||
'''
|
||||
label: str = 'outer'
|
||||
inner: Inner = Inner()
|
||||
|
||||
|
||||
class EmptyStruct(Struct):
|
||||
'''
|
||||
Struct with zero fields.
|
||||
|
||||
'''
|
||||
pass
|
||||
|
||||
|
||||
# ------ tests ------ #
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'struct_and_expected',
|
||||
[
|
||||
(
|
||||
PublicOnly(),
|
||||
{
|
||||
'shown': ['name', 'age'],
|
||||
'hidden': [],
|
||||
},
|
||||
),
|
||||
(
|
||||
MixedFields(),
|
||||
{
|
||||
'shown': ['name', 'value'],
|
||||
'hidden': ['_hidden', '_meta'],
|
||||
},
|
||||
),
|
||||
(
|
||||
PrivateOnly(),
|
||||
{
|
||||
'shown': [],
|
||||
'hidden': ['_secret', '_internal'],
|
||||
},
|
||||
),
|
||||
],
|
||||
ids=[
|
||||
'all-public',
|
||||
'mixed-pub-priv',
|
||||
'all-private',
|
||||
],
|
||||
)
|
||||
def test_field_visibility_in_pformat(
|
||||
struct_and_expected: tuple[
|
||||
Struct,
|
||||
dict[str, list[str]],
|
||||
],
|
||||
):
|
||||
'''
|
||||
Verify `pformat()` shows public fields
|
||||
and hides `_`-prefixed private fields.
|
||||
|
||||
'''
|
||||
(
|
||||
struct,
|
||||
expected,
|
||||
) = struct_and_expected
|
||||
output: str = pformat(struct)
|
||||
|
||||
for field_name in expected['shown']:
|
||||
assert field_name in output, (
|
||||
f'{field_name!r} should appear in:\n'
|
||||
f'{output}'
|
||||
)
|
||||
|
||||
for field_name in expected['hidden']:
|
||||
assert field_name not in output, (
|
||||
f'{field_name!r} should NOT appear in:\n'
|
||||
f'{output}'
|
||||
)
|
||||
|
||||
|
||||
def test_iter_ppfmt_lines_skips_private():
|
||||
'''
|
||||
Directly verify `iter_struct_ppfmt_lines()`
|
||||
never yields tuples with `_`-prefixed field
|
||||
names.
|
||||
|
||||
'''
|
||||
struct = MixedFields()
|
||||
lines: list[tuple[str, str]] = list(
|
||||
iter_struct_ppfmt_lines(
|
||||
struct,
|
||||
field_indent=2,
|
||||
)
|
||||
)
|
||||
# should have lines for public fields only
|
||||
assert len(lines) == 2
|
||||
|
||||
for _prefix, line_content in lines:
|
||||
field_name: str = (
|
||||
line_content.split(':')[0].strip()
|
||||
)
|
||||
assert not field_name.startswith('_'), (
|
||||
f'private field leaked: {field_name!r}'
|
||||
)
|
||||
|
||||
|
||||
def test_nested_struct_filters_inner_private():
|
||||
'''
|
||||
Verify that nested struct's private fields
|
||||
are also filtered out during recursion.
|
||||
|
||||
'''
|
||||
outer = Outer()
|
||||
output: str = pformat(outer)
|
||||
|
||||
# outer's public field
|
||||
assert 'label' in output
|
||||
|
||||
# inner's public field (recursed into)
|
||||
assert 'x' in output
|
||||
|
||||
# inner's private field must be hidden
|
||||
assert '_secret' not in output
|
||||
|
||||
|
||||
def test_empty_struct_pformat():
|
||||
'''
|
||||
An empty struct should produce a valid
|
||||
`pformat()` result with no field lines.
|
||||
|
||||
'''
|
||||
output: str = pformat(EmptyStruct())
|
||||
assert 'EmptyStruct(' in output
|
||||
assert output.rstrip().endswith(')')
|
||||
|
||||
# no field lines => only struct header+footer
|
||||
lines: list[tuple[str, str]] = list(
|
||||
iter_struct_ppfmt_lines(
|
||||
EmptyStruct(),
|
||||
field_indent=2,
|
||||
)
|
||||
)
|
||||
assert lines == []
|
||||
|
||||
|
||||
def test_real_msgdec_pformat_hides_private():
|
||||
'''
|
||||
Verify `pformat()` on a real `MsgDec`
|
||||
hides the `_dec` internal field.
|
||||
|
||||
NOTE: `MsgDec.__repr__` is custom and does
|
||||
NOT call `pformat()`, so we call it directly.
|
||||
|
||||
'''
|
||||
dec: MsgDec = mk_dec(spec=int)
|
||||
output: str = pformat(dec)
|
||||
|
||||
# the private `_dec` field should be filtered
|
||||
assert '_dec' not in output
|
||||
|
||||
# but the struct type name should be present
|
||||
assert 'MsgDec(' in output
|
||||
|
||||
|
||||
def test_pformat_repr_integration():
|
||||
'''
|
||||
Verify that `Struct.__repr__()` (which calls
|
||||
`pformat()`) also hides private fields for
|
||||
custom structs that do NOT override `__repr__`.
|
||||
|
||||
'''
|
||||
mixed = MixedFields()
|
||||
output: str = repr(mixed)
|
||||
|
||||
assert 'name' in output
|
||||
assert 'value' in output
|
||||
assert '_hidden' not in output
|
||||
assert '_meta' not in output
|
||||
|
|
@ -1,12 +1,7 @@
|
|||
'''
|
||||
Audit the simplest inter-actor bidirectional (streaming)
|
||||
msg patterns.
|
||||
"""
|
||||
Bidirectional streaming.
|
||||
|
||||
'''
|
||||
from __future__ import annotations
|
||||
from typing import (
|
||||
Callable,
|
||||
)
|
||||
"""
|
||||
import pytest
|
||||
import trio
|
||||
import tractor
|
||||
|
|
@ -14,8 +9,10 @@ import tractor
|
|||
|
||||
@tractor.context
|
||||
async def simple_rpc(
|
||||
|
||||
ctx: tractor.Context,
|
||||
data: int,
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
Test a small ping-pong server.
|
||||
|
|
@ -42,13 +39,15 @@ async def simple_rpc(
|
|||
|
||||
@tractor.context
|
||||
async def simple_rpc_with_forloop(
|
||||
|
||||
ctx: tractor.Context,
|
||||
data: int,
|
||||
) -> None:
|
||||
'''
|
||||
Same as previous test but using `async for` syntax/api.
|
||||
|
||||
'''
|
||||
) -> None:
|
||||
"""Same as previous test but using ``async for`` syntax/api.
|
||||
|
||||
"""
|
||||
|
||||
# signal to parent that we're up
|
||||
await ctx.started(data + 1)
|
||||
|
||||
|
|
@ -69,37 +68,21 @@ async def simple_rpc_with_forloop(
|
|||
|
||||
@pytest.mark.parametrize(
|
||||
'use_async_for',
|
||||
[
|
||||
True,
|
||||
False,
|
||||
],
|
||||
ids='use_async_for={}'.format,
|
||||
[True, False],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
'server_func',
|
||||
[
|
||||
simple_rpc,
|
||||
simple_rpc_with_forloop,
|
||||
],
|
||||
ids='server_func={}'.format,
|
||||
[simple_rpc, simple_rpc_with_forloop],
|
||||
)
|
||||
def test_simple_rpc(
|
||||
server_func: Callable,
|
||||
use_async_for: bool,
|
||||
loglevel: str,
|
||||
debug_mode: bool,
|
||||
):
|
||||
def test_simple_rpc(server_func, use_async_for):
|
||||
'''
|
||||
The simplest request response pattern.
|
||||
|
||||
'''
|
||||
async def main():
|
||||
with trio.fail_after(6):
|
||||
async with tractor.open_nursery(
|
||||
loglevel=loglevel,
|
||||
debug_mode=debug_mode,
|
||||
) as an:
|
||||
portal: tractor.Portal = await an.start_actor(
|
||||
async with tractor.open_nursery() as n:
|
||||
|
||||
portal = await n.start_actor(
|
||||
'rpc_server',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
|
|
|
|||
|
|
@ -17,8 +17,8 @@ from tractor._testing import (
|
|||
from .conftest import no_windows
|
||||
|
||||
|
||||
_non_linux: bool = platform.system() != 'Linux'
|
||||
_friggin_windows: bool = platform.system() == 'Windows'
|
||||
def is_win():
|
||||
return platform.system() == 'Windows'
|
||||
|
||||
|
||||
async def assert_err(delay=0):
|
||||
|
|
@ -431,7 +431,7 @@ async def test_nested_multierrors(loglevel, start_method):
|
|||
for subexc in err.exceptions:
|
||||
|
||||
# verify first level actor errors are wrapped as remote
|
||||
if _friggin_windows:
|
||||
if is_win():
|
||||
|
||||
# windows is often too slow and cancellation seems
|
||||
# to happen before an actor is spawned
|
||||
|
|
@ -464,7 +464,7 @@ async def test_nested_multierrors(loglevel, start_method):
|
|||
# XXX not sure what's up with this..
|
||||
# on windows sometimes spawning is just too slow and
|
||||
# we get back the (sent) cancel signal instead
|
||||
if _friggin_windows:
|
||||
if is_win():
|
||||
if isinstance(subexc, tractor.RemoteActorError):
|
||||
assert subexc.boxed_type in (
|
||||
BaseExceptionGroup,
|
||||
|
|
@ -507,22 +507,17 @@ def test_cancel_via_SIGINT(
|
|||
|
||||
@no_windows
|
||||
def test_cancel_via_SIGINT_other_task(
|
||||
loglevel: str,
|
||||
start_method: str,
|
||||
spawn_backend: str,
|
||||
loglevel,
|
||||
start_method,
|
||||
spawn_backend,
|
||||
):
|
||||
'''
|
||||
Ensure that a control-C (SIGINT) signal cancels both the parent
|
||||
and child processes in trionic fashion even a subprocess is
|
||||
started from a seperate ``trio`` child task.
|
||||
|
||||
'''
|
||||
pid: int = os.getpid()
|
||||
timeout: float = (
|
||||
4 if _non_linux
|
||||
else 2
|
||||
)
|
||||
if _friggin_windows: # smh
|
||||
"""Ensure that a control-C (SIGINT) signal cancels both the parent
|
||||
and child processes in trionic fashion even a subprocess is started
|
||||
from a seperate ``trio`` child task.
|
||||
"""
|
||||
pid = os.getpid()
|
||||
timeout: float = 2
|
||||
if is_win(): # smh
|
||||
timeout += 1
|
||||
|
||||
async def spawn_and_sleep_forever(
|
||||
|
|
@ -701,7 +696,7 @@ def test_fast_graceful_cancel_when_spawn_task_in_soft_proc_wait_for_daemon(
|
|||
kbi_delay = 0.5
|
||||
timeout: float = 2.9
|
||||
|
||||
if _friggin_windows: # smh
|
||||
if is_win(): # smh
|
||||
timeout += 1
|
||||
|
||||
async def main():
|
||||
|
|
|
|||
|
|
@ -18,15 +18,16 @@ from tractor import RemoteActorError
|
|||
|
||||
|
||||
async def aio_streamer(
|
||||
chan: tractor.to_asyncio.LinkedTaskChannel,
|
||||
from_trio: asyncio.Queue,
|
||||
to_trio: trio.abc.SendChannel,
|
||||
) -> trio.abc.ReceiveChannel:
|
||||
|
||||
# required first msg to sync caller
|
||||
chan.started_nowait(None)
|
||||
to_trio.send_nowait(None)
|
||||
|
||||
from itertools import cycle
|
||||
for i in cycle(range(10)):
|
||||
chan.send_nowait(i)
|
||||
to_trio.send_nowait(i)
|
||||
await asyncio.sleep(0.01)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -9,7 +9,6 @@ from itertools import count
|
|||
import math
|
||||
import platform
|
||||
from pprint import pformat
|
||||
import sys
|
||||
from typing import (
|
||||
Callable,
|
||||
)
|
||||
|
|
@ -942,11 +941,6 @@ def test_one_end_stream_not_opened(
|
|||
from tractor._runtime import Actor
|
||||
buf_size = buf_size_increase + Actor.msg_buffer_size
|
||||
|
||||
timeout: float = (
|
||||
1 if sys.platform == 'linux'
|
||||
else 3
|
||||
)
|
||||
|
||||
async def main():
|
||||
async with tractor.open_nursery(
|
||||
debug_mode=debug_mode,
|
||||
|
|
@ -956,7 +950,7 @@ def test_one_end_stream_not_opened(
|
|||
enable_modules=[__name__],
|
||||
)
|
||||
|
||||
with trio.fail_after(timeout):
|
||||
with trio.fail_after(1):
|
||||
async with portal.open_context(
|
||||
entrypoint,
|
||||
) as (ctx, sent):
|
||||
|
|
|
|||
|
|
@ -1,13 +1,11 @@
|
|||
"""
|
||||
Discovery subsys.
|
||||
|
||||
Actor "discovery" testing
|
||||
"""
|
||||
import os
|
||||
import signal
|
||||
import platform
|
||||
from functools import partial
|
||||
import itertools
|
||||
from typing import Callable
|
||||
|
||||
import psutil
|
||||
import pytest
|
||||
|
|
@ -19,9 +17,7 @@ import trio
|
|||
|
||||
|
||||
@tractor_test
|
||||
async def test_reg_then_unreg(
|
||||
reg_addr: tuple,
|
||||
):
|
||||
async def test_reg_then_unreg(reg_addr):
|
||||
actor = tractor.current_actor()
|
||||
assert actor.is_arbiter
|
||||
assert len(actor._registry) == 1 # only self is registered
|
||||
|
|
@ -86,15 +82,11 @@ async def say_hello_use_wait(
|
|||
|
||||
|
||||
@tractor_test
|
||||
@pytest.mark.parametrize(
|
||||
'func',
|
||||
[say_hello,
|
||||
say_hello_use_wait]
|
||||
)
|
||||
@pytest.mark.parametrize('func', [say_hello, say_hello_use_wait])
|
||||
async def test_trynamic_trio(
|
||||
func: Callable,
|
||||
start_method: str,
|
||||
reg_addr: tuple,
|
||||
func,
|
||||
start_method,
|
||||
reg_addr,
|
||||
):
|
||||
'''
|
||||
Root actor acting as the "director" and running one-shot-task-actors
|
||||
|
|
@ -127,10 +119,7 @@ async def stream_forever():
|
|||
await trio.sleep(0.01)
|
||||
|
||||
|
||||
async def cancel(
|
||||
use_signal: bool,
|
||||
delay: float = 0,
|
||||
):
|
||||
async def cancel(use_signal, delay=0):
|
||||
# hold on there sally
|
||||
await trio.sleep(delay)
|
||||
|
||||
|
|
@ -143,15 +132,13 @@ async def cancel(
|
|||
raise KeyboardInterrupt
|
||||
|
||||
|
||||
async def stream_from(portal: tractor.Portal):
|
||||
async def stream_from(portal):
|
||||
async with portal.open_stream_from(stream_forever) as stream:
|
||||
async for value in stream:
|
||||
print(value)
|
||||
|
||||
|
||||
async def unpack_reg(
|
||||
actor_or_portal: tractor.Portal|tractor.Actor,
|
||||
):
|
||||
async def unpack_reg(actor_or_portal):
|
||||
'''
|
||||
Get and unpack a "registry" RPC request from the "arbiter" registry
|
||||
system.
|
||||
|
|
@ -186,9 +173,7 @@ async def spawn_and_check_registry(
|
|||
registry_addrs=[reg_addr],
|
||||
debug_mode=debug_mode,
|
||||
):
|
||||
async with tractor.get_registry(
|
||||
addr=reg_addr,
|
||||
) as portal:
|
||||
async with tractor.get_registry(reg_addr) as portal:
|
||||
# runtime needs to be up to call this
|
||||
actor = tractor.current_actor()
|
||||
|
||||
|
|
@ -261,10 +246,10 @@ async def spawn_and_check_registry(
|
|||
@pytest.mark.parametrize('with_streaming', [False, True])
|
||||
def test_subactors_unregister_on_cancel(
|
||||
debug_mode: bool,
|
||||
start_method: str,
|
||||
use_signal: bool,
|
||||
reg_addr: tuple,
|
||||
with_streaming: bool,
|
||||
start_method,
|
||||
use_signal,
|
||||
reg_addr,
|
||||
with_streaming,
|
||||
):
|
||||
'''
|
||||
Verify that cancelling a nursery results in all subactors
|
||||
|
|
@ -289,17 +274,15 @@ def test_subactors_unregister_on_cancel(
|
|||
def test_subactors_unregister_on_cancel_remote_daemon(
|
||||
daemon: subprocess.Popen,
|
||||
debug_mode: bool,
|
||||
start_method: str,
|
||||
use_signal: bool,
|
||||
reg_addr: tuple,
|
||||
with_streaming: bool,
|
||||
start_method,
|
||||
use_signal,
|
||||
reg_addr,
|
||||
with_streaming,
|
||||
):
|
||||
'''
|
||||
Verify that cancelling a nursery results in all subactors
|
||||
deregistering themselves with a **remote** (not in the local
|
||||
process tree) arbiter.
|
||||
|
||||
'''
|
||||
"""Verify that cancelling a nursery results in all subactors
|
||||
deregistering themselves with a **remote** (not in the local process
|
||||
tree) arbiter.
|
||||
"""
|
||||
with pytest.raises(KeyboardInterrupt):
|
||||
trio.run(
|
||||
partial(
|
||||
|
|
@ -391,16 +374,14 @@ async def close_chans_before_nursery(
|
|||
|
||||
@pytest.mark.parametrize('use_signal', [False, True])
|
||||
def test_close_channel_explicit(
|
||||
start_method: str,
|
||||
use_signal: bool,
|
||||
reg_addr: tuple,
|
||||
start_method,
|
||||
use_signal,
|
||||
reg_addr,
|
||||
):
|
||||
'''
|
||||
Verify that closing a stream explicitly and killing the actor's
|
||||
"""Verify that closing a stream explicitly and killing the actor's
|
||||
"root nursery" **before** the containing nursery tears down also
|
||||
results in subactor(s) deregistering from the arbiter.
|
||||
|
||||
'''
|
||||
"""
|
||||
with pytest.raises(KeyboardInterrupt):
|
||||
trio.run(
|
||||
partial(
|
||||
|
|
@ -415,16 +396,14 @@ def test_close_channel_explicit(
|
|||
@pytest.mark.parametrize('use_signal', [False, True])
|
||||
def test_close_channel_explicit_remote_arbiter(
|
||||
daemon: subprocess.Popen,
|
||||
start_method: str,
|
||||
use_signal: bool,
|
||||
reg_addr: tuple,
|
||||
start_method,
|
||||
use_signal,
|
||||
reg_addr,
|
||||
):
|
||||
'''
|
||||
Verify that closing a stream explicitly and killing the actor's
|
||||
"""Verify that closing a stream explicitly and killing the actor's
|
||||
"root nursery" **before** the containing nursery tears down also
|
||||
results in subactor(s) deregistering from the arbiter.
|
||||
|
||||
'''
|
||||
"""
|
||||
with pytest.raises(KeyboardInterrupt):
|
||||
trio.run(
|
||||
partial(
|
||||
|
|
|
|||
|
|
@ -9,17 +9,12 @@ import sys
|
|||
import subprocess
|
||||
import platform
|
||||
import shutil
|
||||
from typing import Callable
|
||||
|
||||
import pytest
|
||||
import tractor
|
||||
from tractor._testing import (
|
||||
examples_dir,
|
||||
)
|
||||
|
||||
_non_linux: bool = platform.system() != 'Linux'
|
||||
_friggin_macos: bool = platform.system() == 'Darwin'
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def run_example_in_subproc(
|
||||
|
|
@ -106,10 +101,8 @@ def run_example_in_subproc(
|
|||
ids=lambda t: t[1],
|
||||
)
|
||||
def test_example(
|
||||
run_example_in_subproc: Callable,
|
||||
example_script: str,
|
||||
test_log: tractor.log.StackLevelAdapter,
|
||||
ci_env: bool,
|
||||
run_example_in_subproc,
|
||||
example_script,
|
||||
):
|
||||
'''
|
||||
Load and run scripts from this repo's ``examples/`` dir as a user
|
||||
|
|
@ -123,32 +116,9 @@ def test_example(
|
|||
'''
|
||||
ex_file: str = os.path.join(*example_script)
|
||||
|
||||
if (
|
||||
'rpc_bidir_streaming' in ex_file
|
||||
and
|
||||
sys.version_info < (3, 9)
|
||||
):
|
||||
if 'rpc_bidir_streaming' in ex_file and sys.version_info < (3, 9):
|
||||
pytest.skip("2-way streaming example requires py3.9 async with syntax")
|
||||
|
||||
if (
|
||||
'full_fledged_streaming_service' in ex_file
|
||||
and
|
||||
_friggin_macos
|
||||
and
|
||||
ci_env
|
||||
):
|
||||
pytest.skip(
|
||||
'Streaming example is too flaky in CI\n'
|
||||
'AND their competitor runs this CI service..\n'
|
||||
'This test does run just fine "in person" however..'
|
||||
)
|
||||
|
||||
timeout: float = (
|
||||
60
|
||||
if ci_env and _non_linux
|
||||
else 16
|
||||
)
|
||||
|
||||
with open(ex_file, 'r') as ex:
|
||||
code = ex.read()
|
||||
|
||||
|
|
@ -156,12 +126,9 @@ def test_example(
|
|||
err = None
|
||||
try:
|
||||
if not proc.poll():
|
||||
_, err = proc.communicate(timeout=timeout)
|
||||
_, err = proc.communicate(timeout=15)
|
||||
|
||||
except subprocess.TimeoutExpired as e:
|
||||
test_log.exception(
|
||||
f'Example failed to finish within {timeout}s ??\n'
|
||||
)
|
||||
proc.kill()
|
||||
err = e.stderr
|
||||
|
||||
|
|
|
|||
|
|
@ -47,11 +47,12 @@ async def sleep_and_err(
|
|||
|
||||
# just signature placeholders for compat with
|
||||
# ``to_asyncio.open_channel_from()``
|
||||
chan: to_asyncio.LinkedTaskChannel|None = None,
|
||||
to_trio: trio.MemorySendChannel|None = None,
|
||||
from_trio: asyncio.Queue|None = None,
|
||||
|
||||
):
|
||||
if chan:
|
||||
chan.started_nowait('start')
|
||||
if to_trio:
|
||||
to_trio.send_nowait('start')
|
||||
|
||||
await asyncio.sleep(sleep_for)
|
||||
assert 0
|
||||
|
|
@ -398,7 +399,7 @@ async def no_to_trio_in_args():
|
|||
|
||||
async def push_from_aio_task(
|
||||
sequence: Iterable,
|
||||
chan: to_asyncio.LinkedTaskChannel,
|
||||
to_trio: trio.abc.SendChannel,
|
||||
expect_cancel: False,
|
||||
fail_early: bool,
|
||||
exit_early: bool,
|
||||
|
|
@ -406,12 +407,15 @@ async def push_from_aio_task(
|
|||
) -> None:
|
||||
|
||||
try:
|
||||
# print('trying breakpoint')
|
||||
# breakpoint()
|
||||
|
||||
# sync caller ctx manager
|
||||
chan.started_nowait(True)
|
||||
to_trio.send_nowait(True)
|
||||
|
||||
for i in sequence:
|
||||
print(f'asyncio sending {i}')
|
||||
chan.send_nowait(i)
|
||||
to_trio.send_nowait(i)
|
||||
await asyncio.sleep(0.001)
|
||||
|
||||
if (
|
||||
|
|
@ -728,21 +732,15 @@ def test_aio_errors_and_channel_propagates_and_closes(
|
|||
|
||||
|
||||
async def aio_echo_server(
|
||||
chan: to_asyncio.LinkedTaskChannel,
|
||||
to_trio: trio.MemorySendChannel,
|
||||
from_trio: asyncio.Queue,
|
||||
) -> None:
|
||||
'''
|
||||
An IPC-msg "echo server" with msgs received and relayed by
|
||||
a parent `trio.Task` into a child `asyncio.Task`
|
||||
and then repeated back to that local parent (`trio.Task`)
|
||||
and sent again back to the original calling remote actor.
|
||||
|
||||
'''
|
||||
# same semantics as `trio.TaskStatus.started()`
|
||||
chan.started_nowait('start')
|
||||
to_trio.send_nowait('start')
|
||||
|
||||
while True:
|
||||
try:
|
||||
msg = await chan.get()
|
||||
msg = await from_trio.get()
|
||||
except to_asyncio.TrioTaskExited:
|
||||
print(
|
||||
'breaking aio echo loop due to `trio` exit!'
|
||||
|
|
@ -750,7 +748,7 @@ async def aio_echo_server(
|
|||
break
|
||||
|
||||
# echo the msg back
|
||||
chan.send_nowait(msg)
|
||||
to_trio.send_nowait(msg)
|
||||
|
||||
# if we get the terminate sentinel
|
||||
# break the echo loop
|
||||
|
|
@ -767,10 +765,7 @@ async def trio_to_aio_echo_server(
|
|||
):
|
||||
async with to_asyncio.open_channel_from(
|
||||
aio_echo_server,
|
||||
) as (
|
||||
first, # value from `chan.started_nowait()` above
|
||||
chan,
|
||||
):
|
||||
) as (first, chan):
|
||||
assert first == 'start'
|
||||
|
||||
await ctx.started(first)
|
||||
|
|
@ -781,8 +776,7 @@ async def trio_to_aio_echo_server(
|
|||
await chan.send(msg)
|
||||
|
||||
out = await chan.receive()
|
||||
|
||||
# echo back to parent-actor's remote parent-ctx-task!
|
||||
# echo back to parent actor-task
|
||||
await stream.send(out)
|
||||
|
||||
if out is None:
|
||||
|
|
@ -1096,21 +1090,24 @@ def test_sigint_closes_lifetime_stack(
|
|||
|
||||
|
||||
# ?TODO asyncio.Task fn-deco?
|
||||
# -[ ] do sig checkingat import time like @context?
|
||||
# -[ ] maybe name it @aio_task ??
|
||||
# -[ ] chan: to_asyncio.InterloopChannel ??
|
||||
# -[ ] do fn-sig checking at import time like @context?
|
||||
# |_[ ] maybe name it @a(sync)io_task ??
|
||||
# @asyncio_task <- not bad ??
|
||||
async def raise_before_started(
|
||||
# from_trio: asyncio.Queue,
|
||||
# to_trio: trio.abc.SendChannel,
|
||||
chan: to_asyncio.LinkedTaskChannel,
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
`asyncio.Task` entry point which RTEs before calling
|
||||
`chan.started_nowait()`.
|
||||
`to_trio.send_nowait()`.
|
||||
|
||||
'''
|
||||
await asyncio.sleep(0.2)
|
||||
raise RuntimeError('Some shite went wrong before `.send_nowait()`!!')
|
||||
|
||||
# to_trio.send_nowait('Uhh we shouldve RTE-d ^^ ??')
|
||||
chan.started_nowait('Uhh we shouldve RTE-d ^^ ??')
|
||||
await asyncio.sleep(float('inf'))
|
||||
|
||||
|
|
|
|||
|
|
@ -1,11 +1,9 @@
|
|||
"""
|
||||
Streaming via the, now legacy, "async-gen API".
|
||||
|
||||
Streaming via async gen api
|
||||
"""
|
||||
import time
|
||||
from functools import partial
|
||||
import platform
|
||||
from typing import Callable
|
||||
|
||||
import trio
|
||||
import tractor
|
||||
|
|
@ -21,11 +19,7 @@ def test_must_define_ctx():
|
|||
async def no_ctx():
|
||||
pass
|
||||
|
||||
assert (
|
||||
"no_ctx must be `ctx: tractor.Context"
|
||||
in
|
||||
str(err.value)
|
||||
)
|
||||
assert "no_ctx must be `ctx: tractor.Context" in str(err.value)
|
||||
|
||||
@tractor.stream
|
||||
async def has_ctx(ctx):
|
||||
|
|
@ -75,14 +69,14 @@ async def stream_from_single_subactor(
|
|||
async with tractor.open_nursery(
|
||||
registry_addrs=[reg_addr],
|
||||
start_method=start_method,
|
||||
) as an:
|
||||
) as nursery:
|
||||
|
||||
async with tractor.find_actor('streamerd') as portals:
|
||||
|
||||
if not portals:
|
||||
|
||||
# no brokerd actor found
|
||||
portal = await an.start_actor(
|
||||
portal = await nursery.start_actor(
|
||||
'streamerd',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
|
|
@ -122,22 +116,11 @@ async def stream_from_single_subactor(
|
|||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'stream_func',
|
||||
[
|
||||
async_gen_stream,
|
||||
context_stream,
|
||||
],
|
||||
ids='stream_func={}'.format
|
||||
'stream_func', [async_gen_stream, context_stream]
|
||||
)
|
||||
def test_stream_from_single_subactor(
|
||||
reg_addr: tuple,
|
||||
start_method: str,
|
||||
stream_func: Callable,
|
||||
):
|
||||
'''
|
||||
Verify streaming from a spawned async generator.
|
||||
|
||||
'''
|
||||
def test_stream_from_single_subactor(reg_addr, start_method, stream_func):
|
||||
"""Verify streaming from a spawned async generator.
|
||||
"""
|
||||
trio.run(
|
||||
partial(
|
||||
stream_from_single_subactor,
|
||||
|
|
@ -149,9 +132,10 @@ def test_stream_from_single_subactor(
|
|||
|
||||
|
||||
# this is the first 2 actors, streamer_1 and streamer_2
|
||||
async def stream_data(seed: int):
|
||||
async def stream_data(seed):
|
||||
|
||||
for i in range(seed):
|
||||
|
||||
yield i
|
||||
|
||||
# trigger scheduler to simulate practical usage
|
||||
|
|
@ -159,17 +143,15 @@ async def stream_data(seed: int):
|
|||
|
||||
|
||||
# this is the third actor; the aggregator
|
||||
async def aggregate(seed: int):
|
||||
'''
|
||||
Ensure that the two streams we receive match but only stream
|
||||
async def aggregate(seed):
|
||||
"""Ensure that the two streams we receive match but only stream
|
||||
a single set of values to the parent.
|
||||
|
||||
'''
|
||||
async with tractor.open_nursery() as an:
|
||||
"""
|
||||
async with tractor.open_nursery() as nursery:
|
||||
portals = []
|
||||
for i in range(1, 3):
|
||||
# fork point
|
||||
portal = await an.start_actor(
|
||||
portal = await nursery.start_actor(
|
||||
name=f'streamer_{i}',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
|
|
@ -182,8 +164,7 @@ async def aggregate(seed: int):
|
|||
async with send_chan:
|
||||
|
||||
async with portal.open_stream_from(
|
||||
stream_data,
|
||||
seed=seed,
|
||||
stream_data, seed=seed,
|
||||
) as stream:
|
||||
|
||||
async for value in stream:
|
||||
|
|
@ -193,14 +174,10 @@ async def aggregate(seed: int):
|
|||
print(f"FINISHED ITERATING {portal.channel.uid}")
|
||||
|
||||
# spawn 2 trio tasks to collect streams and push to a local queue
|
||||
async with trio.open_nursery() as tn:
|
||||
async with trio.open_nursery() as n:
|
||||
|
||||
for portal in portals:
|
||||
tn.start_soon(
|
||||
push_to_chan,
|
||||
portal,
|
||||
send_chan.clone(),
|
||||
)
|
||||
n.start_soon(push_to_chan, portal, send_chan.clone())
|
||||
|
||||
# close this local task's reference to send side
|
||||
await send_chan.aclose()
|
||||
|
|
@ -217,21 +194,20 @@ async def aggregate(seed: int):
|
|||
|
||||
print("FINISHED ITERATING in aggregator")
|
||||
|
||||
await an.cancel()
|
||||
await nursery.cancel()
|
||||
print("WAITING on `ActorNursery` to finish")
|
||||
print("AGGREGATOR COMPLETE!")
|
||||
|
||||
|
||||
async def a_quadruple_example() -> list[int]:
|
||||
'''
|
||||
Open the root-actor which is also a "registrar".
|
||||
# this is the main actor and *arbiter*
|
||||
async def a_quadruple_example():
|
||||
# a nursery which spawns "actors"
|
||||
async with tractor.open_nursery() as nursery:
|
||||
|
||||
'''
|
||||
async with tractor.open_nursery() as an:
|
||||
seed = int(1e3)
|
||||
pre_start = time.time()
|
||||
|
||||
portal = await an.start_actor(
|
||||
portal = await nursery.start_actor(
|
||||
name='aggregator',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
|
|
@ -252,14 +228,8 @@ async def a_quadruple_example() -> list[int]:
|
|||
return result_stream
|
||||
|
||||
|
||||
async def cancel_after(
|
||||
wait: float,
|
||||
reg_addr: tuple,
|
||||
) -> list[int]:
|
||||
|
||||
async with tractor.open_root_actor(
|
||||
registry_addrs=[reg_addr],
|
||||
):
|
||||
async def cancel_after(wait, reg_addr):
|
||||
async with tractor.open_root_actor(registry_addrs=[reg_addr]):
|
||||
with trio.move_on_after(wait):
|
||||
return await a_quadruple_example()
|
||||
|
||||
|
|
@ -270,10 +240,6 @@ def time_quad_ex(
|
|||
ci_env: bool,
|
||||
spawn_backend: str,
|
||||
):
|
||||
non_linux: bool = (_sys := platform.system()) != 'Linux'
|
||||
if ci_env and non_linux:
|
||||
pytest.skip(f'Test is too flaky on {_sys!r} in CI')
|
||||
|
||||
if spawn_backend == 'mp':
|
||||
'''
|
||||
no idea but the mp *nix runs are flaking out here often...
|
||||
|
|
@ -281,20 +247,16 @@ def time_quad_ex(
|
|||
'''
|
||||
pytest.skip("Test is too flaky on mp in CI")
|
||||
|
||||
timeout = 7 if non_linux else 4
|
||||
timeout = 7 if platform.system() in ('Windows', 'Darwin') else 4
|
||||
start = time.time()
|
||||
results: list[int] = trio.run(
|
||||
cancel_after,
|
||||
timeout,
|
||||
reg_addr,
|
||||
)
|
||||
diff: float = time.time() - start
|
||||
results = trio.run(cancel_after, timeout, reg_addr)
|
||||
diff = time.time() - start
|
||||
assert results
|
||||
return results, diff
|
||||
|
||||
|
||||
def test_a_quadruple_example(
|
||||
time_quad_ex: tuple[list[int], float],
|
||||
time_quad_ex: tuple,
|
||||
ci_env: bool,
|
||||
spawn_backend: str,
|
||||
):
|
||||
|
|
@ -302,12 +264,13 @@ def test_a_quadruple_example(
|
|||
This also serves as a kind of "we'd like to be this fast test".
|
||||
|
||||
'''
|
||||
non_linux: bool = (_sys := platform.system()) != 'Linux'
|
||||
|
||||
results, diff = time_quad_ex
|
||||
assert results
|
||||
this_fast = (
|
||||
6 if non_linux
|
||||
6 if platform.system() in (
|
||||
'Windows',
|
||||
'Darwin',
|
||||
)
|
||||
else 3
|
||||
)
|
||||
assert diff < this_fast
|
||||
|
|
@ -318,33 +281,19 @@ def test_a_quadruple_example(
|
|||
list(map(lambda i: i/10, range(3, 9)))
|
||||
)
|
||||
def test_not_fast_enough_quad(
|
||||
reg_addr: tuple,
|
||||
time_quad_ex: tuple[list[int], float],
|
||||
cancel_delay: float,
|
||||
ci_env: bool,
|
||||
spawn_backend: str,
|
||||
reg_addr, time_quad_ex, cancel_delay, ci_env, spawn_backend
|
||||
):
|
||||
'''
|
||||
Verify we can cancel midway through the quad example and all
|
||||
actors cancel gracefully.
|
||||
|
||||
'''
|
||||
"""Verify we can cancel midway through the quad example and all actors
|
||||
cancel gracefully.
|
||||
"""
|
||||
results, diff = time_quad_ex
|
||||
delay = max(diff - cancel_delay, 0)
|
||||
results = trio.run(
|
||||
cancel_after,
|
||||
delay,
|
||||
reg_addr,
|
||||
)
|
||||
system: str = platform.system()
|
||||
if (
|
||||
system in ('Windows', 'Darwin')
|
||||
and
|
||||
results is not None
|
||||
):
|
||||
results = trio.run(cancel_after, delay, reg_addr)
|
||||
system = platform.system()
|
||||
if system in ('Windows', 'Darwin') and results is not None:
|
||||
# In CI envoirments it seems later runs are quicker then the first
|
||||
# so just ignore these
|
||||
print(f'Woa there {system} caught your breath eh?')
|
||||
print(f"Woa there {system} caught your breath eh?")
|
||||
else:
|
||||
# should be cancelled mid-streaming
|
||||
assert results is None
|
||||
|
|
@ -352,24 +301,23 @@ def test_not_fast_enough_quad(
|
|||
|
||||
@tractor_test
|
||||
async def test_respawn_consumer_task(
|
||||
reg_addr: tuple,
|
||||
spawn_backend: str,
|
||||
loglevel: str,
|
||||
reg_addr,
|
||||
spawn_backend,
|
||||
loglevel,
|
||||
):
|
||||
'''
|
||||
Verify that ``._portal.ReceiveStream.shield()``
|
||||
"""Verify that ``._portal.ReceiveStream.shield()``
|
||||
sucessfully protects the underlying IPC channel from being closed
|
||||
when cancelling and respawning a consumer task.
|
||||
|
||||
This also serves to verify that all values from the stream can be
|
||||
received despite the respawns.
|
||||
|
||||
'''
|
||||
"""
|
||||
stream = None
|
||||
|
||||
async with tractor.open_nursery() as an:
|
||||
async with tractor.open_nursery() as n:
|
||||
|
||||
portal = await an.start_actor(
|
||||
portal = await n.start_actor(
|
||||
name='streamer',
|
||||
enable_modules=[__name__]
|
||||
)
|
||||
|
|
|
|||
|
|
@ -35,9 +35,6 @@ if TYPE_CHECKING:
|
|||
)
|
||||
|
||||
|
||||
_non_linux: bool = platform.system() != 'Linux'
|
||||
|
||||
|
||||
def test_abort_on_sigint(
|
||||
daemon: subprocess.Popen,
|
||||
):
|
||||
|
|
@ -140,7 +137,6 @@ def test_non_registrar_spawns_child(
|
|||
reg_addr: UnwrappedAddress,
|
||||
loglevel: str,
|
||||
debug_mode: bool,
|
||||
ci_env: bool,
|
||||
):
|
||||
'''
|
||||
Ensure a non-regristar (serving) root actor can spawn a sub and
|
||||
|
|
@ -152,12 +148,6 @@ def test_non_registrar_spawns_child(
|
|||
|
||||
'''
|
||||
async def main():
|
||||
|
||||
# XXX, since apparently on macos in GH's CI it can be a race
|
||||
# with the `daemon` registrar on grabbing the socket-addr..
|
||||
if ci_env and _non_linux:
|
||||
await trio.sleep(.5)
|
||||
|
||||
async with tractor.open_nursery(
|
||||
registry_addrs=[reg_addr],
|
||||
loglevel=loglevel,
|
||||
|
|
|
|||
|
|
@ -91,12 +91,13 @@ def test_infected_root_actor(
|
|||
async def sync_and_err(
|
||||
# just signature placeholders for compat with
|
||||
# ``to_asyncio.open_channel_from()``
|
||||
chan: tractor.to_asyncio.LinkedTaskChannel,
|
||||
to_trio: trio.MemorySendChannel,
|
||||
from_trio: asyncio.Queue,
|
||||
ev: asyncio.Event,
|
||||
|
||||
):
|
||||
if chan:
|
||||
chan.started_nowait('start')
|
||||
if to_trio:
|
||||
to_trio.send_nowait('start')
|
||||
|
||||
await ev.wait()
|
||||
raise RuntimeError('asyncio-side')
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@
|
|||
Shared mem primitives and APIs.
|
||||
|
||||
"""
|
||||
import platform
|
||||
import uuid
|
||||
|
||||
# import numpy
|
||||
|
|
@ -54,18 +53,7 @@ def test_child_attaches_alot():
|
|||
shm_key=shml.key,
|
||||
) as (ctx, start_val),
|
||||
):
|
||||
assert (_key := shml.key) == start_val
|
||||
|
||||
if platform.system() != 'Darwin':
|
||||
# XXX, macOS has a char limit..
|
||||
# see `ipc._shm._shorten_key_for_macos`
|
||||
assert (
|
||||
start_val
|
||||
==
|
||||
key
|
||||
==
|
||||
_key
|
||||
)
|
||||
assert start_val == key
|
||||
await ctx.result()
|
||||
|
||||
await portal.cancel_actor()
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ from __future__ import annotations
|
|||
from contextvars import (
|
||||
ContextVar,
|
||||
)
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import (
|
||||
Any,
|
||||
|
|
@ -29,7 +30,6 @@ from typing import (
|
|||
TYPE_CHECKING,
|
||||
)
|
||||
|
||||
import platformdirs
|
||||
from trio.lowlevel import current_task
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
|
@ -172,56 +172,23 @@ def current_ipc_ctx(
|
|||
return ctx
|
||||
|
||||
|
||||
# std ODE (mutable) app state location
|
||||
_rtdir: Path = Path(os.environ['XDG_RUNTIME_DIR'])
|
||||
|
||||
|
||||
def get_rt_dir(
|
||||
subdir: str|Path|None = None,
|
||||
appname: str = 'tractor',
|
||||
subdir: str = 'tractor'
|
||||
) -> Path:
|
||||
'''
|
||||
Return the user "runtime dir", the file-sys location where most
|
||||
userspace apps stick their IPC and cache related system
|
||||
util-files.
|
||||
|
||||
On linux we use a `${XDG_RUNTIME_DIR}/tractor/` subdir by
|
||||
default, but equivalents are mapped for each platform using
|
||||
the lovely `platformdirs` lib.
|
||||
Return the user "runtime dir" where most userspace apps stick
|
||||
their IPC and cache related system util-files; we take hold
|
||||
of a `'XDG_RUNTIME_DIR'/tractor/` subdir by default.
|
||||
|
||||
'''
|
||||
rt_dir: Path = Path(
|
||||
platformdirs.user_runtime_dir(
|
||||
appname=appname,
|
||||
),
|
||||
)
|
||||
|
||||
# Normalize and validate that `subdir` is a relative path
|
||||
# without any parent-directory ("..") components, to prevent
|
||||
# escaping the runtime directory.
|
||||
if subdir:
|
||||
subdir_path = (
|
||||
subdir
|
||||
if isinstance(subdir, Path)
|
||||
else Path(subdir)
|
||||
)
|
||||
if subdir_path.is_absolute():
|
||||
raise ValueError(
|
||||
f'`subdir` must be a relative path!\n'
|
||||
f'{subdir!r}\n'
|
||||
)
|
||||
if any(part == '..' for part in subdir_path.parts):
|
||||
raise ValueError(
|
||||
"`subdir` must not contain '..' components!\n"
|
||||
f'{subdir!r}\n'
|
||||
)
|
||||
|
||||
rt_dir: Path = rt_dir / subdir_path
|
||||
|
||||
if not rt_dir.is_dir():
|
||||
rt_dir.mkdir(
|
||||
parents=True,
|
||||
exist_ok=True, # avoid `FileExistsError` from conc calls
|
||||
)
|
||||
|
||||
return rt_dir
|
||||
rtdir: Path = _rtdir / subdir
|
||||
if not rtdir.is_dir():
|
||||
rtdir.mkdir()
|
||||
return rtdir
|
||||
|
||||
|
||||
def current_ipc_protos() -> list[str]:
|
||||
|
|
|
|||
|
|
@ -21,7 +21,6 @@ cancellation during REPL interaction.
|
|||
|
||||
'''
|
||||
from __future__ import annotations
|
||||
import platform
|
||||
from typing import (
|
||||
TYPE_CHECKING,
|
||||
)
|
||||
|
|
@ -50,7 +49,6 @@ if TYPE_CHECKING:
|
|||
|
||||
log = get_logger()
|
||||
|
||||
_is_macos: bool = platform.system() == 'Darwin'
|
||||
_ctlc_ignore_header: str = (
|
||||
'Ignoring SIGINT while debug REPL in use'
|
||||
)
|
||||
|
|
@ -302,11 +300,6 @@ def sigint_shield(
|
|||
# XXX: yah, mega hack, but how else do we catch this madness XD
|
||||
if (
|
||||
repl.shname == 'xonsh'
|
||||
or (
|
||||
repl.shname == 'bash'
|
||||
and
|
||||
_is_macos
|
||||
)
|
||||
):
|
||||
flush_status += (
|
||||
'-> ALSO re-flushing due to `xonsh`..\n'
|
||||
|
|
|
|||
|
|
@ -23,7 +23,6 @@ considered optional within the context of this runtime-library.
|
|||
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import hashlib
|
||||
from multiprocessing import shared_memory as shm
|
||||
from multiprocessing.shared_memory import (
|
||||
# SharedMemory,
|
||||
|
|
@ -107,12 +106,11 @@ class NDToken(Struct, frozen=True):
|
|||
This type is msg safe.
|
||||
|
||||
'''
|
||||
shm_name: str # actual OS-level name (may be shortened on macOS)
|
||||
shm_name: str # this servers as a "key" value
|
||||
shm_first_index_name: str
|
||||
shm_last_index_name: str
|
||||
dtype_descr: tuple
|
||||
size: int # in struct-array index / row terms
|
||||
key: str|None = None # original descriptive key (for lookup)
|
||||
|
||||
# TODO: use nptyping here on dtypes
|
||||
@property
|
||||
|
|
@ -126,41 +124,6 @@ class NDToken(Struct, frozen=True):
|
|||
def as_msg(self):
|
||||
return to_builtins(self)
|
||||
|
||||
def __eq__(self, other) -> bool:
|
||||
'''
|
||||
Compare tokens based on shm names and dtype,
|
||||
ignoring the `key` field.
|
||||
|
||||
The `key` field is only used for lookups,
|
||||
not for token identity.
|
||||
|
||||
'''
|
||||
if not isinstance(other, NDToken):
|
||||
return False
|
||||
return (
|
||||
self.shm_name == other.shm_name
|
||||
and self.shm_first_index_name
|
||||
== other.shm_first_index_name
|
||||
and self.shm_last_index_name
|
||||
== other.shm_last_index_name
|
||||
and self.dtype_descr == other.dtype_descr
|
||||
and self.size == other.size
|
||||
)
|
||||
|
||||
def __hash__(self) -> int:
|
||||
'''
|
||||
Hash based on the same fields used
|
||||
in `.__eq__()`.
|
||||
|
||||
'''
|
||||
return hash((
|
||||
self.shm_name,
|
||||
self.shm_first_index_name,
|
||||
self.shm_last_index_name,
|
||||
self.dtype_descr,
|
||||
self.size,
|
||||
))
|
||||
|
||||
@classmethod
|
||||
def from_msg(cls, msg: dict) -> NDToken:
|
||||
if isinstance(msg, NDToken):
|
||||
|
|
@ -197,50 +160,6 @@ def get_shm_token(key: str) -> NDToken | None:
|
|||
return _known_tokens.get(key)
|
||||
|
||||
|
||||
def _shorten_key_for_macos(
|
||||
key: str,
|
||||
prefix: str = '',
|
||||
suffix: str = '',
|
||||
) -> str:
|
||||
'''
|
||||
MacOS has a (hillarious) 31 character limit for POSIX shared
|
||||
memory names. Hash long keys to fit within this limit while
|
||||
maintaining uniqueness.
|
||||
|
||||
'''
|
||||
# macOS shm_open() has a 31 char limit (PSHMNAMLEN)
|
||||
# format: /t_<hash16> = 19 chars, well under limit
|
||||
max_len: int = 31
|
||||
if len(key) <= max_len:
|
||||
return key
|
||||
|
||||
_hash: str = hashlib.sha256(
|
||||
key.encode()
|
||||
).hexdigest()
|
||||
|
||||
hash_len: int = (
|
||||
(max_len - 1)
|
||||
- len(prefix)
|
||||
- len(suffix)
|
||||
)
|
||||
key_hash: str = _hash[:hash_len]
|
||||
short_key = (
|
||||
prefix
|
||||
+
|
||||
f'{key_hash}'
|
||||
+
|
||||
suffix
|
||||
)
|
||||
|
||||
log.debug(
|
||||
f'Shortened shm key for macOS:\n'
|
||||
f' original: {key!r} ({len(key)!r} chars)\n'
|
||||
f' shortened: {short_key!r}'
|
||||
f' ({len(short_key)!r} chars)'
|
||||
)
|
||||
return short_key
|
||||
|
||||
|
||||
def _make_token(
|
||||
key: str,
|
||||
size: int,
|
||||
|
|
@ -252,32 +171,12 @@ def _make_token(
|
|||
to access a shared array.
|
||||
|
||||
'''
|
||||
# On macOS, shorten keys that exceed the
|
||||
# 31 character limit
|
||||
if platform.system() == 'Darwin':
|
||||
shm_name = _shorten_key_for_macos(
|
||||
key=key,
|
||||
)
|
||||
shm_first = _shorten_key_for_macos(
|
||||
key=key,
|
||||
suffix='_first',
|
||||
)
|
||||
shm_last = _shorten_key_for_macos(
|
||||
key=key,
|
||||
suffix='_last',
|
||||
)
|
||||
else:
|
||||
shm_name = key
|
||||
shm_first = key + '_first'
|
||||
shm_last = key + '_last'
|
||||
|
||||
return NDToken(
|
||||
shm_name=shm_name,
|
||||
shm_first_index_name=shm_first,
|
||||
shm_last_index_name=shm_last,
|
||||
shm_name=key,
|
||||
shm_first_index_name=key + "_first",
|
||||
shm_last_index_name=key + "_last",
|
||||
dtype_descr=tuple(np.dtype(dtype).descr),
|
||||
size=size,
|
||||
key=key, # store original key for lookup
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -532,17 +431,9 @@ class ShmArray:
|
|||
|
||||
def destroy(self) -> None:
|
||||
if _USE_POSIX:
|
||||
# We manually unlink to bypass all the
|
||||
# "resource tracker" nonsense meant for
|
||||
# non-SC systems.
|
||||
name = self._shm.name
|
||||
try:
|
||||
shm_unlink(name)
|
||||
except FileNotFoundError:
|
||||
# might be a teardown race here?
|
||||
log.warning(
|
||||
f'Shm for {name} already unlinked?'
|
||||
)
|
||||
# We manually unlink to bypass all the "resource tracker"
|
||||
# nonsense meant for non-SC systems.
|
||||
shm_unlink(self._shm.name)
|
||||
|
||||
self._first.destroy()
|
||||
self._last.destroy()
|
||||
|
|
@ -572,16 +463,8 @@ def open_shm_ndarray(
|
|||
a = np.zeros(size, dtype=dtype)
|
||||
a['index'] = np.arange(len(a))
|
||||
|
||||
# Create token first to get the (possibly
|
||||
# shortened) shm name
|
||||
token = _make_token(
|
||||
key=key,
|
||||
size=size,
|
||||
dtype=dtype,
|
||||
)
|
||||
|
||||
shm = SharedMemory(
|
||||
name=token.shm_name,
|
||||
name=key,
|
||||
create=True,
|
||||
size=a.nbytes
|
||||
)
|
||||
|
|
@ -593,6 +476,12 @@ def open_shm_ndarray(
|
|||
array[:] = a[:]
|
||||
array.setflags(write=int(not readonly))
|
||||
|
||||
token = _make_token(
|
||||
key=key,
|
||||
size=size,
|
||||
dtype=dtype,
|
||||
)
|
||||
|
||||
# create single entry arrays for storing an first and last indices
|
||||
first = SharedInt(
|
||||
shm=SharedMemory(
|
||||
|
|
@ -665,23 +554,13 @@ def attach_shm_ndarray(
|
|||
|
||||
'''
|
||||
token = NDToken.from_msg(token)
|
||||
# Use original key for _known_tokens lookup,
|
||||
# shm_name for OS calls
|
||||
lookup_key = (
|
||||
token.key if token.key
|
||||
else token.shm_name
|
||||
)
|
||||
key = token.shm_name
|
||||
|
||||
if lookup_key in _known_tokens:
|
||||
assert (
|
||||
NDToken.from_msg(
|
||||
_known_tokens[lookup_key]
|
||||
) == token
|
||||
), 'WTF'
|
||||
if key in _known_tokens:
|
||||
assert NDToken.from_msg(_known_tokens[key]) == token, "WTF"
|
||||
|
||||
# XXX: ugh, looks like due to the ``shm_open()``
|
||||
# C api we can't actually place files in a subdir,
|
||||
# see discussion here:
|
||||
# XXX: ugh, looks like due to the ``shm_open()`` C api we can't
|
||||
# actually place files in a subdir, see discussion here:
|
||||
# https://stackoverflow.com/a/11103289
|
||||
|
||||
# attach to array buffer and view as per dtype
|
||||
|
|
@ -689,7 +568,7 @@ def attach_shm_ndarray(
|
|||
for _ in range(3):
|
||||
try:
|
||||
shm = SharedMemory(
|
||||
name=token.shm_name,
|
||||
name=key,
|
||||
create=False,
|
||||
)
|
||||
break
|
||||
|
|
@ -735,10 +614,10 @@ def attach_shm_ndarray(
|
|||
sha.array
|
||||
|
||||
# Stash key -> token knowledge for future queries
|
||||
# via `maybe_open_shm_ndarray()` but only after
|
||||
# we know we can attach.
|
||||
if lookup_key not in _known_tokens:
|
||||
_known_tokens[lookup_key] = token
|
||||
# via `maybe_opepn_shm_array()` but only after we know
|
||||
# we can attach.
|
||||
if key not in _known_tokens:
|
||||
_known_tokens[key] = token
|
||||
|
||||
# "close" attached shm on actor teardown
|
||||
tractor.current_actor().lifetime_stack.callback(sha.close)
|
||||
|
|
@ -782,10 +661,7 @@ def maybe_open_shm_ndarray(
|
|||
False, # not newly opened
|
||||
)
|
||||
except KeyError:
|
||||
log.warning(
|
||||
f'Could not find key in shms cache,\n'
|
||||
f'key: {key!r}\n'
|
||||
)
|
||||
log.warning(f"Could not find {key} in shms cache")
|
||||
if dtype:
|
||||
token = _make_token(
|
||||
key,
|
||||
|
|
@ -895,7 +771,6 @@ def open_shm_list(
|
|||
size: int = int(2 ** 10),
|
||||
dtype: float | int | bool | str | bytes | None = float,
|
||||
readonly: bool = True,
|
||||
prefix: str = 'shml_',
|
||||
|
||||
) -> ShmList:
|
||||
|
||||
|
|
@ -909,12 +784,6 @@ def open_shm_list(
|
|||
}[dtype]
|
||||
sequence = [default] * size
|
||||
|
||||
if platform.system() == 'Darwin':
|
||||
key: str = _shorten_key_for_macos(
|
||||
key=key,
|
||||
prefix=prefix,
|
||||
)
|
||||
|
||||
shml = ShmList(
|
||||
sequence=sequence,
|
||||
name=key,
|
||||
|
|
|
|||
|
|
@ -23,12 +23,12 @@ from contextlib import (
|
|||
)
|
||||
from pathlib import Path
|
||||
import os
|
||||
import sys
|
||||
from socket import (
|
||||
AF_UNIX,
|
||||
SOCK_STREAM,
|
||||
SO_PASSCRED,
|
||||
SO_PEERCRED,
|
||||
SOL_SOCKET,
|
||||
error as socket_error,
|
||||
)
|
||||
import struct
|
||||
from typing import (
|
||||
|
|
@ -53,7 +53,7 @@ from tractor.log import get_logger
|
|||
from tractor.ipc._transport import (
|
||||
MsgpackTransport,
|
||||
)
|
||||
from tractor._state import (
|
||||
from .._state import (
|
||||
get_rt_dir,
|
||||
current_actor,
|
||||
is_root_process,
|
||||
|
|
@ -63,28 +63,6 @@ if TYPE_CHECKING:
|
|||
from ._runtime import Actor
|
||||
|
||||
|
||||
# Platform-specific credential passing constants
|
||||
# See: https://stackoverflow.com/a/7982749
|
||||
if sys.platform == 'linux':
|
||||
from socket import (
|
||||
SO_PASSCRED,
|
||||
SO_PEERCRED,
|
||||
)
|
||||
|
||||
else:
|
||||
# Other (Unix) platforms - though further testing is required and
|
||||
# others may need additional special handling?
|
||||
SO_PASSCRED = None
|
||||
SO_PEERCRED = None
|
||||
|
||||
# NOTE, macOS uses `LOCAL_PEERCRED` instead of `SO_PEERCRED` and
|
||||
# doesn't need `SO_PASSCRED` (credential passing is always enabled).
|
||||
# See code in <sys/un.h>: `#define LOCAL_PEERCRED 0x001`
|
||||
#
|
||||
# XXX INSTEAD we use the (hopefully) more generic
|
||||
# `get_peer_pid()` below for other OSes.
|
||||
|
||||
|
||||
log = get_logger()
|
||||
|
||||
|
||||
|
|
@ -187,11 +165,7 @@ class UDSAddress(
|
|||
err_on_no_runtime=False,
|
||||
)
|
||||
if actor:
|
||||
sockname: str = f'{actor.aid.name}@{pid}'
|
||||
# XXX, orig version which broke both macOS (file-name
|
||||
# length) and `multiaddrs` ('::' invalid separator).
|
||||
# sockname: str = '::'.join(actor.uid) + f'@{pid}'
|
||||
#
|
||||
sockname: str = '::'.join(actor.uid) + f'@{pid}'
|
||||
# ?^TODO, for `multiaddr`'s parser we can't use the `::`
|
||||
# above^, SO maybe a `.` or something else here?
|
||||
# sockname: str = '.'.join(actor.uid) + f'@{pid}'
|
||||
|
|
@ -318,12 +292,7 @@ def close_listener(
|
|||
|
||||
|
||||
async def open_unix_socket_w_passcred(
|
||||
filename: (
|
||||
str
|
||||
|bytes
|
||||
|os.PathLike[str]
|
||||
|os.PathLike[bytes]
|
||||
),
|
||||
filename: str|bytes|os.PathLike[str]|os.PathLike[bytes],
|
||||
) -> trio.SocketStream:
|
||||
'''
|
||||
Literally the exact same as `trio.open_unix_socket()` except we set the additiona
|
||||
|
|
@ -341,66 +310,21 @@ async def open_unix_socket_w_passcred(
|
|||
# much more simplified logic vs tcp sockets - one socket type and only one
|
||||
# possible location to connect to
|
||||
sock = trio.socket.socket(AF_UNIX, SOCK_STREAM)
|
||||
|
||||
# Only set SO_PASSCRED on Linux (not needed/available on macOS)
|
||||
if SO_PASSCRED is not None:
|
||||
sock.setsockopt(SOL_SOCKET, SO_PASSCRED, 1)
|
||||
|
||||
with close_on_error(sock):
|
||||
await sock.connect(os.fspath(filename))
|
||||
|
||||
return trio.SocketStream(sock)
|
||||
|
||||
|
||||
def get_peer_pid(sock) -> int|None:
|
||||
'''
|
||||
Gets the PID of the process connected to the other end of a Unix
|
||||
domain socket on macOS, or `None` if that fails.
|
||||
|
||||
NOTE, should work on MacOS (and others?).
|
||||
|
||||
'''
|
||||
# try to get the peer PID using a naive soln found from,
|
||||
# https://stackoverflow.com/a/67971484
|
||||
#
|
||||
# NOTE, a more correct soln is likely needed here according to
|
||||
# the complaints of `copilot` which led to digging into the
|
||||
# underlying `go`lang issue linked from the above SO answer,
|
||||
|
||||
# XXX, darwin-xnu kernel srces defining these constants,
|
||||
# - SOL_LOCAL
|
||||
# |_https://github.com/apple/darwin-xnu/blob/main/bsd/sys/un.h#L85
|
||||
# - LOCAL_PEERPID
|
||||
# |_https://github.com/apple/darwin-xnu/blob/main/bsd/sys/un.h#L89
|
||||
#
|
||||
SOL_LOCAL: int = 0
|
||||
LOCAL_PEERPID: int = 0x002
|
||||
|
||||
try:
|
||||
pid: int = sock.getsockopt(
|
||||
SOL_LOCAL,
|
||||
LOCAL_PEERPID,
|
||||
)
|
||||
return pid
|
||||
except socket_error as e:
|
||||
log.exception(
|
||||
f"Failed to get peer PID: {e}"
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def get_peer_info(
|
||||
sock: trio.socket.socket,
|
||||
) -> tuple[
|
||||
def get_peer_info(sock: trio.socket.socket) -> tuple[
|
||||
int, # pid
|
||||
int, # uid
|
||||
int, # guid
|
||||
]:
|
||||
'''
|
||||
Deliver the connecting peer's "credentials"-info as defined in
|
||||
a platform-specific way.
|
||||
|
||||
Linux-ONLY, uses SO_PEERCRED.
|
||||
a very Linux specific way..
|
||||
|
||||
For more deats see,
|
||||
- `man accept`,
|
||||
|
|
@ -413,11 +337,6 @@ def get_peer_info(
|
|||
- https://stackoverflow.com/a/7982749
|
||||
|
||||
'''
|
||||
if SO_PEERCRED is None:
|
||||
raise RuntimeError(
|
||||
f'Peer credential retrieval not supported on {sys.platform}!'
|
||||
)
|
||||
|
||||
creds: bytes = sock.getsockopt(
|
||||
SOL_SOCKET,
|
||||
SO_PEERCRED,
|
||||
|
|
@ -521,38 +440,14 @@ class MsgpackUDSStream(MsgpackTransport):
|
|||
match (peername, sockname):
|
||||
case (str(), bytes()):
|
||||
sock_path: Path = Path(peername)
|
||||
|
||||
case (bytes(), str()):
|
||||
sock_path: Path = Path(sockname)
|
||||
|
||||
case (str(), str()): # XXX, likely macOS
|
||||
sock_path: Path = Path(peername)
|
||||
|
||||
case _:
|
||||
raise TypeError(
|
||||
f'Failed to match (peername, sockname) types?\n'
|
||||
f'peername: {peername!r}\n'
|
||||
f'sockname: {sockname!r}\n'
|
||||
)
|
||||
|
||||
if sys.platform == 'linux':
|
||||
(
|
||||
peer_pid,
|
||||
_,
|
||||
_,
|
||||
) = get_peer_info(sock)
|
||||
|
||||
# NOTE known to at least works on,
|
||||
# - macos
|
||||
else:
|
||||
peer_pid: int|None = get_peer_pid(sock)
|
||||
if peer_pid is None:
|
||||
log.warning(
|
||||
f'Unable to get peer PID?\n'
|
||||
f'sock: {sock!r}\n'
|
||||
f'peer_pid: {peer_pid!r}\n'
|
||||
)
|
||||
|
||||
filedir, filename = unwrap_sockpath(sock_path)
|
||||
laddr = UDSAddress(
|
||||
filedir=filedir,
|
||||
|
|
|
|||
|
|
@ -126,17 +126,13 @@ def iter_struct_ppfmt_lines(
|
|||
str(ft)
|
||||
).replace(' ', '')
|
||||
|
||||
if k[0] == '_':
|
||||
continue
|
||||
|
||||
# recurse to get sub-struct's `.pformat()` output Bo
|
||||
elif isinstance(v, Struct):
|
||||
if isinstance(v, Struct):
|
||||
yield from iter_struct_ppfmt_lines(
|
||||
struct=v,
|
||||
field_indent=field_indent+field_indent,
|
||||
)
|
||||
|
||||
else: # top-level field
|
||||
else:
|
||||
val_str: str = repr(v)
|
||||
|
||||
# XXX LOL, below just seems to be f#$%in causing
|
||||
|
|
|
|||
|
|
@ -48,7 +48,7 @@ from tractor._state import (
|
|||
_runtime_vars,
|
||||
)
|
||||
from tractor._context import Unresolved
|
||||
from tractor import devx
|
||||
from tractor.devx import debug
|
||||
from tractor.log import (
|
||||
get_logger,
|
||||
StackLevelAdapter,
|
||||
|
|
@ -94,14 +94,10 @@ else:
|
|||
QueueShutDown = False
|
||||
|
||||
|
||||
# TODO, generally speaking we can generalize this abstraction as,
|
||||
#
|
||||
# > A "SC linked, inter-event-loop" channel for comms between
|
||||
# > a `parent: trio.Task` -> `child: asyncio.Task` pair.
|
||||
#
|
||||
# It is **very similar** in terms of its operation as a "supervision
|
||||
# scope primitive" to that of our `._context.Context` with the only
|
||||
# difference being in how the tasks conduct msg-passing comms.
|
||||
# TODO, generally speaking we can generalize this abstraction, a "SC linked
|
||||
# parent->child task pair", as the same "supervision scope primitive"
|
||||
# **that is** our `._context.Context` with the only difference being
|
||||
# in how the tasks conduct msg-passing comms.
|
||||
#
|
||||
# For `LinkedTaskChannel` we are passing the equivalent of (once you
|
||||
# include all the recently added `._trio/aio_to_raise`
|
||||
|
|
@ -126,7 +122,6 @@ class LinkedTaskChannel(
|
|||
task scheduled in the host loop.
|
||||
|
||||
'''
|
||||
# ?TODO, rename as `._aio_q` since it's 2-way?
|
||||
_to_aio: asyncio.Queue
|
||||
_from_aio: trio.MemoryReceiveChannel
|
||||
|
||||
|
|
@ -240,11 +235,9 @@ class LinkedTaskChannel(
|
|||
#
|
||||
async def receive(self) -> Any:
|
||||
'''
|
||||
Receive a value `trio.Task` <- `asyncio.Task`.
|
||||
|
||||
Note the tasks in each loop are "SC linked" as a pair with
|
||||
Receive a value from the paired `asyncio.Task` with
|
||||
exception/cancel handling to teardown both sides on any
|
||||
unexpected error or cancellation.
|
||||
unexpected error.
|
||||
|
||||
'''
|
||||
try:
|
||||
|
|
@ -268,40 +261,15 @@ class LinkedTaskChannel(
|
|||
):
|
||||
raise err
|
||||
|
||||
async def get(self) -> Any:
|
||||
'''
|
||||
Receive a value `asyncio.Task` <- `trio.Task`.
|
||||
|
||||
This is equiv to `await self._to_aio.get()`.
|
||||
|
||||
'''
|
||||
return await self._to_aio.get()
|
||||
|
||||
async def send(self, item: Any) -> None:
|
||||
'''
|
||||
Send a value `trio.Task` -> `asyncio.Task`
|
||||
by enqueuing `item` onto the internal
|
||||
`asyncio.Queue` via `put_nowait()`.
|
||||
Send a value through to the asyncio task presuming
|
||||
it defines a ``from_trio`` argument, if it does not
|
||||
this method will raise an error.
|
||||
|
||||
'''
|
||||
self._to_aio.put_nowait(item)
|
||||
|
||||
# TODO? could we only compile-in this method on an instance
|
||||
# handed to the `asyncio`-side, i.e. the fn invoked with
|
||||
# `.open_channel_from()`.
|
||||
def send_nowait(
|
||||
self,
|
||||
item: Any,
|
||||
) -> None:
|
||||
'''
|
||||
Send a value through FROM the `asyncio.Task` to
|
||||
the `trio.Task` NON-BLOCKING.
|
||||
|
||||
This is equiv to `self._to_trio.send_nowait()`.
|
||||
|
||||
'''
|
||||
self._to_trio.send_nowait(item)
|
||||
|
||||
# TODO? needed?
|
||||
# async def wait_aio_complete(self) -> None:
|
||||
# await self._aio_task_complete.wait()
|
||||
|
|
@ -369,12 +337,9 @@ def _run_asyncio_task(
|
|||
|
||||
'''
|
||||
__tracebackhide__: bool = hide_tb
|
||||
if not (actor := tractor.current_actor()).is_infected_aio():
|
||||
if not tractor.current_actor().is_infected_aio():
|
||||
raise RuntimeError(
|
||||
f'`infect_asyncio: bool` mode is not enabled ??\n'
|
||||
f'Ensure you pass `ActorNursery.start_actor(infect_asyncio=True)`\n'
|
||||
f'\n'
|
||||
f'{actor}\n'
|
||||
"`infect_asyncio` mode is not enabled!?"
|
||||
)
|
||||
|
||||
# ITC (inter task comms), these channel/queue names are mostly from
|
||||
|
|
@ -437,23 +402,7 @@ def _run_asyncio_task(
|
|||
|
||||
orig = result = id(coro)
|
||||
try:
|
||||
# XXX TODO UGH!
|
||||
# this seems to break a `test_sync_pause_from_aio_task`
|
||||
# in a REALLY weird way where a `dict` value for
|
||||
# `_runtime_vars['_root_addrs']` is delivered from the
|
||||
# parent actor??
|
||||
#
|
||||
# XXX => see masked `.set_trace()` block in
|
||||
# `Actor.from_parent()`..
|
||||
#
|
||||
# with devx.maybe_open_crash_handler(
|
||||
# # XXX, if trio-side exits (intentionally) we
|
||||
# # shouldn't care bc it should have its own crash
|
||||
# # handling logic.
|
||||
# ignore={TrioTaskExited,},
|
||||
# ) as _bxerr:
|
||||
result: Any = await coro
|
||||
|
||||
chan._aio_result = result
|
||||
except BaseException as aio_err:
|
||||
chan._aio_err = aio_err
|
||||
|
|
@ -560,7 +509,7 @@ def _run_asyncio_task(
|
|||
if (
|
||||
debug_mode()
|
||||
and
|
||||
(greenback := devx.debug.maybe_import_greenback(
|
||||
(greenback := debug.maybe_import_greenback(
|
||||
force_reload=True,
|
||||
raise_not_found=False,
|
||||
))
|
||||
|
|
@ -960,11 +909,7 @@ async def translate_aio_errors(
|
|||
except BaseException as _trio_err:
|
||||
trio_err = chan._trio_err = _trio_err
|
||||
# await tractor.pause(shield=True) # workx!
|
||||
|
||||
# !TODO! we need an inter-loop lock here to avoid aio-tasks
|
||||
# clobbering trio ones when both crash in debug-mode!
|
||||
#
|
||||
entered: bool = await devx.debug._maybe_enter_pm(
|
||||
entered: bool = await debug._maybe_enter_pm(
|
||||
trio_err,
|
||||
api_frame=inspect.currentframe(),
|
||||
)
|
||||
|
|
@ -1298,18 +1243,10 @@ async def open_channel_from(
|
|||
suppress_graceful_exits: bool = True,
|
||||
**target_kwargs,
|
||||
|
||||
) -> AsyncIterator[
|
||||
tuple[Any, LinkedTaskChannel]
|
||||
]:
|
||||
) -> AsyncIterator[Any]:
|
||||
'''
|
||||
Start an `asyncio.Task` as `target()` and open an
|
||||
inter-loop (linked) channel for streaming between
|
||||
it and the current `trio.Task`.
|
||||
|
||||
A pair `(Any, chan: LinkedTaskChannel)` is delivered
|
||||
to the caller where the 1st element is the value
|
||||
provided by the `asyncio.Task`'s unblocking call
|
||||
to `chan.started_nowait()`.
|
||||
Open an inter-loop linked task channel for streaming between a target
|
||||
spawned ``asyncio`` task and ``trio``.
|
||||
|
||||
'''
|
||||
chan: LinkedTaskChannel = _run_asyncio_task(
|
||||
|
|
@ -1334,7 +1271,6 @@ async def open_channel_from(
|
|||
|
||||
# deliver stream handle upward
|
||||
yield first, chan
|
||||
# ^TODO! swap these!!
|
||||
except trio.Cancelled as taskc:
|
||||
if cs.cancel_called:
|
||||
if isinstance(chan._trio_to_raise, AsyncioCancelled):
|
||||
|
|
@ -1365,8 +1301,7 @@ async def open_channel_from(
|
|||
)
|
||||
else:
|
||||
# XXX SHOULD NEVER HAPPEN!
|
||||
log.error("SHOULD NEVER GET HERE !?!?")
|
||||
await tractor.pause(shield=True)
|
||||
await tractor.pause()
|
||||
else:
|
||||
chan._to_trio.close()
|
||||
|
||||
|
|
|
|||
11
uv.lock
11
uv.lock
|
|
@ -208,15 +208,6 @@ wheels = [
|
|||
{ url = "https://files.pythonhosted.org/packages/9e/c3/059298687310d527a58bb01f3b1965787ee3b40dce76752eda8b44e9a2c5/pexpect-4.9.0-py2.py3-none-any.whl", hash = "sha256:7236d1e080e4936be2dc3e326cec0af72acf9212a7e1d060210e70a47e253523", size = 63772, upload-time = "2023-11-25T06:56:14.81Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "platformdirs"
|
||||
version = "4.4.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/23/e8/21db9c9987b0e728855bd57bff6984f67952bea55d6f75e055c46b5383e8/platformdirs-4.4.0.tar.gz", hash = "sha256:ca753cf4d81dc309bc67b0ea38fd15dc97bc30ce419a7f58d13eb3bf14c4febf", size = 21634, upload-time = "2025-08-26T14:32:04.268Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/40/4b/2028861e724d3bd36227adfa20d3fd24c3fc6d52032f4a93c133be5d17ce/platformdirs-4.4.0-py3-none-any.whl", hash = "sha256:abd01743f24e5287cd7a5db3752faf1a2d65353f38ec26d98e25a6db65958c85", size = 18654, upload-time = "2025-08-26T14:32:02.735Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pluggy"
|
||||
version = "1.5.0"
|
||||
|
|
@ -385,7 +376,6 @@ dependencies = [
|
|||
{ name = "colorlog" },
|
||||
{ name = "msgspec" },
|
||||
{ name = "pdbp" },
|
||||
{ name = "platformdirs" },
|
||||
{ name = "tricycle" },
|
||||
{ name = "trio" },
|
||||
{ name = "wrapt" },
|
||||
|
|
@ -429,7 +419,6 @@ requires-dist = [
|
|||
{ name = "colorlog", specifier = ">=6.8.2,<7" },
|
||||
{ name = "msgspec", specifier = ">=0.19.0" },
|
||||
{ name = "pdbp", specifier = ">=1.8.2,<2" },
|
||||
{ name = "platformdirs", specifier = ">=4.4.0" },
|
||||
{ name = "tricycle", specifier = ">=0.4.1,<0.5" },
|
||||
{ name = "trio", specifier = ">0.27" },
|
||||
{ name = "wrapt", specifier = ">=1.16.0,<2" },
|
||||
|
|
|
|||
Loading…
Reference in New Issue