Asyncio -- delayed calculation

[Zachary Ware]

On Mon, Nov 28, 2016 at 6:48 AM, Steve D'Aprano
<steve+python at pearwood.info> wrote:
> What am I doing wrong?

Give yourself a bit more to debug with, since you're going to want to
do something with the result your expensive calculation anyway:

import asyncio

class Counter:
    def __init__(self, i):
        self.count = 10
        self.i = i

    async def count_down(self):
        print(self, self.i, "starting")
        while self.count > 0:
            # simulate a computation
            await asyncio.sleep(0.5)
            self.count -= 1
        print(self, self.i, "completed")
        return self.i + self.count

async def main():
    pool = [Counter(i) for i in range(5)]
    results = []
    for obj in pool:
        results.append(obj.count_down())
    return results

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


This gives you:

[<coroutine object Counter.count_down at 0x101631fc0>, <coroutine
object Counter.count_down at 0x101659048>, <coroutine object
Counter.count_down at 0x1016590f8>, <coroutine object
Counter.count_down at 0x101659150>, <coroutine object
Counter.count_down at 0x1016591a8>]
asynctest.py:25: RuntimeWarning: coroutine 'Counter.count_down' was
never awaited
  print(loop.run_until_complete(main()))

Ok, so let's fix that by adding an 'await' on line 21 (it's reported
at line 25 because that's when the unawaited coroutines are gc'd):

        results.append(await obj.count_down())

Running that gives:

<__main__.Counter object at 0x10203f978> 0 starting
<__main__.Counter object at 0x10203f978> 0 completed
<__main__.Counter object at 0x1025af710> 1 starting
<__main__.Counter object at 0x1025af710> 1 completed
<__main__.Counter object at 0x1025b60b8> 2 starting
<__main__.Counter object at 0x1025b60b8> 2 completed
<__main__.Counter object at 0x1025b60f0> 3 starting
<__main__.Counter object at 0x1025b60f0> 3 completed
<__main__.Counter object at 0x1025b6128> 4 starting
<__main__.Counter object at 0x1025b6128> 4 completed
[0, 1, 2, 3, 4]

Still not right, only one count_down is run at a time.  But that's
because we're using a synchronous for loop to await our results and
populate the results list.  Naively, I tried an 'async for', but
that's trying to be asynchronous in the wrong place:

Traceback (most recent call last):
  File "asynctest.py", line 25, in <module>
    print(loop.run_until_complete(main()))
  File "/usr/lib/python3.5/asyncio/base_events.py", line 387, in
run_until_complete
    return future.result()
  File "/usr/lib/python3.5/asyncio/futures.py", line 274, in result
    raise self._exception
  File "/usr/lib/python3.5/asyncio/tasks.py", line 239, in _step
    result = coro.send(None)
  File "asynctest.py", line 20, in main
    async for obj in pool:
TypeError: 'async for' requires an object with __aiter__ method, got list

So instead, checking the docs suggests using asyncio.gather for
parallel execution of tasks, which takes a variable number of
'coros_or_futures'.  On the first attempt, we saw that we had
accidentally created a list of "coroutine objects", so lets go back
and use that:

--- asynctest.py.orig 2016-11-28 21:03:04.000000000 -0600
+++ asynctest.py 2016-11-28 21:03:35.000000000 -0600
@@ -16,9 +16,10 @@

 async def main():
     pool = [Counter(i) for i in range(5)]
-    results = []
+    coros = []
     for obj in pool:
-        results.append(await obj.count_down())
+        coros.append(obj.count_down())
+    results = asyncio.gather(*coros)
     return results

 loop = asyncio.get_event_loop()

Output:

<__main__.Counter object at 0x1026b6160> 4 starting
<__main__.Counter object at 0x10213f978> 0 starting
<__main__.Counter object at 0x1026b6128> 3 starting
<__main__.Counter object at 0x1026af748> 1 starting
<__main__.Counter object at 0x1026b60f0> 2 starting
<_GatheringFuture pending>

Now we've started everything asynchronously, but it exited way too
fast and never completed anything.  But instead of a list of results,
we got a _GatheringFuture at the end of everything.  So let's await
that:

    results = await asyncio.gather(*coros)

And now we get:

<__main__.Counter object at 0x101eb6160> 4 starting
<__main__.Counter object at 0x10063f978> 0 starting
<__main__.Counter object at 0x101eb6128> 3 starting
<__main__.Counter object at 0x101eaf748> 1 starting
<__main__.Counter object at 0x101eb60f0> 2 starting
<__main__.Counter object at 0x101eb6160> 4 completed
<__main__.Counter object at 0x10063f978> 0 completed
<__main__.Counter object at 0x101eb6128> 3 completed
<__main__.Counter object at 0x101eaf748> 1 completed
<__main__.Counter object at 0x101eb60f0> 2 completed
[0, 1, 2, 3, 4]


And there we have it.  Our final script is:

import asyncio

class Counter:
    def __init__(self, i):
        self.count = 10
        self.i = i

    async def count_down(self):
        print(self, self.i, "starting")
        while self.count > 0:
            # simulate a computation
            await asyncio.sleep(0.5)
            self.count -= 1
        print(self, self.i, "completed")
        return self.i + self.count

async def main():
    pool = [Counter(i) for i in range(5)]
    coros = []
    for obj in pool:
        coros.append(obj.count_down())
    results = await asyncio.gather(*coros)
    return results

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


I hope this is a nice prod to your understanding,
-- 
Zach