fix: single task executor getting all tasks from Redis queue

[wanpdsantos]

What problem does this PR solve?

Currently, as long as there are tasks in Redis, this loop will keep getting the tasks. This will lead to a single task executor with many tasks in the pending state. Then we need to wait for the pending tasks to get them back in the queue.

In first place, if we set the MAX_CONCURRENT_TASKS to X, then only X tasks should be picked from the queue, and others should be left in the queue for other task_executors or be picked after 1 of the spots in the current executor gets free. This PR ensures this behavior.

The additional changes were due to the Ruff linting in pre-commit. But I believe these are expected to keep the coding style.

Type of change

• Bug Fix (non-breaking change which fixes an issue)
• New Feature (non-breaking change which adds functionality)
• Documentation Update
• Refactoring
• Performance Improvement
• Other (please describe):

[yuzhichang]

trio.Semaphore is better here. Be careful to ensure handle_task release semaphore finally.

import trio
import random

task_limiter = trio.Semaphore(3)


async def handle_task(id: int):
    try:
        print(f"{trio.current_time()}: Worker {id} started")
        await trio.sleep(random.randint(1, 5))
        print(f"{trio.current_time()}: Worker {id} done")
    finally:
        task_limiter.release()


async def handle_tasks():
    id = 0
    async with trio.open_nursery() as nursery:
        while True:
            await task_limiter.acquire()
            nursery.start_soon(handle_task, id)
            id += 1


trio.run(handle_tasks)

[yuzhichang]

PR #7700 creates task_manager every 0.1s. This can be fixed by acquiring semaphore before creating task_manager.