Macos in ci

.github/workflows/ci.yml

@@ -67,7 +67,39 @@ jobs:
         ]
 
     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
 
+
+  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
 
@@ -85,6 +117,7 @@ jobs:
       - 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.

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 push_to_chan(portal, send_chan):
 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}")

requirements-test.txt

@@ -6,3 +6,4 @@ mypy
 trio_typing
 pexpect
 towncrier
+numpy

tests/test_debugger.py

@@ -36,9 +36,12 @@
 # - 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,
     )
 
@@ -783,8 +786,6 @@ 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(r"\(Pdb\+\+\)")

tests/test_docs_examples.py

@@ -93,14 +93,16 @@ def run(script_code):
     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):
@@ -110,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
-
-                    # 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)
+                out = proc.stdout.read()
+                outmsg = out.decode()
+
+                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

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)

tractor/_runtime.py

@@ -488,7 +488,10 @@ def __init__(
         self._parent_chan: Optional[Channel] = None
         self._forkserver_info: Optional[
             tuple[Any, Any, Any, Any, Any]] = None
-        self._actoruid2nursery: dict[Optional[tuple[str, str]], 'ActorNursery'] = {}  # type: ignore  # noqa
+        self._actoruid2nursery: dict[
+            tuple[str, str],
+            ActorNursery | None,
+        ] = {}  # type: ignore  # noqa
 
     async def wait_for_peer(
         self, uid: tuple[str, str]
@@ -826,7 +829,12 @@ async def _push_result(
 
             if ctx._backpressure:
                 log.warning(text)
-                await send_chan.send(msg)
+                try:
+                    await send_chan.send(msg)
+                except trio.BrokenResourceError:
+                    # XXX: local consumer has closed their side
+                    # so cancel the far end streaming task
+                    log.warning(f"{chan} is already closed")
             else:
                 try:
                     raise StreamOverrun(text) from None
@@ -1371,8 +1379,9 @@ async def async_main(
         actor.lifetime_stack.close()
 
         # Unregister actor from the arbiter
-        if registered_with_arbiter and (
-                actor._arb_addr is not None
+        if (
+            registered_with_arbiter
+            and not actor.is_arbiter
         ):
             failed = False
             with trio.move_on_after(0.5) as cs: