Handle broken mem chan on `Actor._push_result()`

When backpressure is used and a feeder mem chan breaks during msg delivery (usually because the IPC allocating task already terminated) instead of raising we simply warn as we do for the non-backpressure case. Also, add a proper `Actor.is_arbiter` test inside `._invoke()` to avoid doing an arbiter-registry lookup if the current actor **is** the registrar.
Always set the `parent_exit: trio.Event` on exit
2022-12-12 15:04:40 -05:00 · 2022-12-12 15:04:40 -05:00 · 2022-12-12 15:04:40 -05:00 · 2022-12-12 15:04:40 -05:00 · 2022-12-12 15:04:40 -05:00 · 2022-12-12 15:04:40 -05:00
30 changed files with 1456 additions and 1114 deletions
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@ -67,7 +67,6 @@ jobs:
        ]

    steps:
-
      - name: Checkout
        uses: actions/checkout@v2

@ -85,6 +84,40 @@ jobs:
      - name: Run tests
        run: pytest tests/ --spawn-backend=${{ matrix.spawn_backend }} -rsx

+
+  testing-macos:
+    name: '${{ matrix.os }} Python ${{ matrix.python }} - ${{ matrix.spawn_backend }}'
+    timeout-minutes: 10
+    runs-on: ${{ matrix.os }}
+
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [macos-latest]
+        python: ['3.10']
+        spawn_backend: [
+          'trio',
+        ]
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v2
+
+      - name: Setup python
+        uses: actions/setup-python@v2
+        with:
+          python-version: '${{ matrix.python }}'
+
+      - name: Install dependencies
+        run: pip install -U . -r requirements-test.txt -r requirements-docs.txt --upgrade-strategy eager
+
+      - name: List dependencies
+        run: pip list
+
+      - name: Run tests
+        run: pytest tests/ --spawn-backend=${{ matrix.spawn_backend }} -rsx
+
+
  # We skip 3.10 on windows for now due to not having any collabs to
  # debug the CI failures. Anyone wanting to hack and solve them is very
  # welcome, but our primary user base is not using that OS.
--- a/docs/README.rst
+++ b/docs/README.rst
@ -6,14 +6,8 @@
 ``tractor`` is a `structured concurrent`_, multi-processing_ runtime
 built on trio_.

-Fundamentally, ``tractor`` gives you parallelism via
-``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``
+Fundamentally ``tractor`` gives you parallelism via ``trio``-"*actors*":
+our nurseries_ let you spawn new Python processes which each run a ``trio``
 scheduled runtime - a call to ``trio.run()``.

 We believe the system adheres to the `3 axioms`_ of an "`actor model`_"
@ -29,8 +23,7 @@ Features
 - **It's just** a ``trio`` API
 - *Infinitely nesteable* process trees
 - Builtin IPC streaming APIs with task fan-out broadcasting
- A "native" multi-core debugger REPL using `pdbp`_ (a fork & fix of
-  `pdb++`_ thanks to @mdmintz!)
+- A (first ever?) "native" multi-core debugger UX for Python using `pdb++`_
 - Support for a swappable, OS specific, process spawning layer
 - A modular transport stack, allowing for custom serialization (eg. with
  `msgspec`_), communications protocols, and environment specific IPC
@ -125,7 +118,7 @@ Zombie safe: self-destruct a process tree
            f"running in pid {os.getpid()}"
        )

-        await trio.sleep_forever()
+       await trio.sleep_forever()


    async def main():
@ -156,7 +149,7 @@ it **is a bug**.

 "Native" multi-process debugging
 --------------------------------
-Using the magic of `pdbp`_ and our internal IPC, we've
+Using the magic of `pdb++`_ and our internal IPC, we've
 been able to create a native feeling debugging experience for
 any (sub-)process in your ``tractor`` tree.

@ -604,7 +597,6 @@ channel`_!
 .. _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
-.. _pdbp: https://github.com/mdmintz/pdbp
 .. _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
 .. _messages: https://en.wikipedia.org/wiki/Message_passing
--- a/examples/advanced_faults/ipc_failure_during_stream.py
+++ b/examples/advanced_faults/ipc_failure_during_stream.py
@ -1,151 +0,0 @@
-'''
-Complex edge case where during real-time streaming the IPC tranport
-channels are wiped out (purposely in this example though it could have
-been an outage) and we want to ensure that despite being in debug mode
-(or not) the user can sent SIGINT once they notice the hang and the
-actor tree will eventually be cancelled without leaving any zombies.
-
-'''
-import trio
-from tractor import (
-    open_nursery,
-    context,
-    Context,
-    MsgStream,
-)
-
-
-async def break_channel_silently_then_error(
-    stream: MsgStream,
-):
-    async for msg in stream:
-        await stream.send(msg)
-
-        # XXX: close the channel right after an error is raised
-        # purposely breaking the IPC transport to make sure the parent
-        # doesn't get stuck in debug or hang on the connection join.
-        # this more or less simulates an infinite msg-receive hang on
-        # the other end.
-        await stream._ctx.chan.send(None)
-        assert 0
-
-
-async def close_stream_and_error(
-    stream: MsgStream,
-):
-    async for msg in stream:
-        await stream.send(msg)
-
-        # wipe out channel right before raising
-        await stream._ctx.chan.send(None)
-        await stream.aclose()
-        assert 0
-
-
-@context
-async def recv_and_spawn_net_killers(
-
-    ctx: Context,
-    break_ipc_after: bool | int = False,
-
-) -> None:
-    '''
-    Receive stream msgs and spawn some IPC killers mid-stream.
-
-    '''
-    await ctx.started()
-    async with (
-        ctx.open_stream() as stream,
-        trio.open_nursery() as n,
-    ):
-        async for i in stream:
-            print(f'child echoing {i}')
-            await stream.send(i)
-            if (
-                break_ipc_after
-                and i > break_ipc_after
-            ):
-                '#################################\n'
-                'Simulating child-side IPC BREAK!\n'
-                '#################################'
-                n.start_soon(break_channel_silently_then_error, stream)
-                n.start_soon(close_stream_and_error, stream)
-
-
-async def main(
-    debug_mode: bool = False,
-    start_method: str = 'trio',
-
-    # by default we break the parent IPC first (if configured to break
-    # at all), but this can be changed so the child does first (even if
-    # both are set to break).
-    break_parent_ipc_after: int | bool = False,
-    break_child_ipc_after: int | bool = False,
-
-) -> None:
-
-    async with (
-        open_nursery(
-            start_method=start_method,
-
-            # NOTE: even debugger is used we shouldn't get
-            # a hang since it never engages due to broken IPC
-            debug_mode=debug_mode,
-            loglevel='warning',
-
-        ) as an,
-    ):
-        portal = await an.start_actor(
-            'chitty_hijo',
-            enable_modules=[__name__],
-        )
-
-        async with portal.open_context(
-            recv_and_spawn_net_killers,
-            break_ipc_after=break_child_ipc_after,
-
-        ) as (ctx, sent):
-            async with ctx.open_stream() as stream:
-                for i in range(1000):
-
-                    if (
-                        break_parent_ipc_after
-                        and i > break_parent_ipc_after
-                    ):
-                        print(
-                            '#################################\n'
-                            'Simulating parent-side IPC BREAK!\n'
-                            '#################################'
-                        )
-                        await stream._ctx.chan.send(None)
-
-                    # it actually breaks right here in the
-                    # mp_spawn/forkserver backends and thus the zombie
-                    # reaper never even kicks in?
-                    print(f'parent sending {i}')
-                    await stream.send(i)
-
-                    with trio.move_on_after(2) as cs:
-
-                        # NOTE: in the parent side IPC failure case this
-                        # will raise an ``EndOfChannel`` after the child
-                        # is killed and sends a stop msg back to it's
-                        # caller/this-parent.
-                        rx = await stream.receive()
-
-                        print(f"I'm a happy user and echoed to me is {rx}")
-
-                    if cs.cancelled_caught:
-                        # pretend to be a user seeing no streaming action
-                        # thinking it's a hang, and then hitting ctl-c..
-                        print("YOO i'm a user anddd thingz hangin..")
-
-                print(
-                    "YOO i'm mad send side dun but thingz hangin..\n"
-                    'MASHING CTlR-C Ctl-c..'
-                )
-                raise KeyboardInterrupt
-
-
-if __name__ == '__main__':
-    trio.run(main)
--- a/examples/debugging/restore_builtin_breakpoint.py
+++ b/examples/debugging/restore_builtin_breakpoint.py
@ -1,24 +0,0 @@
-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)
--- a/examples/full_fledged_streaming_service.py
+++ b/examples/full_fledged_streaming_service.py
@ -12,9 +12,11 @@ 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
+    '''
+    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:
        portals = []
        for i in range(1, 3):
@ -69,7 +71,8 @@ async def aggregate(seed):
 async def main():
    # a nursery which spawns "actors"
    async with tractor.open_nursery(
-        arbiter_addr=('127.0.0.1', 1616)
+        arbiter_addr=('127.0.0.1', 1616),
+        loglevel='cancel',
    ) as nursery:

        seed = int(1e3)
@ -92,6 +95,9 @@ async def main():
            async for value in stream:
                result_stream.append(value)

+            print("ROOT STREAM CONSUMER COMPLETE")
+
+        print("ROOT CANCELLING AGGREGATOR CHILD")
        await portal.cancel_actor()

        print(f"STREAM TIME = {time.time() - start}")
--- a/nooz/343.trivial.rst
+++ b/nooz/343.trivial.rst
@ -1,19 +0,0 @@
-Rework our ``.trionics.BroadcastReceiver`` internals to avoid method
-recursion and approach a design and interface closer to ``trio``'s
-``MemoryReceiveChannel``.
-
-The details of the internal changes include:
-
- implementing a ``BroadcastReceiver.receive_nowait()`` and using it
-  within the async ``.receive()`` thus avoiding recursion from
-  ``.receive()``.
- failing over to an internal ``._receive_from_underlying()`` when the
-  ``_nowait()`` call raises ``trio.WouldBlock``
- adding ``BroadcastState.statistics()`` for debugging and testing both
-  internals and by users.
- add an internal ``BroadcastReceiver._raise_on_lag: bool`` which can be
-  set to avoid ``Lagged`` raising for possible use cases where a user
-  wants to choose between a [cheap or nasty
-  pattern](https://zguide.zeromq.org/docs/chapter7/#The-Cheap-or-Nasty-Pattern)
-  the the particular stream (we use this in ``piker``'s dark clearing
-  engine to avoid fast feeds breaking during HFT periods).
--- a/nooz/346.bugfix.rst
+++ b/nooz/346.bugfix.rst
@ -1,15 +0,0 @@
-Fixes to ensure IPC (channel) breakage doesn't result in hung actor
-trees; the zombie reaping and general supervision machinery will always
-clean up and terminate.
-
-This includes not only the (mostly minor) fixes to solve these cases but
-also a new extensive test suite in `test_advanced_faults.py` with an
-accompanying highly configurable example module-script in
-`examples/advanced_faults/ipc_failure_during_stream.py`. Tests ensure we
-never get hang or zombies despite operating in debug mode and attempt to
-simulate all possible IPC transport failure cases for a local-host actor
-tree.
-
-Further we simplify `Context.open_stream.__aexit__()` to just call
-`MsgStream.aclose()` directly more or less avoiding a pure duplicate
-code path.
--- a/nooz/349.trivial.rst
+++ b/nooz/349.trivial.rst
@ -1,10 +0,0 @@
-Always redraw the `pdbpp` prompt on `SIGINT` during REPL use.
-
-There was recent changes todo with Python 3.10 that required us to pin
-to a specific commit in `pdbpp` which have recently been fixed minus
-this last issue with `SIGINT` shielding: not clobbering or not
-showing the `(Pdb++)` prompt on ctlr-c by the user. This repairs all
-that by firstly removing the standard KBI intercepting of the std lib's
-`pdb.Pdb._cmdloop()` as well as ensuring that only the actor with REPL
-control ever reports `SIGINT` handler log msgs and prompt redraws. With
-this we move back to using pypi `pdbpp` release.
--- a/nooz/356.trivial.rst
+++ b/nooz/356.trivial.rst
@ -1,7 +0,0 @@
-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.
--- a/nooz/358.feature.rst
+++ b/nooz/358.feature.rst
@ -1,15 +0,0 @@
-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.
--- a/requirements-test.txt
+++ b/requirements-test.txt
@ -1,8 +1,9 @@
 pytest
 pytest-trio
 pytest-timeout
-pdbp
+pdbpp
 mypy
 trio_typing
 pexpect
 towncrier
+numpy
--- a/setup.py
+++ b/setup.py
@ -26,12 +26,12 @@ with open('docs/README.rst', encoding='utf-8') as f:
 setup(
    name="tractor",
    version='0.1.0a6dev0',  # alpha zone
-    description='structured concurrrent `trio`-"actors"',
+    description='structured concurrrent "actors"',
    long_description=readme,
    license='AGPLv3',
    author='Tyler Goodlet',
    maintainer='Tyler Goodlet',
-    maintainer_email='goodboy_foss@protonmail.com',
+    maintainer_email='jgbt@protonmail.com',
    url='https://github.com/goodboy/tractor',
    platforms=['linux', 'windows'],
    packages=[
@ -52,15 +52,14 @@ setup(
        # tooling
        'tricycle',
        'trio_typing',
+
+        # tooling
        'colorlog',
        'wrapt',

-        # IPC serialization
+        # serialization
        'msgspec',

-        # debug mode REPL
-        'pdbp',
-
        # pip ref docs on these specs:
        # https://pip.pypa.io/en/stable/reference/requirement-specifiers/#examples
        # and pep:
@ -71,9 +70,14 @@ setup(
        # https://github.com/pdbpp/fancycompleter/issues/37
        'pyreadline3 ; platform_system == "Windows"',

+        # 3.10 has an outstanding unreleased issue and `pdbpp` itself
+        #   pins to patched forks of its own dependencies as well..and
+        #   we need a specific patch on master atm.
+        'pdbpp @ git+https://github.com/pdbpp/pdbpp@76c4be5#egg=pdbpp ; python_version > "3.9"',  # noqa: E501
+
    ],
    tests_require=['pytest'],
-    python_requires=">=3.10",
+    python_requires=">=3.9",
    keywords=[
        'trio',
        'async',
--- a/tests/conftest.py
+++ b/tests/conftest.py
@ -7,7 +7,6 @@ import os
 import random
 import signal
 import platform
-import pathlib
 import time
 import inspect
 from functools import partial, wraps
@ -114,21 +113,14 @@ no_windows = pytest.mark.skipif(
 )


-def repodir() -> pathlib.Path:
-    '''
-    Return the abspath to the repo directory.
-
-    '''
-    # 2 parents up to step up through tests/<repo_dir>
-    return pathlib.Path(__file__).parent.parent.absolute()
-
-
-def examples_dir() -> pathlib.Path:
-    '''
-    Return the abspath to the examples directory as `pathlib.Path`.
-
-    '''
-    return repodir() / 'examples'
+def repodir():
+    """Return the abspath to the repo directory.
+    """
+    dirname = os.path.dirname
+    dirpath = os.path.abspath(
+        dirname(dirname(os.path.realpath(__file__)))
+        )
+    return dirpath


 def pytest_addoption(parser):
@ -159,7 +151,7 @@ def loglevel(request):


@pytest.fixture(scope='session')
-def spawn_backend(request) -> str:
+def spawn_backend(request):
    return request.config.option.spawn_backend


@ -213,22 +205,16 @@ def sig_prog(proc, sig):


@pytest.fixture
-def daemon(
-    loglevel: str,
-    testdir,
-    arb_addr: tuple[str, int],
-):
-    '''
-    Run a daemon actor as a "remote arbiter".
-
-    '''
+def daemon(loglevel, testdir, arb_addr):
+    """Run a daemon actor as a "remote arbiter".
+    """
    if loglevel in ('trace', 'debug'):
        # too much logging will lock up the subproc (smh)
        loglevel = 'info'

    cmdargs = [
        sys.executable, '-c',
-        "import tractor; tractor.run_daemon([], registry_addr={}, loglevel={})"
+        "import tractor; tractor.run_daemon([], arbiter_addr={}, loglevel={})"
        .format(
            arb_addr,
            "'{}'".format(loglevel) if loglevel else None)
--- a/tests/test_advanced_faults.py
+++ b/tests/test_advanced_faults.py
@ -1,193 +0,0 @@
-'''
-Sketchy network blackoutz, ugly byzantine gens, puedes eschuchar la
-cancelacion?..
-
-'''
-from functools import partial
-
-import pytest
-from _pytest.pathlib import import_path
-import trio
-import tractor
-
-from conftest import (
-    examples_dir,
-)
-
-
-@pytest.mark.parametrize(
-    'debug_mode',
-    [False, True],
-    ids=['no_debug_mode', 'debug_mode'],
-)
-@pytest.mark.parametrize(
-    'ipc_break',
-    [
-        # no breaks
-        {
-            'break_parent_ipc_after': False,
-            'break_child_ipc_after': False,
-        },
-
-        # only parent breaks
-        {
-            'break_parent_ipc_after': 500,
-            'break_child_ipc_after': False,
-        },
-
-        # only child breaks
-        {
-            'break_parent_ipc_after': False,
-            'break_child_ipc_after': 500,
-        },
-
-        # both: break parent first
-        {
-            'break_parent_ipc_after': 500,
-            'break_child_ipc_after': 800,
-        },
-        # both: break child first
-        {
-            'break_parent_ipc_after': 800,
-            'break_child_ipc_after': 500,
-        },
-
-    ],
-    ids=[
-        'no_break',
-        'break_parent',
-        'break_child',
-        'break_both_parent_first',
-        'break_both_child_first',
-    ],
-)
-def test_ipc_channel_break_during_stream(
-    debug_mode: bool,
-    spawn_backend: str,
-    ipc_break: dict | None,
-):
-    '''
-    Ensure we can have an IPC channel break its connection during
-    streaming and it's still possible for the (simulated) user to kill
-    the actor tree using SIGINT.
-
-    We also verify the type of connection error expected in the parent
-    depending on which side if the IPC breaks first.
-
-    '''
-    if spawn_backend != 'trio':
-        if debug_mode:
-            pytest.skip('`debug_mode` only supported on `trio` spawner')
-
-        # non-`trio` spawners should never hit the hang condition that
-        # requires the user to do ctl-c to cancel the actor tree.
-        expect_final_exc = trio.ClosedResourceError
-
-    mod = import_path(
-        examples_dir() / 'advanced_faults' / 'ipc_failure_during_stream.py',
-        root=examples_dir(),
-    )
-
-    expect_final_exc = KeyboardInterrupt
-
-    # when ONLY the child breaks we expect the parent to get a closed
-    # resource error on the next `MsgStream.receive()` and then fail out
-    # and cancel the child from there.
-    if (
-
-        # only child breaks
-        (
-            ipc_break['break_child_ipc_after']
-            and ipc_break['break_parent_ipc_after'] is False
-        )
-
-        # both break but, parent breaks first
-        or (
-            ipc_break['break_child_ipc_after'] is not False
-            and (
-                ipc_break['break_parent_ipc_after']
-                > ipc_break['break_child_ipc_after']
-            )
-        )
-
-    ):
-        expect_final_exc = trio.ClosedResourceError
-
-    # when the parent IPC side dies (even if the child's does as well
-    # but the child fails BEFORE the parent) we expect the channel to be
-    # sent a stop msg from the child at some point which will signal the
-    # parent that the stream has been terminated.
-    # NOTE: when the parent breaks "after" the child you get this same
-    # case as well, the child breaks the IPC channel with a stop msg
-    # before any closure takes place.
-    elif (
-        # only parent breaks
-        (
-            ipc_break['break_parent_ipc_after']
-            and ipc_break['break_child_ipc_after'] is False
-        )
-
-        # both break but, child breaks first
-        or (
-            ipc_break['break_parent_ipc_after'] is not False
-            and (
-                ipc_break['break_child_ipc_after']
-                > ipc_break['break_parent_ipc_after']
-            )
-        )
-    ):
-        expect_final_exc = trio.EndOfChannel
-
-    with pytest.raises(expect_final_exc):
-        trio.run(
-            partial(
-                mod.main,
-                debug_mode=debug_mode,
-                start_method=spawn_backend,
-                **ipc_break,
-            )
-        )
-
-
-@tractor.context
-async def break_ipc_after_started(
-    ctx: tractor.Context,
-) -> None:
-    await ctx.started()
-    async with ctx.open_stream() as stream:
-        await stream.aclose()
-        await trio.sleep(0.2)
-        await ctx.chan.send(None)
-        print('child broke IPC and terminating')
-
-
-def test_stream_closed_right_after_ipc_break_and_zombie_lord_engages():
-    '''
-    Verify that is a subactor's IPC goes down just after bringing up a stream
-    the parent can trigger a SIGINT and the child will be reaped out-of-IPC by
-    the localhost process supervision machinery: aka "zombie lord".
-
-    '''
-    async def main():
-        async with tractor.open_nursery() as n:
-            portal = await n.start_actor(
-                'ipc_breaker',
-                enable_modules=[__name__],
-            )
-
-            with trio.move_on_after(1):
-                async with (
-                    portal.open_context(
-                        break_ipc_after_started
-                    ) as (ctx, sent),
-                ):
-                    async with ctx.open_stream():
-                        await trio.sleep(0.5)
-
-                    print('parent waiting on context')
-
-            print('parent exited context')
-            raise KeyboardInterrupt
-
-    with pytest.raises(KeyboardInterrupt):
-        trio.run(main)
--- a/tests/test_advanced_streaming.py
+++ b/tests/test_advanced_streaming.py
@ -14,7 +14,7 @@ def is_win():
    return platform.system() == 'Windows'


-_registry: dict[str, set[tractor.MsgStream]] = {
+_registry: dict[str, set[tractor.ReceiveMsgStream]] = {
    'even': set(),
    'odd': set(),
 }
--- a/tests/test_debugger.py
+++ b/tests/test_debugger.py
@ -14,7 +14,6 @@ import itertools
 from os import path
 from typing import Optional
 import platform
-import pathlib
 import sys
 import time

@ -25,10 +24,7 @@ from pexpect.exceptions import (
    EOF,
 )

-from conftest import (
-    examples_dir,
-    _ci_env,
-)
+from conftest import repodir, _ci_env

 # TODO: The next great debugger audit could be done by you!
 # - recurrent entry to breakpoint() from single actor *after* and an
@ -40,20 +36,29 @@ from conftest import (
 # - recurrent root errors


-if platform.system() == 'Windows':
+if osname := platform.system() in (
+    'Windows',
+    'Darwin',
+):
    pytest.skip(
-        'Debugger tests have no windows support (yet)',
+        'Debugger tests have no {osname} support (yet)',
        allow_module_level=True,
    )


-def mk_cmd(ex_name: str) -> str:
-    '''
-    Generate a command suitable to pass to ``pexpect.spawn()``.
+def examples_dir():
+    """Return the abspath to the examples directory.
+    """
+    return path.join(repodir(), 'examples', 'debugging/')

-    '''
-    script_path: pathlib.Path = examples_dir() / 'debugging' / f'{ex_name}.py'
-    return ' '.join(['python', str(script_path)])
+
+def mk_cmd(ex_name: str) -> str:
+    """Generate a command suitable to pass to ``pexpect.spawn()``.
+    """
+    return ' '.join(
+        ['python',
+         path.join(examples_dir(), f'{ex_name}.py')]
+    )


 # TODO: was trying to this xfail style but some weird bug i see in CI
@ -95,7 +100,7 @@ def spawn(
    return _spawn


-PROMPT = r"\(Pdb\+\)"
+PROMPT = r"\(Pdb\+\+\)"


 def expect(
@ -151,16 +156,29 @@ def ctlc(

    use_ctlc = request.param

-    node = request.node
-    markers = node.own_markers
-    for mark in markers:
-        if mark.name == 'has_nested_actors':
-            pytest.skip(
-                f'Test {node} has nested actors and fails with Ctrl-C.\n'
-                f'The test can sometimes run fine locally but until'
-                ' we solve' 'this issue this CI test will be xfail:\n'
-                'https://github.com/goodboy/tractor/issues/320'
-            )
+    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..')
+
+    if ci_env:
+        node = request.node
+        markers = node.own_markers
+        for mark in markers:
+            if mark.name == 'has_nested_actors':
+                pytest.skip(
+                    f'Test for {node} uses nested actors and fails in CI\n'
+                    f'The test seems to run fine locally but until we solve'
+                    'this issue this CI test will be xfail:\n'
+                    'https://github.com/goodboy/tractor/issues/320'
+                )

    if use_ctlc:
        # XXX: disable pygments highlighting for auto-tests
@ -181,15 +199,13 @@ def ctlc(
    ids=lambda item: f'{item[0]} -> {item[1]}',
 )
 def test_root_actor_error(spawn, user_in_out):
-    '''
-    Demonstrate crash handler entering pdb from basic error in root actor.
-
-    '''
+    """Demonstrate crash handler entering pdbpp from basic error in root actor.
+    """
    user_input, expect_err_str = user_in_out

    child = spawn('root_actor_error')

-    # scan for the prompt
+    # scan for the pdbpp prompt
    expect(child, PROMPT)

    before = str(child.before.decode())
@ -220,8 +236,8 @@ def test_root_actor_bp(spawn, user_in_out):
    user_input, expect_err_str = user_in_out
    child = spawn('root_actor_breakpoint')

-    # scan for the prompt
-    child.expect(PROMPT)
+    # scan for the pdbpp prompt
+    child.expect(r"\(Pdb\+\+\)")

    assert 'Error' not in str(child.before)

@ -262,7 +278,7 @@ def do_ctlc(
        if expect_prompt:
            before = str(child.before.decode())
            time.sleep(delay)
-            child.expect(PROMPT)
+            child.expect(r"\(Pdb\+\+\)")
            time.sleep(delay)

            if patt:
@ -281,7 +297,7 @@ def test_root_actor_bp_forever(
    # entries
    for _ in range(10):

-        child.expect(PROMPT)
+        child.expect(r"\(Pdb\+\+\)")

        if ctlc:
            do_ctlc(child)
@ -291,7 +307,7 @@ def test_root_actor_bp_forever(
    # do one continue which should trigger a
    # new task to lock the tty
    child.sendline('continue')
-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")

    # seems that if we hit ctrl-c too fast the
    # sigint guard machinery might not kick in..
@ -302,10 +318,10 @@ def test_root_actor_bp_forever(

    # XXX: this previously caused a bug!
    child.sendline('n')
-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")

    child.sendline('n')
-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")

    # quit out of the loop
    child.sendline('q')
@ -328,8 +344,8 @@ def test_subactor_error(
    '''
    child = spawn('subactor_error')

-    # scan for the prompt
-    child.expect(PROMPT)
+    # scan for the pdbpp prompt
+    child.expect(r"\(Pdb\+\+\)")

    before = str(child.before.decode())
    assert "Attaching to pdb in crashed actor: ('name_error'" in before
@ -349,7 +365,7 @@ def test_subactor_error(
        # creating actor
        child.sendline('continue')

-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")
    before = str(child.before.decode())

    # root actor gets debugger engaged
@ -376,8 +392,8 @@ def test_subactor_breakpoint(

    child = spawn('subactor_breakpoint')

-    # scan for the prompt
-    child.expect(PROMPT)
+    # scan for the pdbpp prompt
+    child.expect(r"\(Pdb\+\+\)")

    before = str(child.before.decode())
    assert "Attaching pdb to actor: ('breakpoint_forever'" in before
@ -386,7 +402,7 @@ def test_subactor_breakpoint(
    # entries
    for _ in range(10):
        child.sendline('next')
-        child.expect(PROMPT)
+        child.expect(r"\(Pdb\+\+\)")

        if ctlc:
            do_ctlc(child)
@ -394,7 +410,7 @@ def test_subactor_breakpoint(
    # now run some "continues" to show re-entries
    for _ in range(5):
        child.sendline('continue')
-        child.expect(PROMPT)
+        child.expect(r"\(Pdb\+\+\)")
        before = str(child.before.decode())
        assert "Attaching pdb to actor: ('breakpoint_forever'" in before

@ -405,7 +421,7 @@ def test_subactor_breakpoint(
    child.sendline('q')

    # child process should exit but parent will capture pdb.BdbQuit
-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")

    before = str(child.before.decode())
    assert "RemoteActorError: ('breakpoint_forever'" in before
@ -437,8 +453,8 @@ def test_multi_subactors(
    '''
    child = spawn(r'multi_subactors')

-    # scan for the prompt
-    child.expect(PROMPT)
+    # scan for the pdbpp prompt
+    child.expect(r"\(Pdb\+\+\)")

    before = str(child.before.decode())
    assert "Attaching pdb to actor: ('breakpoint_forever'" in before
@ -450,7 +466,7 @@ def test_multi_subactors(
    # entries
    for _ in range(10):
        child.sendline('next')
-        child.expect(PROMPT)
+        child.expect(r"\(Pdb\+\+\)")

        if ctlc:
            do_ctlc(child)
@ -459,7 +475,7 @@ def test_multi_subactors(
    child.sendline('c')

    # first name_error failure
-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")
    before = str(child.before.decode())
    assert "Attaching to pdb in crashed actor: ('name_error'" in before
    assert "NameError" in before
@ -471,7 +487,7 @@ def test_multi_subactors(
    child.sendline('c')

    # 2nd name_error failure
-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")

    # TODO: will we ever get the race where this crash will show up?
    # blocklist strat now prevents this crash
@ -485,7 +501,7 @@ def test_multi_subactors(

    # breakpoint loop should re-engage
    child.sendline('c')
-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")
    before = str(child.before.decode())
    assert "Attaching pdb to actor: ('breakpoint_forever'" in before

@ -501,7 +517,7 @@ def test_multi_subactors(
    ):
        child.sendline('c')
        time.sleep(0.1)
-        child.expect(PROMPT)
+        child.expect(r"\(Pdb\+\+\)")
        before = str(child.before.decode())

        if ctlc:
@ -520,11 +536,11 @@ def test_multi_subactors(
    # now run some "continues" to show re-entries
    for _ in range(5):
        child.sendline('c')
-        child.expect(PROMPT)
+        child.expect(r"\(Pdb\+\+\)")

    # quit the loop and expect parent to attach
    child.sendline('q')
-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")
    before = str(child.before.decode())

    assert_before(child, [
@ -568,7 +584,7 @@ def test_multi_daemon_subactors(
    '''
    child = spawn('multi_daemon_subactors')

-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")

    # there can be a race for which subactor will acquire
    # the root's tty lock first so anticipate either crash
@ -598,7 +614,7 @@ def test_multi_daemon_subactors(
    # second entry by `bp_forever`.

    child.sendline('c')
-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")
    assert_before(child, [next_msg])

    # XXX: hooray the root clobbering the child here was fixed!
@ -620,7 +636,7 @@ def test_multi_daemon_subactors(

    # expect another breakpoint actor entry
    child.sendline('c')
-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")

    try:
        assert_before(child, [bp_forever_msg])
@ -636,7 +652,7 @@ def test_multi_daemon_subactors(
        # after 1 or more further bp actor entries.

        child.sendline('c')
-        child.expect(PROMPT)
+        child.expect(r"\(Pdb\+\+\)")
        assert_before(child, [name_error_msg])

    # wait for final error in root
@ -644,7 +660,7 @@ def test_multi_daemon_subactors(
    while True:
        try:
            child.sendline('c')
-            child.expect(PROMPT)
+            child.expect(r"\(Pdb\+\+\)")
            assert_before(
                child,
                [bp_forever_msg]
@ -677,8 +693,8 @@ def test_multi_subactors_root_errors(
    '''
    child = spawn('multi_subactor_root_errors')

-    # scan for the prompt
-    child.expect(PROMPT)
+    # scan for the pdbpp prompt
+    child.expect(r"\(Pdb\+\+\)")

    # at most one subactor should attach before the root is cancelled
    before = str(child.before.decode())
@ -693,7 +709,7 @@ def test_multi_subactors_root_errors(

    # due to block list strat from #337, this will no longer
    # propagate before the root errors and cancels the spawner sub-tree.
-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")

    # only if the blocking condition doesn't kick in fast enough
    before = str(child.before.decode())
@ -708,7 +724,7 @@ def test_multi_subactors_root_errors(
            do_ctlc(child)

        child.sendline('c')
-        child.expect(PROMPT)
+        child.expect(r"\(Pdb\+\+\)")

    # check if the spawner crashed or was blocked from debug
    # and if this intermediary attached check the boxed error
@ -725,7 +741,7 @@ def test_multi_subactors_root_errors(
            do_ctlc(child)

        child.sendline('c')
-        child.expect(PROMPT)
+        child.expect(r"\(Pdb\+\+\)")

    # expect a root actor crash
    assert_before(child, [
@ -770,11 +786,9 @@ def test_multi_nested_subactors_error_through_nurseries(

    child = spawn('multi_nested_subactors_error_up_through_nurseries')

-    timed_out_early: bool = False
-
    for send_char in itertools.cycle(['c', 'q']):
        try:
-            child.expect(PROMPT)
+            child.expect(r"\(Pdb\+\+\)")
            child.sendline(send_char)
            time.sleep(0.01)

@ -816,7 +830,7 @@ def test_root_nursery_cancels_before_child_releases_tty_lock(

    child = spawn('root_cancelled_but_child_is_in_tty_lock')

-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")

    before = str(child.before.decode())
    assert "NameError: name 'doggypants' is not defined" in before
@ -831,7 +845,7 @@ def test_root_nursery_cancels_before_child_releases_tty_lock(
    for i in range(4):
        time.sleep(0.5)
        try:
-            child.expect(PROMPT)
+            child.expect(r"\(Pdb\+\+\)")

        except (
            EOF,
@ -888,7 +902,7 @@ def test_root_cancels_child_context_during_startup(
    '''
    child = spawn('fast_error_in_root_after_spawn')

-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")

    before = str(child.before.decode())
    assert "AssertionError" in before
@ -905,7 +919,7 @@ def test_different_debug_mode_per_actor(
    ctlc: bool,
 ):
    child = spawn('per_actor_debug')
-    child.expect(PROMPT)
+    child.expect(r"\(Pdb\+\+\)")

    # only one actor should enter the debugger
    before = str(child.before.decode())
--- a/tests/test_docs_examples.py
+++ b/tests/test_docs_examples.py
@ -12,17 +12,17 @@ import shutil

 import pytest

-from conftest import (
-    examples_dir,
-)
+from conftest import repodir
+
+
+def examples_dir():
+    """Return the abspath to the examples directory.
+    """
+    return os.path.join(repodir(), 'examples')


@pytest.fixture
-def run_example_in_subproc(
-    loglevel: str,
-    testdir,
-    arb_addr: tuple[str, int],
-):
+def run_example_in_subproc(loglevel, testdir, arb_addr):

    @contextmanager
    def run(script_code):
@ -32,8 +32,8 @@ def run_example_in_subproc(
            # on windows we need to create a special __main__.py which will
            # be executed with ``python -m <modulename>`` on windows..
            shutil.copyfile(
-                examples_dir() / '__main__.py',
-                str(testdir / '__main__.py'),
+                os.path.join(examples_dir(), '__main__.py'),
+                os.path.join(str(testdir), '__main__.py')
            )

            # drop the ``if __name__ == '__main__'`` guard onwards from
@ -88,20 +88,21 @@ def run_example_in_subproc(
        and f[0] != '_'
        and 'debugging' not in p[0]
        and 'integration' not in p[0]
-        and 'advanced_faults' not in p[0]
    ],

    ids=lambda t: t[1],
 )
 def test_example(run_example_in_subproc, example_script):
-    """Load and run scripts from this repo's ``examples/`` dir as a user
+    '''
+    Load and run scripts from this repo's ``examples/`` dir as a user
    would copy and pasing them into their editor.

    On windows a little more "finessing" is done to make
    ``multiprocessing`` play nice: we copy the ``__main__.py`` into the
    test directory and invoke the script as a module with ``python -m
    test_example``.
-    """
+
+    '''
    ex_file = os.path.join(*example_script)

    if 'rpc_bidir_streaming' in ex_file and sys.version_info < (3, 9):
@ -111,25 +112,32 @@ def test_example(run_example_in_subproc, example_script):
        code = ex.read()

        with run_example_in_subproc(code) as proc:
-            proc.wait()
-            err, _ = proc.stderr.read(), proc.stdout.read()
-            # print(f'STDERR: {err}')
-            # print(f'STDOUT: {out}')
-
-            # if we get some gnarly output let's aggregate and raise
-            if err:
+            try:
+                proc.wait(timeout=5)
+            finally:
+                err = proc.stderr.read()
                errmsg = err.decode()
-                errlines = errmsg.splitlines()
-                last_error = errlines[-1]
-                if (
-                    'Error' in last_error
+                out = proc.stdout.read()
+                outmsg = out.decode()

-                    # XXX: currently we print this to console, but maybe
-                    # shouldn't eventually once we figure out what's
-                    # a better way to be explicit about aio side
-                    # cancels?
-                    and 'asyncio.exceptions.CancelledError' not in last_error
-                ):
-                    raise Exception(errmsg)
+                if out:
+                    print(f'STDOUT: {out.decode()}')
+
+                # if we get some gnarly output let's aggregate and raise
+                if err:
+                    print(f'STDERR:\n{errmsg}')
+                    errmsg = err.decode()
+                    errlines = errmsg.splitlines()
+                    last_error = errlines[-1]
+                    if (
+                        'Error' in last_error
+
+                        # XXX: currently we print this to console, but maybe
+                        # shouldn't eventually once we figure out what's
+                        # a better way to be explicit about aio side
+                        # cancels?
+                        and 'asyncio.exceptions.CancelledError' not in last_error
+                    ):
+                        raise Exception(errmsg)

            assert proc.returncode == 0
--- a/tests/test_shm.py
+++ b/tests/test_shm.py
@ -0,0 +1,167 @@
+"""
+Shared mem primitives and APIs.
+
+"""
+import uuid
+
+# import numpy
+import pytest
+import trio
+import tractor
+from tractor._shm import (
+    open_shm_list,
+    attach_shm_list,
+)
+
+
+@tractor.context
+async def child_attach_shml_alot(
+    ctx: tractor.Context,
+    shm_key: str,
+) -> None:
+
+    await ctx.started(shm_key)
+
+    # now try to attach a boatload of times in a loop..
+    for _ in range(1000):
+        shml = attach_shm_list(
+            key=shm_key,
+            readonly=False,
+        )
+        assert shml.shm.name == shm_key
+        await trio.sleep(0.001)
+
+
+def test_child_attaches_alot():
+    async def main():
+        async with tractor.open_nursery() as an:
+
+            # allocate writeable list in parent
+            key = f'shml_{uuid.uuid4()}'
+            shml = open_shm_list(
+                key=key,
+            )
+
+            portal = await an.start_actor(
+                'shm_attacher',
+                enable_modules=[__name__],
+            )
+
+            async with (
+                portal.open_context(
+                    child_attach_shml_alot,
+                    shm_key=shml.key,
+                ) as (ctx, start_val),
+            ):
+                assert start_val == key
+                await ctx.result()
+
+            await portal.cancel_actor()
+
+    trio.run(main)
+
+
+@tractor.context
+async def child_read_shm_list(
+    ctx: tractor.Context,
+    shm_key: str,
+    use_str: bool,
+    frame_size: int,
+) -> None:
+
+    # attach in child
+    shml = attach_shm_list(
+        key=shm_key,
+        # dtype=str if use_str else float,
+    )
+    await ctx.started(shml.key)
+
+    async with ctx.open_stream() as stream:
+        async for i in stream:
+            print(f'(child): reading shm list index: {i}')
+
+            if use_str:
+                expect = str(float(i))
+            else:
+                expect = float(i)
+
+            if frame_size == 1:
+                val = shml[i]
+                assert expect == val
+                print(f'(child): reading value: {val}')
+            else:
+                frame = shml[i - frame_size:i]
+                print(f'(child): reading frame: {frame}')
+
+
+@pytest.mark.parametrize(
+    'use_str',
+    [False, True],
+    ids=lambda i: f'use_str_values={i}',
+)
+@pytest.mark.parametrize(
+    'frame_size',
+    [1, 2**6, 2**10],
+    ids=lambda i: f'frame_size={i}',
+)
+def test_parent_writer_child_reader(
+    use_str: bool,
+    frame_size: int,
+):
+
+    async def main():
+        async with tractor.open_nursery(
+            # debug_mode=True,
+        ) as an:
+
+            portal = await an.start_actor(
+                'shm_reader',
+                enable_modules=[__name__],
+                debug_mode=True,
+            )
+
+            # allocate writeable list in parent
+            key = 'shm_list'
+            seq_size = int(2 * 2 ** 10)
+            shml = open_shm_list(
+                key=key,
+                size=seq_size,
+                dtype=str if use_str else float,
+                readonly=False,
+            )
+
+            async with (
+                portal.open_context(
+                    child_read_shm_list,
+                    shm_key=key,
+                    use_str=use_str,
+                    frame_size=frame_size,
+                ) as (ctx, sent),
+
+                ctx.open_stream() as stream,
+            ):
+
+                assert sent == key
+
+                for i in range(seq_size):
+
+                    val = float(i)
+                    if use_str:
+                        val = str(val)
+
+                    # print(f'(parent): writing {val}')
+                    shml[i] = val
+
+                    # only on frame fills do we
+                    # signal to the child that a frame's
+                    # worth is ready.
+                    if (i % frame_size) == 0:
+                        print(f'(parent): signalling frame full on {val}')
+                        await stream.send(i)
+                else:
+                    print(f'(parent): signalling final frame on {val}')
+                    await stream.send(i)
+
+            await portal.cancel_actor()
+
+    trio.run(main)
--- a/tests/test_legacy_one_way_streaming.py
+++ b/tests/test_legacy_one_way_streaming.py
@ -251,7 +251,7 @@ def test_a_quadruple_example(time_quad_ex, ci_env, spawn_backend):

    results, diff = time_quad_ex
    assert results
-    this_fast = 6 if platform.system() in ('Windows', 'Darwin') else 3
+    this_fast = 6 if platform.system() in ('Windows', 'Darwin') else 2.666
    assert diff < this_fast


--- a/tests/test_task_broadcasting.py
+++ b/tests/test_task_broadcasting.py
@ -12,10 +12,7 @@ import pytest
 import trio
 from trio.lowlevel import current_task
 import tractor
-from tractor.trionics import (
-    broadcast_receiver,
-    Lagged,
-)
+from tractor.trionics import broadcast_receiver, Lagged


@tractor.context
@ -40,7 +37,7 @@ async def echo_sequences(

 async def ensure_sequence(

-    stream: tractor.MsgStream,
+    stream: tractor.ReceiveMsgStream,
    sequence: list,
    delay: Optional[float] = None,

@ -214,8 +211,7 @@ def test_faster_task_to_recv_is_cancelled_by_slower(
    arb_addr,
    start_method,
 ):
-    '''
-    Ensure that if a faster task consuming from a stream is cancelled
+    '''Ensure that if a faster task consuming from a stream is cancelled
    the slower task can continue to receive all expected values.

    '''
@ -464,51 +460,3 @@ def test_first_recver_is_cancelled():
                    assert value == 1

    trio.run(main)
-
-
-def test_no_raise_on_lag():
-    '''
-    Run a simple 2-task broadcast where one task is slow but configured
-    so that it does not raise `Lagged` on overruns using
-    `raise_on_lasg=False` and verify that the task does not raise.
-
-    '''
-    size = 100
-    tx, rx = trio.open_memory_channel(size)
-    brx = broadcast_receiver(rx, size)
-
-    async def slow():
-        async with brx.subscribe(
-            raise_on_lag=False,
-        ) as br:
-            async for msg in br:
-                print(f'slow task got: {msg}')
-                await trio.sleep(0.1)
-
-    async def fast():
-        async with brx.subscribe() as br:
-            async for msg in br:
-                print(f'fast task got: {msg}')
-
-    async def main():
-        async with (
-            tractor.open_root_actor(
-                # NOTE: so we see the warning msg emitted by the bcaster
-                # internals when the no raise flag is set.
-                loglevel='warning',
-            ),
-            trio.open_nursery() as n,
-        ):
-            n.start_soon(slow)
-            n.start_soon(fast)
-
-            for i in range(1000):
-                await tx.send(i)
-
-            # simulate user nailing ctl-c after realizing
-            # there's a lag in the slow task.
-            await trio.sleep(1)
-            raise KeyboardInterrupt
-
-    with pytest.raises(KeyboardInterrupt):
-        trio.run(main)
--- a/tractor/init.py
+++ b/tractor/init.py
@ -24,6 +24,7 @@ from ._clustering import open_actor_cluster
 from ._ipc import Channel
 from ._streaming import (
    Context,
+    ReceiveMsgStream,
    MsgStream,
    stream,
    context,
@ -44,10 +45,7 @@ from ._exceptions import (
    ModuleNotExposed,
    ContextCancelled,
 )
-from ._debug import (
-    breakpoint,
-    post_mortem,
-)
+from ._debug import breakpoint, post_mortem
 from . import msg
 from ._root import (
    run_daemon,
@ -66,6 +64,7 @@ __all__ = [
    'MsgStream',
    'BaseExceptionGroup',
    'Portal',
+    'ReceiveMsgStream',
    'RemoteActorError',
    'breakpoint',
    'context',
--- a/tractor/_debug.py
+++ b/tractor/_debug.py
@ -20,13 +20,9 @@ Multi-core debugging for da peeps!
 """
 from __future__ import annotations
 import bdb
-import os
 import sys
 import signal
-from functools import (
-    partial,
-    cached_property,
-)
+from functools import partial
 from contextlib import asynccontextmanager as acm
 from typing import (
    Any,
@ -37,7 +33,6 @@ from typing import (
 )
 from types import FrameType

-import pdbp
 import tractor
 import trio
 from trio_typing import TaskStatus
@ -54,6 +49,17 @@ from ._exceptions import (
 )
 from ._ipc import Channel

+
+try:
+    # wtf: only exported when installed in dev mode?
+    import pdbpp
+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__)


@ -67,7 +73,6 @@ class Lock:
    Mostly to avoid a lot of ``global`` declarations for now XD.

    '''
-    repl: MultiActorPdb | None = None
    # placeholder for function to set a ``trio.Event`` on debugger exit
    # pdb_release_hook: Optional[Callable] = None

@ -106,7 +111,7 @@ class Lock:
    def shield_sigint(cls):
        cls._orig_sigint_handler = signal.signal(
            signal.SIGINT,
-            shield_sigint_handler,
+            shield_sigint,
        )

    @classmethod
@ -141,29 +146,24 @@ class Lock:
        finally:
            # restore original sigint handler
            cls.unshield_sigint()
-            cls.repl = None


-class TractorConfig(pdbp.DefaultConfig):
+class TractorConfig(pdbpp.DefaultConfig):
    '''
-    Custom ``pdbp`` goodness :surfer:
+    Custom ``pdbpp`` goodness.

    '''
-    use_pygments: bool = True
-    sticky_by_default: bool = False
-    enable_hidden_frames: bool = False
-
-    # much thanks @mdmintz for the hot tip!
-    # fixes line spacing issue when resizing terminal B)
-    truncate_long_lines: bool = False
+    # use_pygments = True
+    # sticky_by_default = True
+    enable_hidden_frames = False


-class MultiActorPdb(pdbp.Pdb):
+class MultiActorPdb(pdbpp.Pdb):
    '''
-    Add teardown hooks to the regular ``pdbp.Pdb``.
+    Add teardown hooks to the regular ``pdbpp.Pdb``.

    '''
-    # override the pdbp config with our coolio one
+    # override the pdbpp config with our coolio one
    DefaultConfig = TractorConfig

    # def preloop(self):
@ -184,35 +184,6 @@ class MultiActorPdb(pdbp.Pdb):
        finally:
            Lock.release()

-    # XXX NOTE: we only override this because apparently the stdlib pdb
-    # bois likes to touch the SIGINT handler as much as i like to touch
-    # my d$%&.
-    def _cmdloop(self):
-        self.cmdloop()
-
-    @cached_property
-    def shname(self) -> str | None:
-        '''
-        Attempt to return the login shell name with a special check for
-        the infamous `xonsh` since it seems to have some issues much
-        different from std shells when it comes to flushing the prompt?
-
-        '''
-        # SUPER HACKY and only really works if `xonsh` is not used
-        # before spawning further sub-shells..
-        shpath = os.getenv('SHELL', None)
-
-        if shpath:
-            if (
-                os.getenv('XONSH_LOGIN', default=False)
-                or 'xonsh' in shpath
-            ):
-                return 'xonsh'
-
-            return os.path.basename(shpath)
-
-        return None
-

@acm
 async def _acquire_debug_lock_from_root_task(
@ -307,7 +278,7 @@ async def lock_tty_for_child(
 ) -> str:
    '''
    Lock the TTY in the root process of an actor tree in a new
-    inter-actor-context-task such that the ``pdbp`` debugger console
+    inter-actor-context-task such that the ``pdbpp`` debugger console
    can be mutex-allocated to the calling sub-actor for REPL control
    without interference by other processes / threads.

@ -417,7 +388,6 @@ async def wait_for_parent_stdin_hijack(

        except ContextCancelled:
            log.warning('Root actor cancelled debug lock')
-            raise

        finally:
            Lock.local_task_in_debug = None
@ -427,7 +397,7 @@ async def wait_for_parent_stdin_hijack(
 def mk_mpdb() -> tuple[MultiActorPdb, Callable]:

    pdb = MultiActorPdb()
-    # signal.signal = pdbp.hideframe(signal.signal)
+    # signal.signal = pdbpp.hideframe(signal.signal)

    Lock.shield_sigint()

@ -465,10 +435,7 @@ async def _breakpoint(
    # with trio.CancelScope(shield=shield):
    #     await trio.lowlevel.checkpoint()

-    if (
-        not Lock.local_pdb_complete
-        or Lock.local_pdb_complete.is_set()
-    ):
+    if not Lock.local_pdb_complete or Lock.local_pdb_complete.is_set():
        Lock.local_pdb_complete = trio.Event()

    # TODO: need a more robust check for the "root" actor
@ -517,7 +484,6 @@ async def _breakpoint(
                    wait_for_parent_stdin_hijack,
                    actor.uid,
                )
-                Lock.repl = pdb
        except RuntimeError:
            Lock.release()

@ -556,7 +522,6 @@ async def _breakpoint(

        Lock.global_actor_in_debug = actor.uid
        Lock.local_task_in_debug = task_name
-        Lock.repl = pdb

    try:
        # block here one (at the appropriate frame *up*) where
@ -577,13 +542,13 @@ async def _breakpoint(
    #     # frame = sys._getframe()
    #     # last_f = frame.f_back
    #     # last_f.f_globals['__tracebackhide__'] = True
-    #     # signal.signal = pdbp.hideframe(signal.signal)
+    #     # signal.signal = pdbpp.hideframe(signal.signal)


-def shield_sigint_handler(
+def shield_sigint(
    signum: int,
    frame: 'frame',  # type: ignore # noqa
-    # pdb_obj: Optional[MultiActorPdb] = None,
+    pdb_obj: Optional[MultiActorPdb] = None,
    *args,

 ) -> None:
@ -600,7 +565,6 @@ def shield_sigint_handler(
    uid_in_debug = Lock.global_actor_in_debug

    actor = tractor.current_actor()
-    # print(f'{actor.uid} in HANDLER with ')

    def do_cancel():
        # If we haven't tried to cancel the runtime then do that instead
@ -634,9 +598,6 @@ def shield_sigint_handler(
                )
                return do_cancel()

-    # only set in the actor actually running the REPL
-    pdb_obj = Lock.repl
-
    # root actor branch that reports whether or not a child
    # has locked debugger.
    if (
@ -651,34 +612,32 @@ def shield_sigint_handler(
    ):
        # we are root and some actor is in debug mode
        # if uid_in_debug is not None:
+        name = uid_in_debug[0]
+        if name != 'root':
+            log.pdb(
+                f"Ignoring SIGINT while child in debug mode: `{uid_in_debug}`"
+            )

-        if pdb_obj:
-            name = uid_in_debug[0]
-            if name != 'root':
-                log.pdb(
-                    f"Ignoring SIGINT, child in debug mode: `{uid_in_debug}`"
-                )
-
-            else:
-                log.pdb(
-                    "Ignoring SIGINT while in debug mode"
-                )
+        else:
+            log.pdb(
+                "Ignoring SIGINT while in debug mode"
+            )
    elif (
        is_root_process()
    ):
-        if pdb_obj:
-            log.pdb(
-                "Ignoring SIGINT since debug mode is enabled"
-            )
+        log.pdb(
+            "Ignoring SIGINT since debug mode is enabled"
+        )

+        # revert back to ``trio`` handler asap!
+        Lock.unshield_sigint()
        if (
            Lock._root_local_task_cs_in_debug
            and not Lock._root_local_task_cs_in_debug.cancel_called
        ):
            Lock._root_local_task_cs_in_debug.cancel()

-            # revert back to ``trio`` handler asap!
-            Lock.unshield_sigint()
+        # raise KeyboardInterrupt

    # child actor that has locked the debugger
    elif not is_root_process():
@ -694,10 +653,7 @@ def shield_sigint_handler(
            return do_cancel()

        task = Lock.local_task_in_debug
-        if (
-            task
-            and pdb_obj
-        ):
+        if task:
            log.pdb(
                f"Ignoring SIGINT while task in debug mode: `{task}`"
            )
@ -712,21 +668,14 @@ def shield_sigint_handler(
        raise KeyboardInterrupt

    # NOTE: currently (at least on ``fancycompleter`` 0.9.2)
-    # it looks to be that the last command that was run (eg. ll)
+    # it lookks to be that the last command that was run (eg. ll)
    # will be repeated by default.

-    # maybe redraw/print last REPL output to console since
-    # we want to alert the user that more input is expect since
-    # nothing has been done dur to ignoring sigint.
+    # TODO: maybe redraw/print last REPL output to console
    if (
-        pdb_obj  # only when this actor has a REPL engaged
+        pdb_obj
+        and sys.version_info <= (3, 10)
    ):
-        # XXX: yah, mega hack, but how else do we catch this madness XD
-        if pdb_obj.shname == 'xonsh':
-            pdb_obj.stdout.write(pdb_obj.prompt)
-
-        pdb_obj.stdout.flush()
-
        # TODO: make this work like sticky mode where if there is output
        # detected as written to the tty we redraw this part underneath
        # and erase the past draw of this same bit above?
@ -737,13 +686,21 @@ def shield_sigint_handler(
        # https://github.com/goodboy/tractor/issues/130#issuecomment-663752040
        # https://github.com/prompt-toolkit/python-prompt-toolkit/blob/c2c6af8a0308f9e5d7c0e28cb8a02963fe0ce07a/prompt_toolkit/patch_stdout.py

-        # XXX LEGACY: lol, see ``pdbpp`` issue:
+        # XXX: lol, see ``pdbpp`` issue:
        # https://github.com/pdbpp/pdbpp/issues/496

+        # TODO: pretty sure this is what we should expect to have to run
+        # in total but for now we're just going to wait until `pdbpp`
+        # figures out it's own stuff on 3.10 (and maybe we'll help).
+        # pdb_obj.do_longlist(None)
+
+        # XXX: we were doing this but it shouldn't be required..
+        print(pdb_obj.prompt, end='', flush=True)
+

 def _set_trace(
-    actor: tractor.Actor | None = None,
-    pdb: MultiActorPdb | None = None,
+    actor: Optional[tractor.Actor] = None,
+    pdb: Optional[MultiActorPdb] = None,
 ):
    __tracebackhide__ = True
    actor = actor or tractor.current_actor()
@ -753,11 +710,7 @@ def _set_trace(
    if frame:
        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")
        # no f!#$&* idea, but when we're in async land
        # we need 2x frames up?
@ -766,8 +719,7 @@ def _set_trace(
    else:
        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'

    pdb.set_trace(frame=frame)
@ -797,7 +749,7 @@ def _post_mortem(
    # https://github.com/pdbpp/pdbpp/issues/480
    # TODO: help with a 3.10+ major release if/when it arrives.

-    pdbp.xpm(Pdb=lambda: pdb)
+    pdbpp.xpm(Pdb=lambda: pdb)


 post_mortem = partial(
@ -868,10 +820,7 @@ async def maybe_wait_for_debugger(

 ) -> None:

-    if (
-        not debug_mode()
-        and not child_in_debug
-    ):
+    if not debug_mode() and not child_in_debug:
        return

    if (
--- a/tractor/_portal.py
+++ b/tractor/_portal.py
@ -45,10 +45,7 @@ from ._exceptions import (
    NoResult,
    ContextCancelled,
 )
-from ._streaming import (
-    Context,
-    MsgStream,
-)
+from ._streaming import Context, ReceiveMsgStream


 log = get_logger(__name__)
@ -104,7 +101,7 @@ class Portal:
        # it is expected that ``result()`` will be awaited at some
        # point.
        self._expect_result: Optional[Context] = None
-        self._streams: set[MsgStream] = set()
+        self._streams: set[ReceiveMsgStream] = set()
        self.actor = current_actor()

    async def _submit_for_result(
@ -319,7 +316,7 @@ class Portal:
        async_gen_func: Callable,  # typing: ignore
        **kwargs,

-    ) -> AsyncGenerator[MsgStream, None]:
+    ) -> AsyncGenerator[ReceiveMsgStream, None]:

        if not inspect.isasyncgenfunction(async_gen_func):
            if not (
@ -344,7 +341,7 @@ class Portal:

        try:
            # deliver receive only stream
-            async with MsgStream(
+            async with ReceiveMsgStream(
                ctx, ctx._recv_chan,
            ) as rchan:
                self._streams.add(rchan)
@ -500,10 +497,6 @@ class Portal:
                    f'actor: {uid}'
                )
                result = await ctx.result()
-                log.runtime(
-                    f'Context {fn_name} returned '
-                    f'value from callee `{result}`'
-                )

            # though it should be impossible for any tasks
            # operating *in* this scope to have survived
@ -525,6 +518,12 @@ class Portal:
                        f'task:{cid}\n'
                        f'actor:{uid}'
                    )
+            else:
+                log.runtime(
+                    f'Context {fn_name} returned '
+                    f'value from callee `{result}`'
+                )
+
            # XXX: (MEGA IMPORTANT) if this is a root opened process we
            # wait for any immediate child in debug before popping the
            # context from the runtime msg loop otherwise inside
--- a/tractor/_root.py
+++ b/tractor/_root.py
@ -22,9 +22,11 @@ from contextlib import asynccontextmanager
 from functools import partial
 import importlib
 import logging
-import signal
-import sys
 import os
+import signal
+from typing import (
+    Optional,
+)
 import typing
 import warnings

@ -56,28 +58,27 @@ logger = log.get_logger('tractor')
@asynccontextmanager
 async def open_root_actor(

-    *,
    # defaults are above
-    arbiter_addr: tuple[str, int] | None = None,
+    arbiter_addr: Optional[tuple[str, int]] = (
+        _default_arbiter_host,
+        _default_arbiter_port,
+    ),

-    # defaults are above
-    registry_addr: tuple[str, int] | None = None,
-
-    name: str | None = 'root',
+    name: Optional[str] = 'root',

    # either the `multiprocessing` start method:
    # https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods
    # OR `trio` (the new default).
-    start_method: _spawn.SpawnMethodKey | None = None,
+    start_method: Optional[_spawn.SpawnMethodKey] = None,

    # enables the multi-process debugger support
    debug_mode: bool = False,

    # internal logging
-    loglevel: str | None = None,
+    loglevel: Optional[str] = None,

-    enable_modules: list | None = None,
-    rpc_module_paths: list | None = None,
+    enable_modules: Optional[list] = None,
+    rpc_module_paths: Optional[list] = None,

 ) -> typing.Any:
    '''
@ -85,15 +86,13 @@ async def open_root_actor(

    '''
    # Override the global debugger hook to make it play nice with
-    # ``trio``, see much discussion in:
+    # ``trio``, see:
    # 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'

    # attempt to retreive ``trio``'s sigint handler and stash it
    # on our debugger lock state.
-    _debug.Lock._trio_handler = signal.getsignal(signal.SIGINT)
+    _debug.Lock._trio_handler =  signal.getsignal(signal.SIGINT)

    # mark top most level process as root actor
    _state._runtime_vars['_is_root'] = True
@ -113,21 +112,9 @@ async def open_root_actor(
    if start_method is not None:
        _spawn.try_set_start_method(start_method)

-    if arbiter_addr is not None:
-        warnings.warn(
-            '`arbiter_addr` is now deprecated and has been renamed to'
-            '`registry_addr`.\nUse that instead..',
-            DeprecationWarning,
-            stacklevel=2,
-        )
-
-    registry_addr = (host, port) = (
-        registry_addr
-        or arbiter_addr
-        or (
-            _default_arbiter_host,
-            _default_arbiter_port,
-        )
+    arbiter_addr = (host, port) = arbiter_addr or (
+        _default_arbiter_host,
+        _default_arbiter_port,
    )

    loglevel = (loglevel or log._default_loglevel).upper()
@ -173,7 +160,7 @@ async def open_root_actor(

    except OSError:
        # TODO: make this a "discovery" log level?
-        logger.warning(f"No actor registry found @ {host}:{port}")
+        logger.warning(f"No actor could be found @ {host}:{port}")

    # create a local actor and start up its main routine/task
    if arbiter_found:
@ -183,7 +170,7 @@ async def open_root_actor(

        actor = Actor(
            name or 'anonymous',
-            arbiter_addr=registry_addr,
+            arbiter_addr=arbiter_addr,
            loglevel=loglevel,
            enable_modules=enable_modules,
        )
@ -199,7 +186,7 @@ async def open_root_actor(

        actor = Arbiter(
            name or 'arbiter',
-            arbiter_addr=registry_addr,
+            arbiter_addr=arbiter_addr,
            loglevel=loglevel,
            enable_modules=enable_modules,
        )
@ -257,15 +244,6 @@ async def open_root_actor(
                await actor.cancel()
    finally:
        _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")


@ -273,13 +251,13 @@ def run_daemon(
    enable_modules: list[str],

    # runtime kwargs
-    name: str | None = 'root',
-    registry_addr: tuple[str, int] = (
+    name: Optional[str] = 'root',
+    arbiter_addr: tuple[str, int] = (
        _default_arbiter_host,
        _default_arbiter_port,
    ),

-    start_method: str | None = None,
+    start_method: Optional[str] = None,
    debug_mode: bool = False,
    **kwargs

@ -301,7 +279,7 @@ def run_daemon(
    async def _main():

        async with open_root_actor(
-            registry_addr=registry_addr,
+            arbiter_addr=arbiter_addr,
            name=name,
            start_method=start_method,
            debug_mode=debug_mode,
--- a/tractor/_runtime.py
+++ b/tractor/_runtime.py
@ -228,11 +228,11 @@ async def _invoke(

                fname = func.__name__
                if ctx._cancel_called:
-                    msg = f'`{fname}()` cancelled itself'
+                    msg = f'{fname} cancelled itself'

                elif cs.cancel_called:
                    msg = (
-                        f'`{fname}()` was remotely cancelled by its caller '
+                        f'{fname} was remotely cancelled by its caller '
                        f'{ctx.chan.uid}'
                    )

@ -319,7 +319,7 @@ async def _invoke(
            BrokenPipeError,
        ):
            # if we can't propagate the error that's a big boo boo
-            log.exception(
+            log.error(
                f"Failed to ship error to caller @ {chan.uid} !?"
            )

@ -455,7 +455,7 @@ class Actor:
        self._mods: dict[str, ModuleType] = {}
        self.loglevel = loglevel

-        self._arb_addr: tuple[str, int] | None = (
+        self._arb_addr = (
            str(arbiter_addr[0]),
            int(arbiter_addr[1])
        ) if arbiter_addr else None
@ -1384,7 +1384,6 @@ async def async_main(
            and not actor.is_arbiter
        ):
            failed = False
-            assert isinstance(actor._arb_addr, tuple)
            with trio.move_on_after(0.5) as cs:
                cs.shield = True
                try:
@ -1609,10 +1608,7 @@ async def process_messages(
        # handshake for them (yet) and instead we simply bail out of
        # the message loop and expect the teardown sequence to clean
        # up.
-        log.runtime(
-            f'channel from {chan.uid} closed abruptly:\n'
-            f'-> {chan.raddr}\n'
-        )
+        log.runtime(f'channel from {chan.uid} closed abruptly:\n{chan}')

        # transport **was** disconnected
        return True
--- a/tractor/_shm.py
+++ b/tractor/_shm.py
@ -0,0 +1,828 @@
+# tractor: structured concurrent "actors".
+# Copyright 2018-eternity Tyler Goodlet.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+"""
+SC friendly shared memory management geared at real-time
+processing.
+
+Support for ``numpy`` compatible array-buffers is provided but is
+considered optional within the context of this runtime-library.
+
+"""
+from __future__ import annotations
+from sys import byteorder
+import time
+from typing import Optional
+from multiprocessing import shared_memory as shm
+from multiprocessing.shared_memory import (
+    SharedMemory,
+    ShareableList,
+)
+
+from msgspec import Struct
+import tractor
+
+from .log import get_logger
+
+
+_USE_POSIX = getattr(shm, '_USE_POSIX', False)
+if _USE_POSIX:
+    from _posixshmem import shm_unlink
+
+
+try:
+    import numpy as np
+    from numpy.lib import recfunctions as rfn
+    import nptyping
+except ImportError:
+    pass
+
+
+log = get_logger(__name__)
+
+
+def disable_mantracker():
+    '''
+    Disable all ``multiprocessing``` "resource tracking" machinery since
+    it's an absolute multi-threaded mess of non-SC madness.
+
+    '''
+    from multiprocessing import resource_tracker as mantracker
+
+    # Tell the "resource tracker" thing to fuck off.
+    class ManTracker(mantracker.ResourceTracker):
+        def register(self, name, rtype):
+            pass
+
+        def unregister(self, name, rtype):
+            pass
+
+        def ensure_running(self):
+            pass
+
+    # "know your land and know your prey"
+    # https://www.dailymotion.com/video/x6ozzco
+    mantracker._resource_tracker = ManTracker()
+    mantracker.register = mantracker._resource_tracker.register
+    mantracker.ensure_running = mantracker._resource_tracker.ensure_running
+    mantracker.unregister = mantracker._resource_tracker.unregister
+    mantracker.getfd = mantracker._resource_tracker.getfd
+
+
+disable_mantracker()
+
+
+class SharedInt:
+    '''
+    Wrapper around a single entry shared memory array which
+    holds an ``int`` value used as an index counter.
+
+    '''
+    def __init__(
+        self,
+        shm: SharedMemory,
+    ) -> None:
+        self._shm = shm
+
+    @property
+    def value(self) -> int:
+        return int.from_bytes(self._shm.buf, byteorder)
+
+    @value.setter
+    def value(self, value) -> None:
+        self._shm.buf[:] = value.to_bytes(self._shm.size, byteorder)
+
+    def destroy(self) -> None:
+        if _USE_POSIX:
+            # 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?')
+
+
+class NDToken(Struct, frozen=True):
+    '''
+    Internal represenation of a shared memory ``numpy`` array "token"
+    which can be used to key and load a system (OS) wide shm entry
+    and correctly read the array by type signature.
+
+    This type is msg safe.
+
+    '''
+    shm_name: str  # this servers as a "key" value
+    shm_first_index_name: str
+    shm_last_index_name: str
+    dtype_descr: tuple
+    size: int  # in struct-array index / row terms
+
+    # TODO: use nptyping here on dtypes
+    @property
+    def dtype(self) -> list[tuple[str, str, tuple[int, ...]]]:
+        return np.dtype(
+            list(
+                map(tuple, self.dtype_descr)
+            )
+        ).descr
+
+    def as_msg(self):
+        return self.to_dict()
+
+    @classmethod
+    def from_msg(cls, msg: dict) -> NDToken:
+        if isinstance(msg, NDToken):
+            return msg
+
+        # TODO: native struct decoding
+        # return _token_dec.decode(msg)
+
+        msg['dtype_descr'] = tuple(map(tuple, msg['dtype_descr']))
+        return NDToken(**msg)
+
+
+# _token_dec = msgspec.msgpack.Decoder(NDToken)
+
+# TODO: this api?
+# _known_tokens = tractor.ActorVar('_shm_tokens', {})
+# _known_tokens = tractor.ContextStack('_known_tokens', )
+# _known_tokens = trio.RunVar('shms', {})
+
+# TODO: this should maybe be provided via
+# a `.trionics.maybe_open_context()` wrapper factory?
+# process-local store of keys to tokens
+_known_tokens: dict[str, NDToken] = {}
+
+
+def get_shm_token(key: str) -> NDToken | None:
+    '''
+    Convenience func to check if a token
+    for the provided key is known by this process.
+
+    Returns either the ``numpy`` token or a string for a shared list.
+
+    '''
+    return _known_tokens.get(key)
+
+
+def _make_token(
+    key: str,
+    size: int,
+    dtype: np.dtype,
+
+) -> NDToken:
+    '''
+    Create a serializable token that can be used
+    to access a shared array.
+
+    '''
+    return NDToken(
+        shm_name=key,
+        shm_first_index_name=key + "_first",
+        shm_last_index_name=key + "_last",
+        dtype_descr=tuple(np.dtype(dtype).descr),
+        size=size,
+    )
+
+
+class ShmArray:
+    '''
+    A shared memory ``numpy.ndarray`` API.
+
+    An underlying shared memory buffer is allocated based on
+    a user specified ``numpy.ndarray``. This fixed size array
+    can be read and written to by pushing data both onto the "front"
+    or "back" of a set index range. The indexes for the "first" and
+    "last" index are themselves stored in shared memory (accessed via
+    ``SharedInt`` interfaces) values such that multiple processes can
+    interact with the same array using a synchronized-index.
+
+    '''
+    def __init__(
+        self,
+        shmarr: np.ndarray,
+        first: SharedInt,
+        last: SharedInt,
+        shm: SharedMemory,
+        # readonly: bool = True,
+    ) -> None:
+        self._array = shmarr
+
+        # indexes for first and last indices corresponding
+        # to fille data
+        self._first = first
+        self._last = last
+
+        self._len = len(shmarr)
+        self._shm = shm
+        self._post_init: bool = False
+
+        # pushing data does not write the index (aka primary key)
+        self._write_fields: list[str] | None = None
+        dtype = shmarr.dtype
+        if dtype.fields:
+            self._write_fields = list(shmarr.dtype.fields.keys())[1:]
+
+    # TODO: ringbuf api?
+
+    @property
+    def _token(self) -> NDToken:
+        return NDToken(
+            shm_name=self._shm.name,
+            shm_first_index_name=self._first._shm.name,
+            shm_last_index_name=self._last._shm.name,
+            dtype_descr=tuple(self._array.dtype.descr),
+            size=self._len,
+        )
+
+    @property
+    def token(self) -> dict:
+        """Shared memory token that can be serialized and used by
+        another process to attach to this array.
+        """
+        return self._token.as_msg()
+
+    @property
+    def index(self) -> int:
+        return self._last.value % self._len
+
+    @property
+    def array(self) -> np.ndarray:
+        '''
+        Return an up-to-date ``np.ndarray`` view of the
+        so-far-written data to the underlying shm buffer.
+
+        '''
+        a = self._array[self._first.value:self._last.value]
+
+        # first, last = self._first.value, self._last.value
+        # a = self._array[first:last]
+
+        # TODO: eventually comment this once we've not seen it in the
+        # wild in a long time..
+        # XXX: race where first/last indexes cause a reader
+        # to load an empty array..
+        if len(a) == 0 and self._post_init:
+            raise RuntimeError('Empty array race condition hit!?')
+            # breakpoint()
+
+        return a
+
+    def ustruct(
+        self,
+        fields: Optional[list[str]] = None,
+
+        # type that all field values will be cast to in the returned
+        # view.
+        common_dtype: np.dtype = np.float64,  # type: ignore
+
+    ) -> np.ndarray:
+
+        array = self._array
+
+        if fields:
+            selection = array[fields]
+            # fcount = len(fields)
+        else:
+            selection = array
+            # fcount = len(array.dtype.fields)
+
+        # XXX: manual ``.view()`` attempt that also doesn't work.
+        # uview = selection.view(
+        #     dtype='<f16',
+        # ).reshape(-1, 4, order='A')
+
+        # assert len(selection) == len(uview)
+
+        u = rfn.structured_to_unstructured(
+            selection,
+            # dtype=float,
+            copy=True,
+        )
+
+        # unstruct = np.ndarray(u.shape, dtype=a.dtype, buffer=shm.buf)
+        # array[:] = a[:]
+        return u
+        # return ShmArray(
+        #     shmarr=u,
+        #     first=self._first,
+        #     last=self._last,
+        #     shm=self._shm
+        # )
+
+    def last(
+        self,
+        length: int = 1,
+
+    ) -> np.ndarray:
+        '''
+        Return the last ``length``'s worth of ("row") entries from the
+        array.
+
+        '''
+        return self.array[-length:]
+
+    def push(
+        self,
+        data: np.ndarray,
+
+        field_map: Optional[dict[str, str]] = None,
+        prepend: bool = False,
+        update_first: bool = True,
+        start: Optional[int] = None,
+
+    ) -> int:
+        '''
+        Ring buffer like "push" to append data
+        into the buffer and return updated "last" index.
+
+        NB: no actual ring logic yet to give a "loop around" on overflow
+        condition, lel.
+
+        '''
+        length = len(data)
+
+        if prepend:
+            index = (start or self._first.value) - length
+
+            if index < 0:
+                raise ValueError(
+                    f'Array size of {self._len} was overrun during prepend.\n'
+                    f'You have passed {abs(index)} too many datums.'
+                )
+
+        else:
+            index = start if start is not None else self._last.value
+
+        end = index + length
+
+        if field_map:
+            src_names, dst_names = zip(*field_map.items())
+        else:
+            dst_names = src_names = self._write_fields
+
+        try:
+            self._array[
+                list(dst_names)
+            ][index:end] = data[list(src_names)][:]
+
+            # NOTE: there was a race here between updating
+            # the first and last indices and when the next reader
+            # tries to access ``.array`` (which due to the index
+            # overlap will be empty). Pretty sure we've fixed it now
+            # but leaving this here as a reminder.
+            if prepend and update_first and length:
+                assert index < self._first.value
+
+            if (
+                index < self._first.value
+                and update_first
+            ):
+                assert prepend, 'prepend=True not passed but index decreased?'
+                self._first.value = index
+
+            elif not prepend:
+                self._last.value = end
+
+            self._post_init = True
+            return end
+
+        except ValueError as err:
+            if field_map:
+                raise
+
+            # should raise if diff detected
+            self.diff_err_fields(data)
+            raise err
+
+    def diff_err_fields(
+        self,
+        data: np.ndarray,
+    ) -> None:
+        # reraise with any field discrepancy
+        our_fields, their_fields = (
+            set(self._array.dtype.fields),
+            set(data.dtype.fields),
+        )
+
+        only_in_ours = our_fields - their_fields
+        only_in_theirs = their_fields - our_fields
+
+        if only_in_ours:
+            raise TypeError(
+                f"Input array is missing field(s): {only_in_ours}"
+            )
+        elif only_in_theirs:
+            raise TypeError(
+                f"Input array has unknown field(s): {only_in_theirs}"
+            )
+
+    # TODO: support "silent" prepends that don't update ._first.value?
+    def prepend(
+        self,
+        data: np.ndarray,
+    ) -> int:
+        end = self.push(data, prepend=True)
+        assert end
+
+    def close(self) -> None:
+        self._first._shm.close()
+        self._last._shm.close()
+        self._shm.close()
+
+    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)
+
+        self._first.destroy()
+        self._last.destroy()
+
+    def flush(self) -> None:
+        # TODO: flush to storage backend like markestore?
+        ...
+
+
+def open_shm_ndarray(
+    key: Optional[str] = None,
+    size: int = int(2 ** 10),
+    dtype: np.dtype | None = None,
+    append_start_index: int = 0,
+    readonly: bool = False,
+
+) -> ShmArray:
+    '''
+    Open a memory shared ``numpy`` using the standard library.
+
+    This call unlinks (aka permanently destroys) the buffer on teardown
+    and thus should be used from the parent-most accessor (process).
+
+    '''
+    # create new shared mem segment for which we
+    # have write permission
+    a = np.zeros(size, dtype=dtype)
+    a['index'] = np.arange(len(a))
+
+    shm = SharedMemory(
+        name=key,
+        create=True,
+        size=a.nbytes
+    )
+    array = np.ndarray(
+        a.shape,
+        dtype=a.dtype,
+        buffer=shm.buf
+    )
+    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(
+            name=token.shm_first_index_name,
+            create=True,
+            size=4,  # std int
+        )
+    )
+
+    last = SharedInt(
+        shm=SharedMemory(
+            name=token.shm_last_index_name,
+            create=True,
+            size=4,  # std int
+        )
+    )
+
+    # Start the "real-time" append-updated (or "pushed-to") section
+    # after some start index: ``append_start_index``. This allows appending
+    # from a start point in the array which isn't the 0 index and looks
+    # something like,
+    # -------------------------
+    # |              |        i
+    # _________________________
+    # <-------------> <------->
+    #  history         real-time
+    #
+    # Once fully "prepended", the history section will leave the
+    # ``ShmArray._start.value: int = 0`` and the yet-to-be written
+    # real-time section will start at ``ShmArray.index: int``.
+
+    # this sets the index to 3/4 of the length of the buffer
+    # leaving a "days worth of second samples" for the real-time
+    # section.
+    last.value = first.value = append_start_index
+
+    shmarr = ShmArray(
+        array,
+        first,
+        last,
+        shm,
+    )
+
+    assert shmarr._token == token
+    _known_tokens[key] = shmarr.token
+
+    # "unlink" created shm on process teardown by
+    # pushing teardown calls onto actor context stack
+    stack = tractor.current_actor().lifetime_stack
+    stack.callback(shmarr.close)
+    stack.callback(shmarr.destroy)
+
+    return shmarr
+
+
+def attach_shm_ndarray(
+    token: tuple[str, str, tuple[str, str]],
+    readonly: bool = True,
+
+) -> ShmArray:
+    '''
+    Attach to an existing shared memory array previously
+    created by another process using ``open_shared_array``.
+
+    No new shared mem is allocated but wrapper types for read/write
+    access are constructed.
+
+    '''
+    token = NDToken.from_msg(token)
+    key = token.shm_name
+
+    if key in _known_tokens:
+        assert NDToken.from_msg(_known_tokens[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:
+    # https://stackoverflow.com/a/11103289
+
+    # attach to array buffer and view as per dtype
+    _err: Optional[Exception] = None
+    for _ in range(3):
+        try:
+            shm = SharedMemory(
+                name=key,
+                create=False,
+            )
+            break
+        except OSError as oserr:
+            _err = oserr
+            time.sleep(0.1)
+    else:
+        if _err:
+            raise _err
+
+    shmarr = np.ndarray(
+        (token.size,),
+        dtype=token.dtype,
+        buffer=shm.buf
+    )
+    shmarr.setflags(write=int(not readonly))
+
+    first = SharedInt(
+        shm=SharedMemory(
+            name=token.shm_first_index_name,
+            create=False,
+            size=4,  # std int
+        ),
+    )
+    last = SharedInt(
+        shm=SharedMemory(
+            name=token.shm_last_index_name,
+            create=False,
+            size=4,  # std int
+        ),
+    )
+
+    # make sure we can read
+    first.value
+
+    sha = ShmArray(
+        shmarr,
+        first,
+        last,
+        shm,
+    )
+    # read test
+    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
+
+    # "close" attached shm on actor teardown
+    tractor.current_actor().lifetime_stack.callback(sha.close)
+
+    return sha
+
+
+def maybe_open_shm_ndarray(
+    key: str,  # unique identifier for segment
+
+    # from ``open_shm_array()``
+    size: int = int(2 ** 10),  # array length in index terms
+    dtype: np.dtype | None = None,
+    append_start_index: int = 0,
+    readonly: bool = True,
+
+) -> tuple[ShmArray, bool]:
+    '''
+    Attempt to attach to a shared memory block using a "key" lookup
+    to registered blocks in the users overall "system" registry
+    (presumes you don't have the block's explicit token).
+
+    This function is meant to solve the problem of discovering whether
+    a shared array token has been allocated or discovered by the actor
+    running in **this** process. Systems where multiple actors may seek
+    to access a common block can use this function to attempt to acquire
+    a token as discovered by the actors who have previously stored
+    a "key" -> ``NDToken`` map in an actor local (aka python global)
+    variable.
+
+    If you know the explicit ``NDToken`` for your memory segment instead
+    use ``attach_shm_array``.
+
+    '''
+    try:
+        # see if we already know this key
+        token = _known_tokens[key]
+        return (
+            attach_shm_ndarray(
+                token=token,
+                readonly=readonly,
+            ),
+            False,  # not newly opened
+        )
+    except KeyError:
+        log.warning(f"Could not find {key} in shms cache")
+        if dtype:
+            token = _make_token(
+                key,
+                size=size,
+                dtype=dtype,
+            )
+        else:
+
+            try:
+                return (
+                    attach_shm_ndarray(
+                        token=token,
+                        readonly=readonly,
+                    ),
+                    False,
+                )
+            except FileNotFoundError:
+                log.warning(f"Could not attach to shm with token {token}")
+
+        # This actor does not know about memory
+        # associated with the provided "key".
+        # Attempt to open a block and expect
+        # to fail if a block has been allocated
+        # on the OS by someone else.
+        return (
+            open_shm_ndarray(
+                key=key,
+                size=size,
+                dtype=dtype,
+                append_start_index=append_start_index,
+                readonly=readonly,
+            ),
+            True,
+        )
+
+
+class ShmList(ShareableList):
+    '''
+    Carbon copy of ``.shared_memory.ShareableList`` with a few
+    enhancements:
+
+    - readonly mode via instance var flag  `._readonly: bool`
+    - ``.__getitem__()`` accepts ``slice`` inputs
+    - exposes the underlying buffer "name" as a ``.key: str``
+
+    '''
+    def __init__(
+        self,
+        sequence: list | None = None,
+        *,
+        name: str | None = None,
+        readonly: bool = True
+
+    ) -> None:
+        self._readonly = readonly
+        self._key = name
+        return super().__init__(
+            sequence=sequence,
+            name=name,
+        )
+
+    @property
+    def key(self) -> str:
+        return self._key
+
+    @property
+    def readonly(self) -> bool:
+        return self._readonly
+
+    def __setitem__(
+        self,
+        position,
+        value,
+
+    ) -> None:
+
+        # mimick ``numpy`` error
+        if self._readonly:
+            raise ValueError('assignment destination is read-only')
+
+        return super().__setitem__(position, value)
+
+    def __getitem__(
+        self,
+        indexish,
+    ) -> list:
+
+        # NOTE: this is a non-writeable view (copy?) of the buffer
+        # in a new list instance.
+        if isinstance(indexish, slice):
+            return list(self)[indexish]
+
+        return super().__getitem__(indexish)
+
+    # TODO: should we offer a `.array` and `.push()` equivalent
+    # to the `ShmArray`?
+    # currently we have the following limitations:
+    # - can't write slices of input using traditional slice-assign
+    #   syntax due to the ``ShareableList.__setitem__()`` implementation.
+    # - ``list(shmlist)`` returns a non-mutable copy instead of
+    #   a writeable view which would be handier numpy-style ops.
+
+
+def open_shm_list(
+    key: str,
+    sequence: list | None = None,
+    size: int = int(2 ** 10),
+    dtype: float | int | bool | str | bytes | None = float,
+    readonly: bool = True,
+
+) -> ShmList:
+
+    if sequence is None:
+        default = {
+            float: 0.,
+            int: 0,
+            bool: True,
+            str: 'doggy',
+            None: None,
+        }[dtype]
+        sequence = [default] * size
+
+    shml = ShmList(
+        sequence=sequence,
+        name=key,
+        readonly=readonly,
+    )
+
+    # "close" attached shm on actor teardown
+    try:
+        actor = tractor.current_actor()
+        actor.lifetime_stack.callback(shml.shm.close)
+        actor.lifetime_stack.callback(shml.shm.unlink)
+    except RuntimeError:
+        log.warning('tractor runtime not active, skipping teardown steps')
+
+    return shml
+
+
+def attach_shm_list(
+    key: str,
+    readonly: bool = False,
+
+) -> ShmList:
+
+    return ShmList(
+        name=key,
+        readonly=readonly,
+    )
--- a/tractor/_spawn.py
+++ b/tractor/_spawn.py
@ -23,12 +23,13 @@ import sys
 import platform
 from typing import (
    Any,
-    Awaitable,
    Literal,
+    Optional,
    Callable,
    TypeVar,
    TYPE_CHECKING,
 )
+from collections.abc import Awaitable

 from exceptiongroup import BaseExceptionGroup
 import trio
@ -59,7 +60,7 @@ if TYPE_CHECKING:
 log = get_logger('tractor')

 # placeholder for an mp start context if so using that backend
-_ctx: mp.context.BaseContext | None = None
+_ctx: Optional[mp.context.BaseContext] = None
 SpawnMethodKey = Literal[
    'trio',  # supported on all platforms
    'mp_spawn',
@ -85,7 +86,7 @@ else:
 def try_set_start_method(
    key: SpawnMethodKey

-) -> mp.context.BaseContext | None:
+) -> Optional[mp.context.BaseContext]:
    '''
    Attempt to set the method for process starting, aka the "actor
    spawning backend".
@ -199,37 +200,16 @@ async def cancel_on_completion(
 async def do_hard_kill(
    proc: trio.Process,
    terminate_after: int = 3,
-
 ) -> None:
    # NOTE: this timeout used to do nothing since we were shielding
    # the ``.wait()`` inside ``new_proc()`` which will pretty much
    # never release until the process exits, now it acts as
    # a hard-kill time ultimatum.
-    log.debug(f"Terminating {proc}")
    with trio.move_on_after(terminate_after) as cs:

-        # NOTE: code below was copied verbatim from the now deprecated
-        # (in 0.20.0) ``trio._subrocess.Process.aclose()``, orig doc
-        # 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()
+        # NOTE: This ``__aexit__()`` shields internally.
+        async with proc:  # calls ``trio.Process.aclose()``
+            log.debug(f"Terminating {proc}")

    if cs.cancelled_caught:
        # XXX: should pretty much never get here unless we have
@ -280,9 +260,7 @@ async def soft_wait(

            if proc.poll() is None:  # type: ignore
                log.warning(
-                    'Actor still alive after cancel request:\n'
-                    f'{uid}'
-                )
+                    f'Process still alive after cancel request:\n{uid}')

                n.cancel_scope.cancel()
        raise
@ -375,11 +353,12 @@ async def trio_proc(
        spawn_cmd.append("--asyncio")

    cancelled_during_spawn: bool = False
-    proc: trio.Process | None = None
+    proc: Optional[trio.Process] = None
    try:
        try:
            # TODO: needs ``trio_typing`` patch?
-            proc = await trio.lowlevel.open_process(spawn_cmd)
+            proc = await trio.lowlevel.open_process(    # type: ignore
+                spawn_cmd)

            log.runtime(f"Started {proc}")

@ -463,8 +442,8 @@ async def trio_proc(
            nursery.cancel_scope.cancel()

    finally:
-        # XXX NOTE XXX: The "hard" reap since no actor zombies are
-        # allowed! Do this **after** cancellation/teardown to avoid
+        # The "hard" reap since no actor zombies are allowed!
+        # XXX: do this **after** cancellation/tearfown to avoid
        # killing the process too early.
        if proc:
            log.cancel(f'Hard reap sequence starting for {subactor.uid}')
@ -478,13 +457,6 @@ async def trio_proc(
                            await proc.wait()

                if is_root_process():
-                    # TODO: solve the following issue where we need
-                    # to do a similar wait like this but in an
-                    # "intermediary" parent actor that itself isn't
-                    # in debug but has a child that is, and we need
-                    # to hold off on relaying SIGINT until that child
-                    # is complete.
-                    # https://github.com/goodboy/tractor/issues/320
                    await maybe_wait_for_debugger(
                        child_in_debug=_runtime_vars.get(
                            '_debug_mode', False),
--- a/tractor/_streaming.py
+++ b/tractor/_streaming.py
@ -50,13 +50,12 @@ log = get_logger(__name__)
 # - use __slots__ on ``Context``?


-class MsgStream(trio.abc.Channel):
+class ReceiveMsgStream(trio.abc.ReceiveChannel):
    '''
-    A bidirectional message stream for receiving logically sequenced
-    values over an inter-actor IPC ``Channel``.
-
-    This is the type returned to a local task which entered either
-    ``Portal.open_stream_from()`` or ``Context.open_stream()``.
+    A IPC message stream for receiving logically sequenced values over
+    an inter-actor ``Channel``. This is the type returned to a local
+    task which entered either ``Portal.open_stream_from()`` or
+    ``Context.open_stream()``.

    Termination rules:

@ -98,9 +97,6 @@ class MsgStream(trio.abc.Channel):
        if self._eoc:
            raise trio.EndOfChannel

-        if self._closed:
-            raise trio.ClosedResourceError('This stream was closed')
-
        try:
            msg = await self._rx_chan.receive()
            return msg['yield']
@ -114,9 +110,6 @@ class MsgStream(trio.abc.Channel):
            # - 'error'
            # possibly just handle msg['stop'] here!

-            if self._closed:
-                raise trio.ClosedResourceError('This stream was closed')
-
            if msg.get('stop') or self._eoc:
                log.debug(f"{self} was stopped at remote end")

@ -196,6 +189,7 @@ class MsgStream(trio.abc.Channel):
            return

        self._eoc = True
+        self._closed = True

        # NOTE: this is super subtle IPC messaging stuff:
        # Relay stop iteration to far end **iff** we're
@ -212,32 +206,29 @@ class MsgStream(trio.abc.Channel):

        # In the bidirectional case, `Context.open_stream()` will create
        # the `Actor._cids2qs` entry from a call to
-        # `Actor.get_context()` and will call us here to send the stop
-        # msg in ``__aexit__()`` on teardown.
-        try:
-            # NOTE: if this call is cancelled we expect this end to
-            # handle as though the stop was never sent (though if it
-            # was it shouldn't matter since it's unlikely a user
-            # will try to re-use a stream after attemping to close
-            # it).
-            with trio.CancelScope(shield=True):
-                await self._ctx.send_stop()
+        # `Actor.get_context()` and will send the stop message in
+        # ``__aexit__()`` on teardown so it **does not** need to be
+        # called here.
+        if not self._ctx._portal:
+            # Only for 2 way streams can we can send stop from the
+            # caller side.
+            try:
+                # NOTE: if this call is cancelled we expect this end to
+                # handle as though the stop was never sent (though if it
+                # was it shouldn't matter since it's unlikely a user
+                # will try to re-use a stream after attemping to close
+                # it).
+                with trio.CancelScope(shield=True):
+                    await self._ctx.send_stop()

-        except (
-            trio.BrokenResourceError,
-            trio.ClosedResourceError
-        ):
-            # the underlying channel may already have been pulled
-            # in which case our stop message is meaningless since
-            # it can't traverse the transport.
-            ctx = self._ctx
-            log.warning(
-                f'Stream was already destroyed?\n'
-                f'actor: {ctx.chan.uid}\n'
-                f'ctx id: {ctx.cid}'
-            )
-
-        self._closed = True
+            except (
+                trio.BrokenResourceError,
+                trio.ClosedResourceError
+            ):
+                # the underlying channel may already have been pulled
+                # in which case our stop message is meaningless since
+                # it can't traverse the transport.
+                log.debug(f'Channel for {self} was already closed')

        # Do we close the local mem chan ``self._rx_chan`` ??!?

@ -280,8 +271,7 @@ class MsgStream(trio.abc.Channel):
        self,

    ) -> AsyncIterator[BroadcastReceiver]:
-        '''
-        Allocate and return a ``BroadcastReceiver`` which delegates
+        '''Allocate and return a ``BroadcastReceiver`` which delegates
        to this message stream.

        This allows multiple local tasks to receive each their own copy
@ -318,15 +308,15 @@ class MsgStream(trio.abc.Channel):
        async with self._broadcaster.subscribe() as bstream:
            assert bstream.key != self._broadcaster.key
            assert bstream._recv == self._broadcaster._recv
-
-            # NOTE: we patch on a `.send()` to the bcaster so that the
-            # caller can still conduct 2-way streaming using this
-            # ``bstream`` handle transparently as though it was the msg
-            # stream instance.
-            bstream.send = self.send  # type: ignore
-
            yield bstream

+
+class MsgStream(ReceiveMsgStream, trio.abc.Channel):
+    '''
+    Bidirectional message stream for use within an inter-actor actor
+    ``Context```.
+
+    '''
    async def send(
        self,
        data: Any
@ -603,23 +593,23 @@ class Context:
        async with MsgStream(
            ctx=self,
            rx_chan=ctx._recv_chan,
-        ) as stream:
+        ) as rchan:

            if self._portal:
-                self._portal._streams.add(stream)
+                self._portal._streams.add(rchan)

            try:
                self._stream_opened = True

-                # XXX: do we need this?
-                # ensure we aren't cancelled before yielding the stream
+                # ensure we aren't cancelled before delivering
+                # the stream
                # await trio.lowlevel.checkpoint()
-                yield stream
+                yield rchan

-                # NOTE: Make the stream "one-shot use".  On exit, signal
+                # XXX: Make the stream "one-shot use".  On exit, signal
                # ``trio.EndOfChannel``/``StopAsyncIteration`` to the
                # far end.
-                await stream.aclose()
+                await self.send_stop()

            finally:
                if self._portal:
--- a/tractor/_supervise.py
+++ b/tractor/_supervise.py
@ -302,7 +302,7 @@ async def _open_and_supervise_one_cancels_all_nursery(
 ) -> typing.AsyncGenerator[ActorNursery, None]:

    # TODO: yay or nay?
-    __tracebackhide__ = True
+    # __tracebackhide__ = True

    # the collection of errors retreived from spawned sub-actors
    errors: dict[tuple[str, str], BaseException] = {}
--- a/tractor/trionics/_broadcast.py
+++ b/tractor/trionics/_broadcast.py
@ -23,6 +23,7 @@ from __future__ import annotations
 from abc import abstractmethod
 from collections import deque
 from contextlib import asynccontextmanager
+from dataclasses import dataclass
 from functools import partial
 from operator import ne
 from typing import Optional, Callable, Awaitable, Any, AsyncIterator, Protocol
@ -32,10 +33,7 @@ import trio
 from trio._core._run import Task
 from trio.abc import ReceiveChannel
 from trio.lowlevel import current_task
-from msgspec import Struct
-from tractor.log import get_logger

-log = get_logger(__name__)

 # A regular invariant generic type
 T = TypeVar("T")
@ -88,7 +86,8 @@ class Lagged(trio.TooSlowError):
    '''


-class BroadcastState(Struct):
+@dataclass
+class BroadcastState:
    '''
    Common state to all receivers of a broadcast.

@ -111,35 +110,7 @@ class BroadcastState(Struct):
    eoc: bool = False

    # If the broadcaster was cancelled, we might as well track it
-    cancelled: dict[int, Task] = {}
-
-    def statistics(self) -> dict[str, Any]:
-        '''
-        Return broadcast receiver group "statistics" like many of
-        ``trio``'s internal task-sync primitives.
-
-        '''
-        key: int | None
-        ev: trio.Event | None
-
-        subs = self.subs
-        if self.recv_ready is not None:
-            key, ev = self.recv_ready
-        else:
-            key = ev = None
-
-        qlens: dict[int, int] = {}
-        for tid, sz in subs.items():
-            qlens[tid] = sz if sz != -1 else 0
-
-        return {
-            'open_consumers': len(subs),
-            'queued_len_by_task': qlens,
-            'max_buffer_size': self.maxlen,
-            'tasks_waiting': ev.statistics().tasks_waiting if ev else 0,
-            'tasks_cancelled': self.cancelled,
-            'next_value_receiver_id': key,
-        }
+    cancelled: bool = False


 class BroadcastReceiver(ReceiveChannel):
@ -157,40 +128,23 @@ class BroadcastReceiver(ReceiveChannel):
        rx_chan: AsyncReceiver,
        state: BroadcastState,
        receive_afunc: Optional[Callable[[], Awaitable[Any]]] = None,
-        raise_on_lag: bool = True,

    ) -> None:

        # register the original underlying (clone)
        self.key = id(self)
        self._state = state
-
-        # each consumer has an int count which indicates
-        # which index contains the next value that the task has not yet
-        # consumed and thus should read. In the "up-to-date" case the
-        # consumer task must wait for a new value from the underlying
-        # receiver and we use ``-1`` as the sentinel for this state.
        state.subs[self.key] = -1

        # underlying for this receiver
        self._rx = rx_chan
        self._recv = receive_afunc or rx_chan.receive
        self._closed: bool = False
-        self._raise_on_lag = raise_on_lag

-    def receive_nowait(
-        self,
-        _key: int | None = None,
-        _state: BroadcastState | None = None,
+    async def receive(self) -> ReceiveType:

-    ) -> Any:
-        '''
-        Sync version of `.receive()` which does all the low level work
-        of receiving from the underlying/wrapped receive channel.
-
-        '''
-        key = _key or self.key
-        state = _state or self._state
+        key = self.key
+        state = self._state

        # TODO: ideally we can make some way to "lock out" the
        # underlying receive channel in some way such that if some task
@ -223,173 +177,128 @@ class BroadcastReceiver(ReceiveChannel):
                # return this value."
                # https://docs.rs/tokio/1.11.0/tokio/sync/broadcast/index.html#lagging

-                mxln = state.maxlen
-                lost = seq - mxln
-
                # decrement to the last value and expect
                # consumer to either handle the ``Lagged`` and come back
                # or bail out on its own (thus un-subscribing)
-                state.subs[key] = mxln - 1
+                state.subs[key] = state.maxlen - 1

                # this task was overrun by the producer side
                task: Task = current_task()
-                msg = f'Task `{task.name}` overrun and dropped `{lost}` values'
-
-                if self._raise_on_lag:
-                    raise Lagged(msg)
-                else:
-                    log.warning(msg)
-                    return self.receive_nowait(_key, _state)
+                raise Lagged(f'Task {task.name} was overrun')

            state.subs[key] -= 1
            return value

-        raise trio.WouldBlock
+        # current task already has the latest value **and** is the
+        # first task to begin waiting for a new one
+        if state.recv_ready is None:

-    async def _receive_from_underlying(
-        self,
-        key: int,
-        state: BroadcastState,
+            if self._closed:
+                raise trio.ClosedResourceError

-    ) -> ReceiveType:
+            event = trio.Event()
+            state.recv_ready = key, event

-        if self._closed:
-            raise trio.ClosedResourceError
-
-        event = trio.Event()
-        assert state.recv_ready is None
-        state.recv_ready = key, event
-
-        try:
            # if we're cancelled here it should be
            # fine to bail without affecting any other consumers
            # right?
-            value = await self._recv()
+            try:
+                value = await self._recv()

-            # items with lower indices are "newer"
-            # NOTE: ``collections.deque`` implicitly takes care of
-            # trucating values outside our ``state.maxlen``. In the
-            # alt-backend-array-case we'll need to make sure this is
-            # implemented in similar ringer-buffer-ish style.
-            state.queue.appendleft(value)
+                # items with lower indices are "newer"
+                # NOTE: ``collections.deque`` implicitly takes care of
+                # trucating values outside our ``state.maxlen``. In the
+                # alt-backend-array-case we'll need to make sure this is
+                # implemented in similar ringer-buffer-ish style.
+                state.queue.appendleft(value)

-            # broadcast new value to all subscribers by increasing
-            # all sequence numbers that will point in the queue to
-            # their latest available value.
+                # broadcast new value to all subscribers by increasing
+                # all sequence numbers that will point in the queue to
+                # their latest available value.

-            # don't decrement the sequence for this task since we
-            # already retreived the last value
+                # don't decrement the sequence for this task since we
+                # already retreived the last value

-            # XXX: which of these impls is fastest?
-            # subs = state.subs.copy()
-            # subs.pop(key)
+                # XXX: which of these impls is fastest?

-            for sub_key in filter(
-                # lambda k: k != key, state.subs,
-                partial(ne, key), state.subs,
+                # subs = state.subs.copy()
+                # subs.pop(key)
+
+                for sub_key in filter(
+                    # lambda k: k != key, state.subs,
+                    partial(ne, key), state.subs,
+                ):
+                    state.subs[sub_key] += 1
+
+                # NOTE: this should ONLY be set if the above task was *NOT*
+                # cancelled on the `._recv()` call.
+                event.set()
+                return value
+
+            except trio.EndOfChannel:
+                # if any one consumer gets an EOC from the underlying
+                # receiver we need to unblock and send that signal to
+                # all other consumers.
+                self._state.eoc = True
+                if event.statistics().tasks_waiting:
+                    event.set()
+                raise
+
+            except (
+                trio.Cancelled,
            ):
-                state.subs[sub_key] += 1
+                # handle cancelled specially otherwise sibling
+                # consumers will be awoken with a sequence of -1
+                # and will potentially try to rewait the underlying
+                # receiver instead of just cancelling immediately.
+                self._state.cancelled = True
+                if event.statistics().tasks_waiting:
+                    event.set()
+                raise

-            # NOTE: this should ONLY be set if the above task was *NOT*
-            # cancelled on the `._recv()` call.
-            event.set()
-            return value
+            finally:

-        except trio.EndOfChannel:
-            # if any one consumer gets an EOC from the underlying
-            # receiver we need to unblock and send that signal to
-            # all other consumers.
-            self._state.eoc = True
-            if event.statistics().tasks_waiting:
-                event.set()
-            raise
-
-        except (
-            trio.Cancelled,
-        ):
-            # handle cancelled specially otherwise sibling
-            # consumers will be awoken with a sequence of -1
-            # and will potentially try to rewait the underlying
-            # receiver instead of just cancelling immediately.
-            self._state.cancelled[key] = current_task()
-            if event.statistics().tasks_waiting:
-                event.set()
-            raise
-
-        finally:
-            # Reset receiver waiter task event for next blocking condition.
-            # this MUST be reset even if the above ``.recv()`` call
-            # was cancelled to avoid the next consumer from blocking on
-            # an event that won't be set!
-            state.recv_ready = None
-
-    async def receive(self) -> ReceiveType:
-        key = self.key
-        state = self._state
-
-        try:
-            return self.receive_nowait(
-                _key=key,
-                _state=state,
-            )
-        except trio.WouldBlock:
-            pass
-
-        # current task already has the latest value **and** is the
-        # first task to begin waiting for a new one so we begin blocking
-        # until rescheduled with the a new value from the underlying.
-        if state.recv_ready is None:
-            return await self._receive_from_underlying(key, state)
+                # Reset receiver waiter task event for next blocking condition.
+                # this MUST be reset even if the above ``.recv()`` call
+                # was cancelled to avoid the next consumer from blocking on
+                # an event that won't be set!
+                state.recv_ready = None

        # This task is all caught up and ready to receive the latest
-        # value, so queue/schedule it to be woken on the next internal
-        # event.
+        # value, so queue sched it on the internal event.
        else:
-            while state.recv_ready is not None:
-                # seq = state.subs[key]
-                # assert seq == -1  # sanity
-                _, ev = state.recv_ready
-                await ev.wait()
-                try:
-                    return self.receive_nowait(
-                        _key=key,
-                        _state=state,
-                    )
-                except trio.WouldBlock:
-                    if self._closed:
-                        raise trio.ClosedResourceError
+            seq = state.subs[key]
+            assert seq == -1  # sanity
+            _, ev = state.recv_ready
+            await ev.wait()

-                    subs = state.subs
-                    if (
-                        len(subs) == 1
-                        and key in subs
-                        # or cancelled
-                    ):
-                        # XXX: we are the last and only user of this BR so
-                        # likely it makes sense to unwind back to the
-                        # underlying?
-                        # import tractor
-                        # await tractor.breakpoint()
-                        log.warning(
-                            f'Only one sub left for {self}?\n'
-                            'We can probably unwind from breceiver?'
-                        )
+            # NOTE: if we ever would like the behaviour where if the
+            # first task to recv on the underlying is cancelled but it
+            # still DOES trigger the ``.recv_ready``, event we'll likely need
+            # this logic:

-                    # XXX: In the case where the first task to allocate the
-                    # ``.recv_ready`` event is cancelled we will be woken
-                    # with a non-incremented sequence number (the ``-1``
-                    # sentinel) and thus will read the oldest value if we
-                    # use that. Instead we need to detect if we have not
-                    # been incremented and then receive again.
-                    # return await self.receive()
+            if seq > -1:
+                # stuff from above..
+                seq = state.subs[key]

-            return await self._receive_from_underlying(key, state)
+                value = state.queue[seq]
+                state.subs[key] -= 1
+                return value
+
+            elif seq == -1:
+                # XXX: In the case where the first task to allocate the
+                # ``.recv_ready`` event is cancelled we will be woken with
+                # a non-incremented sequence number and thus will read the
+                # oldest value if we use that. Instead we need to detect if
+                # we have not been incremented and then receive again.
+                return await self.receive()
+
+            else:
+                raise ValueError(f'Invalid sequence {seq}!?')

    @asynccontextmanager
    async def subscribe(
        self,
-        raise_on_lag: bool = True,
-
    ) -> AsyncIterator[BroadcastReceiver]:
        '''
        Subscribe for values from this broadcast receiver.
@ -407,7 +316,6 @@ class BroadcastReceiver(ReceiveChannel):
            rx_chan=self._rx,
            state=state,
            receive_afunc=self._recv,
-            raise_on_lag=raise_on_lag,
        )
        # assert clone in state.subs
        assert br.key in state.subs
@ -444,8 +352,7 @@ def broadcast_receiver(

    recv_chan: AsyncReceiver,
    max_buffer_size: int,
-    receive_afunc: Optional[Callable[[], Awaitable[Any]]] = None,
-    raise_on_lag: bool = True,
+    **kwargs,

 ) -> BroadcastReceiver:

@ -456,6 +363,5 @@ def broadcast_receiver(
            maxlen=max_buffer_size,
            subs={},
        ),
-        receive_afunc=receive_afunc,
-        raise_on_lag=raise_on_lag,
+        **kwargs,
    )
Author	SHA1	Message	Date
Tyler Goodlet	a89799b682	Handle broken mem chan on `Actor._push_result()` When backpressure is used and a feeder mem chan breaks during msg delivery (usually because the IPC allocating task already terminated) instead of raising we simply warn as we do for the non-backpressure case. Also, add a proper `Actor.is_arbiter` test inside `._invoke()` to avoid doing an arbiter-registry lookup if the current actor is the registrar.	2022-12-12 15:04:40 -05:00
Tyler Goodlet	45a9aaf6e9	Always set the `parent_exit: trio.Event` on exit	2022-12-12 15:04:40 -05:00
Tyler Goodlet	b624ebba21	Add logging and teardown prints to full-fledged-streaming example	2022-12-12 15:04:40 -05:00
Tyler Goodlet	e12def51a8	Always print any std streams to console in docs examples tests	2022-12-12 15:04:40 -05:00
Tyler Goodlet	64819b2acb	Skip debugger tests on OS X for now	2022-12-12 15:04:40 -05:00
Tyler Goodlet	a5e3cf4ecf	Add macos run using only the `trio` spawner	2022-12-12 15:04:40 -05:00
Tyler Goodlet	7c42d2510c	Add back `pytest` full trace flag to debug CI hangs	2022-12-12 15:04:40 -05:00
Tyler Goodlet	9c336ec064	Add `numpy` for testing optional integrated shm API layer	2022-12-12 15:04:11 -05:00
Tyler Goodlet	6453195e97	Pass `str` dtype for `use_str` case	2022-12-12 15:04:11 -05:00
Tyler Goodlet	54322f2bae	Allocate size-specced "empty" sequence from default values by type	2022-12-12 15:04:11 -05:00
Tyler Goodlet	1be3f4115d	Mod define `_USE_POSIX`, add a of of todos	2022-12-12 15:04:11 -05:00
Tyler Goodlet	3bdd04ef4a	Parametrize rw test with variable frame sizes Demonstrates fixed size frame-oriented reads by the child where the parent only transmits a "read" stream msg on "frame fill events" such that the child incrementally reads the shm list data (much like in a real-time-buffered streaming system).	2022-12-12 15:04:11 -05:00
Tyler Goodlet	2ac19b2e96	Add `ShmList` slice support in `.__getitem__()`	2022-12-12 15:04:11 -05:00
Tyler Goodlet	60f8f110e8	Rename token type to `NDToken` in the style of `nptyping`	2022-12-12 15:04:11 -05:00
Tyler Goodlet	6d4d428205	Fix uid2nursery lookup table type annot	2022-12-12 15:04:11 -05:00
Tyler Goodlet	ecea1e1658	Don't require runtime (for now), type annot fixing	2022-12-12 15:04:11 -05:00
Tyler Goodlet	971ac50756	Add repetitive attach to existing segment test	2022-12-12 15:04:11 -05:00
Tyler Goodlet	c01c22769b	Add initial readers-writer shm list tests	2022-12-12 15:04:11 -05:00
Tyler Goodlet	0e4b37d122	Add `ShmList` wrapping the stdlib's `ShareableList` First attempt at getting `multiprocessing.shared_memory.ShareableList` working; we wrap the stdlib type with a readonly attr and a `.key` for cross-actor lookup. Also, rename all `numpy` specific routines to have a `ndarray` suffix in the func names.	2022-12-12 15:04:11 -05:00
Tyler Goodlet	901353c213	Initial module import from `piker.data._sharemem` More or less a verbatim copy-paste minus some edgy variable naming and internal `piker` module imports. There is a bunch of OHLC related defaults that need to be dropped and we need to adjust to an optional dependence on `numpy` by supporting shared lists as per the mp docs.	2022-12-12 15:04:11 -05:00