Uasyncio V2.0 code changes.

Author: peterhinch

FASTPOLL.md

@@ -238,10 +238,11 @@ priority, one with higher and one with lower priority than standard. It
 provides a replacement for `uasyncio.get_event_loop()` enabling the queue
 sizes to be set.
 
-`aysncio_priority.get_event_loop(len, lpqlen)`  
+`aysncio_priority.get_event_loop(runq_len, waitq_len, lpqlen)`  
 Arguments:  
- 1. `len` Length of normal queue. Default 42 tasks.
- 2. `lpqlen` Length of low priority queue. Default 42.
+ 1. `runq_len` Length of normal queue. Default 16 tasks.
+ 2. `waitq_len` Length of wait queue. Default 16.
+ 3. `lpqlen` Length of low priority queue. Default 16.
 
 The low priority solution is based on the notion of "after" implying a time
 delay which can be expected to be less precise than the asyncio standard calls.
@@ -348,12 +349,11 @@ priority tasks which are pending execution.
 A simple demo of this is `benchmarks/call_lp.py`. Documentation is in the
 code.
 
-`call_after_ms(delay, callback, args=())` Call with low priority. Positional
+`call_after_ms(delay, callback, *args)` Call with low priority. Positional
 args:  
  1. `delay` Integer. Minimum delay in millisecs before callback runs.
  2. `callback` The callback to run.
- 3. `args` Optional tuple or other iterable with positional args for the
- callback.
+ 3. `*args` Optional positional args for the callback.
 
 ###### [Jump to Contents](./FASTPOLL.md#contents)
 

asyncio_priority.py

@@ -28,43 +28,35 @@
 
 import utime as time
 import utimeq
+import ucollections
 from uasyncio import *
 
 class PriorityEventLoop(PollEventLoop):
-    def __init__(self, len=42, lpqlen=42):
-        super().__init__(len)
+    def __init__(self, runq_len=16, waitq_len=16, lpqlen=42):
+        super().__init__(runq_len, waitq_len)
         self._max_overdue_ms = 0
         self.lpq = utimeq.utimeq(lpqlen)
-        self.hp_tasks = None
-        self.create_task(self.lp_monitor())
-
-    # Monitor low priority tasks. If one can be scheduled, remove from LP queue
-    # and queue it for scheduling.
-    # If a normal task is ready we only queue an LP one which is due by more
-    # than the max_overdue_ms threshold.
-    # If no normal task is ready we queue the most overdue LP task.
-    # Readiness is determined by whether it is actually due. An option would be
-    # to check if it's due in less than N ms. This would reduce competition at
-    # the cost of the application having to consider tight loops which pend for
-    # < N ms.
-    async def lp_monitor(self):
-        this_task = [0, 0, 0]
-        while True:
-            if self.lpq:
-                tnow = self.time()
-                t = self.lpq.peektime()
-                tim = time.ticks_diff(t, tnow)
-                to_run = self._max_overdue_ms > 0 and tim < -self._max_overdue_ms
-                if not to_run:  # No overdue LP task. Are any normal tasks due?
-                    can_run = True  # If q is empty can run an LP task
-                    if self.q:
-                        t = self.q.peektime()
-                        can_run = time.ticks_diff(t, tnow) > 0 # No normal task is ready -
-                    to_run = can_run and tim <= 0  # run if so and an LP one is ready
-                if to_run:
-                    self.lpq.pop(this_task)
-                    self.q.push(*this_task)
-            yield
+        self.hp_tasks = []
+
+    # Schedule a single low priority task if one is ready or overdue.
+    # The most overdue task is scheduled even if normal tasks are pending.
+    # The most due task is scheduled only if no normal tasks are pending.
+    def schedule_lp_task(self, cur_task, tnow):
+        t = self.lpq.peektime()
+        tim = time.ticks_diff(t, tnow)
+        to_run = self._max_overdue_ms > 0 and tim < -self._max_overdue_ms
+        if not to_run:  # No overdue LP task.
+            if len(self.runq):
+                return False
+            to_run = tim <= 0  # True if LP task is due
+            if to_run and self.waitq:  # Set False if a normal tasks is due.
+                t = self.waitq.peektime()
+                to_run = time.ticks_diff(t, tnow) > 0 # No normal task is ready
+        if to_run:
+            self.lpq.pop(cur_task)
+            self.call_soon(cur_task[1], *cur_task[2])
+            return True
+        return False
 
     def max_overdue_ms(self, t=None):
         if t is not None:
@@ -73,17 +65,16 @@ def max_overdue_ms(self, t=None):
 
     # Low priority versions of call_later() call_later_ms() and call_at_()
     def call_after_ms(self, delay, callback, *args):
-        self.call_at_lp_(time.ticks_add(self.time(), delay), callback, args)
+        self.call_at_lp_(time.ticks_add(self.time(), delay), callback, *args)
+
 
     def call_after(self, delay, callback, *args):
-        self.call_at_lp_(time.ticks_add(self.time(), int(delay * 1000)), callback, args)
+        self.call_at_lp_(time.ticks_add(self.time(), int(delay * 1000)), callback, *args)
 
-    def call_at_lp_(self, time, callback, args=()):
+    def call_at_lp_(self, time, callback, *args):
         self.lpq.push(time, callback, args)
 
-    def _schedule_hp(self, func, callback, args=()):
-        if self.hp_tasks is None:
-            self.hp_tasks = []
+    def _schedule_hp(self, func, callback, *args):
         # If there's an empty slot, assign without allocation
         for entry in self.hp_tasks:  # O(N) search - but N is typically 1 or 2...
             if not entry[0]:
@@ -97,80 +88,79 @@ def _schedule_hp(self, func, callback, args=()):
     def run_forever(self):
         cur_task = [0, 0, 0]
         while True:
-            if self.q:
-                # wait() may finish prematurely due to I/O completion,
-                # and schedule new, earlier than before tasks to run.
-                while 1:
-                    # Check list of high priority tasks
-                    if self.hp_tasks is not None:
-                        hp_found = False
-                        for entry in self.hp_tasks:
-                            if entry[0] and entry[0]():
-                                hp_found = True
-                                entry[0] = 0
-                                cur_task[0] = 0
-                                cur_task[1] = entry[1] # ??? quick non-allocating copy
-                                cur_task[2] = entry[2]
-                                break
-                        if hp_found:
-                            break
-
-                    # Schedule any due normal task
-                    t = self.q.peektime()
-                    tnow = self.time()
+            tnow = self.time()
+            # Schedule a LP task if no normal task is ready
+            l = len(self.lpq)
+            if (l and not self.schedule_lp_task(cur_task, tnow)) or l == 0:
+                # Expire entries in waitq and move them to runq
+                while self.waitq:
+                    t = self.waitq.peektime()
                     delay = time.ticks_diff(t, tnow)
-                    if delay <= 0:
-                        # Always call wait(), to give a chance to I/O scheduling
-                        self.wait(0)
-                        self.q.pop(cur_task)
+                    if delay > 0:
                         break
+                    self.waitq.pop(cur_task)
+                    if __debug__ and DEBUG:
+                        log.debug("Moving from waitq to runq: %s", cur_task[1])
+                    self.call_soon(cur_task[1], *cur_task[2])
+
+            # Process runq
+            l = len(self.runq)
+            if __debug__ and DEBUG:
+                log.debug("Entries in runq: %d", l)
+            while l:
+                # Check list of high priority tasks
+                cb = None
+                for entry in self.hp_tasks:
+                    if entry[0] and entry[0]():  # Ready to run
+                        entry[0] = 0
+                        cb = entry[1]
+                        args = entry[2]
+                        break
+
+                if cb is None:
+                    cb = self.runq.popleft()
+                    l -= 1
+                    args = ()
+                    if not isinstance(cb, type_gen):
+                        args = self.runq.popleft()
+                        l -= 1
+                        if __debug__ and DEBUG:
+                            log.info("Next callback to run: %s", (cb, args))
+                        cb(*args)
+                        continue
 
-                    self.wait(delay)  # Handled in superclass
-                t = cur_task[0]
-                cb = cur_task[1]
-                args = cur_task[2]
                 if __debug__ and DEBUG:
-                    log.debug("Next coroutine to run: %s", (t, cb, args))
-#                __main__.mem_info()
+                    log.info("Next coroutine to run: %s", (cb, args))
                 self.cur_task = cb
-            else:
-                self.wait(-1)
-                # Assuming IO completion scheduled some tasks
-                continue
-            if callable(cb):
-                cb(*args)
-            else:
                 delay = 0
                 func = None
-                priority = True
+                low_priority = False  # Assume normal priority
                 try:
-                    if __debug__ and DEBUG:
-                        log.debug("Coroutine %s send args: %s", cb, args)
-                    if args == ():
-                        ret = next(cb)  # See notes at end of code
+                    if args is ():
+                        ret = next(cb)
                     else:
                         ret = cb.send(*args)
                     if __debug__ and DEBUG:
-                        log.debug("Coroutine %s yield result: %s", cb, ret)
+                        log.info("Coroutine %s yield result: %s", cb, ret)
                     if isinstance(ret, SysCall1):
                         arg = ret.arg
-                        if isinstance(ret, After):
-                            delay = int(arg * 1000)
-                            priority = False
-                        elif isinstance(ret, AfterMs):
-                            delay = int(arg)
-                            priority = False
-                        elif isinstance(ret, When):
-                            if callable(arg):
-                                func = arg
-                            else:
-                                assert False, "Argument to 'when' must be a function or method."
-                        elif isinstance(ret, SleepMs):
+                        if isinstance(ret, SleepMs):
                             delay = arg
+                            if isinstance(ret, AfterMs):
+                                low_priority = True
+                                if isinstance(ret, After):
+                                    delay = int(delay*1000)
+                            elif isinstance(ret, When):
+                                if callable(arg):
+                                    func = arg
+                                else:
+                                    assert False, "Argument to 'when' must be a function or method."
                         elif isinstance(ret, IORead):
+                            cb.pend_throw(False)
                             self.add_reader(arg, cb)
                             continue
                         elif isinstance(ret, IOWrite):
+                            cb.pend_throw(False)
                             self.add_writer(arg, cb)
                             continue
                         elif isinstance(ret, IOReadDone):
@@ -202,18 +192,37 @@ def run_forever(self):
                     if __debug__ and DEBUG:
                         log.debug("Coroutine cancelled: %s", cb)
                     continue
-
                 if func is not None:
                     self._schedule_hp(func, cb)
+                    continue
+                # Currently all syscalls don't return anything, so we don't
+                # need to feed anything to the next invocation of coroutine.
+                # If that changes, need to pass that value below.
+                if low_priority:
+                    self.call_after_ms(delay, cb)
+                elif delay:
+                    self.call_later_ms(delay, cb)
                 else:
-                    # Currently all syscalls don't return anything, so we don't
-                    # need to feed anything to the next invocation of coroutine.
-                    # If that changes, need to pass that value below.
-                    if priority:
-                        self.call_later_ms(delay, cb)
-                    else:
-                        self.call_after_ms(delay, cb)
+                    self.call_soon(cb)
 
+            # Wait until next waitq task or I/O availability
+            delay = 0
+            if not self.runq:
+                delay = -1
+                tnow = self.time()
+                if self.waitq:
+                    t = self.waitq.peektime()
+                    delay = time.ticks_diff(t, tnow)
+                    if delay < 0:
+                        delay = 0
+                if self.lpq:
+                    t = self.lpq.peektime()
+                    lpdelay = time.ticks_diff(t, tnow)
+                    if lpdelay < 0:
+                        lpdelay = 0
+                    if lpdelay < delay or delay < 0:
+                        delay = lpdelay
+            self.wait(delay)
 
 # Low priority
 class AfterMs(SleepMs):
@@ -232,7 +241,7 @@ class When(SleepMs):
 
 import uasyncio.core
 uasyncio.core._event_loop_class = PriorityEventLoop
-def get_event_loop(len=42, lpqlen=42):
+def get_event_loop(runq_len=16, waitq_len=16, lpqlen=16):
     if uasyncio.core._event_loop is None:  # Add a q entry for lp_monitor()
-        uasyncio.core._event_loop = uasyncio.core._event_loop_class(len + 1, lpqlen)
+        uasyncio.core._event_loop = uasyncio.core._event_loop_class(runq_len, waitq_len, lpqlen)
     return uasyncio.core._event_loop

asyntest.py

@@ -152,24 +152,20 @@ async def run_event_test(lp):
     loop.create_task(run_lock(3, lock))
     print('Test Event class')
     event = asyn.Event(lp)
-    print('got here')
     loop.create_task(eventset(event))
-    print('gh1')
     await eventwait(event)  # run_event_test runs fast until this point
     print('Event status {}'.format('Incorrect' if event.is_set() else 'OK'))
     print('Tasks complete')
 
 def event_test(lp=True):  # Option to use low priority scheduling
     printexp('''Test Lock class
 Test Event class
-got here
-gh1
 waiting for event
 run_lock 1 waiting for lock
+run_lock 1 acquired lock
 run_lock 2 waiting for lock
 run_lock 3 waiting for lock
 Waiting 5 secs before setting event
-run_lock 1 acquired lock
 run_lock 1 released lock
 run_lock 2 acquired lock
 run_lock 2 released lock

benchmarks/latency.py

@@ -102,14 +102,14 @@ def test(use_priority=True):
     else:
         ntasks = max(num_coros) + 10 #4
         if use_priority:
-            loop = asyncio.get_event_loop(ntasks, ntasks)
+            loop = asyncio.get_event_loop(ntasks, ntasks, ntasks)
             after = asyncio.after
             after_ms = asyncio.after_ms
         else:
             lp_version = False
             after = asyncio.sleep
             after_ms = asyncio.sleep_ms
-            loop = asyncio.get_event_loop(ntasks)
+            loop = asyncio.get_event_loop(ntasks, ntasks)
         s = 'Testing latency of priority task with coros blocking for {}ms.'
         print(s.format(processing_delay))
         if lp_version:

benchmarks/rate.py

@@ -1,10 +1,9 @@
-# rate.py Benchmark for uasyncio. Author Peter Hinch April 2017.
+# rate.py Benchmark for uasyncio. Author Peter Hinch Feb 2018.
 # Benchmark uasyncio round-robin scheduling performance
 # This measures the rate at which uasyncio can schedule a minimal coro which
 # mereley increments a global.
 
-# Outcome: minimal coros are scheduled at an interval of ~208us, independent of
-# the number of instances.
+# Outcome: minimal coros are scheduled at an interval of ~150us
 
 import uasyncio as asyncio
 
@@ -40,7 +39,8 @@ async def test():
         iterations[n] = count
     done = True
 
-loop = asyncio.get_event_loop(max(num_coros) + 2)
+ntasks = max(num_coros) + 2
+loop = asyncio.get_event_loop(ntasks, ntasks)
 loop.create_task(test())
 loop.run_until_complete(report())