Infect asyncio

.github/workflows/ci.yml

@@ -25,7 +25,7 @@ jobs:
 
   testing-linux:
     name: '${{ matrix.os }} Python ${{ matrix.python }} - ${{ matrix.spawn_backend }}'
-    timeout-minutes: 9
+    timeout-minutes: 10
     runs-on: ${{ matrix.os }}
 
     strategy:

docs/README.rst

@@ -27,7 +27,9 @@ Features
 - A modular transport stack, allowing for custom serialization (eg.
   `msgspec`_), communications protocols, and environment specific IPC
   primitives
-- `structured concurrency`_ from the ground up
+- Support for spawning process-level-SC, inter-loop one-to-one-task oriented
+  ``asyncio`` actors via "infected ``asyncio``" mode
+- `structured chadcurrency`_ from the ground up
 
 
 Run a func in a process
@@ -313,6 +315,117 @@ real time::
 This uses no extra threads, fancy semaphores or futures; all we need
 is ``tractor``'s IPC!
 
+"Infected ``asyncio``" mode
+---------------------------
+Have a bunch of ``asyncio`` code you want to force to be SC at the process level?
+
+Check out our experimental system for `guest-mode`_ controlled
+``asyncio`` actors:
+
+.. code:: python
+
+    import asyncio
+    from statistics import mean
+    import time
+
+    import trio
+    import tractor
+
+
+    async def aio_echo_server(
+        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():``
+        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
+            to_trio.send_nowait(await from_trio.get())
+            await asyncio.sleep(0)
+
+
+    @tractor.context
+    async def trio_to_aio_echo_server(
+        ctx: tractor.Context,
+    ):
+        # this will block until the ``asyncio`` task sends a "first"
+        # message.
+        async with tractor.to_asyncio.open_channel_from(
+            aio_echo_server,
+        ) as (first, chan):
+
+            assert first == 'start'
+            await ctx.started(first)
+
+            async with ctx.open_stream() as stream:
+
+                async for msg in stream:
+                    await chan.send(msg)
+
+                    out = await chan.receive()
+                    # echo back to parent actor-task
+                    await stream.send(out)
+
+
+    async def main():
+
+        async with tractor.open_nursery() as n:
+            p = await n.start_actor(
+                'aio_server',
+                enable_modules=[__name__],
+                infect_asyncio=True,
+            )
+            async with p.open_context(
+                trio_to_aio_echo_server,
+            ) as (ctx, first):
+
+                assert first == 'start'
+
+                count = 0
+                async with ctx.open_stream() as stream:
+
+                    delays = []
+                    send = time.time()
+
+                    await stream.send(count)
+                    async for msg in stream:
+                        recv = time.time()
+                        delays.append(recv - send)
+                        assert msg == count
+                        count += 1
+                        send = time.time()
+                        await stream.send(count)
+
+                        if count >= 1e3:
+                            break
+
+            print(f'mean round trip rate (Hz): {1/mean(delays)}')
+            await p.cancel_actor()
+
+
+    if __name__ == '__main__':
+        trio.run(main)
+
+
+Yes, we spawn a python process, run ``asyncio``, start ``trio`` on the
+``asyncio`` loop, then send commands to the ``trio`` scheduled tasks to
+tell ``asyncio`` tasks what to do XD
+
+We need help refining the `asyncio`-side channel API to be more
+`trio`-like. Feel free to sling your opinion in `#273`_!
+
+
+.. _#273: https://github.com/goodboy/tractor/issues/273
+
+
+Higher level "cluster" APIs
+---------------------------
 To be extra terse the ``tractor`` devs have started hacking some "higher
 level" APIs for managing actor trees/clusters. These interfaces should
 generally be condsidered provisional for now but we encourage you to try
@@ -476,6 +589,7 @@ channel`_!
 .. _async sandwich: https://trio.readthedocs.io/en/latest/tutorial.html#async-sandwich
 .. _structured concurrent: https://trio.discourse.group/t/concise-definition-of-structured-concurrency/228
 .. _3 axioms: https://www.youtube.com/watch?v=7erJ1DV_Tlo&t=162s
+.. .. _3 axioms: https://en.wikipedia.org/wiki/Actor_model#Fundamental_concepts
 .. _adherance to: https://www.youtube.com/watch?v=7erJ1DV_Tlo&t=1821s
 .. _trio gitter channel: https://gitter.im/python-trio/general
 .. _matrix channel: https://matrix.to/#/!tractor:matrix.org
@@ -484,11 +598,14 @@ channel`_!
 .. _messages: https://en.wikipedia.org/wiki/Message_passing
 .. _trio docs: https://trio.readthedocs.io/en/latest/
 .. _blog post: https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/
-.. _structured concurrency: https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/
+.. _structured concurrency: https://en.wikipedia.org/wiki/Structured_concurrency
+.. _structured chadcurrency: https://en.wikipedia.org/wiki/Structured_concurrency
+.. _structured concurrency: https://en.wikipedia.org/wiki/Structured_concurrency
 .. _unrequirements: https://en.wikipedia.org/wiki/Actor_model#Direct_communication_and_asynchrony
 .. _async generators: https://www.python.org/dev/peps/pep-0525/
 .. _trio-parallel: https://github.com/richardsheridan/trio-parallel
 .. _msgspec: https://jcristharif.com/msgspec/
+.. _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
 
 
 .. |gh_actions| image:: https://img.shields.io/endpoint.svg?url=https%3A%2F%2Factions-badge.atrox.dev%2Fgoodboy%2Ftractor%2Fbadge&style=popout-square

examples/infected_asyncio_echo_server.py

@@ -0,0 +1,91 @@
+'''
+An SC compliant infected ``asyncio`` echo server.
+
+'''
+import asyncio
+from statistics import mean
+import time
+
+import trio
+import tractor
+
+
+async def aio_echo_server(
+    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():``
+    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
+        to_trio.send_nowait(await from_trio.get())
+        await asyncio.sleep(0)
+
+
+@tractor.context
+async def trio_to_aio_echo_server(
+    ctx: tractor.Context,
+):
+    # this will block until the ``asyncio`` task sends a "first"
+    # message.
+    async with tractor.to_asyncio.open_channel_from(
+        aio_echo_server,
+    ) as (first, chan):
+
+        assert first == 'start'
+        await ctx.started(first)
+
+        async with ctx.open_stream() as stream:
+
+            async for msg in stream:
+                await chan.send(msg)
+
+                out = await chan.receive()
+                # echo back to parent actor-task
+                await stream.send(out)
+
+
+async def main():
+
+    async with tractor.open_nursery() as n:
+        p = await n.start_actor(
+            'aio_server',
+            enable_modules=[__name__],
+            infect_asyncio=True,
+        )
+        async with p.open_context(
+            trio_to_aio_echo_server,
+        ) as (ctx, first):
+
+            assert first == 'start'
+
+            count = 0
+            async with ctx.open_stream() as stream:
+
+                delays = []
+                send = time.time()
+
+                await stream.send(count)
+                async for msg in stream:
+                    recv = time.time()
+                    delays.append(recv - send)
+                    assert msg == count
+                    count += 1
+                    send = time.time()
+                    await stream.send(count)
+
+                    if count >= 1e3:
+                        break
+
+        print(f'mean round trip rate (Hz): {1/mean(delays)}')
+        await p.cancel_actor()
+
+
+if __name__ == '__main__':
+    trio.run(main)

newsfragments/121.feature.rst

@@ -0,0 +1,28 @@
+Add "infected ``asyncio`` mode; a sub-system to spawn and control
+``asyncio`` actors using ``trio``'s guest-mode.
+
+This gets us the following very interesting functionality:
+
+- ability to spawn an actor that has a process entry point of
+  ``asyncio.run()`` by passing ``infect_asyncio=True`` to
+  ``Portal.start_actor()`` (and friends).
+- the ``asyncio`` actor embeds ``trio`` using guest-mode and starts
+  a main ``trio`` task which runs the ``tractor.Actor._async_main()``
+  entry point engages all the normal ``tractor`` runtime IPC/messaging
+  machinery; for all purposes the actor is now running normally on
+  a ``trio.run()``.
+- the actor can now make one-to-one task spawning requests to the
+  underlying ``asyncio`` event loop using either of:
+  * ``to_asyncio.run_task()`` to spawn and run an ``asyncio`` task to
+    completion and block until a return value is delivered.
+  * ``async with to_asyncio.open_channel_from():`` which spawns a task
+    and hands it a pair of "memory channels" to allow for bi-directional
+    streaming between the now SC-linked ``trio`` and ``asyncio`` tasks.
+
+The output from any call(s) to ``asyncio`` can be handled as normal in
+``trio``/``tractor`` task operation with the caveat of the overhead due
+to guest-mode use.
+
+For more details see the `original PR
+<https://github.com/goodboy/tractor/pull/121>`_ and `issue
+<https://github.com/goodboy/tractor/issues/120>`_.

tests/test_cancellation.py

@@ -523,7 +523,7 @@ def test_fast_graceful_cancel_when_spawn_task_in_soft_proc_wait_for_daemon(
     cancellation, and it's faster, we might as well do it.
 
     '''
-    kbi_delay = 0.2
+    kbi_delay = 0.5
 
     async def main():
         start = time.time()

tests/test_docs_examples.py

@@ -1,6 +1,7 @@
-"""
+'''
 Let's make sure them docs work yah?
-"""
+
+'''
 from contextlib import contextmanager
 import itertools
 import os