V3 primitives replace polling with Event.

Author: peterhinch

v3/docs/TUTORIAL.md

@@ -500,8 +500,8 @@ target. A primitive is loaded by issuing (for example):
 from primitives.semaphore import Semaphore, BoundedSemaphore
 from primitives.queue import Queue
 ```
-When `uasyncio` acquires an official version (which will be more efficient) the
-invocation lines alone should be changed:
+When `uasyncio` acquires official versions of the CPython primitives the
+invocation lines alone should be changed. e.g. :
 ```python
 from uasyncio import Semaphore, BoundedSemaphore
 from uasyncio import Queue
@@ -848,7 +848,7 @@ asyncio.run(queue_go(4))
 
 ## 3.6 Message
 
-This is an unofficial primitive and has no analog in CPython asyncio.
+This is an unofficial primitive and has no counterpart in CPython asyncio.
 
 This is a minor adaptation of the `Event` class. It provides the following:
  * `.set()` has an optional data payload.
@@ -910,8 +910,8 @@ Secondly it can allow a task to pause until one or more other tasks have
 terminated or passed a particular point. For example an application might want
 to shut down various peripherals before starting a sleep period. The task
 wanting to sleep initiates several shut down tasks and waits until they have
-triggered the barrier to indicate completion. This use case may be better
-served by `gather`.
+triggered the barrier to indicate completion. This use case may also be served
+by `gather`.
 
 The key difference between `Barrier` and `gather` is symmetry: `gather` is
 asymmetrical. One task owns the `gather` and awaits completion of a set of
@@ -921,9 +921,8 @@ the `while True:` constructs common in firmware applications. Use of `gather`
 would imply instantiating a set of tasks on every pass of the loop.
 
 `gather` provides access to return values; irrelevant to `Barrier` because
-passing a barrier does not imply return. 
-
-Currently `gather` is more efficient.
+passing a barrier does not imply return. `Barrier` now has an efficient
+implementation using `Event` to suspend waiting tasks.
 
 Constructor.  
 Mandatory arg:  

v3/primitives/barrier.py

@@ -2,6 +2,8 @@
 # Copyright (c) 2018-2020 Peter Hinch
 # Released under the MIT License (MIT) - see LICENSE file
 
+# Now uses Event rather than polling.
+
 try:
     import uasyncio as asyncio
 except ImportError:
@@ -14,20 +16,14 @@
 # At that point the callback is executed. Then the barrier is 'opened' and
 # execution of all participants resumes.
 
-# The nowait arg is to support task cancellation. It enables usage where one or
-# more coros can register that they have reached the barrier without waiting
-# for it. Any 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 which have been
-# cancelled to leave the task queue as soon as possible.
-
 class Barrier():
     def __init__(self, participants, func=None, args=()):
         self._participants = participants
         self._func = func
         self._args = args
         self._reset(True)
         self._res = None
+        self._evt = asyncio.Event()
 
     def __await__(self):
         if self.trigger():
@@ -37,15 +33,19 @@ def __await__(self):
         while True:  # Wait until last waiting task changes the direction
             if direction != self._down:
                 return
-            await asyncio.sleep_ms(0)
+            await self._evt.wait()
+            self._evt.clear()
 
     __iter__ = __await__
 
     def result(self):
         return self._res
 
     def trigger(self):
-        self._update()
+        self._count += -1 if self._down else 1
+        if self._count < 0 or self._count > self._participants:
+            raise ValueError('Too many tasks accessing Barrier')
+        self._evt.set()
         if self._at_limit():  # All other tasks are also at limit
             if self._func is not None:
                 self._res = launch(self._func, self._args)
@@ -67,8 +67,3 @@ def busy(self):
     def _at_limit(self):  # Has count reached up or down limit?
         limit = 0 if self._down else self._participants
         return self._count == limit
-
-    def _update(self):
-        self._count += -1 if self._down else 1
-        if self._count < 0 or self._count > self._participants:
-            raise ValueError('Too many tasks accessing Barrier')

v3/primitives/queue.py

@@ -23,14 +23,18 @@ class Queue:
     def __init__(self, maxsize=0):
         self.maxsize = maxsize
         self._queue = []
+        self._evput = asyncio.Event()  # Triggered by put, tested by get
+        self._evget = asyncio.Event()  # Triggered by get, tested by put
 
     def _get(self):
+        self._evget.set()
         return self._queue.pop(0)
 
     async def get(self):  #  Usage: item = await queue.get()
-        while self.empty():
+        if self.empty():
             # Queue is empty, put the calling Task on the waiting queue
-            await asyncio.sleep_ms(0)
+            await self._evput.wait()
+            self._evput.clear()
         return self._get()
 
     def get_nowait(self):  # Remove and return an item from the queue.
@@ -40,12 +44,14 @@ def get_nowait(self):  # Remove and return an item from the queue.
         return self._get()
 
     def _put(self, val):
+        self._evput.set()
         self._queue.append(val)
 
     async def put(self, val):  # Usage: await queue.put(item)
-        while self.qsize() >= self.maxsize and self.maxsize:
+        if self.qsize() >= self.maxsize and self.maxsize:
             # Queue full
-            await asyncio.sleep_ms(0)
+            await self._evget.wait()
+            self._evget.clear()
             # Task(s) waiting to get from queue, schedule first Task
         self._put(val)
 

v3/primitives/semaphore.py

@@ -13,6 +13,7 @@
 class Semaphore():
     def __init__(self, value=1):
         self._count = value
+        self._event = asyncio.Event()
 
     async def __aenter__(self):
         await self.acquire()
@@ -23,11 +24,16 @@ async def __aexit__(self, *args):
         await asyncio.sleep(0)
 
     async def acquire(self):
-        while self._count == 0:
-            await asyncio.sleep_ms(0)
+        self._event.clear()
+        while self._count == 0:  # Multiple tasks may be waiting for
+            await self._event.wait()  # a release
+            self._event.clear()
+            # When we yield, another task may succeed. In this case
+            await asyncio.sleep(0)  # the loop repeats
         self._count -= 1
 
     def release(self):
+        self._event.set()
         self._count += 1
 
 class BoundedSemaphore(Semaphore):
@@ -37,6 +43,6 @@ def __init__(self, value=1):
 
     def release(self):
         if self._count < self._initial_value:
-            self._count += 1
+            super().release()
         else:
             raise ValueError('Semaphore released more than acquired')