Support for forthcoming timeouts and task cancellation. Barrier supports nowait.

peterhinch · peterhinch · commit 115e4c7202e7 · 2017-11-12T11:47:28.000Z
diff --git a/TUTORIAL.md b/TUTORIAL.md
@@ -52,13 +52,17 @@ guides to this may be found online.
 
   3.5 [Queue](./TUTORIAL.md#35-queue)
 
+  3.6 [Task cancellation](./TUTORIAL.md#36-task-cancellation)
+
  4. [Designing classes for asyncio](./TUTORIAL.md#4-designing-classes-for-asyncio)
 
   4.1 [Awaitable classes](./TUTORIAL.md#41-awaitable-classes)
 
   4.2 [Asynchronous iterators](./TUTORIAL.md#42-asynchronous-iterators)
 
   4.3 [Asynchronous context managers](./TUTORIAL.md#43-asynchronous-context-managers)
+  
+  4.4 [Coroutines with timeouts](./TUTORIAL.md#44-coroutines-with-timeouts)
 
  5. [Device driver examples](./TUTORIAL.md#5-device-driver-examples)
 
@@ -440,6 +444,13 @@ multiple instances of ``report`` print their result and pause until the other
 instances are also complete. At that point the callback runs. On its completion
 the coros resume.
 
+A special case of `Barrier` usage is where some coros are allowed to pass the
+barrier, registering the fact that they have done so. At least one coro must
+wait on the barrier. It will continue execution when all non-waiting coros have
+passed the barrier, and all other waiting coros have reached it. This can be of
+use when cancelling coros. A coro which cancels others might wait until all
+cancelled coros have passed the barrier as they quit.
+
 ###### [Jump to Contents](./TUTORIAL.md#contents)
 
 ## 3.4 Semaphore
@@ -498,6 +509,71 @@ An example of its use is provided in ``aqtest.py``.
 
 ###### [Jump to Contents](./TUTORIAL.md#contents)
 
+## 3.6 Task cancellation
+
+At the time of writing (12th Nov 2017) this requires PR #3380 and
+micropython-lib PR #221 which are yet to be merged.
+
+The `uasyncio` library supports task cancellation by throwing an exception to
+the coro which is to be cancelled. The latter must trap the exception and
+(after performing any cleanup) terminate. The use of this mechanism is
+facilitated by the `Cancellable` class which enables a coro to be associated
+with a user-defined name for cancellation. Examples of its usage may be found
+in `asyntest.py`.
+
+A cancellable coro is instantiated from a normal coro `foo()` by means of
+`Cancellable(foo(5), 'foo')`. Note the passing of an argument to the coro. A
+coro can `await` such a task with `await Cancellable(foo(5), 'foo')`.
+Alternatively a cancellable task can be scheduled for execution with
+`loop.create_task(Cancellable(foo(5), 'foo').task)`
+
+In either case the coro with the user-defined name 'foo' is cancelled with
+`Cancellable.cancel('foo')`. The coro `foo` will receive the `CancelError` when
+it next runs. This means that in real time, and from the point of view of the
+coro which has cancelled it, cancellation may not be immediate. In some
+situations this may matter. Synchronisation may be achieved using the `Barrier`
+class, with the cancelling task pausing until all the coros it has cancelled
+have processed the exception. The following - adapted from `asyntest.py` - 
+illustrates this.
+
+```python
+import uasyncio as asyncio
+from asyn import Barrier, Cancellable, CancelError
+
+async def forever(n):
+    print('Started forever() instance', n)
+    while True:  # Run until cancelled. Error propagates to caller.
+        await asyncio.sleep(7 + n)
+        print('Running instance', n)
+
+barrier = Barrier(3)  # 3 tasks share the barrier
+
+async def rats(n):
+    # Cancellable coros must trap the CancelError
+    try:
+        await forever(n)  # Error propagates up from forever()
+    except CancelError:
+        await barrier(nowait = True)  # Quit immediately
+        print('Instance', n, 'was cancelled')
+
+async def run_cancel_test2():
+    loop = asyncio.get_event_loop()
+    loop.create_task(Cancellable(rats(1), 'rats_1').task)
+    loop.create_task(Cancellable(rats(2), 'rats_2').task)
+    print('Running two tasks')
+    await asyncio.sleep(10)
+    print('About to cancel tasks')
+    Cancellable.cancel('rats_1')
+    Cancellable.cancel('rats_2')
+    await barrier  # Continue when dependent tasks have quit
+    print('tasks were cancelled')
+
+loop = asyncio.get_event_loop()
+loop.run_until_complete(run_cancel_test2())
+```
+
+###### [Jump to Contents](./TUTORIAL.md#contents)
+
 # 4 Designing classes for asyncio
 
 In the context of device drivers the aim is to ensure nonblocking operation.
@@ -628,6 +704,43 @@ to completion. The error appears to be in PEP492. See
 
 ###### [Jump to Contents](./TUTORIAL.md#contents)
 
+## 4.4 Coroutines with timeouts
+
+At the time of writing (12th Nov 2017) this requires PR #3380 and
+micropython-lib PR #221 which are yet to be merged.
+
+Timeouts are implemented by means of `uasyncio.wait_for()`. This takes as
+arguments a coroutine and a timeout in seconds. If the timeout expires a
+`TimeoutError` will be thrown to the coro. The next time the coro is scheduled
+for execution the exception will be raised: the coro should trap this and quit.
+
+```python
+import uasyncio as asyncio
+
+async def forever():
+    print('Starting')
+    try:
+        while True:
+            await asyncio.sleep_ms(300)
+            print('Got here')
+    except asyncio.TimeoutError:
+        print('Got timeout')
+
+async def foo():
+    await asyncio.wait_for(forever(), 5)
+    await asyncio.sleep(2)
+
+loop = asyncio.get_event_loop()
+loop.run_until_complete(foo())
+```
+
+Note that if the coro awaits a long delay, it will not be rescheduled until the
+time has elapsed. The `TimeoutError` will occur as soon as it is scheduled. But
+in real time and from the point of view of the calling coro, its response to
+the `TimeoutError` will be correspondingly delayed.
+
+###### [Jump to Contents](./TUTORIAL.md#contents)
+
 # 5 Device driver examples
 
 Many devices such as sensors are read-only in nature and need to be polled to
diff --git a/asyn.py b/asyn.py
@@ -109,29 +109,49 @@ def value(self):
 # A Barrier synchronises N coros. Each issues await barrier
 # execution pauses until all other participant coros are waiting on it.
 # At that point the callback is executed. Then the barrier is 'opened' and
-# excution of all participants resumes.
+# execution of all participants resumes.
+# The nowait arg enables usage where one or more coros can register that
+# they have reached the barrier without waiting for it. Other coros waiting
+# normally on the barrier will pause until all non-waiting coros have passed
+# the barrier and all waiting ones have reached it.
+# The use of nowait promotes efficiency by enabling tasks to leave the task
+# queue as soon as possible.
+
 # Uses low_priority if available
+
 class Barrier():
     def __init__(self, participants, func=None, args=()):
         self._participants = participants
         self._func = func
         self._args = args
         self._reset(True)
+        self._nowait = False
 
     def __await__(self):
         self._update()
         if self._at_limit():  # All other threads are also at limit
+            self._nowait = False
             if self._func is not None:
                 launch(self._func, self._args)
             self._reset(not self._down)
             return
 
+        if self._nowait:
+            self._nowait = False
+            if self._func is not None:
+                launch(self._func, self._args)
+            return
+
         direction = self._down
         while True:  # Wait until last waiting thread changes the direction
             if direction != self._down:
                 return
             yield from after(0)
 
+    def __call__(self, nowait=False):  # Enable await barrier(nowait = True)
+        self._nowait = nowait
+        return self
+
     __iter__ = __await__
 
     def _reset(self, down):
@@ -179,6 +199,36 @@ def release(self):
         else:
             raise ValueError('Semaphore released more than acquired')
 
+# Task cancellation. At the time of writing this is dependent on PR #3380 and
+# micropython-lib PR #221 which are not yet merged.
+# A coro foo is made cancellable by isssuing Cancellable(foo(5), 'foo'), the
+# second arg being a name used when the task is to be cancelled. This is done
+# by issuing Cancellable.cancel('foo'). See asyntest.py and tutorial.
+
+class CancelError(Exception):
+    pass
+
+class Cancellable():
+    tasks = {}
+    @classmethod
+    def cancel(cls, taskname):
+        if taskname in cls.tasks:
+            cls.tasks.pop(taskname).pend_throw(CancelError)
+            return True
+        return False
+
+    def __init__(self, task, name):
+        self.tasks[name] = task
+        self.task = task
+
+    def __await__(self):
+        res = yield from self.task
+        return res
+
+    __iter__ = __await__
+
+# ExitGate is obsolescent - task cancellation is about to be implemented.
+# Retained for compatibilty with existing applications.
 # uasyncio does not have a mechanism whereby a task can be terminated by another.
 # This can cause an issue if a task instantiates other tasks then terminates:
 # the child tasks continue to run, which may not be desired. The ExitGate helps
diff --git a/asyntest.py b/asyntest.py
@@ -7,6 +7,8 @@
 # event_test()
 # barrier_test()
 # semaphore_test()  Pass True to test BoundedSemaphore class
+# cancel_test1()  Awaiting cancellable coros
+# cancel_test2() Cancellable coros as tasks and using barrier to synchronise.
 # Issue ctrl-D after running each test
 
 # CPython 3.5 compatibility
@@ -16,7 +18,7 @@
 except ImportError:
     import asyncio
 
-from asyn import Lock, Event, Semaphore, BoundedSemaphore, Barrier
+from asyn import Lock, Event, Semaphore, BoundedSemaphore, Barrier, Cancellable, CancelError
 
 # ************ Test Event class ************
 # Demo use of acknowledge event
@@ -150,3 +152,71 @@ async def run_sema_test(bounded):
 def semaphore_test(bounded=False):
     loop = asyncio.get_event_loop()
     loop.run_until_complete(run_sema_test(bounded))
+
+# ************ Cancellation tests ************
+
+async def foo(num):
+    try:
+        await asyncio.sleep(4)
+        return num + 42
+    except CancelError:
+        print('foo was cancelled.')
+        return -1
+
+def kill(task_name):
+    if Cancellable.cancel(task_name):
+        print(task_name, 'will be cancelled when next scheduled')
+    else:
+        print(task_name, 'was not cancellable.')
+
+# Example of a task which cancels another
+async def bar():
+    await asyncio.sleep(1)
+    kill('foo')
+    kill('not_me')  # Will fail because not yet scheduled
+
+async def run_cancel_test1():
+    loop = asyncio.get_event_loop()
+    loop.create_task(bar())
+    res = await Cancellable(foo(5), 'foo')
+    print(res)
+    res = await Cancellable(foo(0), 'not_me')  # Runs to completion
+    print(res)
+
+def cancel_test1():
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(run_cancel_test1())
+
+# TEST with barrier
+
+async def forever(n):
+    print('Started forever() instance', n)
+    while True:
+        await asyncio.sleep(7 + n)
+        print('Running instance', n)
+
+barrier = Barrier(3)
+
+# Cancellable coros must trap the CancelError
+async def rats(n):
+    try:
+        await forever(n)
+    except CancelError:
+        await barrier(nowait = True)
+        print('Instance', n, 'was cancelled')
+
+async def run_cancel_test2():
+    loop = asyncio.get_event_loop()
+    loop.create_task(Cancellable(rats(1), 'rats_1').task)
+    loop.create_task(Cancellable(rats(2), 'rats_2').task)
+    print('Running two tasks')
+    await asyncio.sleep(10)
+    print('About to cancel tasks')
+    Cancellable.cancel('rats_1')
+    Cancellable.cancel('rats_2')
+    await barrier
+    print('tasks were cancelled')
+
+def cancel_test2():
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(run_cancel_test2())
