Taming blocking functions in asyncio results in error after some time

[saraverbeecke]

Hi,

I'm following @peterhinch's great approach to tame blocking functions in asyncio.

Peter's example code:

import uasyncio as asyncio
from threadsafe import Message
import _thread
from time import sleep

def slow_add(a, b, *, c, d):  # Blocking function.
    sleep(5)
    return a + b + c + d

# Convert a blocking function to a nonblocking one using threading.
async def unblock(func, *args, **kwargs):
    def wrap(func, message, args, kwargs):
        message.set(func(*args, **kwargs))  # Run the blocking function.
    msg = Message()
    _thread.start_new_thread(wrap, (func, msg, args, kwargs))
    return await msg

async def busywork():  # Prove uasyncio is running.
    while True:
        print("#", end="")
        await asyncio.sleep_ms(200)

async def main():
    bw = asyncio.create_task(busywork())
    res = await unblock(slow_add, 1, 2, c = 3, d = 4)
    bw.cancel()
    print(f"\nDone. Result = {res}")

asyncio.run(main())

Now, I wanted to execute the unblock function in the main repeatedly using the asyncio.create_task() without explicitly awaiting it, which at first sight works:

async def main():
    bw = asyncio.create_task(busywork())
    my_blocking_repeating_task = None
    while True:
        if my_blocking_repeating_task == None:
            my_blocking_repeating_task = asyncio.create_task(unblock(slow_add, 1, 2, c = 3, d = 4))
        else:
            if my_blocking_repeating_task.done():
                print(f"\nDone. Result = {my_blocking_repeating_task.data}")
                my_blocking_repeating_task = None
        await asyncio.sleep_ms(0)
                
asyncio.run(main())

After some time though I get the following error:

Task exception wasn't retrieved
future: <Task> coro= <generator object 'unblock' at 3d83ab50>
Traceback (most recent call last):
  File "asyncio/core.py", line 1, in run_until_complete
  File "<stdin>", line 58, in unblock
OSError: can't create thread

When setting the sleep time in the busy_work function to smaller amounts the error occurs quicker... I tested this code first to see how it works. Eventually I want to process some data from sensors and do a calculation on the data in an unblocking way without blocking the main event loop. This approach seems very nice, but I don't understand why I get the error after some time in this modified example code. (FYI I'm running this code on a ESP32S3, MicroPython version v1.20.0)

[peterhinch]

It will take me a little time to figure out what's going on, but I can state emphatically that the unblock function is designed to be awaited. The idea is that one task pauses for the duration of blocking while others continue to run.

I think we need to step back and examine what you're trying to achieve.

In general having one function polling the outcome of another is almost always a sign of sub-optimal asyncio design. The Event mechanism is much more efficient. Please explain what is your aim and I will try to suggest an efficient way to implement it.

[saraverbeecke]

Thank you for your reply @peterhinch.
The application that I'm trying to build involves processing audio from an I2S buffer. This data has to be processed further and involves mathematics that takes a couple of hundred ms which is blocking the rest of the asyncio tasks. I'm looking for a way to have this calculation step to be unblocked.

[peterhinch]

I have replicated your issue and "fixed" it with the commented line:

async def foo():
    global res
    res = await unblock(slow_add, 1, 2, c = 3, d = 4)

async def main():
    global res
    res = None
    bw = asyncio.create_task(busywork())
    stopped = True
    while True:
        if stopped:
            asyncio.create_task(foo())
            stopped = False
        else:
            if res is not None:
                print(f"\nDone. Result = {res}")
                res = None
                stopped = True
                await asyncio.sleep_ms(10)  # Makes the fault go away
        await asyncio.sleep(0)

asyncio.run(main())

This looks like a timing issue with the (largely undocumented) _thread module.

I must stress that this kind of polling is not good practice, but without a clearer understanding of what you aim to achieve by using create_task rather than await I can't advise further.