forked from goodboy/tractor
Compare commits
22 Commits
switch_to_
...
master
Author | SHA1 | Date |
---|---|---|
goodboy | e5ee2e3de8 | |
Tyler Goodlet | 41aa91c8eb | |
Tyler Goodlet | 6758e4487c | |
Tyler Goodlet | 1c3893a383 | |
Tyler Goodlet | 73befac9bc | |
Tyler Goodlet | 79622bbeea | |
Tyler Goodlet | 95535b2226 | |
Tyler Goodlet | 87c6e09d6b | |
Tyler Goodlet | 9ccd3a74b6 | |
Tyler Goodlet | ae4ff5dc8d | |
Tyler Goodlet | 705538398f | |
Tyler Goodlet | 86aef5238d | |
Tyler Goodlet | cc82447db6 | |
Tyler Goodlet | 23cffbd940 | |
Tyler Goodlet | 3d202272c4 | |
Tyler Goodlet | 63cdb0891f | |
goodboy | 0f7db27b68 | |
Tyler Goodlet | c53d62d2f7 | |
Tyler Goodlet | f667d16d66 | |
Tyler Goodlet | 24a062341e | |
goodboy | e714bec8db | |
Igor | 009cd6552e |
|
@ -6,8 +6,14 @@
|
||||||
``tractor`` is a `structured concurrent`_, multi-processing_ runtime
|
``tractor`` is a `structured concurrent`_, multi-processing_ runtime
|
||||||
built on trio_.
|
built on trio_.
|
||||||
|
|
||||||
Fundamentally ``tractor`` gives you parallelism via ``trio``-"*actors*":
|
Fundamentally, ``tractor`` gives you parallelism via
|
||||||
our nurseries_ let you spawn new Python processes which each run a ``trio``
|
``trio``-"*actors*": independent Python processes (aka
|
||||||
|
non-shared-memory threads) which maintain structured
|
||||||
|
concurrency (SC) *end-to-end* inside a *supervision tree*.
|
||||||
|
|
||||||
|
Cross-process (and thus cross-host) SC is accomplished through the
|
||||||
|
combined use of our "actor nurseries_" and an "SC-transitive IPC
|
||||||
|
protocol" constructed on top of multiple Pythons each running a ``trio``
|
||||||
scheduled runtime - a call to ``trio.run()``.
|
scheduled runtime - a call to ``trio.run()``.
|
||||||
|
|
||||||
We believe the system adheres to the `3 axioms`_ of an "`actor model`_"
|
We believe the system adheres to the `3 axioms`_ of an "`actor model`_"
|
||||||
|
@ -23,7 +29,8 @@ Features
|
||||||
- **It's just** a ``trio`` API
|
- **It's just** a ``trio`` API
|
||||||
- *Infinitely nesteable* process trees
|
- *Infinitely nesteable* process trees
|
||||||
- Builtin IPC streaming APIs with task fan-out broadcasting
|
- Builtin IPC streaming APIs with task fan-out broadcasting
|
||||||
- A (first ever?) "native" multi-core debugger UX for Python using `pdb++`_
|
- A "native" multi-core debugger REPL using `pdbp`_ (a fork & fix of
|
||||||
|
`pdb++`_ thanks to @mdmintz!)
|
||||||
- Support for a swappable, OS specific, process spawning layer
|
- Support for a swappable, OS specific, process spawning layer
|
||||||
- A modular transport stack, allowing for custom serialization (eg. with
|
- A modular transport stack, allowing for custom serialization (eg. with
|
||||||
`msgspec`_), communications protocols, and environment specific IPC
|
`msgspec`_), communications protocols, and environment specific IPC
|
||||||
|
@ -149,7 +156,7 @@ it **is a bug**.
|
||||||
|
|
||||||
"Native" multi-process debugging
|
"Native" multi-process debugging
|
||||||
--------------------------------
|
--------------------------------
|
||||||
Using the magic of `pdb++`_ and our internal IPC, we've
|
Using the magic of `pdbp`_ and our internal IPC, we've
|
||||||
been able to create a native feeling debugging experience for
|
been able to create a native feeling debugging experience for
|
||||||
any (sub-)process in your ``tractor`` tree.
|
any (sub-)process in your ``tractor`` tree.
|
||||||
|
|
||||||
|
@ -597,6 +604,7 @@ channel`_!
|
||||||
.. _adherance to: https://www.youtube.com/watch?v=7erJ1DV_Tlo&t=1821s
|
.. _adherance to: https://www.youtube.com/watch?v=7erJ1DV_Tlo&t=1821s
|
||||||
.. _trio gitter channel: https://gitter.im/python-trio/general
|
.. _trio gitter channel: https://gitter.im/python-trio/general
|
||||||
.. _matrix channel: https://matrix.to/#/!tractor:matrix.org
|
.. _matrix channel: https://matrix.to/#/!tractor:matrix.org
|
||||||
|
.. _pdbp: https://github.com/mdmintz/pdbp
|
||||||
.. _pdb++: https://github.com/pdbpp/pdbpp
|
.. _pdb++: https://github.com/pdbpp/pdbpp
|
||||||
.. _guest mode: https://trio.readthedocs.io/en/stable/reference-lowlevel.html?highlight=guest%20mode#using-guest-mode-to-run-trio-on-top-of-other-event-loops
|
.. _guest mode: https://trio.readthedocs.io/en/stable/reference-lowlevel.html?highlight=guest%20mode#using-guest-mode-to-run-trio-on-top-of-other-event-loops
|
||||||
.. _messages: https://en.wikipedia.org/wiki/Message_passing
|
.. _messages: https://en.wikipedia.org/wiki/Message_passing
|
||||||
|
|
|
@ -0,0 +1,24 @@
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
|
||||||
|
import trio
|
||||||
|
import tractor
|
||||||
|
|
||||||
|
|
||||||
|
async def main() -> None:
|
||||||
|
async with tractor.open_nursery(debug_mode=True) as an:
|
||||||
|
|
||||||
|
assert os.environ['PYTHONBREAKPOINT'] == 'tractor._debug._set_trace'
|
||||||
|
|
||||||
|
# TODO: an assert that verifies the hook has indeed been, hooked
|
||||||
|
# XD
|
||||||
|
assert sys.breakpointhook is not tractor._debug._set_trace
|
||||||
|
|
||||||
|
breakpoint()
|
||||||
|
|
||||||
|
# TODO: an assert that verifies the hook is unhooked..
|
||||||
|
assert sys.breakpointhook
|
||||||
|
breakpoint()
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
trio.run(main)
|
|
@ -0,0 +1,7 @@
|
||||||
|
Drop `trio.Process.aclose()` usage, copy into our spawning code.
|
||||||
|
|
||||||
|
The details are laid out in https://github.com/goodboy/tractor/issues/330.
|
||||||
|
`trio` changed is process running quite some time ago, this just copies
|
||||||
|
out the small bit we needed (from the old `.aclose()`) for hard kills
|
||||||
|
where a soft runtime cancel request fails and our "zombie killer"
|
||||||
|
implementation kicks in.
|
|
@ -0,0 +1,15 @@
|
||||||
|
Switch to using the fork & fix of `pdb++`, `pdbp`:
|
||||||
|
https://github.com/mdmintz/pdbp
|
||||||
|
|
||||||
|
Allows us to sidestep a variety of issues that aren't being maintained
|
||||||
|
in the upstream project thanks to the hard work of @mdmintz!
|
||||||
|
|
||||||
|
We also include some default settings adjustments as per recent
|
||||||
|
development on the fork:
|
||||||
|
|
||||||
|
- sticky mode is still turned on by default but now activates when
|
||||||
|
a using the `ll` repl command.
|
||||||
|
- turn off line truncation by default to avoid inter-line gaps when
|
||||||
|
resizing the terimnal during use.
|
||||||
|
- when using the backtrace cmd either by `w` or `bt`, the config
|
||||||
|
automatically switches to non-sticky mode.
|
|
@ -1,7 +1,7 @@
|
||||||
pytest
|
pytest
|
||||||
pytest-trio
|
pytest-trio
|
||||||
pytest-timeout
|
pytest-timeout
|
||||||
pdbpp
|
pdbp
|
||||||
mypy
|
mypy
|
||||||
trio_typing
|
trio_typing
|
||||||
pexpect
|
pexpect
|
||||||
|
|
|
@ -151,19 +151,6 @@ def ctlc(
|
||||||
|
|
||||||
use_ctlc = request.param
|
use_ctlc = request.param
|
||||||
|
|
||||||
# TODO: we can remove this bc pdbp right?
|
|
||||||
if (
|
|
||||||
sys.version_info <= (3, 10)
|
|
||||||
and use_ctlc
|
|
||||||
):
|
|
||||||
# on 3.9 it seems the REPL UX
|
|
||||||
# is highly unreliable and frankly annoying
|
|
||||||
# to test for. It does work from manual testing
|
|
||||||
# but i just don't think it's wroth it to try
|
|
||||||
# and get this working especially since we want to
|
|
||||||
# be 3.10+ mega-asap.
|
|
||||||
pytest.skip('Py3.9 and `pdbpp` son no bueno..')
|
|
||||||
|
|
||||||
node = request.node
|
node = request.node
|
||||||
markers = node.own_markers
|
markers = node.own_markers
|
||||||
for mark in markers:
|
for mark in markers:
|
||||||
|
@ -194,13 +181,15 @@ def ctlc(
|
||||||
ids=lambda item: f'{item[0]} -> {item[1]}',
|
ids=lambda item: f'{item[0]} -> {item[1]}',
|
||||||
)
|
)
|
||||||
def test_root_actor_error(spawn, user_in_out):
|
def test_root_actor_error(spawn, user_in_out):
|
||||||
"""Demonstrate crash handler entering pdbpp from basic error in root actor.
|
'''
|
||||||
"""
|
Demonstrate crash handler entering pdb from basic error in root actor.
|
||||||
|
|
||||||
|
'''
|
||||||
user_input, expect_err_str = user_in_out
|
user_input, expect_err_str = user_in_out
|
||||||
|
|
||||||
child = spawn('root_actor_error')
|
child = spawn('root_actor_error')
|
||||||
|
|
||||||
# scan for the pdbpp prompt
|
# scan for the prompt
|
||||||
expect(child, PROMPT)
|
expect(child, PROMPT)
|
||||||
|
|
||||||
before = str(child.before.decode())
|
before = str(child.before.decode())
|
||||||
|
@ -231,7 +220,7 @@ def test_root_actor_bp(spawn, user_in_out):
|
||||||
user_input, expect_err_str = user_in_out
|
user_input, expect_err_str = user_in_out
|
||||||
child = spawn('root_actor_breakpoint')
|
child = spawn('root_actor_breakpoint')
|
||||||
|
|
||||||
# scan for the pdbpp prompt
|
# scan for the prompt
|
||||||
child.expect(PROMPT)
|
child.expect(PROMPT)
|
||||||
|
|
||||||
assert 'Error' not in str(child.before)
|
assert 'Error' not in str(child.before)
|
||||||
|
@ -339,7 +328,7 @@ def test_subactor_error(
|
||||||
'''
|
'''
|
||||||
child = spawn('subactor_error')
|
child = spawn('subactor_error')
|
||||||
|
|
||||||
# scan for the pdbpp prompt
|
# scan for the prompt
|
||||||
child.expect(PROMPT)
|
child.expect(PROMPT)
|
||||||
|
|
||||||
before = str(child.before.decode())
|
before = str(child.before.decode())
|
||||||
|
@ -387,7 +376,7 @@ def test_subactor_breakpoint(
|
||||||
|
|
||||||
child = spawn('subactor_breakpoint')
|
child = spawn('subactor_breakpoint')
|
||||||
|
|
||||||
# scan for the pdbpp prompt
|
# scan for the prompt
|
||||||
child.expect(PROMPT)
|
child.expect(PROMPT)
|
||||||
|
|
||||||
before = str(child.before.decode())
|
before = str(child.before.decode())
|
||||||
|
@ -448,7 +437,7 @@ def test_multi_subactors(
|
||||||
'''
|
'''
|
||||||
child = spawn(r'multi_subactors')
|
child = spawn(r'multi_subactors')
|
||||||
|
|
||||||
# scan for the pdbpp prompt
|
# scan for the prompt
|
||||||
child.expect(PROMPT)
|
child.expect(PROMPT)
|
||||||
|
|
||||||
before = str(child.before.decode())
|
before = str(child.before.decode())
|
||||||
|
@ -688,7 +677,7 @@ def test_multi_subactors_root_errors(
|
||||||
'''
|
'''
|
||||||
child = spawn('multi_subactor_root_errors')
|
child = spawn('multi_subactor_root_errors')
|
||||||
|
|
||||||
# scan for the pdbpp prompt
|
# scan for the prompt
|
||||||
child.expect(PROMPT)
|
child.expect(PROMPT)
|
||||||
|
|
||||||
# at most one subactor should attach before the root is cancelled
|
# at most one subactor should attach before the root is cancelled
|
||||||
|
|
|
@ -54,18 +54,6 @@ from ._exceptions import (
|
||||||
)
|
)
|
||||||
from ._ipc import Channel
|
from ._ipc import Channel
|
||||||
|
|
||||||
|
|
||||||
# TODO: we can drop this now yah?
|
|
||||||
# try:
|
|
||||||
# # wtf: only exported when installed in dev mode?
|
|
||||||
# import pdbp
|
|
||||||
# except ImportError:
|
|
||||||
# # pdbpp is installed in regular mode...it monkey patches stuff
|
|
||||||
# import pdb
|
|
||||||
# xpm = getattr(pdb, 'xpm', None)
|
|
||||||
# assert xpm, "pdbpp is not installed?" # type: ignore
|
|
||||||
# pdbpp = pdb
|
|
||||||
|
|
||||||
log = get_logger(__name__)
|
log = get_logger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
@ -754,8 +742,8 @@ def shield_sigint_handler(
|
||||||
|
|
||||||
|
|
||||||
def _set_trace(
|
def _set_trace(
|
||||||
actor: Optional[tractor.Actor] = None,
|
actor: tractor.Actor | None = None,
|
||||||
pdb: Optional[MultiActorPdb] = None,
|
pdb: MultiActorPdb | None = None,
|
||||||
):
|
):
|
||||||
__tracebackhide__ = True
|
__tracebackhide__ = True
|
||||||
actor = actor or tractor.current_actor()
|
actor = actor or tractor.current_actor()
|
||||||
|
@ -765,7 +753,11 @@ def _set_trace(
|
||||||
if frame:
|
if frame:
|
||||||
frame = frame.f_back # type: ignore
|
frame = frame.f_back # type: ignore
|
||||||
|
|
||||||
if frame and pdb and actor is not None:
|
if (
|
||||||
|
frame
|
||||||
|
and pdb
|
||||||
|
and actor is not None
|
||||||
|
):
|
||||||
log.pdb(f"\nAttaching pdb to actor: {actor.uid}\n")
|
log.pdb(f"\nAttaching pdb to actor: {actor.uid}\n")
|
||||||
# no f!#$&* idea, but when we're in async land
|
# no f!#$&* idea, but when we're in async land
|
||||||
# we need 2x frames up?
|
# we need 2x frames up?
|
||||||
|
@ -774,7 +766,8 @@ def _set_trace(
|
||||||
else:
|
else:
|
||||||
pdb, undo_sigint = mk_mpdb()
|
pdb, undo_sigint = mk_mpdb()
|
||||||
|
|
||||||
# we entered the global ``breakpoint()`` built-in from sync code?
|
# we entered the global ``breakpoint()`` built-in from sync
|
||||||
|
# code?
|
||||||
Lock.local_task_in_debug = 'sync'
|
Lock.local_task_in_debug = 'sync'
|
||||||
|
|
||||||
pdb.set_trace(frame=frame)
|
pdb.set_trace(frame=frame)
|
||||||
|
|
|
@ -22,8 +22,9 @@ from contextlib import asynccontextmanager
|
||||||
from functools import partial
|
from functools import partial
|
||||||
import importlib
|
import importlib
|
||||||
import logging
|
import logging
|
||||||
import os
|
|
||||||
import signal
|
import signal
|
||||||
|
import sys
|
||||||
|
import os
|
||||||
import typing
|
import typing
|
||||||
import warnings
|
import warnings
|
||||||
|
|
||||||
|
@ -84,8 +85,10 @@ async def open_root_actor(
|
||||||
|
|
||||||
'''
|
'''
|
||||||
# Override the global debugger hook to make it play nice with
|
# Override the global debugger hook to make it play nice with
|
||||||
# ``trio``, see:
|
# ``trio``, see much discussion in:
|
||||||
# https://github.com/python-trio/trio/issues/1155#issuecomment-742964018
|
# https://github.com/python-trio/trio/issues/1155#issuecomment-742964018
|
||||||
|
builtin_bp_handler = sys.breakpointhook
|
||||||
|
orig_bp_path: str | None = os.environ.get('PYTHONBREAKPOINT', None)
|
||||||
os.environ['PYTHONBREAKPOINT'] = 'tractor._debug._set_trace'
|
os.environ['PYTHONBREAKPOINT'] = 'tractor._debug._set_trace'
|
||||||
|
|
||||||
# attempt to retreive ``trio``'s sigint handler and stash it
|
# attempt to retreive ``trio``'s sigint handler and stash it
|
||||||
|
@ -254,6 +257,15 @@ async def open_root_actor(
|
||||||
await actor.cancel()
|
await actor.cancel()
|
||||||
finally:
|
finally:
|
||||||
_state._current_actor = None
|
_state._current_actor = None
|
||||||
|
|
||||||
|
# restore breakpoint hook state
|
||||||
|
sys.breakpointhook = builtin_bp_handler
|
||||||
|
if orig_bp_path is not None:
|
||||||
|
os.environ['PYTHONBREAKPOINT'] = orig_bp_path
|
||||||
|
else:
|
||||||
|
# clear env back to having no entry
|
||||||
|
os.environ.pop('PYTHONBREAKPOINT')
|
||||||
|
|
||||||
logger.runtime("Root actor terminated")
|
logger.runtime("Root actor terminated")
|
||||||
|
|
||||||
|
|
||||||
|
@ -289,7 +301,7 @@ def run_daemon(
|
||||||
async def _main():
|
async def _main():
|
||||||
|
|
||||||
async with open_root_actor(
|
async with open_root_actor(
|
||||||
arbiter_addr=registry_addr,
|
registry_addr=registry_addr,
|
||||||
name=name,
|
name=name,
|
||||||
start_method=start_method,
|
start_method=start_method,
|
||||||
debug_mode=debug_mode,
|
debug_mode=debug_mode,
|
||||||
|
|
|
@ -23,13 +23,12 @@ import sys
|
||||||
import platform
|
import platform
|
||||||
from typing import (
|
from typing import (
|
||||||
Any,
|
Any,
|
||||||
|
Awaitable,
|
||||||
Literal,
|
Literal,
|
||||||
Optional,
|
|
||||||
Callable,
|
Callable,
|
||||||
TypeVar,
|
TypeVar,
|
||||||
TYPE_CHECKING,
|
TYPE_CHECKING,
|
||||||
)
|
)
|
||||||
from collections.abc import Awaitable
|
|
||||||
|
|
||||||
from exceptiongroup import BaseExceptionGroup
|
from exceptiongroup import BaseExceptionGroup
|
||||||
import trio
|
import trio
|
||||||
|
@ -60,7 +59,7 @@ if TYPE_CHECKING:
|
||||||
log = get_logger('tractor')
|
log = get_logger('tractor')
|
||||||
|
|
||||||
# placeholder for an mp start context if so using that backend
|
# placeholder for an mp start context if so using that backend
|
||||||
_ctx: Optional[mp.context.BaseContext] = None
|
_ctx: mp.context.BaseContext | None = None
|
||||||
SpawnMethodKey = Literal[
|
SpawnMethodKey = Literal[
|
||||||
'trio', # supported on all platforms
|
'trio', # supported on all platforms
|
||||||
'mp_spawn',
|
'mp_spawn',
|
||||||
|
@ -86,7 +85,7 @@ else:
|
||||||
def try_set_start_method(
|
def try_set_start_method(
|
||||||
key: SpawnMethodKey
|
key: SpawnMethodKey
|
||||||
|
|
||||||
) -> Optional[mp.context.BaseContext]:
|
) -> mp.context.BaseContext | None:
|
||||||
'''
|
'''
|
||||||
Attempt to set the method for process starting, aka the "actor
|
Attempt to set the method for process starting, aka the "actor
|
||||||
spawning backend".
|
spawning backend".
|
||||||
|
@ -200,16 +199,37 @@ async def cancel_on_completion(
|
||||||
async def do_hard_kill(
|
async def do_hard_kill(
|
||||||
proc: trio.Process,
|
proc: trio.Process,
|
||||||
terminate_after: int = 3,
|
terminate_after: int = 3,
|
||||||
|
|
||||||
) -> None:
|
) -> None:
|
||||||
# NOTE: this timeout used to do nothing since we were shielding
|
# NOTE: this timeout used to do nothing since we were shielding
|
||||||
# the ``.wait()`` inside ``new_proc()`` which will pretty much
|
# the ``.wait()`` inside ``new_proc()`` which will pretty much
|
||||||
# never release until the process exits, now it acts as
|
# never release until the process exits, now it acts as
|
||||||
# a hard-kill time ultimatum.
|
# a hard-kill time ultimatum.
|
||||||
|
log.debug(f"Terminating {proc}")
|
||||||
with trio.move_on_after(terminate_after) as cs:
|
with trio.move_on_after(terminate_after) as cs:
|
||||||
|
|
||||||
# NOTE: This ``__aexit__()`` shields internally.
|
# NOTE: code below was copied verbatim from the now deprecated
|
||||||
async with proc: # calls ``trio.Process.aclose()``
|
# (in 0.20.0) ``trio._subrocess.Process.aclose()``, orig doc
|
||||||
log.debug(f"Terminating {proc}")
|
# string:
|
||||||
|
#
|
||||||
|
# Close any pipes we have to the process (both input and output)
|
||||||
|
# and wait for it to exit. If cancelled, kills the process and
|
||||||
|
# waits for it to finish exiting before propagating the
|
||||||
|
# cancellation.
|
||||||
|
with trio.CancelScope(shield=True):
|
||||||
|
if proc.stdin is not None:
|
||||||
|
await proc.stdin.aclose()
|
||||||
|
if proc.stdout is not None:
|
||||||
|
await proc.stdout.aclose()
|
||||||
|
if proc.stderr is not None:
|
||||||
|
await proc.stderr.aclose()
|
||||||
|
try:
|
||||||
|
await proc.wait()
|
||||||
|
finally:
|
||||||
|
if proc.returncode is None:
|
||||||
|
proc.kill()
|
||||||
|
with trio.CancelScope(shield=True):
|
||||||
|
await proc.wait()
|
||||||
|
|
||||||
if cs.cancelled_caught:
|
if cs.cancelled_caught:
|
||||||
# XXX: should pretty much never get here unless we have
|
# XXX: should pretty much never get here unless we have
|
||||||
|
@ -355,12 +375,11 @@ async def trio_proc(
|
||||||
spawn_cmd.append("--asyncio")
|
spawn_cmd.append("--asyncio")
|
||||||
|
|
||||||
cancelled_during_spawn: bool = False
|
cancelled_during_spawn: bool = False
|
||||||
proc: Optional[trio.Process] = None
|
proc: trio.Process | None = None
|
||||||
try:
|
try:
|
||||||
try:
|
try:
|
||||||
# TODO: needs ``trio_typing`` patch?
|
# TODO: needs ``trio_typing`` patch?
|
||||||
proc = await trio.lowlevel.open_process( # type: ignore
|
proc = await trio.lowlevel.open_process(spawn_cmd)
|
||||||
spawn_cmd)
|
|
||||||
|
|
||||||
log.runtime(f"Started {proc}")
|
log.runtime(f"Started {proc}")
|
||||||
|
|
||||||
|
@ -444,8 +463,8 @@ async def trio_proc(
|
||||||
nursery.cancel_scope.cancel()
|
nursery.cancel_scope.cancel()
|
||||||
|
|
||||||
finally:
|
finally:
|
||||||
# The "hard" reap since no actor zombies are allowed!
|
# XXX NOTE XXX: The "hard" reap since no actor zombies are
|
||||||
# XXX: do this **after** cancellation/tearfown to avoid
|
# allowed! Do this **after** cancellation/teardown to avoid
|
||||||
# killing the process too early.
|
# killing the process too early.
|
||||||
if proc:
|
if proc:
|
||||||
log.cancel(f'Hard reap sequence starting for {subactor.uid}')
|
log.cancel(f'Hard reap sequence starting for {subactor.uid}')
|
||||||
|
|
Loading…
Reference in New Issue