.travis.yml

@@ -25,16 +25,16 @@ matrix:
         - name: "Python 3.7: multiprocessing"
           python: 3.7  # this works for Linux but is ignored on macOS or Windows
           env: SPAWN_BACKEND="mp"
-        - name: "Python 3.7: trio"
+        - name: "Python 3.7: trio-run-in-process"
           python: 3.7  # this works for Linux but is ignored on macOS or Windows
-          env: SPAWN_BACKEND="trio"
+          env: SPAWN_BACKEND="trio_run_in_process"
 
         - name: "Python 3.8: multiprocessing"
           python: 3.8  # this works for Linux but is ignored on macOS or Windows
           env: SPAWN_BACKEND="mp"
-        - name: "Python 3.8: trio"
+        - name: "Python 3.8: trio-run-in-process"
           python: 3.8  # this works for Linux but is ignored on macOS or Windows
-          env: SPAWN_BACKEND="trio"
+          env: SPAWN_BACKEND="trio_run_in_process"
 
 install:
     - cd $TRAVIS_BUILD_DIR

setup.py

@@ -39,7 +39,7 @@ setup(
     ],
     install_requires=[
         'msgpack', 'trio>0.8', 'async_generator', 'colorlog', 'wrapt',
-        'trio_typing', 'cloudpickle',
+        'trio_typing', 'trio-run-in-process',
     ],
     tests_require=['pytest'],
     python_requires=">=3.7",

tests/conftest.py

@@ -21,7 +21,7 @@ def pytest_addoption(parser):
 
     parser.addoption(
         "--spawn-backend", action="store", dest='spawn_backend',
-        default='trio',
+        default='trio_run_in_process',
         help="Processing spawning backend to use for test run",
     )
 
@@ -34,7 +34,7 @@ def pytest_configure(config):
 
     if backend == 'mp':
         tractor._spawn.try_set_start_method('spawn')
-    elif backend == 'trio':
+    elif backend == 'trio_run_in_process':
         tractor._spawn.try_set_start_method(backend)
 
 
@@ -56,7 +56,7 @@ def pytest_generate_tests(metafunc):
     if not spawn_backend:
         # XXX some weird windows bug with `pytest`?
         spawn_backend = 'mp'
-    assert spawn_backend in ('mp', 'trio')
+    assert spawn_backend in ('mp', 'trio_run_in_process')
 
     if 'start_method' in metafunc.fixturenames:
         if spawn_backend == 'mp':
@@ -67,11 +67,11 @@ def pytest_generate_tests(metafunc):
                 # removing XXX: the fork method is in general
                 # incompatible with trio's global scheduler state
                 methods.remove('fork')
-        elif spawn_backend == 'trio':
+        elif spawn_backend == 'trio_run_in_process':
             if platform.system() == "Windows":
                 pytest.fail(
                     "Only `--spawn-backend=mp` is supported on Windows")
 
-            methods = ['trio']
+            methods = ['trio_run_in_process']
 
         metafunc.parametrize("start_method", methods, scope='module')

tests/test_cancellation.py

@@ -197,7 +197,7 @@ async def test_cancel_infinite_streamer(start_method):
     ],
 )
 @tractor_test
-async def test_some_cancels_all(num_actors_and_errs, start_method, loglevel):
+async def test_some_cancels_all(num_actors_and_errs, start_method):
     """Verify a subset of failed subactors causes all others in
     the nursery to be cancelled just like the strategy in trio.
 

tractor/_actor.py

@@ -179,9 +179,6 @@ class Actor:
     # Information about `__main__` from parent
     _parent_main_data: Dict[str, str]
 
-    # if started on ``asycio`` running ``trio`` in guest mode
-    _infected_aio: bool = False
-
     def __init__(
         self,
         name: str,
@@ -259,7 +256,7 @@ class Actor:
         code (if it exists).
         """
         try:
-            if self._spawn_method == 'trio':
+            if self._spawn_method == 'trio_run_in_process':
                 parent_data = self._parent_main_data
                 if 'init_main_from_name' in parent_data:
                     _mp_fixup_main._fixup_main_from_name(
@@ -542,6 +539,58 @@ class Actor:
                 f"Exiting msg loop for {chan} from {chan.uid} "
                 f"with last msg:\n{msg}")
 
+    def _mp_main(
+        self,
+        accept_addr: Tuple[str, int],
+        forkserver_info: Tuple[Any, Any, Any, Any, Any],
+        start_method: str,
+        parent_addr: Tuple[str, int] = None
+    ) -> None:
+        """The routine called *after fork* which invokes a fresh ``trio.run``
+        """
+        self._forkserver_info = forkserver_info
+        from ._spawn import try_set_start_method
+        spawn_ctx = try_set_start_method(start_method)
+
+        if self.loglevel is not None:
+            log.info(
+                f"Setting loglevel for {self.uid} to {self.loglevel}")
+            get_console_log(self.loglevel)
+
+        assert spawn_ctx
+        log.info(
+            f"Started new {spawn_ctx.current_process()} for {self.uid}")
+
+        _state._current_actor = self
+
+        log.debug(f"parent_addr is {parent_addr}")
+        try:
+            trio.run(partial(
+                self._async_main, accept_addr, parent_addr=parent_addr))
+        except KeyboardInterrupt:
+            pass  # handle it the same way trio does?
+        log.info(f"Actor {self.uid} terminated")
+
+    async def _trip_main(
+        self,
+        accept_addr: Tuple[str, int],
+        parent_addr: Tuple[str, int] = None
+    ) -> None:
+        """Entry point for a `trio_run_in_process` subactor.
+
+        Here we don't need to call `trio.run()` since trip does that as
+        part of its subprocess startup sequence.
+        """
+        if self.loglevel is not None:
+            log.info(
+                f"Setting loglevel for {self.uid} to {self.loglevel}")
+            get_console_log(self.loglevel)
+
+        log.info(f"Started new TRIP process for {self.uid}")
+        _state._current_actor = self
+        await self._async_main(accept_addr, parent_addr=parent_addr)
+        log.info(f"Actor {self.uid} terminated")
+
     async def _async_main(
         self,
         accept_addr: Tuple[str, int],
@@ -797,8 +846,6 @@ class Actor:
         log.info(f"Handshake with actor {uid}@{chan.raddr} complete")
         return uid
 
-    def is_infected_aio(self) -> bool:
-        return self._infected_aio
 
 class Arbiter(Actor):
     """A special actor who knows all the other actors and always has

tractor/_child.py

@@ -1,13 +0,0 @@
-import sys
-import trio
-import cloudpickle
-
-if __name__ == "__main__":
-    job = cloudpickle.load(sys.stdin.detach())
-
-    try:
-        result = trio.run(job)
-        cloudpickle.dump(sys.stdout.detach(), result)
-
-    except BaseException as err:
-        cloudpickle.dump(sys.stdout.detach(), err)

tractor/_entry.py

@@ -1,20 +1,60 @@
 """
 Process entry points.
 """
+import asyncio
 from functools import partial
-from typing import Tuple, Any
+from typing import Tuple, Any, Awaitable
 
 import trio  # type: ignore
 
 from ._actor import Actor
 from .log import get_console_log, get_logger
 from . import _state
-from .to_asyncio import run_as_asyncio_guest
 
 
 log = get_logger(__name__)
 
 
+def _asyncio_main(
+    trio_main: Awaitable,
+) -> None:
+    """Entry for an "infected ``asyncio`` actor".
+
+    Uh, oh. :o
+
+    It looks like your event loop has caught a case of the ``trio``s.
+
+    :()
+
+    Don't worry, we've heard you'll barely notice. You might hallucinate
+    a few more propagating errors and feel like your digestion has
+    slowed but if anything get's too bad your parents will know about
+    it.
+
+    :)
+    """
+    async def aio_main(trio_main):
+        loop = asyncio.get_running_loop()
+
+        trio_done_fut = asyncio.Future()
+
+        def trio_done_callback(main_outcome):
+            log.info(f"trio_main finished: {main_outcome!r}")
+            trio_done_fut.set_result(main_outcome)
+
+        # start the infection: run trio on the asyncio loop in "guest mode"
+        log.info(f"Infecting asyncio process with {trio_main}")
+        trio.lowlevel.start_guest_run(
+            trio_main,
+            run_sync_soon_threadsafe=loop.call_soon_threadsafe,
+            done_callback=trio_done_callback,
+        )
+
+        (await trio_done_fut).unwrap()
+
+    asyncio.run(aio_main(trio_main))
+
+
 def _mp_main(
     actor: 'Actor',
     accept_addr: Tuple[str, int],
@@ -49,7 +89,7 @@ def _mp_main(
     try:
         if infect_asyncio:
             actor._infected_aio = True
-            run_as_asyncio_guest(trio_main)
+            _asyncio_main(trio_main)
         else:
             trio.run(trio_main)
     except KeyboardInterrupt:
@@ -57,7 +97,7 @@ def _mp_main(
     log.info(f"Actor {actor.uid} terminated")
 
 
-async def _trio_main(
+async def _trip_main(
     actor: 'Actor',
     accept_addr: Tuple[str, int],
     parent_addr: Tuple[str, int] = None
@@ -72,9 +112,7 @@ async def _trio_main(
             f"Setting loglevel for {actor.uid} to {actor.loglevel}")
         get_console_log(actor.loglevel)
 
-    log.info(f"Started new trio process for {actor.uid}")
-
+    log.info(f"Started new TRIP process for {actor.uid}")
     _state._current_actor = actor
-
     await actor._async_main(accept_addr, parent_addr=parent_addr)
     log.info(f"Actor {actor.uid} terminated")

tractor/_spawn.py

@@ -1,18 +1,14 @@
 """
 Machinery for actor process spawning using multiple backends.
 """
-import sys
 import inspect
-import subprocess
 import multiprocessing as mp
 import platform
 from typing import Any, Dict, Optional
-from functools import partial
 
 import trio
-import cloudpickle
 from trio_typing import TaskStatus
-from async_generator import aclosing, asynccontextmanager
+from async_generator import aclosing
 
 try:
     from multiprocessing import semaphore_tracker  # type: ignore
@@ -25,12 +21,12 @@ except ImportError:
 from multiprocessing import forkserver  # type: ignore
 from typing import Tuple
 
-from . import _forkserver_override, _child
+from . import _forkserver_override
 from ._state import current_actor
 from .log import get_logger
 from ._portal import Portal
 from ._actor import Actor, ActorFailure
-from ._entry import _mp_main, _trio_main
+from ._entry import _mp_main, _trip_main
 
 
 log = get_logger('tractor')
@@ -45,13 +41,14 @@ if platform.system() == 'Windows':
     _ctx = mp.get_context("spawn")
 
     async def proc_waiter(proc: mp.Process) -> None:
-        await trio.lowlevel.WaitForSingleObject(proc.sentinel)
+        await trio.hazmat.WaitForSingleObject(proc.sentinel)
 else:
-    # *NIX systems use ``trio`` primitives as our default
-    _spawn_method = "trio"
+    # *NIX systems use ``trio_run_in_process` as our default (for now)
+    import trio_run_in_process
+    _spawn_method = "trio_run_in_process"
 
     async def proc_waiter(proc: mp.Process) -> None:
-        await trio.lowlevel.wait_readable(proc.sentinel)
+        await trio.hazmat.wait_readable(proc.sentinel)
 
 
 def try_set_start_method(name: str) -> Optional[mp.context.BaseContext]:
@@ -60,7 +57,7 @@ def try_set_start_method(name: str) -> Optional[mp.context.BaseContext]:
 
     If the desired method is not supported this function will error. On
     Windows the only supported option is the ``multiprocessing`` "spawn"
-    method. The default on *nix systems is ``trio``.
+    method. The default on *nix systems is ``trio_run_in_process``.
     """
     global _ctx
     global _spawn_method
@@ -72,7 +69,7 @@ def try_set_start_method(name: str) -> Optional[mp.context.BaseContext]:
 
     # no Windows support for trip yet
     if platform.system() != 'Windows':
-        methods += ['trio']
+        methods += ['trio_run_in_process']
 
     if name not in methods:
         raise ValueError(
@@ -81,7 +78,7 @@ def try_set_start_method(name: str) -> Optional[mp.context.BaseContext]:
     elif name == 'forkserver':
         _forkserver_override.override_stdlib()
         _ctx = mp.get_context(name)
-    elif name == 'trio':
+    elif name == 'trio_run_in_process':
         _ctx = None
     else:
         _ctx = mp.get_context(name)
@@ -156,27 +153,6 @@ async def cancel_on_completion(
         await portal.cancel_actor()
 
 
-@asynccontextmanager
-async def run_in_process(async_fn, *args, **kwargs):
-    encoded_job = cloudpickle.dumps(partial(async_fn, *args, **kwargs))
-    p = await trio.open_process(
-        [
-            sys.executable,
-            "-m",
-            _child.__name__
-        ],
-        stdin=subprocess.PIPE
-    )
-
-    # send over func to call
-    await p.stdin.send_all(encoded_job)
-
-    yield p
-
-    # wait for termination
-    await p.wait()
-
-
 async def new_proc(
     name: str,
     actor_nursery: 'ActorNursery',  # type: ignore
@@ -198,9 +174,12 @@ async def new_proc(
     subactor._spawn_method = _spawn_method
 
     async with trio.open_nursery() as nursery:
-        if use_trio_run_in_process or _spawn_method == 'trio':
-            async with run_in_process(
-                _trio_main,
+        if use_trio_run_in_process or _spawn_method == 'trio_run_in_process':
+            if infect_asyncio:
+                raise NotImplementedError("Asyncio is incompatible with trip")
+            # trio_run_in_process
+            async with trio_run_in_process.open_in_process(
+                _trip_main,
                 subactor,
                 bind_addr,
                 parent_addr,
@@ -224,7 +203,7 @@ async def new_proc(
                     cancel_scope = await nursery.start(
                         cancel_on_completion, portal, subactor, errors)
 
-                # run_in_process blocks here until process is complete
+                # TRIP blocks here until process is complete
         else:
             # `multiprocessing`
             assert _ctx

tractor/_state.py

@@ -30,7 +30,7 @@ class ActorContextInfo(Mapping):
     def __getitem__(self, key: str):
         try:
             return {
-                'task': trio.lowlevel.current_task,
+                'task': trio.hazmat.current_task,
                 'actor': current_actor
             }[key]().name
         except RuntimeError:

tractor/_streaming.py

@@ -41,11 +41,9 @@ def stream(func):
     """
     func._tractor_stream_function = True
     sig = inspect.signature(func)
-    params = sig.parameters
-    if 'ctx' not in params and 'to_trio' not in params:
+    if 'ctx' not in sig.parameters:
         raise TypeError(
             "The first argument to the stream function "
-            f"{func.__name__} must be `ctx: tractor.Context` "
-            "(Or ``to_trio`` if using ``asyncio`` in guest mode)."
+            f"{func.__name__} must be `ctx: tractor.Context`"
         )
     return func

tractor/testing/_tractor_test.py

@@ -47,7 +47,7 @@ def tractor_test(fn):
             if platform.system() == "Windows":
                 start_method = 'spawn'
             else:
-                start_method = 'trio'
+                start_method = 'trio_run_in_process'
 
         if 'start_method' in inspect.signature(fn).parameters:
             # set of subprocess spawning backends

tractor/to_asyncio.py

@@ -4,6 +4,7 @@ Infection apis for ``asyncio`` loops running ``trio`` using guest mode.
 import asyncio
 import inspect
 from typing import (
+    Any,
     Callable,
     AsyncGenerator,
     Awaitable,
@@ -12,61 +13,50 @@ from typing import (
 
 import trio
 
-from .log import get_logger
 from ._state import current_actor
 
-log = get_logger(__name__)
 
-
-__all__ = ['run_task', 'run_as_asyncio_guest']
+__all__ = ['run']
 
 
 async def _invoke(
     from_trio: trio.abc.ReceiveChannel,
     to_trio: asyncio.Queue,
     coro: Awaitable,
-) -> None:
-    """Await or stream awaiable object based on ``coro`` type into
-    ``trio`` memory channel.
-
-    ``from_trio`` might eventually be used here for bidirectional streaming.
-    """
-    if inspect.isasyncgen(coro):
-        async for item in coro:
-            to_trio.send_nowait(item)
-    elif inspect.iscoroutine(coro):
-        to_trio.send_nowait(await coro)
-
-
-async def run_task(
-    func: Callable,
-    *,
-    qsize: int = 2**10,
-    _treat_as_stream: bool = False,
-    **kwargs,
 ) -> Union[AsyncGenerator, Awaitable]:
+    """Await or stream awaiable object based on type into
+    ``trio`` memory channel.
+    """
+    async def stream_from_gen(c):
+        async for item in c:
+            to_trio.send_nowait(item)
+
+    async def just_return(c):
+        to_trio.send_nowait(await c)
+
+    if inspect.isasyncgen(coro):
+        return await stream_from_gen(coro)
+    elif inspect.iscoroutine(coro):
+        return await coro
+
+
+async def run(
+    func: Callable,
+    qsize: int = 2**10,
+    **kwargs,
+) -> Any:
     """Run an ``asyncio`` async function or generator in a task, return
     or stream the result back to ``trio``.
     """
-    assert current_actor().is_infected_aio()
+    assert current_actor()._infected_aio
 
     # ITC (inter task comms)
     from_trio = asyncio.Queue(qsize)
     to_trio, from_aio = trio.open_memory_channel(qsize)
 
-    args = tuple(inspect.getfullargspec(func).args)
-
-    if getattr(func, '_tractor_steam_function', None):
-        # the assumption is that the target async routine accepts the
-        # send channel then it intends to yield more then one return
-        # value otherwise it would just return ;P
-        _treat_as_stream = True
-
-    # allow target func to accept/stream results manually by name
-    if 'to_trio' in args:
-        kwargs['to_trio'] = to_trio
-    if 'from_trio' in args:
-        kwargs['from_trio'] = to_trio
+    # allow target func to accept/stream results manually
+    kwargs['to_trio'] = to_trio
+    kwargs['from_trio'] = to_trio
 
     coro = func(**kwargs)
 
@@ -77,6 +67,7 @@ async def run_task(
     task = asyncio.create_task(_invoke(from_trio, to_trio, coro))
     err = None
 
+    # XXX: I'm not sure this actually does anything...
     def cancel_trio(task):
         """Cancel the calling ``trio`` task on error.
         """
@@ -86,8 +77,17 @@ async def run_task(
 
     task.add_done_callback(cancel_trio)
 
-    # asycn gen
-    if inspect.isasyncgen(coro) or _treat_as_stream:
+    # determine return type async func vs. gen
+    if inspect.isasyncgen(coro):
+        # simple async func
+        async def result():
+            with cancel_scope:
+                return await from_aio.get()
+            if cancel_scope.cancelled_caught and err:
+                raise err
+
+    elif inspect.iscoroutine(coro):
+        # asycn gen
         async def result():
             with cancel_scope:
                 async with from_aio:
@@ -96,60 +96,4 @@ async def run_task(
             if cancel_scope.cancelled_caught and err:
                 raise err
 
-        return result()
-
-    # simple async func
-    elif inspect.iscoroutine(coro):
-        with cancel_scope:
-            result = await from_aio.receive()
-            return result
-        if cancel_scope.cancelled_caught and err:
-            raise err
-
-
-def run_as_asyncio_guest(
-    trio_main: Awaitable,
-) -> None:
-    """Entry for an "infected ``asyncio`` actor".
-
-    Uh, oh. :o
-
-    It looks like your event loop has caught a case of the ``trio``s.
-
-    :()
-
-    Don't worry, we've heard you'll barely notice. You might hallucinate
-    a few more propagating errors and feel like your digestion has
-    slowed but if anything get's too bad your parents will know about
-    it.
-
-    :)
-    """
-    async def aio_main(trio_main):
-        loop = asyncio.get_running_loop()
-
-        trio_done_fut = asyncio.Future()
-
-        def trio_done_callback(main_outcome):
-            log.info(f"trio_main finished: {main_outcome!r}")
-            trio_done_fut.set_result(main_outcome)
-
-        # start the infection: run trio on the asyncio loop in "guest mode"
-        log.info(f"Infecting asyncio process with {trio_main}")
-        trio.lowlevel.start_guest_run(
-            trio_main,
-            run_sync_soon_threadsafe=loop.call_soon_threadsafe,
-            done_callback=trio_done_callback,
-        )
-
-        (await trio_done_fut).unwrap()
-
-    # might as well if it's installed.
-    try:
-        import uvloop
-        loop = uvloop.new_event_loop()
-        asyncio.set_event_loop(loop)
-    except ImportError:
-        pass
-
-    asyncio.run(aio_main(trio_main))
+    return result()