CODEWITHSUNDEEP
pythonpython-asyncioconcurrent.futures
user2416984
user2416984

Reputation: 971

Concurrent future polling of series of blocking calls

I'm trying to generate a polling mechanism for a long running task in Python. To do this, I'm using a concurrent Future and poll with .done(). The task exists of many iterations that are themselves blocking, which I wrapped in an async function. I don't have access to the code of the blocking functions as I'm calling third-party software. This is a minimal example of my current approach:

import asyncio
import time

async def blocking_iteration():
    time.sleep(1)

async def long_running():
    for i in range(5):
        print(f"sleeping {i}")
        await blocking_iteration()

async def poll_run():
    future = asyncio.ensure_future(long_running())
    while not future.done():
        print("before polling")
        await asyncio.sleep(0.05)
        print("polling")
    future.result()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(poll_run())
    loop.close()

The result of this is:

before polling
sleeping 0
sleeping 1
sleeping 2
sleeping 3
sleeping 4
polling

From my current understanding of the asyncio mechanism in Python, I had expected the loop to unblock after the first sleep, return control to the loop that would go back to the poll_run await statement and would only run the second iteration of the long_running function after the subsequent poll. So desired output is something like this:

before polling
sleeping 0
polling
before polling
sleeping 1
polling
before polling
sleeping 2
polling
before polling
sleeping 3
polling
before polling
sleeping 4
polling

Can this be achieved with the current approach somehow, or is it possible in a different way?

EDIT

Thanks to @drjackild was able to solve it by changing

async def blocking_iteration():
    time.sleep(1)

into

def blocking():
    time.sleep(1)

async def blocking_iteration():
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, blocking)

Upvotes: 1

Views: 1262

Answers (1)

drjackild
drjackild

Reputation: 472

time is synchronous library and block whole main thread when executing. If you have such blocking calls in your program you can avoid blocking with thread or process pool executors (you can read about it here). Or, change your blocking_iteration to use asyncio.sleep instead of time.sleep

UPD. Just to make it clear, here is non-blocking version, which use loop.run_in_executor with default executor. Please, pay attention, that blocking_iteration now without async

import asyncio
import concurrent.futures
import time

def blocking_iteration():
    time.sleep(1)

async def long_running():
    loop = asyncio.get_event_loop()
    for i in range(5):
        print(f"sleeping {i}")
        await loop.run_in_executor(None, blocking_iteration)

async def poll_run():
    task = asyncio.create_task(long_running())
    while not task.done():
        print("before polling")
        await asyncio.sleep(0.05)
        print("polling")
    print(task.result())

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(poll_run())
    loop.close()

Upvotes: 4

Related Questions