How to optimize a service so that a synchronous CPU task (model prediction) does not block the FastAPI event loop?

In FastAPI, which is based on asyncio and event loop, CPU-intensive synchronous tasks such as calling a predict model can block the event loop and reduce service performance. To avoid this, you can:

• Offload the CPU task to a separate thread or process. For this, use:

  • run_in_threadpool from fastapi.concurrency to run in a thread pool (suitable if the task is not too heavy and GIL does not interfere much).
  • concurrent.futures.ProcessPoolExecutor to run in a separate process, which is effective for CPU-intensive tasks as it bypasses GIL.

• Use a task queue (e.g., Celery, RQ) and perform predictions asynchronously in workers, returning the result later.

Example with ProcessPoolExecutor:

from fastapi import FastAPI
import asyncio
from concurrent.futures import ProcessPoolExecutor

app = FastAPI()
executor = ProcessPoolExecutor()

# Synchronous CPU task
def predict_model(data):
    # heavy computations
    return "prediction"

@app.get("/predict")
async def predict(data: str):
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(executor, predict_model, data)
    return {"result": result}

This approach does not block the event loop and allows other requests to be processed in parallel.