Merge pull request #68 from tgoodlet/close_mem_chans

Use "clean channel shutdown" in streaming example

README.rst

@@ -391,30 +391,32 @@ and print the results to your screen:
 
            send_chan, recv_chan = trio.open_memory_channel(500)
 
-           async def push_to_q(portal):
+           async def push_to_chan(portal, send_chan):
-               async for value in await portal.run(
+               async with send_chan:
-                   __name__, 'stream_data', seed=seed
+                   async for value in await portal.run(
-               ):
+                       __name__, 'stream_data', seed=seed
-                   # leverage trio's built-in backpressure
+                   ):
-                   await send_chan.send(value)
+                       # leverage trio's built-in backpressure
+                       await send_chan.send(value)
 
-               await send_chan.send(None)
                print(f"FINISHED ITERATING {portal.channel.uid}")
 
            # spawn 2 trio tasks to collect streams and push to a local queue
            async with trio.open_nursery() as n:
+
                for portal in portals:
-                   n.start_soon(push_to_q, portal)
+                   n.start_soon(push_to_chan, portal, send_chan.clone())
+
+               # close this local task's reference to send side
+               await send_chan.aclose()
 
                unique_vals = set()
-               async for value in recv_chan:
+               async with recv_chan:
-                   if value not in unique_vals:
+                   async for value in recv_chan:
-                       unique_vals.add(value)
+                       if value not in unique_vals:
-                       # yield upwards to the spawning parent actor
+                           unique_vals.add(value)
-                       yield value
+                           # yield upwards to the spawning parent actor
-
+                           yield value
-                       if value is None:
-                           break
 
                    assert value in unique_vals
 
@@ -449,7 +451,7 @@ and print the results to your screen:
 
            print(f"STREAM TIME = {time.time() - start}")
            print(f"STREAM + SPAWN TIME = {time.time() - pre_start}")
-           assert result_stream == list(range(seed)) + [None]
+           assert result_stream == list(range(seed))
            return result_stream
 
 

tests/test_streaming.py

@@ -76,7 +76,8 @@ def test_stream_from_single_subactor(arb_addr, start_method):
 async def stream_data(seed):
     for i in range(seed):
         yield i
-        await trio.sleep(0)  # trigger scheduler
+        # trigger scheduler to simulate practical usage
+        await trio.sleep(0)
 
 
 # this is the third actor; the aggregator
@@ -97,30 +98,32 @@ async def aggregate(seed):
 
         send_chan, recv_chan = trio.open_memory_channel(500)
 
-        async def push_to_chan(portal):
+        async def push_to_chan(portal, send_chan):
-            async for value in await portal.run(
+            async with send_chan:
-                __name__, 'stream_data', seed=seed
+                async for value in await portal.run(
-            ):
+                    __name__, 'stream_data', seed=seed
-                # leverage trio's built-in backpressure
+                ):
-                await send_chan.send(value)
+                    # leverage trio's built-in backpressure
+                    await send_chan.send(value)
 
-            await send_chan.send(None)
             print(f"FINISHED ITERATING {portal.channel.uid}")
 
         # spawn 2 trio tasks to collect streams and push to a local queue
         async with trio.open_nursery() as n:
+
             for portal in portals:
-                n.start_soon(push_to_chan, portal)
+                n.start_soon(push_to_chan, portal, send_chan.clone())
+
+            # close this local task's reference to send side
+            await send_chan.aclose()
 
             unique_vals = set()
-            async for value in recv_chan:
+            async with recv_chan:
-                if value not in unique_vals:
+                async for value in recv_chan:
-                    unique_vals.add(value)
+                    if value not in unique_vals:
-                    # yield upwards to the spawning parent actor
+                        unique_vals.add(value)
-                    yield value
+                        # yield upwards to the spawning parent actor
-
+                        yield value
-                    if value is None:
-                        break
 
                 assert value in unique_vals
 
@@ -154,7 +157,7 @@ async def a_quadruple_example():
 
         print(f"STREAM TIME = {time.time() - start}")
         print(f"STREAM + SPAWN TIME = {time.time() - pre_start}")
-        assert result_stream == list(range(seed)) + [None]
+        assert result_stream == list(range(seed))
         return result_stream