concurrent async BackgroundTasks design pattern

[LuchiLucs]

If I have a list of Callable Async tasks and I wish to run them in a background and also restart them after a duration period, how can I design this?

Using the framework FastAPI I tried to code this toy example but as far as I understood, the BackgroundTasks list is run not concurrently but in a FIFO order. How can I fix this example or design this alternatively in a simple and elegant manner?

I wish to replace some legacy code base where there were a huge codebase dedicated to define custome Thread/Daemon objects with background/periodic properties but I feel like it was not consistent and it was overcomplex too.

from fastapi import FastAPI, BackgroundTasks
# from fastapi_utils.tasks import repeat_every
from src.utils.logger_utils import get_logger
import asyncio
from enum import Enum

app = FastAPI()
logger = get_logger(__name__)

class BaseTask:
    def __init__(self, name, wait_duration: int):
        self.name = name
        self.running = False
        self.wait_duration = wait_duration

    async def __call__(self):
        while self.running:
            await self.run_task()
            logger.debug(f"{self.name} completed, waiting for {self.wait_duration} seconds...")
            await asyncio.sleep(self.wait_duration)

    async def run_task(self):
        raise NotImplementedError

    def start(self):
        self.running = True

    def stop(self):
        self.running = False

class TaskB(BaseTask):
    async def run_task(self):
        logger.debug(f"{self.name} running...")
        await asyncio.sleep(2)

class TaskC(BaseTask):
    async def run_task(self):
        logger.debug(f"{self.name} running...")
        await asyncio.sleep(5)

period_in_seconds = 10

class TaskList(str, Enum):
    TASK_A = "TASK_A"
    TASK_B = "TASK_B"
    TASK_C = "TASK_C"
    ALL = "ALL"

tasks = [
    TaskB(TaskList.TASK_B.value, period_in_seconds),
    TaskC(TaskList.TASK_C.value, period_in_seconds),
]


# @repeat_every(seconds=period_in_seconds, wait_first=True)
# async def periodic_tasks():
#     for task in tasks:
#         if task.running:
#             asyncio.create_task(task())

@app.post("/start-tasks/{task_names}")
async def start_task(task_names: TaskList, background_tasks: BackgroundTasks):
    task_names = task_names.value
    if task_names == TaskList.ALL.value:
        to_be_started = tasks
    else:
        to_be_started = [t for t in tasks if t.name == task_names]

    for task in to_be_started:
        task.start()
        background_tasks.add_task(task)
    return {"message": f"{[t.name for t in to_be_started]} started"}


@app.post("/stop-tasks/{task_names}")
async def stop_task(task_names: TaskList, background_tasks: BackgroundTasks):
    task_names = task_names.value
    if task_names == TaskList.ALL.value:
        to_be_stopped = tasks
    else:
        to_be_stopped = [t for t in tasks if t.name == task_names]

    for task in to_be_stopped:
        task.stop()
    return {"message": f"{to_be_stopped} started"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8085)

[adriangb]

I don't think this is what Starlette's background tasks are meant for. Try taking control of the event loop and running your own async tasks.

[LuchiLucs]

As far as I understood Starlette/FastAPI servers, when they are started an event loop is automatically created in order to process the async requests without blocking. If I wish to manage my periodic async tasks with start/stop APIs as in the toy example, I should run this tasks in an event loop, but:

• If I use the same loop of Starlette server, I believe it could be blocked by my async task (let's assume they are async but contains blocking code, and they take a long time to complete).
• If I wish to use another event loop, as far as I understood 1 loop = 1 thread, I should create as many threads as many tasks I wish to manage, and under these threads I should start my own event loop.

Is that correct? Anyways, could it be possible to be helped by providing another toy example where these concepts are integrated and/or shown?

Another solution could be to turn by periodic background tasks into sync code by convertin internal async code to sync. Is it possible? If so, how?

[adriangb]

let's assume they are async but contains blocking code, and they take a long time to complete

This seems like a problem in general. You should never run blocking code in an async context. Move it to a worker thread.

[adriangb]

It seems like you have two unrelated problems:

• You have sync code you are running or want to run in an async context: see https://anyio.readthedocs.io/en/stable/threads.html and beware of the default capacity limiter
• You want to run long-running or infinite background tasks alongside your FastAPI app: consider something like https://github.com/adriangb/asgi-lifespan or https://github.com/adriangb/asapi