V3 Barrier primitive: access callback return value, improve docs.

Author: peterhinch

v3/docs/TUTORIAL.md

@@ -927,18 +927,25 @@ Constructor.
 Mandatory arg:  
  * `participants` The number of coros which will use the barrier.  
 Optional args:  
- * `func` Callback to run. Default `None`.  
+ * `func` Callback or coroutine to run. Default `None`.  
  * `args` Tuple of args for the callback. Default `()`.
 
 Public synchronous methods:  
  * `busy` No args. Returns `True` if at least one coro is waiting on the
  barrier, or if at least one non-waiting coro has not triggered it.
  * `trigger` No args. The barrier records that the coro has passed the critical
  point. Returns "immediately".
+ * `result` No args. If a callback was provided, returns the return value from
+ the callback. If a coro, returns the `Task` instance. See below.
 
 The callback can be a function or a coro. Typically a function will be used; it
 must run to completion beore the barrier is released. A coro will be promoted
-to a `Task` and run asynchronously.
+to a `Task` and run asynchronously. The `Task` may be retrieved (e.g. for
+cancellation) using the `result` method.
+
+If a coro waits on a barrier, it should issue an `await` prior to accessing the
+`result` method. To guarantee that the callback has run it is necessary to wait
+until all participant coros have passed the barrier.
 
 Participant coros issue `await my_barrier` whereupon execution pauses until all
 other participants are also waiting on it. At this point any callback will run

v3/primitives/barrier.py

@@ -27,29 +27,31 @@ def __init__(self, participants, func=None, args=()):
         self._func = func
         self._args = args
         self._reset(True)
+        self._res = None
 
     def __await__(self):
-        self._update()
-        if self._at_limit():  # All other threads are also at limit
-            if self._func is not None:
-                launch(self._func, self._args)
-            self._reset(not self._down)  # Toggle direction to release others
+        if self.trigger():
             return
 
         direction = self._down
-        while True:  # Wait until last waiting thread changes the direction
+        while True:  # Wait until last waiting task changes the direction
             if direction != self._down:
                 return
             await asyncio.sleep_ms(0)
 
     __iter__ = __await__
 
+    def result(self):
+        return self._res
+
     def trigger(self):
         self._update()
-        if self._at_limit():  # All other threads are also at limit
+        if self._at_limit():  # All other tasks are also at limit
             if self._func is not None:
-                launch(self._func, self._args)
+                self._res = launch(self._func, self._args)
             self._reset(not self._down)  # Toggle direction to release others
+            return True
+        return False
 
     def _reset(self, down):
         self._down = down

v3/primitives/tests/asyntest.py

@@ -24,11 +24,12 @@ def print_tests():
 test(0)  Print this list.
 test(1)  Test message acknowledge.
 test(2)  Test Messge and Lock objects.
-test(3)  Test the Barrier class.
-test(4)  Test Semaphore
-test(5)  Test BoundedSemaphore.
-test(6)  Test the Condition class.
-test(7)  Test the Queue class.
+test(3)  Test the Barrier class with callback.
+test(4)  Test the Barrier class with coroutine.
+test(5)  Test Semaphore
+test(6)  Test BoundedSemaphore.
+test(7)  Test the Condition class.
+test(8)  Test the Queue class.
 '''
     print('\x1b[32m')
     print(st)
@@ -186,6 +187,49 @@ def barrier_test():
         asyncio.create_task(report(barrier))
     asyncio.run(killer(2))
 
+# ************ Barrier test 1 ************
+
+async def my_coro(text):
+    try:
+        await asyncio.sleep_ms(0)
+        while True:
+            await asyncio.sleep(1)
+            print(text)
+    except asyncio.CancelledError:
+        print('my_coro was cancelled.')
+
+async def report1(barrier, x):
+    await asyncio.sleep(x)
+    print('report instance', x, 'waiting')
+    await barrier
+    print('report instance', x, 'done')
+
+async def bart():
+    barrier = Barrier(4, my_coro, ('my_coro running',))
+    for x in range(3):
+        asyncio.create_task(report1(barrier, x))
+    await barrier
+    # Must yield before reading result(). Here we wait long enough for
+    await asyncio.sleep_ms(1500)  # coro to print
+    barrier.result().cancel()
+    await asyncio.sleep(2)
+
+def barrier_test1():
+    printexp('''Running (runtime = 5s):
+report instance 0 waiting
+report instance 1 waiting
+report instance 2 waiting
+report instance 2 done
+report instance 1 done
+report instance 0 done
+my_coro running
+my_coro was cancelled.
+
+Exact report instance done sequence may vary, but 3 instances should report
+done before my_coro runs.
+''', 5)
+    asyncio.run(bart())
+
 # ************ Semaphore test ************
 
 async def run_sema(n, sema, barrier):
@@ -373,12 +417,14 @@ def test(n):
         elif n == 3:
             barrier_test()  # Test the Barrier class.
         elif n == 4:
-            semaphore_test(False) # Test Semaphore
+            barrier_test1()  # Test the Barrier class.
         elif n == 5:
-            semaphore_test(True)  # Test BoundedSemaphore.
+            semaphore_test(False) # Test Semaphore
         elif n == 6:
-            condition_test()  # Test the Condition class.
+            semaphore_test(True)  # Test BoundedSemaphore.
         elif n == 7:
+            condition_test()  # Test the Condition class.
+        elif n == 8:
             queue_test()  # Test the Queue class.
     except KeyboardInterrupt:
         print('Interrupted')