Learning uasyncio and futures

[apois]

Hi, I've been playing with micropython and have a situation where I want to collect results concurrently from a number of slow (ie. a few seconds) sensors. I thought this could be a job for uasyncio and futures, but I have seen that uasyncio focuses on coroutines in preference to futures for performance reasons. I've dabbled with Lua coroutines but never found them intuitive to work with, and I find futures/promises much easier to work with conceptually.

I'm aware of the additional asyn.py library of concurrency primitives, and the implementation of asynio.gather() using "barriers", which I don't fully understand but looks quite elegant and is probably the "right" way to do it.

But it seemed like quite a lot of machinery to do something I was hoping was going to be quite simple, so I thought I'd have a go at doing some "poor man's" futures using "awaitable" classes and polling. I think it is kind of working - it is not the most elegant, and is probably broken in a number of regards, but it has the advantage that I understand what is going on!

I'm posting the code here for critique - please feel free to rip it to shreds! Is this a bad approach? I've made no attempt to deal with exceptions or timeouts. Is there a simple way to check for completion without polling? (I think the barrier way is probably right, I'm just curious about other approaches). Should I just use asyn.py?

[code]
import uasyncio as asyncio


class PoorFuture(object):

def __init__(self, coro):
self.coro = coro
self.started = False
self.completed = False
self.result = None

def __await__(self):
self.started = True
self.result = await self.coro
self.completed = True
return self.result

__iter__ = __await__

async def as_task(self):
'''Convenience method to turn the future into a task'''
return await self



async def slow_task(seconds, answer):
'''Demo function to build coros that simulate slow-running tasks'''
for i in range(seconds):
print(i, 'of', seconds)
await asyncio.sleep(1)
return answer



async def poor_gather(futures):
'''Schedule a list of futures, wait for them to complete, and return the list of results'''

loop = asyncio.get_event_loop()

# start all the futures as tasks on the event loop
for f in futures:
loop.create_task(f.as_task())

# poll until all the futures have completed
finished = False
while not(finished):
await asyncio.sleep_ms(100)
print([f.completed for f in futures])
# check if they have all finished
finished = all(f.completed for f in futures)

results = [f.result for f in futures]
print('results:', results)
return results


futures = [
PoorFuture(slow_task(3, 'banana')),
PoorFuture(slow_task(5, 'satsuma')),
PoorFuture(slow_task(1, 'mango')),
]


loop = asyncio.get_event_loop()
loop.run_until_complete(poor_gather(futures))
[/code]

[pythoncoder]

That is quite a neat solution.

Is your aim to write a general-purpose Future class and to replicate my asyn library with Future compatible primitives? If so, I'd suggest as a minimum, a mechanism for cancelling Future instances and a means of creating a Future with a timeout. Both of these are required in communications protocols.

This could be a useful project, especially if you can get close to CPython syntax.

[apois]

Thanks - glad to hear I'm not totally off-course

No - I don't have any ambitions beyond my simple "gather" use case for collecting results concurrently from a number of sensors - I just wanted a simple way of doing this. I'm not familiar enough with the asyncio library or futures in general to produce a decent general-purpose library.

Thank you for the suggestions - yes, I think both cancelling and timeouts could be useful. But by the time I have implemented those properly it may be just as easy to use the asyn.py library?

[pythoncoder]

Well, that's your call I found cancellation and timeouts rather tricky to implement, but then I'm far too old for this kind of thing; a younger mind might polish them off in a day...

The following shows gather being used with timeouts and cancellation (tested on Unix and Pyboard):

Code: Select all

import uasyncio as asyncio
import asyn

async def foo(n):
    while True:
        try:
            await asyncio.sleep(1)
            n += 1
        except asyncio.TimeoutError:
            print('foo timeout')
            return n

@asyn.cancellable
async def bar(n):
    while True:
        try:
            await asyncio.sleep(1)
            n += 1
        except asyn.StopTask:
            print('bar stopped')
            return n

async def do_cancel():
    await asyncio.sleep(5)
    await asyn.Cancellable.cancel_all()

async def main(loop):
    bar_task = asyn.Cancellable(bar, 70)
    gatherables = [asyn.Gatherable(bar_task),]
    gatherables.append(asyn.Gatherable(foo, 10, timeout=7))
    loop.create_task(do_cancel())
    res = await asyn.Gather(gatherables)
    print('Result: ', res)

loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))

[HermannSW]

Hi apois,

I tried to get your code running with unix port of MicroPython.
But unfortunately you did not use this forum's code block correctly.
I tried to figure out the whitespaces needed to insert and came up with this:

Code: Select all

import uasyncio as asyncio


class PoorFuture(object):

    def __init__(self, coro):
        self.coro = coro
        self.started = False
        self.completed = False
        self.result = None

    def __await__(self):
        self.started = True
        self.result = await self.coro
        self.completed = True
        return self.result

    __iter__ = __await__

    async def as_task(self):
        '''Convenience method to turn the future into a task'''
        return await self



async def slow_task(seconds, answer):
    '''Demo function to build coros that simulate slow-running tasks'''
    for i in range(seconds):
        print(i, 'of', seconds)
        await asyncio.sleep(1)
    return answer



async def poor_gather(futures):
    '''Schedule a list of futures, wait for them to complete, and return the list of results'''

    loop = asyncio.get_event_loop()

    # start all the futures as tasks on the event loop
    for f in futures:
        loop.create_task(f.as_task())

        # poll until all the futures have completed
        finished = False
        while not(finished):
            await asyncio.sleep_ms(100)
            print([f.completed for f in futures])
            # check if they have all finished
            finished = all(f.completed for f in futures)

        results = [f.result for f in futures]
        print('results:', results)
        return results


futures = [
    PoorFuture(slow_task(3, 'banana')),
    PoorFuture(slow_task(5, 'satsuma')),
    PoorFuture(slow_task(1, 'mango')),
]


loop = asyncio.get_event_loop()
loop.run_until_complete(poor_gather(futures))

It works partially, but does never end.
Please show where I did adding spaces wrong:

Code: Select all

$ ./micropython poorfutures.py 
0 of 3
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
1 of 3
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
2 of 3
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
[True, False, False]
...

P.S:
This is my setup:

Code: Select all

$ pwd
/home/stammw/micropython/ports/unix
$ ll logging.py 
-rw-rw-r--. 1 stammw stammw 2094 Nov 13 14:38 logging.py
$ ll uasyncio/
total 24
-rw-rw-r--. 1 stammw stammw 9509 Nov 16 15:15 core.py
-rw-rw-r--. 1 stammw stammw 8400 Nov 16 15:15 __init__.py
$ 

[apois]

Thanks - yes, the editor didn't seem to let me insert code blocks properly for some reason - maybe because I am new? It seems to be working now however.

You are nearly there with your indentation, but I think you have some extra indention after this for loop - it should be:

Code: Select all

    # start all the futures as tasks on the event loop
    for f in futures:
        loop.create_task(f.as_task())

    # poll until all the futures have completed
    finished = False
    while not(finished):
        await asyncio.sleep_ms(100)
        print([f.completed for f in futures])
        # check if they have all finished
        finished = all(f.completed for f in futures)

    results = [f.result for f in futures]
    print('results:', results)
    return results 

Thanks pythoncoder - I will try the asyn.py version.

[HermannSW]

Thanks, with that changes it works.
Unfortunately importing takes so much RAM that it runs oom on ESP-01s with 28KB free RAM.
On ESP32 it works, and shows that it takes 15584 bytes (see at bottom):

Code: Select all

MicroPython v1.9.4-623-g34af10d2e on 2018-10-03; ESP32 module with ESP32
Type "help()" for more information.
>>> gc.collect(); m0=gc.mem_free()
>>> import poorfutures
0 of 3
0 of 5
0 of 1
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
[False, False, False]
1 of 3
1 of 5
[False, False, True]
[False, False, True]
[False, False, True]
[False, False, True]
[False, False, True]
[False, False, True]
[False, False, True]
[False, False, True]
[False, False, True]
2 of 3
2 of 5
[False, False, True]
[False, False, True]
[False, False, True]
[False, False, True]
[False, False, True]
[False, False, True]
[False, False, True]
[False, False, True]
[True, False, True]
3 of 5
[True, False, True]
[True, False, True]
[True, False, True]
[True, False, True]
[True, False, True]
[True, False, True]
[True, False, True]
[True, False, True]
4 of 5
[True, False, True]
[True, False, True]
[True, False, True]
[True, False, True]
[True, False, True]
[True, False, True]
[True, False, True]
[True, False, True]
[True, False, True]
[True, True, True]
results: ['banana', 'satsuma', 'mango']
>>> gc.collect(); m1=gc.mem_free()
>>> print(m0,m1,m0-m1)
100944 85360 15584
>>> 

P.S:
I used this script to copy over to module what is needed -- perhaps I should learn how to use upip on my modules:

Code: Select all

$ cat do_poor 
#!/bin/bash
~/webrepl/webrepl_cli.py -p abcd poorfutures.py 192.168.4.1:

./micropython mp.py 192.168.4.1 'uos.mkdir("uasyncio")'

cd ~/micropython-lib
~/webrepl/webrepl_cli.py -p abcd logging/logging.py 192.168.4.1:
~/webrepl/webrepl_cli.py -p abcd uasyncio.core/uasyncio/core.py 192.168.4.1:uasyncio/
~/webrepl/webrepl_cli.py -p abcd uasyncio/uasyncio/__init__.py 192.168.4.1:uasyncio/
$ 

[apois]

I've been trying to simplify my gather() and have come up with a version using just coros that doesn't need the PoorFuture wrapper class.

Code: Select all

import uasyncio as asyncio


async def slow_task(seconds, answer):
    '''Demo function to build coros that simulate slow-running tasks'''
    for i in range(seconds):
        print(i, 'of', seconds)
        await asyncio.sleep(1)
    return answer


async def gather(coros):
    # count how many coros we have left to complete
    count = len(coros)

    # prepare a list to hold the results
    results = [None for i in range(count)]

    loop = asyncio.get_event_loop()

    # use a closure to wrap a coro into a task that does
    # the necessary record-keeping upon completion
    async def wrap(coro, i):
        result = await coro
        results[i] = result
        count = count - 1

    # wrap all the coros and kick them off on the event loop
    for i in range(count):
        task = wrap(coros[i], i)
        loop.create_task(task)
    
    # keep polling until they are all done
    while count > 0 :
        print(count)
        await asyncio.sleep_ms(100)
    
    print(results)
    return results
        

coros = [
    slow_task(3, 'banana'),
    slow_task(5, 'satsuma'),
    slow_task(1, 'mango'),
]


loop = asyncio.get_event_loop()
loop.run_until_complete(gather(coros))

I think this is simpler, but I quite like the idea of the PoorFuture object keeping track of its own status and result. I haven't done any investigation regarding memory usage - I'm targetting an ESP32 so I hope this isn't an issue.

I haven't invested any thought into timeouts / cancelling / exception handling yet.

[HermannSW]

Running on ESP32 is fine, but micropython unix port reveals a problem:

Code: Select all

...
1 of 3
1 of 5
Traceback (most recent call last):
  File "coros.py", line 50, in <module>
  File "/home/stammw/micropython/ports/unix/uasyncio/core.py", line 180, in run_until_complete
  File "/home/stammw/micropython/ports/unix/uasyncio/core.py", line 154, in run_forever
  File "/home/stammw/micropython/ports/unix/uasyncio/core.py", line 109, in run_forever
  File "coros.py", line 26, in wrap
NameError: local variable referenced before assignment
$

[pythoncoder]

To avoid the error wrap should read:

Code: Select all

    async def wrap(coro, i):
        nonlocal count
        result = await coro
        results[i] = result
        count = count - 1

Note that

Code: Select all

   results = [None for i in range(count)]

could be written

Code: Select all

   results = [None] * count