Adapted for uasyncio V1.7.1 - uses official means of task cancellation.

Author: peterhinch

PRIMITIVES.md

@@ -1,7 +1,9 @@
 # 1. The asyn.py library
 
 This provides five simple synchronisation primitives, together with an API for
-task monitoring and cancellation.
+task monitoring and cancellation. Task cancellation requires usayncio V 1.7.1
+or higher. At the time of writing (7th Jan 2018) it requires a daily build of
+MicroPython firmware or one built from source.
 
 ###### [Main README](./README.md)
 
@@ -528,10 +530,6 @@ Class methods:
  See note in 4.3.1 below.
  * `end` Asynchronous. Arg: A coro name. Run by the `NamedTask` instance to
  inform the class that the instance has ended. Completes quickly.
- * `pend_throw` Synchronous. Args: 1. A coro name 2. An exception passed by
- exception class name (not an instance). The named coro will receive an
- instance of the exception the next time it is scheduled. This method should be
- regarded as **experimental** - its use is discouraged.
 
 Bound method:
  * `__call__` This returns the coro and is used to schedule the task using the

asyn.py

@@ -180,6 +180,13 @@ def _reset(self, down):
         self._down = down
         self._count = self._participants if down else 0
 
+    def busy(self):
+        if self._down:
+            done = self._count == self._participants
+        else:
+            done = self._count == 0
+        return not done
+
     def _at_limit(self):  # Has count reached up or down limit?
         limit = 0 if self._down else self._participants
         return self._count == limit
@@ -223,8 +230,7 @@ def release(self):
 
 # Task Cancellation
 
-class StopTask(Exception):
-    pass
+StopTask = asyncio.CancelledError  # More descriptive name
 
 class TaskId():
     def __init__(self, taskid):
@@ -247,73 +253,6 @@ async def sleep(t, granularity=100):  # 100ms default
             await asyncio.sleep_ms(granularity)
         await asyncio.sleep_ms(rem)
 
-# The NamedTask class enables a coro to be identified by a user defined name.
-# It maintains a dict of [coroutine instance, barrier] indexed by name.
-
-class NamedTask():
-    tasks = {}
-    @classmethod
-    def cancel(cls, name):
-        return cls.pend_throw(name, StopTask)
-
-    @classmethod
-    def pend_throw(cls, taskname, ClsException):
-        if taskname in cls.tasks:
-            # pend_throw() does nothing if the task does not exist
-            # (because it has terminated).
-            # Enable throwing arbitrary exceptions
-            loop = asyncio.get_event_loop()
-            task = cls.tasks[taskname][0]  # coro
-            prev = task.pend_throw(ClsException())  # Instantiate exception
-            if prev is False:
-                loop.call_soon(task)
-            return True
-        return False
-
-    @classmethod
-    def is_running(cls, name):
-        return name in cls.tasks
-
-    @classmethod
-    async def end(cls, name):  # On completion remove it
-        if name in cls.tasks:
-            barrier = cls.tasks[name][1]
-            if barrier is not None:
-                await barrier(nowait = True)
-            del cls.tasks[name]
-
-    def __init__(self, name, gf, *args, barrier=None):
-        if name in self.tasks:
-            raise ValueError('Task name "{}" already exists.'.format(name))
-        task = gf(TaskId(name), *args)
-        self.tasks[name] = [task, barrier]
-        self.task = task
-
-    def __call__(self):
-        return self.task
-
-    def __await__(self):
-        return (yield from self.task)
-
-    __iter__ = __await__
-
-# @namedtask
-
-def namedtask(f):
-    def new_gen(*args):
-        g = f(*args)
-        # Task ID is args[1] if a bound method (else args[0])
-        idx = 0 if isinstance(args[0], TaskId) else 1
-        name = args[idx]()
-        try:
-            res = await g
-            return res
-        except StopTask:
-            pass
-        finally:
-            await NamedTask.end(name)
-    return new_gen
-
 # Anonymous cancellable tasks. These are members of a group which is identified
 # by a user supplied name/number (default 0). Class method cancel_all() cancels
 # all tasks in a group and awaits confirmation (signalled by a task awaiting
@@ -322,27 +261,36 @@ def new_gen(*args):
 
 class Cancellable():
     task_no = 0  # Generated task ID, index of tasks dict
-    tasks = {}  # Value is [coro, group]
-    barrier = None  # Instantiated by the cancel_all method
+    tasks = {}  # Value is [coro, group, barrier] indexed by task_id
 
     @classmethod
     def _cancel(cls, task_no):
-        # pend_throw() does nothing if the task does not exist
-        # (because it has terminated).
         task = cls.tasks[task_no][0]
-        prev = task.pend_throw(StopTask())  # Instantiate exception
-        if prev is False:
-            loop = asyncio.get_event_loop()
-            loop.call_soon(task)
+        asyncio.cancel(task)
 
     @classmethod
-    async def cancel_all(cls, group=0):
-        tokill = [task_no for task_no in cls.tasks if cls.tasks[task_no][1] == group]
-        cls.barrier = Barrier(len(tokill) + 1)  # Include this task
+    async def cancel_all(cls, group=0, nowait=False):
+        tokill = cls._get_task_nos(group)
+        barrier = Barrier(len(tokill) + 1)  # Include this task
         for task_no in tokill:
+            cls.tasks[task_no][2] = barrier
             cls._cancel(task_no)
-        await cls.barrier
-        cls.barrier = None  # Make available for GC
+        await barrier(nowait=nowait)
+
+    @classmethod
+    def is_running(cls, group=0):
+        tasks = cls._get_task_nos(group)
+        if tasks is []:
+            return False
+        for task_no in tasks:
+            barrier = Cancellable.tasks[task_no][2]
+            if barrier.busy():
+                return True
+        return False
+
+    @classmethod
+    def _get_task_nos(cls, group):
+        return [task_no for task_no in cls.tasks if cls.tasks[task_no][1] == group]
 
     @classmethod
     def end(cls, task_no):
@@ -352,14 +300,15 @@ def end(cls, task_no):
     @classmethod
     async def stopped(cls, task_no):
         if task_no in cls.tasks:
+            barrier = Cancellable.tasks[task_no][2]
+            await barrier(nowait=True)
             del cls.tasks[task_no]
-            await Cancellable.barrier(nowait=True)
 
     def __init__(self, gf, *args, group=0):
         task = gf(TaskId(Cancellable.task_no), *args)
         if task in self.tasks:
             raise ValueError('Task already exists.')
-        self.tasks[Cancellable.task_no] = [task, group]
+        self.tasks[Cancellable.task_no] = [task, group, None]
         Cancellable.task_no += 1
         self.task = task
 
@@ -388,48 +337,59 @@ def new_gen(*args):
             Cancellable.end(task_no)
     return new_gen
 
+# The NamedTask class enables a coro to be identified by a user defined name.
+# It maintains a dict of [coroutine instance, barrier] indexed by name.
+
+class NamedTask():
+    tasks = {}
+    @classmethod
+    def cancel(cls, name):
+        if name in cls.tasks:
+            task = cls.tasks[name][0]  # coro
+            asyncio.cancel(task)
+            return True
+        return False
+
+    @classmethod
+    def is_running(cls, name):
+        return name in cls.tasks
 
-# ExitGate is ***obsolete*** since uasyncio now supports task cancellation.
+    @classmethod
+    async def end(cls, name):  # On completion remove it
+        if name in cls.tasks:
+            barrier = cls.tasks[name][1]
+            if barrier is not None:
+                await barrier(nowait = True)
+            del cls.tasks[name]
 
-class ExitGate():
-    def __init__(self, granularity=100):
-        self._ntasks = 0
-        self._ending = False
-        self._granularity = granularity
+    def __init__(self, name, gf, *args, barrier=None):
+        if name in self.tasks:
+            raise ValueError('Task name "{}" already exists.'.format(name))
+        task = gf(TaskId(name), *args)
+        self.tasks[name] = [task, barrier]
+        self.task = task
 
-    def ending(self):
-        return self._ending
+    def __call__(self):
+        return self.task
 
     def __await__(self):
-        self._ending = True
-        while self._ntasks:
-            yield from asyncio.sleep_ms(self._granularity)
-        self._ending = False  # May want to re-use
+        return (yield from self.task)
 
     __iter__ = __await__
 
-    async def __aenter__(self):
-        self._ntasks += 1
-        await asyncio.sleep_ms(0)
+# @namedtask
 
-    async def __aexit__(self, *args):
-        self._ntasks -= 1
-        await asyncio.sleep_ms(0)
-
-    # Sleep while checking for premature ending. Return True on normal ending,
-    # False if premature.
-    async def sleep(self, t):
-        t *= 1000  # ms
-        granularity = self._granularity
-        if t <= granularity:
-            await asyncio.sleep_ms(t)
-        else:
-            n, rem = divmod(t, granularity)
-            for _ in range(n):
-                if self._ending:
-                    return False
-                await asyncio.sleep_ms(granularity)
-            if self._ending:
-                return False
-            await asyncio.sleep_ms(rem)
-        return not self._ending
+def namedtask(f):
+    def new_gen(*args):
+        g = f(*args)
+        # Task ID is args[1] if a bound method (else args[0])
+        idx = 0 if isinstance(args[0], TaskId) else 1
+        name = args[idx]()
+        try:
+            res = await g
+            return res
+        except StopTask:
+            pass
+        finally:
+            await NamedTask.end(name)
+    return new_gen

asyncio_priority.py

@@ -198,6 +198,10 @@ def run_forever(self):
                     if __debug__ and DEBUG:
                         log.debug("Coroutine finished: %s", cb)
                     continue
+                except CancelledError as e:
+                    if __debug__ and DEBUG:
+                        log.debug("Coroutine cancelled: %s", cb)
+                    continue
 
                 if func is not None:
                     self._schedule_hp(func, cb)