Add asyncio echo server ex to readme; fix cluster section

docs/README.rst

@@ -313,6 +313,110 @@ 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())
+
+
+    @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
+
+
+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
@@ -489,6 +593,7 @@ channel`_!
 .. _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