goodboy
/
tractor
1
0
Fork 2
Code Issues 5 Pull Requests 7 Packages Projects Releases Wiki Activity

Merge pull request #406 from goodboy/macos_support 

macOS support (for the fancy peeps with nice furniture)
ns_aware
Bd 2026-03-10 00:28:54 -04:00 committed by GitHub
parent 683476cc96 88b084802f
commit c1c4d85958
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
23 changed files with 917 additions and 243 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
.github/workflows/ci.yml vendored
View File

@ -74,16 +74,22 @@ jobs:
# run: mypy tractor/ --ignore-missing-imports --show-traceback
testing-linux:
name: '${{ matrix.os }} Python ${{ matrix.python }} - ${{ matrix.spawn_backend }}'
timeout-minutes: 10
testing:
name: '${{ matrix.os }} Python${{ matrix.python-version }} - spawn_backend=${{ matrix.spawn_backend }}'
timeout-minutes: 16
runs-on: ${{ matrix.os }}
strategy:
fail-fast: false
matrix:
os: [ubuntu-latest]
python-version: ['3.13']
os: [
ubuntu-latest,
macos-latest,
]
python-version: [
'3.13',
# '3.14',
]
spawn_backend: [
'trio',
# 'mp_spawn',
@ -91,7 +97,6 @@ jobs:
]
steps:
- uses: actions/checkout@v4
- name: 'Install uv + py-${{ matrix.python-version }}'

16
docs/README.rst
View File

@ -641,13 +641,15 @@ 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>`_)
- 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..
- **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..
Feel like saying hi?

16
examples/debugging/shield_hang_in_sub.py
View File

@ -3,6 +3,7 @@ Verify we can dump a `stackscope` tree on a hang.
'''
import os
import platform
import signal
import trio
@ -31,13 +32,26 @@ 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=['uds'],
enable_transports=[tpt],
) as an,
):
ptl: tractor.Portal = await an.start_actor(

17
examples/debugging/subactor_bp_in_ctx.py
View File

@ -1,3 +1,5 @@
import platform
import tractor
import trio
@ -34,9 +36,22 @@ 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=['uds'],
enable_transports=[tpt],
loglevel='devx',
) as n:
p = await n.start_actor(

8
examples/full_fledged_streaming_service.py
View File

@ -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='cancel',
loglevel='error',
# debug_mode=True,
) as an:
@ -118,8 +118,10 @@ async def main() -> list[int]:
cancelled: bool = await portal.cancel_actor()
assert cancelled
print(f"STREAM TIME = {time.time() - start}")
print(f"STREAM + SPAWN TIME = {time.time() - pre_start}")
print(
f"STREAM TIME = {time.time() - start}\n"
f"STREAM + SPAWN TIME = {time.time() - pre_start}\n"
)
assert result_stream == list(range(seed))
return result_stream

2
pyproject.toml
View File

@ -24,6 +24,7 @@ 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",
@ -48,6 +49,7 @@ dependencies = [
"msgspec>=0.19.0",
"cffi>=1.17.1",
"bidict>=0.23.1",
"platformdirs>=4.4.0",
]
# ------ project ------

86
tests/conftest.py
View File

@ -11,6 +11,7 @@ import platform
import time
import pytest
import tractor
from tractor._testing import (
examples_dir as examples_dir,
tractor_test as tractor_test,
@ -22,6 +23,7 @@ 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':
@ -44,6 +46,10 @@ 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(
@ -61,7 +67,7 @@ def pytest_addoption(
@pytest.fixture(scope='session', autouse=True)
def loglevel(request):
def loglevel(request) -> str:
import tractor
orig = tractor.log._default_loglevel
level = tractor.log._default_loglevel = request.config.option.loglevel
@ -69,11 +75,46 @@ def loglevel(request):
level=level,
name='tractor', # <- enable root logger
)
log.info(f'Test-harness logging level: {level}\n')
log.info(
f'Test-harness set runtime loglevel: {level!r}\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)
@ -110,6 +151,7 @@ def daemon(
testdir: pytest.Pytester,
reg_addr: tuple[str, int],
tpt_proto: str,
ci_env: bool,
) -> subprocess.Popen:
'''
@ -147,13 +189,25 @@ 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
@ -163,18 +217,30 @@ 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()
if stderr:
stdout: str = proc.stdout.read().decode()
if (
stderr
or
stdout
):
print(
f'Daemon actor tree produced STDERR:\n'
f'Daemon actor tree produced output:\n'
f'{proc.args}\n'
f'\n'
f'{stderr}\n'
f'stderr: {stderr!r}\n'
f'stdout: {stdout!r}\n'
)
if proc.returncode != -2:
raise RuntimeError(
'Daemon actor tree failed !?\n'
f'{proc.args}\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 rc < 0:
raise RuntimeError(msg)
log.error(msg)
# @pytest.fixture(autouse=True)

33
tests/devx/conftest.py
View File

@ -3,8 +3,9 @@
'''
from __future__ import annotations
import time
import platform
import signal
import time
from typing import (
Callable,
TYPE_CHECKING,
@ -33,6 +34,17 @@ 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[
@ -68,7 +80,10 @@ 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
@ -83,7 +98,10 @@ def spawn(
cmd,
**mkcmd_kwargs,
),
expect_timeout=3,
expect_timeout=(
10 if _non_linux and _ci_env
else 3
),
# preexec_fn=unset_colors,
# ^TODO? get `pytest` core to expose underlying
# `pexpect.spawn()` stuff?
@ -146,6 +164,8 @@ 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'
@ -251,12 +271,13 @@ def assert_before(
err_on_false=True,
**kwargs
)
return str(child.before.decode())
def do_ctlc(
child,
count: int = 3,
delay: float = 0.1,
delay: float|None = None,
patt: str|None = None,
# expect repl UX to reprint the prompt after every
@ -268,6 +289,7 @@ 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):
@ -278,7 +300,10 @@ def do_ctlc(
# if you run this test manually it works just fine..
if expect_prompt:
time.sleep(delay)
child.expect(PROMPT)
child.expect(
PROMPT,
timeout=(child.timeout * 2) if _ci_env else child.timeout,
)
before = str(child.before.decode())
time.sleep(delay)

123
tests/devx/test_debugger.py
View File

@ -37,6 +37,9 @@ from .conftest import (
in_prompt_msg,
assert_before,
)
from ..conftest import (
_ci_env,
)
if TYPE_CHECKING:
from ..conftest import PexpectSpawner
@ -51,13 +54,14 @@ 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...
@ -193,6 +197,11 @@ 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),
@ -258,6 +267,11 @@ 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,
@ -480,8 +494,24 @@ def test_multi_daemon_subactors(
stream.
'''
child = spawn('multi_daemon_subactors')
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
@ -511,8 +541,19 @@ def test_multi_daemon_subactors(
else:
raise ValueError('Neither log msg was found !?')
non_linux_delay: float = 0.3
if ctlc:
do_ctlc(child)
do_ctlc(
child,
delay=(
non_linux_delay
if non_linux
else None
),
)
if non_linux:
time.sleep(1)
# NOTE: previously since we did not have clobber prevention
# in the root actor this final resume could result in the debugger
@ -543,33 +584,66 @@ def test_multi_daemon_subactors(
# assert "in use by child ('bp_forever'," in before
if ctlc:
do_ctlc(child)
do_ctlc(
child,
delay=(
non_linux_delay
if non_linux
else None
),
)
if non_linux:
time.sleep(1)
# expect another breakpoint actor entry
child.sendline('c')
child.expect(PROMPT)
try:
assert_before(
before: str = assert_before(
child,
bp_forev_parts,
)
except AssertionError:
assert_before(
before: str = assert_before(
child,
name_error_parts,
)
else:
if ctlc:
do_ctlc(child)
before: str = do_ctlc(
child,
delay=(
non_linux_delay
if non_linux
else None
),
)
if non_linux:
time.sleep(1)
# 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')
child.expect(PROMPT)
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
assert_before(
child,
name_error_parts,
@ -689,7 +763,8 @@ def test_multi_subactors_root_errors(
@has_nested_actors
def test_multi_nested_subactors_error_through_nurseries(
spawn,
ci_env: bool,
spawn: PexpectSpawner,
# TODO: address debugger issue for nested tree:
# https://github.com/goodboy/tractor/issues/320
@ -712,7 +787,16 @@ def test_multi_nested_subactors_error_through_nurseries(
for send_char in itertools.cycle(['c', 'q']):
try:
child.expect(PROMPT)
child.expect(
PROMPT,
timeout=(
6 if (
_non_linux
and
ci_env
) else -1
),
)
child.sendline(send_char)
time.sleep(0.01)
@ -889,6 +973,11 @@ 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,
@ -1133,14 +1222,20 @@ 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?',
"'MsgpackUDSStream' was already closed locally?",
"TransportClosed: 'MsgpackUDSStream' was already closed 'by peer'?",
# ?TODO^? match depending on `tpt_proto(s)`?
f"'Msgpack{tpt}Stream' was already closed locally?",
f"TransportClosed: 'Msgpack{tpt}Stream' was already closed 'by peer'?",
]
# XXX races on whether these show/hit?

5
tests/devx/test_tooling.py
View File

@ -31,6 +31,9 @@ from .conftest import (
PROMPT,
_pause_msg,
)
from ..conftest import (
no_macos,
)
import pytest
from pexpect.exceptions import (
@ -42,6 +45,7 @@ if TYPE_CHECKING:
from ..conftest import PexpectSpawner
@no_macos
def test_shield_pause(
spawn: PexpectSpawner,
):
@ -57,6 +61,7 @@ def test_shield_pause(
expect(
child,
'Yo my child hanging..?',
timeout=3,
)
assert_before(
child,

107
tests/test_2way.py
View File

@ -1,7 +1,12 @@
"""
Bidirectional streaming.
'''
Audit the simplest inter-actor bidirectional (streaming)
msg patterns.
"""
'''
from __future__ import annotations
from typing import (
Callable,
)
import pytest
import trio
import tractor
@ -9,10 +14,8 @@ import tractor
@tractor.context
async def simple_rpc(
ctx: tractor.Context,
data: int,
) -> None:
'''
Test a small ping-pong server.
@ -39,15 +42,13 @@ 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.
"""
'''
Same as previous test but using `async for` syntax/api.
'''
# signal to parent that we're up
await ctx.started(data + 1)
@ -68,62 +69,78 @@ async def simple_rpc_with_forloop(
@pytest.mark.parametrize(
'use_async_for',
[True, False],
[
True,
False,
],
ids='use_async_for={}'.format,
)
@pytest.mark.parametrize(
'server_func',
[simple_rpc, simple_rpc_with_forloop],
[
simple_rpc,
simple_rpc_with_forloop,
],
ids='server_func={}'.format,
)
def test_simple_rpc(server_func, use_async_for):
def test_simple_rpc(
server_func: Callable,
use_async_for: bool,
loglevel: str,
debug_mode: bool,
):
'''
The simplest request response pattern.
'''
async def main():
async with tractor.open_nursery() as n:
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(
'rpc_server',
enable_modules=[__name__],
)
portal = await n.start_actor(
'rpc_server',
enable_modules=[__name__],
)
async with portal.open_context(
server_func, # taken from pytest parameterization
data=10,
) as (ctx, sent):
async with portal.open_context(
server_func, # taken from pytest parameterization
data=10,
) as (ctx, sent):
assert sent == 11
assert sent == 11
async with ctx.open_stream() as stream:
async with ctx.open_stream() as stream:
if use_async_for:
if use_async_for:
count = 0
# receive msgs using async for style
print('ping')
await stream.send('ping')
async for msg in stream:
assert msg == 'pong'
count = 0
# receive msgs using async for style
print('ping')
await stream.send('ping')
count += 1
if count >= 9:
break
async for msg in stream:
assert msg == 'pong'
print('ping')
await stream.send('ping')
count += 1
else:
# classic send/receive style
for _ in range(10):
if count >= 9:
break
print('ping')
await stream.send('ping')
assert await stream.receive() == 'pong'
else:
# classic send/receive style
for _ in range(10):
# stream should terminate here
print('ping')
await stream.send('ping')
assert await stream.receive() == 'pong'
# final context result(s) should be consumed here in __aexit__()
# stream should terminate here
await portal.cancel_actor()
# final context result(s) should be consumed here in __aexit__()
await portal.cancel_actor()
trio.run(main)

35
tests/test_cancellation.py
View File

@ -17,8 +17,8 @@ from tractor._testing import (
from .conftest import no_windows
def is_win():
return platform.system() == 'Windows'
_non_linux: bool = platform.system() != 'Linux'
_friggin_windows: bool = 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 is_win():
if _friggin_windows:
# 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 is_win():
if _friggin_windows:
if isinstance(subexc, tractor.RemoteActorError):
assert subexc.boxed_type in (
BaseExceptionGroup,
@ -507,17 +507,22 @@ def test_cancel_via_SIGINT(
@no_windows
def test_cancel_via_SIGINT_other_task(
loglevel,
start_method,
spawn_backend,
loglevel: str,
start_method: str,
spawn_backend: str,
):
"""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
'''
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
timeout += 1
async def spawn_and_sleep_forever(
@ -696,7 +701,7 @@ def test_fast_graceful_cancel_when_spawn_task_in_soft_proc_wait_for_daemon(
kbi_delay = 0.5
timeout: float = 2.9
if is_win(): # smh
if _friggin_windows: # smh
timeout += 1
async def main():

8
tests/test_context_stream_semantics.py
View File

@ -9,6 +9,7 @@ from itertools import count
import math
import platform
from pprint import pformat
import sys
from typing import (
Callable,
)
@ -941,6 +942,11 @@ 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,
@ -950,7 +956,7 @@ def test_one_end_stream_not_opened(
enable_modules=[__name__],
)
with trio.fail_after(1):
with trio.fail_after(timeout):
async with portal.open_context(
entrypoint,
) as (ctx, sent):

85
tests/test_discovery.py
View File

@ -1,11 +1,13 @@
"""
Actor "discovery" testing
Discovery subsys.
"""
import os
import signal
import platform
from functools import partial
import itertools
from typing import Callable
import psutil
import pytest
@ -17,7 +19,9 @@ import trio
@tractor_test
async def test_reg_then_unreg(reg_addr):
async def test_reg_then_unreg(
reg_addr: tuple,
):
actor = tractor.current_actor()
assert actor.is_arbiter
assert len(actor._registry) == 1 # only self is registered
@ -82,11 +86,15 @@ 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,
start_method,
reg_addr,
func: Callable,
start_method: str,
reg_addr: tuple,
):
'''
Root actor acting as the "director" and running one-shot-task-actors
@ -119,7 +127,10 @@ async def stream_forever():
await trio.sleep(0.01)
async def cancel(use_signal, delay=0):
async def cancel(
use_signal: bool,
delay: float = 0,
):
# hold on there sally
await trio.sleep(delay)
@ -132,13 +143,15 @@ async def cancel(use_signal, delay=0):
raise KeyboardInterrupt
async def stream_from(portal):
async def stream_from(portal: tractor.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):
async def unpack_reg(
actor_or_portal: tractor.Portal|tractor.Actor,
):
'''
Get and unpack a "registry" RPC request from the "arbiter" registry
system.
@ -173,7 +186,9 @@ async def spawn_and_check_registry(
registry_addrs=[reg_addr],
debug_mode=debug_mode,
):
async with tractor.get_registry(reg_addr) as portal:
async with tractor.get_registry(
addr=reg_addr,
) as portal:
# runtime needs to be up to call this
actor = tractor.current_actor()
@ -246,10 +261,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,
use_signal,
reg_addr,
with_streaming,
start_method: str,
use_signal: bool,
reg_addr: tuple,
with_streaming: bool,
):
'''
Verify that cancelling a nursery results in all subactors
@ -274,15 +289,17 @@ def test_subactors_unregister_on_cancel(
def test_subactors_unregister_on_cancel_remote_daemon(
daemon: subprocess.Popen,
debug_mode: bool,
start_method,
use_signal,
reg_addr,
with_streaming,
start_method: str,
use_signal: bool,
reg_addr: tuple,
with_streaming: bool,
):
"""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(
@ -374,14 +391,16 @@ async def close_chans_before_nursery(
@pytest.mark.parametrize('use_signal', [False, True])
def test_close_channel_explicit(
start_method,
use_signal,
reg_addr,
start_method: str,
use_signal: bool,
reg_addr: tuple,
):
"""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(
@ -396,14 +415,16 @@ def test_close_channel_explicit(
@pytest.mark.parametrize('use_signal', [False, True])
def test_close_channel_explicit_remote_arbiter(
daemon: subprocess.Popen,
start_method,
use_signal,
reg_addr,
start_method: str,
use_signal: bool,
reg_addr: tuple,
):
"""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(

41
tests/test_docs_examples.py
View File

@ -9,12 +9,17 @@ 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(
@ -101,8 +106,10 @@ def run_example_in_subproc(
ids=lambda t: t[1],
)
def test_example(
run_example_in_subproc,
example_script,
run_example_in_subproc: Callable,
example_script: str,
test_log: tractor.log.StackLevelAdapter,
ci_env: bool,
):
'''
Load and run scripts from this repo's ``examples/`` dir as a user
@ -116,9 +123,32 @@ 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()
@ -126,9 +156,12 @@ def test_example(
err = None
try:
if not proc.poll():
_, err = proc.communicate(timeout=15)
_, err = proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired as e:
test_log.exception(
f'Example failed to finish within {timeout}s ??\n'
)
proc.kill()
err = e.stderr

150
tests/test_legacy_one_way_streaming.py
View File

@ -1,9 +1,11 @@
"""
Streaming via async gen api
Streaming via the, now legacy, "async-gen API".
"""
import time
from functools import partial
import platform
from typing import Callable
import trio
import tractor
@ -19,7 +21,11 @@ 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):
@ -69,14 +75,14 @@ async def stream_from_single_subactor(
async with tractor.open_nursery(
registry_addrs=[reg_addr],
start_method=start_method,
) as nursery:
) as an:
async with tractor.find_actor('streamerd') as portals:
if not portals:
# no brokerd actor found
portal = await nursery.start_actor(
portal = await an.start_actor(
'streamerd',
enable_modules=[__name__],
)
@ -116,11 +122,22 @@ async def stream_from_single_subactor(
@pytest.mark.parametrize(
'stream_func', [async_gen_stream, context_stream]
'stream_func',
[
async_gen_stream,
context_stream,
],
ids='stream_func={}'.format
)
def test_stream_from_single_subactor(reg_addr, start_method, stream_func):
"""Verify streaming from a spawned async generator.
"""
def test_stream_from_single_subactor(
reg_addr: tuple,
start_method: str,
stream_func: Callable,
):
'''
Verify streaming from a spawned async generator.
'''
trio.run(
partial(
stream_from_single_subactor,
@ -132,10 +149,9 @@ def test_stream_from_single_subactor(reg_addr, start_method, stream_func):
# this is the first 2 actors, streamer_1 and streamer_2
async def stream_data(seed):
async def stream_data(seed: int):
for i in range(seed):
yield i
# trigger scheduler to simulate practical usage
@ -143,15 +159,17 @@ async def stream_data(seed):
# this is the third actor; the aggregator
async def aggregate(seed):
"""Ensure that the two streams we receive match but only stream
async def aggregate(seed: int):
'''
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 nursery:
'''
async with tractor.open_nursery() as an:
portals = []
for i in range(1, 3):
# fork point
portal = await nursery.start_actor(
portal = await an.start_actor(
name=f'streamer_{i}',
enable_modules=[__name__],
)
@ -164,7 +182,8 @@ async def aggregate(seed):
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:
@ -174,10 +193,14 @@ async def aggregate(seed):
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 n:
async with trio.open_nursery() as tn:
for portal in portals:
n.start_soon(push_to_chan, portal, send_chan.clone())
tn.start_soon(
push_to_chan,
portal,
send_chan.clone(),
)
# close this local task's reference to send side
await send_chan.aclose()
@ -194,20 +217,21 @@ async def aggregate(seed):
print("FINISHED ITERATING in aggregator")
await nursery.cancel()
await an.cancel()
print("WAITING on `ActorNursery` to finish")
print("AGGREGATOR COMPLETE!")
# 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 def a_quadruple_example() -> list[int]:
'''
Open the root-actor which is also a "registrar".
'''
async with tractor.open_nursery() as an:
seed = int(1e3)
pre_start = time.time()
portal = await nursery.start_actor(
portal = await an.start_actor(
name='aggregator',
enable_modules=[__name__],
)
@ -228,8 +252,14 @@ async def a_quadruple_example():
return result_stream
async def cancel_after(wait, reg_addr):
async with tractor.open_root_actor(registry_addrs=[reg_addr]):
async def cancel_after(
wait: float,
reg_addr: tuple,
) -> list[int]:
async with tractor.open_root_actor(
registry_addrs=[reg_addr],
):
with trio.move_on_after(wait):
return await a_quadruple_example()
@ -240,6 +270,10 @@ 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...
@ -247,16 +281,20 @@ def time_quad_ex(
'''
pytest.skip("Test is too flaky on mp in CI")
timeout = 7 if platform.system() in ('Windows', 'Darwin') else 4
timeout = 7 if non_linux else 4
start = time.time()
results = trio.run(cancel_after, timeout, reg_addr)
diff = time.time() - start
results: list[int] = trio.run(
cancel_after,
timeout,
reg_addr,
)
diff: float = time.time() - start
assert results
return results, diff
def test_a_quadruple_example(
time_quad_ex: tuple,
time_quad_ex: tuple[list[int], float],
ci_env: bool,
spawn_backend: str,
):
@ -264,13 +302,12 @@ 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 platform.system() in (
'Windows',
'Darwin',
)
6 if non_linux
else 3
)
assert diff < this_fast
@ -281,19 +318,33 @@ def test_a_quadruple_example(
list(map(lambda i: i/10, range(3, 9)))
)
def test_not_fast_enough_quad(
reg_addr, time_quad_ex, cancel_delay, ci_env, spawn_backend
reg_addr: tuple,
time_quad_ex: tuple[list[int], float],
cancel_delay: float,
ci_env: bool,
spawn_backend: str,
):
"""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 = platform.system()
if system in ('Windows', 'Darwin') and results is not None:
results = trio.run(
cancel_after,
delay,
reg_addr,
)
system: str = 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
@ -301,23 +352,24 @@ def test_not_fast_enough_quad(
@tractor_test
async def test_respawn_consumer_task(
reg_addr,
spawn_backend,
loglevel,
reg_addr: tuple,
spawn_backend: str,
loglevel: str,
):
"""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 n:
async with tractor.open_nursery() as an:
portal = await n.start_actor(
portal = await an.start_actor(
name='streamer',
enable_modules=[__name__]
)

10
tests/test_multi_program.py
View File

@ -35,6 +35,9 @@ if TYPE_CHECKING:
)
_non_linux: bool = platform.system() != 'Linux'
def test_abort_on_sigint(
daemon: subprocess.Popen,
):
@ -137,6 +140,7 @@ 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
@ -148,6 +152,12 @@ 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,

14
tests/test_shm.py
View File

@ -2,6 +2,7 @@
Shared mem primitives and APIs.
"""
import platform
import uuid
# import numpy
@ -53,7 +54,18 @@ def test_child_attaches_alot():
shm_key=shml.key,
) as (ctx, start_val),
):
assert start_val == key
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
)
await ctx.result()
await portal.cancel_actor()

57
tractor/_state.py
View File

@ -22,7 +22,6 @@ from __future__ import annotations
from contextvars import (
ContextVar,
)
import os
from pathlib import Path
from typing import (
Any,
@ -30,6 +29,7 @@ from typing import (
TYPE_CHECKING,
)
import platformdirs
from trio.lowlevel import current_task
if TYPE_CHECKING:
@ -172,23 +172,56 @@ 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 = 'tractor'
subdir: str|Path|None = None,
appname: str = 'tractor',
) -> Path:
'''
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.
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.
'''
rtdir: Path = _rtdir / subdir
if not rtdir.is_dir():
rtdir.mkdir()
return rtdir
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
def current_ipc_protos() -> list[str]:

7
tractor/devx/debug/_sigint.py
View File

@ -21,6 +21,7 @@ cancellation during REPL interaction.
'''
from __future__ import annotations
import platform
from typing import (
TYPE_CHECKING,
)
@ -49,6 +50,7 @@ if TYPE_CHECKING:
log = get_logger()
_is_macos: bool = platform.system() == 'Darwin'
_ctlc_ignore_header: str = (
'Ignoring SIGINT while debug REPL in use'
)
@ -300,6 +302,11 @@ 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'

181
tractor/ipc/_shm.py
View File

@ -23,6 +23,7 @@ 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,
@ -106,11 +107,12 @@ class NDToken(Struct, frozen=True):
This type is msg safe.
'''
shm_name: str # this servers as a "key" value
shm_name: str # actual OS-level name (may be shortened on macOS)
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
@ -124,6 +126,41 @@ 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):
@ -160,6 +197,50 @@ 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,
@ -171,12 +252,32 @@ 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=key,
shm_first_index_name=key + "_first",
shm_last_index_name=key + "_last",
shm_name=shm_name,
shm_first_index_name=shm_first,
shm_last_index_name=shm_last,
dtype_descr=tuple(np.dtype(dtype).descr),
size=size,
key=key, # store original key for lookup
)
@ -431,9 +532,17 @@ class ShmArray:
def destroy(self) -> None:
if _USE_POSIX:
# We manually unlink to bypass all the "resource tracker"
# nonsense meant for non-SC systems.
shm_unlink(self._shm.name)
# 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?'
)
self._first.destroy()
self._last.destroy()
@ -463,8 +572,16 @@ 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=key,
name=token.shm_name,
create=True,
size=a.nbytes
)
@ -476,12 +593,6 @@ 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(
@ -554,13 +665,23 @@ def attach_shm_ndarray(
'''
token = NDToken.from_msg(token)
key = token.shm_name
# Use original key for _known_tokens lookup,
# shm_name for OS calls
lookup_key = (
token.key if token.key
else token.shm_name
)
if key in _known_tokens:
assert NDToken.from_msg(_known_tokens[key]) == token, "WTF"
if lookup_key in _known_tokens:
assert (
NDToken.from_msg(
_known_tokens[lookup_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
@ -568,7 +689,7 @@ def attach_shm_ndarray(
for _ in range(3):
try:
shm = SharedMemory(
name=key,
name=token.shm_name,
create=False,
)
break
@ -614,10 +735,10 @@ def attach_shm_ndarray(
sha.array
# Stash key -> token knowledge for future queries
# via `maybe_opepn_shm_array()` but only after we know
# we can attach.
if key not in _known_tokens:
_known_tokens[key] = token
# 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
# "close" attached shm on actor teardown
tractor.current_actor().lifetime_stack.callback(sha.close)
@ -661,7 +782,10 @@ def maybe_open_shm_ndarray(
False, # not newly opened
)
except KeyError:
log.warning(f"Could not find {key} in shms cache")
log.warning(
f'Could not find key in shms cache,\n'
f'key: {key!r}\n'
)
if dtype:
token = _make_token(
key,
@ -771,6 +895,7 @@ def open_shm_list(
size: int = int(2 ** 10),
dtype: float | int | bool | str | bytes | None = float,
readonly: bool = True,
prefix: str = 'shml_',
) -> ShmList:
@ -784,6 +909,12 @@ 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,

131
tractor/ipc/_uds.py
View File

@ -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 .._state import (
from tractor._state import (
get_rt_dir,
current_actor,
is_root_process,
@ -63,6 +63,28 @@ 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()
@ -165,7 +187,11 @@ class UDSAddress(
err_on_no_runtime=False,
)
if actor:
sockname: str = '::'.join(actor.uid) + f'@{pid}'
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}'
#
# ?^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}'
@ -292,7 +318,12 @@ 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
@ -310,21 +341,66 @@ 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)
sock.setsockopt(SOL_SOCKET, SO_PASSCRED, 1)
# 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_info(sock: trio.socket.socket) -> tuple[
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[
int, # pid
int, # uid
int, # guid
]:
'''
Deliver the connecting peer's "credentials"-info as defined in
a very Linux specific way..
a platform-specific way.
Linux-ONLY, uses SO_PEERCRED.
For more deats see,
- `man accept`,
@ -337,6 +413,11 @@ def get_peer_info(sock: trio.socket.socket) -> tuple[
- 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,
@ -440,13 +521,37 @@ class MsgpackUDSStream(MsgpackTransport):
match (peername, sockname):
case (str(), bytes()):
sock_path: Path = Path(peername)
case (bytes(), str()):
sock_path: Path = Path(sockname)
(
peer_pid,
_,
_,
) = get_peer_info(sock)
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(

11
uv.lock
View File

@ -208,6 +208,15 @@ 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"
@ -376,6 +385,7 @@ dependencies = [
{ name = "colorlog" },
{ name = "msgspec" },
{ name = "pdbp" },
{ name = "platformdirs" },
{ name = "tricycle" },
{ name = "trio" },
{ name = "wrapt" },
@ -419,6 +429,7 @@ 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" },