[FIXED] How to increase asyncio thread limits in an existing co-routine

Issue

I am currently working on a program using concurrency with asyncio. The code here is over simplified in order to show the problem. It is runnable without any dependencies if you need so.

You have two tasks levels:

• 1: One in operation_loop which creates asyncio workers.
• 2: One in worker which calls synchronous functions with blockings points, and requests to db (simulated here with time.sleep). keep in mind that custom_sleep function is in reality a complex job with blocking parts: that's why it is in a thread. it is alimented by a queue.
  • The queue is elimented each second.

But it seems that asyncio.to_thread can handle 20 threads at max, so, everything is fine with one worker and 20 threads, but the gather operation time is doubled when using 21 threads. And so on with 61... Is we add a worker, we are still limited to 20 threads.

import asyncio
import logging
import os
import threading
import time

def perf_counter_ms() -> int:
    return time.perf_counter_ns() // 1000000

def custom_sleep(worker_id, machine_name):
    logging.debug("Start sleep for worker_id : {worker_id} for machine : {machine_name}")
    time.sleep(1)
    thread_name = threading.current_thread().name
    logging.debug(f"Thread {thread_name} is running on worker_id : {worker_id} for machine : {machine_name}")
    # return worker_id, 1, 1

async def worker(
    worker_id,
    queue: asyncio.Queue,
):
    logging.info("Starting worker")

    while True:
        logging.debug(f"There are {queue.qsize()} items in the queue")
        machines, now, nominal_intensities = await queue.get()

        tasks = []

        for machine in machines:
            # FIXME: Running only a limited number of tasks in parallel.
            tasks.append(
                asyncio.to_thread(
                    custom_sleep,
                    worker_id,
                    machine
                )
            )

        bench_ms = perf_counter_ms()
        logging.debug(f"PERF: Getting all operations for worker_id : {worker_id}")
        try:
            results = await asyncio.wait_for(asyncio.gather(*tasks), 5)
            logging.debug(f"PERF: Getting all operations succeed in time: {perf_counter_ms() - bench_ms} with worker_id : {worker_id}")
        except asyncio.TimeoutError as e:
            logging.error(f"TimeoutError on operation for worker_id : {worker_id}")

        queue.task_done()


async def operation_loop(
):
    # Create a queue that we will use to store our "workload".
    queue = asyncio.Queue()

    for id in range(1):
        asyncio.create_task(
            worker(
                id,
                queue,
            )
        )
    while True:
        await asyncio.sleep(1)

        try:
            queue.put_nowait(([str(i) for i in range(0,40)], 1, 2))
        except asyncio.QueueFull:
            logging.warning(
                "Queue full, performance issue... operations might be lost"
            )


def main():
    # Execute when the module is not initialized from an import statement.

    logging.basicConfig(
        level=os.getenv("LOG_LEVEL", "INFO"),
        format="%(levelname)s [%(asctime)s] [%(name)s.%(funcName)s:%(lineno)d] %(message)s",
    )

    asyncio.run(
        operation_loop(
        )
    )


if __name__ == "__main__":
    main()

DEBUG [2023-03-28 21:38:10,205] [root.custom_sleep:14] Thread asyncio_0 is running on worker_id : 0 for machine : 0
DEBUG [2023-03-28 21:38:10,206] [root.custom_sleep:14] Thread asyncio_1 is running on worker_id : 0 for machine : 1
DEBUG [2023-03-28 21:38:10,206] [root.custom_sleep:14] Thread asyncio_2 is running on worker_id : 0 for machine : 2
DEBUG [2023-03-28 21:38:10,207] [root.custom_sleep:14] Thread asyncio_3 is running on worker_id : 0 for machine : 3
DEBUG [2023-03-28 21:38:10,207] [root.custom_sleep:14] Thread asyncio_4 is running on worker_id : 0 for machine : 4
DEBUG [2023-03-28 21:38:10,208] [root.custom_sleep:14] Thread asyncio_5 is running on worker_id : 0 for machine : 5
DEBUG [2023-03-28 21:38:10,208] [root.custom_sleep:14] Thread asyncio_6 is running on worker_id : 0 for machine : 6
DEBUG [2023-03-28 21:38:10,209] [root.custom_sleep:14] Thread asyncio_7 is running on worker_id : 0 for machine : 7
DEBUG [2023-03-28 21:38:10,209] [root.custom_sleep:14] Thread asyncio_8 is running on worker_id : 0 for machine : 8
DEBUG [2023-03-28 21:38:10,210] [root.custom_sleep:14] Thread asyncio_9 is running on worker_id : 0 for machine : 9
DEBUG [2023-03-28 21:38:10,211] [root.custom_sleep:14] Thread asyncio_13 is running on worker_id : 0 for machine : 13
DEBUG [2023-03-28 21:38:10,211] [root.custom_sleep:14] Thread asyncio_10 is running on worker_id : 0 for machine : 10
DEBUG [2023-03-28 21:38:10,211] [root.custom_sleep:14] Thread asyncio_11 is running on worker_id : 0 for machine : 11
DEBUG [2023-03-28 21:38:10,211] [root.custom_sleep:14] Thread asyncio_14 is running on worker_id : 0 for machine : 14
DEBUG [2023-03-28 21:38:10,211] [root.custom_sleep:14] Thread asyncio_12 is running on worker_id : 0 for machine : 12
DEBUG [2023-03-28 21:38:10,212] [root.custom_sleep:14] Thread asyncio_15 is running on worker_id : 0 for machine : 15
DEBUG [2023-03-28 21:38:10,213] [root.custom_sleep:14] Thread asyncio_17 is running on worker_id : 0 for machine : 17
DEBUG [2023-03-28 21:38:10,213] [root.custom_sleep:14] Thread asyncio_16 is running on worker_id : 0 for machine : 16
DEBUG [2023-03-28 21:38:10,214] [root.custom_sleep:14] Thread asyncio_18 is running on worker_id : 0 for machine : 18
DEBUG [2023-03-28 21:38:10,214] [root.custom_sleep:14] Thread asyncio_19 is running on worker_id : 0 for machine : 19
DEBUG [2023-03-28 21:38:11,207] [root.custom_sleep:14] Thread asyncio_0 is running on worker_id : 0 for machine : 20
DEBUG [2023-03-28 21:38:11,208] [root.custom_sleep:14] Thread asyncio_2 is running on worker_id : 0 for machine : 21
DEBUG [2023-03-28 21:38:11,208] [root.custom_sleep:14] Thread asyncio_1 is running on worker_id : 0 for machine : 22
DEBUG [2023-03-28 21:38:11,209] [root.custom_sleep:14] Thread asyncio_4 is running on worker_id : 0 for machine : 24
DEBUG [2023-03-28 21:38:11,209] [root.custom_sleep:14] Thread asyncio_5 is running on worker_id : 0 for machine : 25
DEBUG [2023-03-28 21:38:11,210] [root.custom_sleep:14] Thread asyncio_7 is running on worker_id : 0 for machine : 26
DEBUG [2023-03-28 21:38:11,211] [root.custom_sleep:14] Thread asyncio_3 is running on worker_id : 0 for machine : 23
DEBUG [2023-03-28 21:38:11,211] [root.custom_sleep:14] Thread asyncio_6 is running on worker_id : 0 for machine : 27
DEBUG [2023-03-28 21:38:11,212] [root.custom_sleep:14] Thread asyncio_8 is running on worker_id : 0 for machine : 28
DEBUG [2023-03-28 21:38:11,213] [root.custom_sleep:14] Thread asyncio_9 is running on worker_id : 0 for machine : 29
DEBUG [2023-03-28 21:38:11,213] [root.custom_sleep:14] Thread asyncio_13 is running on worker_id : 0 for machine : 30
DEBUG [2023-03-28 21:38:11,213] [root.custom_sleep:14] Thread asyncio_10 is running on worker_id : 0 for machine : 31
DEBUG [2023-03-28 21:38:11,214] [root.custom_sleep:14] Thread asyncio_11 is running on worker_id : 0 for machine : 32
DEBUG [2023-03-28 21:38:11,214] [root.custom_sleep:14] Thread asyncio_12 is running on worker_id : 0 for machine : 33
DEBUG [2023-03-28 21:38:11,215] [root.custom_sleep:14] Thread asyncio_14 is running on worker_id : 0 for machine : 34
DEBUG [2023-03-28 21:38:11,215] [root.custom_sleep:14] Thread asyncio_17 is running on worker_id : 0 for machine : 37
DEBUG [2023-03-28 21:38:11,216] [root.custom_sleep:14] Thread asyncio_18 is running on worker_id : 0 for machine : 39
DEBUG [2023-03-28 21:38:11,216] [root.custom_sleep:14] Thread asyncio_19 is running on worker_id : 0 for machine : 38
DEBUG [2023-03-28 21:38:11,216] [root.custom_sleep:14] Thread asyncio_15 is running on worker_id : 0 for machine : 35
DEBUG [2023-03-28 21:38:11,216] [root.custom_sleep:14] Thread asyncio_16 is running on worker_id : 0 for machine : 36

I have seen that I can use a concurrent.futures.ThreadPoolExecuter context manager with a max_worker value, but I want to stick with asyncio as much as possible. (except if there is no other options).

Another option could be to fully convert in asyncio the code but it should take some time to do.

Another option could be to use multiprocess but it seems to be an overkill for now.

Do you have an idea of why the threads are limited? And how to increase this limit?

Thank you in advance

Solution

The asyncio function to_thread is actually a wrapper around ThreadPoolExecutor. Here is its code after removing some details that don't concern this question (from asyncio.thread.py, Python3.11):

loop = events.get_running_loop()
return await loop.run_in_executor(None, func_call)

Here func_call is basically the function that you passed to to_thread. When None is the first argument, the default ThreadPoolExecutor is used. So even though you said you didn't want to use ThreadPoolExecutor, you are doing so in fact. And you are using the default one.

Now here is a snippet from the documentation for ThreadPoolExecutor:

"If max_workers is None or not given, it will default to the number of processors on the machine, multiplied by 5."

So you see that this will limit the number of threads, in your case to 20. You can define a ThreadPoolExecutor with a higher thread limit and set that using the function asyncio.loop.set_default_executor. https://docs.python.org/3/library/asyncio-eventloop.html?highlight=threadpoolexecutor#asyncio.loop.set_default_executor

Try adding these 2 lines at the beginning of operation_loop:

async def operation_loop():
    loop = asyncio.get_running_loop()
    loop.set_default_executor(ThreadPoolExecutor(max_workers=40))

Subsequent calls to to_thread will now use an executor with a thread limit of 40.

I hope you realize that Python threads do not actually run in parallel. Python switches between threads rapidly to give the appearance of parallelism. Your test code, using the sleep function, does not illustrate this situation. In your real program this might be very important.

If you need true parallel processing you must create separate Python Processes.

Answered By - Paul Cornelius