.github/workflows/ci.yml

@@ -67,6 +67,7 @@ jobs:
         ]
 
     steps:
+
       - name: Checkout
         uses: actions/checkout@v2
 
@@ -84,40 +85,6 @@ 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.

examples/full_fledged_streaming_service.py

@@ -12,11 +12,9 @@ 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):
@@ -71,8 +69,7 @@ async def aggregate(seed):
 async def main():
     # a nursery which spawns "actors"
     async with tractor.open_nursery(
-        arbiter_addr=('127.0.0.1', 1616),
-        loglevel='cancel',
+        arbiter_addr=('127.0.0.1', 1616)
     ) as nursery:
 
         seed = int(1e3)
@@ -95,9 +92,6 @@ 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,4 +6,3 @@ mypy
 trio_typing
 pexpect
 towncrier
-numpy

tests/test_debugger.py

@@ -36,12 +36,9 @@ from conftest import repodir, _ci_env
 # - recurrent root errors
 
 
-if osname := platform.system() in (
-    'Windows',
-    'Darwin',
-):
+if platform.system() == 'Windows':
     pytest.skip(
-        'Debugger tests have no {osname} support (yet)',
+        'Debugger tests have no windows support (yet)',
         allow_module_level=True,
     )
 
@@ -786,6 +783,8 @@ 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,16 +93,14 @@ def run_example_in_subproc(loglevel, testdir, arb_addr):
     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):
@@ -112,32 +110,25 @@ def test_example(run_example_in_subproc, example_script):
         code = ex.read()
 
         with run_example_in_subproc(code) as proc:
-            try:
-                proc.wait(timeout=5)
-            finally:
-                err = proc.stderr.read()
+            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:
                 errmsg = err.decode()
-                out = proc.stdout.read()
-                outmsg = out.decode()
+                errlines = errmsg.splitlines()
+                last_error = errlines[-1]
+                if (
+                    'Error' in last_error
 
-                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)
+                    # 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

@@ -70,10 +70,7 @@ async def child_read_shm_list(
 ) -> None:
 
     # attach in child
-    shml = attach_shm_list(
-        key=shm_key,
-        # dtype=str if use_str else float,
-    )
+    shml = attach_shm_list(key=shm_key)
     await ctx.started(shml.key)
 
     async with ctx.open_stream() as stream:
@@ -95,9 +92,7 @@ async def child_read_shm_list(
 
 
 @pytest.mark.parametrize(
-    'use_str',
-    [False, True],
-    ids=lambda i: f'use_str_values={i}',
+    'use_str', [False, True],
 )
 @pytest.mark.parametrize(
     'frame_size',
@@ -111,7 +106,7 @@ def test_parent_writer_child_reader(
 
     async def main():
         async with tractor.open_nursery(
-            # debug_mode=True,
+            debug_mode=True,
         ) as an:
 
             portal = await an.start_actor(
@@ -126,7 +121,6 @@ def test_parent_writer_child_reader(
             shml = open_shm_list(
                 key=key,
                 size=seq_size,
-                dtype=str if use_str else float,
                 readonly=False,
             )
 
@@ -149,7 +143,7 @@ def test_parent_writer_child_reader(
                     if use_str:
                         val = str(val)
 
-                    # print(f'(parent): writing {val}')
+                    print(f'(parent): writing {val}')
                     shml[i] = val
 
                     # only on frame fills do we

tractor/_runtime.py

@@ -829,12 +829,7 @@ class Actor:
 
             if ctx._backpressure:
                 log.warning(text)
-                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")
+                await send_chan.send(msg)
             else:
                 try:
                     raise StreamOverrun(text) from None
@@ -1379,9 +1374,8 @@ async def async_main(
         actor.lifetime_stack.close()
 
         # Unregister actor from the arbiter
-        if (
-            registered_with_arbiter
-            and not actor.is_arbiter
+        if registered_with_arbiter and (
+                actor._arb_addr is not None
         ):
             failed = False
             with trio.move_on_after(0.5) as cs:

tractor/_shm.py

@@ -460,6 +460,7 @@ class ShmArray:
 
 
 def open_shm_ndarray(
+
     key: Optional[str] = None,
     size: int = int(2 ** 10),
     dtype: np.dtype | None = None,
@@ -718,7 +719,7 @@ class ShmList(ShareableList):
     Carbon copy of ``.shared_memory.ShareableList`` with a few
     enhancements:
 
-    - readonly mode via instance var flag  `._readonly: bool`
+    - readonly mode via instance var flag
     - ``.__getitem__()`` accepts ``slice`` inputs
     - exposes the underlying buffer "name" as a ``.key: str``
 
@@ -742,10 +743,6 @@ class ShmList(ShareableList):
     def key(self) -> str:
         return self._key
 
-    @property
-    def readonly(self) -> bool:
-        return self._readonly
-
     def __setitem__(
         self,
         position,
@@ -784,20 +781,13 @@ def open_shm_list(
     key: str,
     sequence: list | None = None,
     size: int = int(2 ** 10),
-    dtype: float | int | bool | str | bytes | None = float,
+    dtype: np.dtype | None = None,
     readonly: bool = True,
 
 ) -> ShmList:
 
     if sequence is None:
-        default = {
-            float: 0.,
-            int: 0,
-            bool: True,
-            str: 'doggy',
-            None: None,
-        }[dtype]
-        sequence = [default] * size
+        sequence = list(map(float, range(size)))
 
     shml = ShmList(
         sequence=sequence,

tractor/trionics/_mngrs.py

@@ -133,13 +133,13 @@ async def gather_contexts(
         # deliver control once all managers have started up
         await all_entered.wait()
 
-        try:
-            yield tuple(unwrapped.values())
-        finally:
-            # NOTE: this is ABSOLUTELY REQUIRED to avoid
-            # the following wacky bug:
-            # <tractorbugurlhere>
-            parent_exit.set()
+        # NOTE: order *should* be preserved in the output values
+        # since ``dict``s are now implicitly ordered.
+        yield tuple(unwrapped.values())
+
+        # we don't need a try/finally since cancellation will be triggered
+        # by the surrounding nursery on error.
+        parent_exit.set()
 
 
 # Per actor task caching helpers.