Why are there up to 40 tasks running on anyio?

Question

import logging
import time
from datetime import datetime

import anyio
import numpy as np
from anyio.streams.memory import MemoryObjectReceiveStream as rstream
from anyio.streams.memory import MemoryObjectSendStream as sstream

logging.getLogger().setLevel(logging.INFO)
logger = logging.getLogger(__name__)


async def ping_num(send: sstream):
    async with send:
        for num in range(100):
            await send.send(num)


async def pong_num(receive_num: rstream, send_other_num: sstream, key: int):
    async with receive_num, send_other_num:
        async for num in receive_num:
            await send_other_num.send((key, num, np.sqrt(num)))
        send_other_num.close()


async def async_sleep_5_and_print(receive_other: rstream):
    async with receive_other:
        async with anyio.create_task_group() as task_group:
            async for other in receive_other:
                task_group.start_soon(
                    anyio.to_thread.run_sync, sync_sleep_5_and_print, other
                )


def sync_sleep_5_and_print(item):
    logging.info(f"start:: {datetime.today()} {item=}")
    time.sleep(5)
    logging.info(f"  end:: {datetime.today()} {item=}")


async def main():
    send, receive = anyio.create_memory_object_stream()
    send_other, receive_other = anyio.create_memory_object_stream()
    async with anyio.create_task_group() as task_group:
        async with send:
            task_group.start_soon(ping_num, send.clone())
        async with receive, send_other:
            for key in range(5):
                task_group.start_soon(
                    pong_num, receive.clone(), send_other.clone(), key
                )
        async with receive_other:
            task_group.start_soon(async_sleep_5_and_print, receive_other.clone())
    logger.info("main end")


if __name__ == "__main__":
    anyio.run(main)

logs:

INFO:root:start:: 2021-11-21 14:45:48.113164 item=(0, 0, 0.0)
INFO:root:start:: 2021-11-21 14:45:48.117558 item=(1, 1, 1.0)
INFO:root:start:: 2021-11-21 14:45:48.122124 item=(2, 2, 1.4142135623730951)

...

INFO:root:start:: 2021-11-21 14:45:48.149377 item=(3, 38, 6.164414002968976)
INFO:root:start:: 2021-11-21 14:45:48.154694 item=(4, 39, 6.244997998398398)
INFO:root:  end:: 2021-11-21 14:45:53.115420 item=(0, 0, 0.0)
INFO:root:start:: 2021-11-21 14:45:53.116359 item=(4, 99, 9.9498743710662)

...

It was expected that 100 tasks would run together and end in about 5 seconds, but it took 15 seconds.

As can be seen from the log, it seems to run together up to 40 tasks at the same time.

I changed the backend to trio, but the same problem arises.

Why is this happening?

Is there a way to fix this in above code?

Answer 1

When I learned more about anyio.to_thread.run_sync, I found out that It was receiving a keyword parameter called limiter.

If it receives the None value, it use the default limiter.

Presumably, this basic limiter is designated as 40.

So I added and modified the following code to implement the action I wanted.

There may be someone who has the same problem as me, so I leave it as an answer.

from functools import partial

async def async_sleep_5_and_print(receive_other: rstream):
    limit = anyio.CapacityLimiter(100)
    run_sync_with_limit = partial(anyio.to_thread.run_sync, limiter=limit)
    async with receive_other:
        async with anyio.create_task_group() as task_group:
            async for other in receive_other:
                task_group.start_soon(
                    run_sync_with_limit, sync_sleep_5_and_print, other
                )

logs:

INFO:root:start:: 2021-11-21 21:40:24.235837 item=(0, 0, 0.0)
INFO:root:start:: 2021-11-21 21:40:24.237306 item=(1, 1, 1.0)
INFO:root:start:: 2021-11-21 21:40:24.237800 item=(2, 2, 1.4142135623730951)
INFO:root:start:: 2021-11-21 21:40:24.238302 item=(3, 3, 1.7320508075688772)
INFO:root:start:: 2021-11-21 21:40:24.238695 item=(4, 4, 2.0)

...

INFO:root:  end:: 2021-11-21 21:40:29.302220 item=(4, 89, 9.433981132056603)
INFO:root:  end:: 2021-11-21 21:40:29.302260 item=(4, 94, 9.695359714832659)
INFO:root:  end:: 2021-11-21 21:40:29.303010 item=(0, 95, 9.746794344808963)
INFO:root:  end:: 2021-11-21 21:40:29.303086 item=(2, 97, 9.848857801796104)
INFO:__main__:main end

i found it anyio._backends_._asyncio.current_default_thread_limiter