Feature request: Warn on blocked event loop in dev mode

[Otto-AA]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched in the documentation/README.
• I already searched in Google "How to do X" and didn't find any information.
• I already read and followed all the tutorial in the docs/README and didn't find an answer.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from contextlib import asynccontextmanager
from fastapi import FastAPI
import time
import asyncio
import asyncio.exceptions
import logging


@asynccontextmanager
async def lifespan(app: FastAPI):
    # PoC event loop blocking checker
    # runs in the background until the application shuts down
    task = asyncio.create_task(check_blocked(interval_ms=100, max_delay_ms=1000))
    asyncio.gather(task)

    yield

    task.cancel()

app = FastAPI(lifespan=lifespan)


@app.get("/sleep_sync")
async def sleep_sync(seconds: int):
    # blocks the event loop for {seconds}s
    time.sleep(seconds)

@app.get("/hello")
def hello():
    # cannot be called while /sleep_sync is processed
    return "hello"

async def check_blocked(interval_ms: int, max_delay_ms: int):
    """Verify that tasks on the event loop take at most {max_delay_ms}ms until they are executed"""
    last_run_time_ms = time.time() * 1000

    while True:
        try:
            # wait a bit, so we do not consume too many resources
            # (we may want to use loop.call_at instead of sleeping)
            await asyncio.sleep(interval_ms / 1000)
        except asyncio.exceptions.CancelledError:
            break
        elapsed_ms = time.time() * 1000 - last_run_time_ms
        # after the sleep should have finished, how long did it take until we execute?
        elapsed_between_awake_and_schedule_ms = int(elapsed_ms - interval_ms)

        if elapsed_between_awake_and_schedule_ms > max_delay_ms:
            logging.warning(f'Event loop has been blocked for {elapsed_between_awake_and_schedule_ms/1000:.3}s.')
        
        last_run_time_ms = time.time() * 1000

Description

(Originally posted in the fastapi repo: fastapi/fastapi#14603)

Feature request

I suggest to automatically warn when the event loop is stuck for 1+ seconds. I think this should be enabled by default when running fastapi dev for a better UX.

Why

The async docs explain that sync IO/... should not be used in async path operations. However, I think it is:

• easy to get wrong for beginners
• hard to troubleshoot, because there is no exception/warning/... pointing to the cause of the performance issue

I've seen this happen in production applications and also found several issues related to this when searching the discussions here (e.g. fastapi/fastapi#8842 with a good overview of the problem).

How

Measure time between task schedule and task execution

For detection, I think we could regularly schedule a task and measure how long it takes until it gets executed. My assumption is that if a task takes a long time until it is executed, then FastAPI would also not be able to process requests for roughly the same time.

Example with blocking sync code within async operator:

1. schedule task detect_blocking to run at 12:00:00
2. a blocking endpoint is called at 11:59:59 running for 5 seconds
3. at 12:00:04 detect_blocking is called
4. calculate execution time minus scheduled time: 12:00:04 - 12:00:00 = 4s
5. issue a warning if this time is > 1s: 4s > 1s -> warning
6. schedule task detect_blocking again and continue...

Non-blocking example:

1. schedule task detect_blocking to run at 12:00:00
2. a (good) non-blocking endpoint is called at 11:59:59 running for 5 seconds
3. at 12:00:00.123 the non-blocking endpoint awaits something (and thus the next task in the event loop is executed)
4. at 12:00:00.123 detect_blocking is called
5. calculate execution time minus scheduled time: 12:00:00.123 - 12:00:00 = 0.123s
6. issue a warning if this time is > 1s: 0.123s < 1s -> no warning
7. schedule task detect_blocking again and continue...

You can take a look at the example code. After starting go to http://127.0.0.1:8000/sleep_sync?seconds=2 and observe the WARNING:root:Event loop has been blocked for 1.9s. log. I did not yet look at how to do this internally within FastAPI and to restrict it to dev-only.

Run asyncio in debug mode

Alternatively, there is an option to run asyncio in debug mode which would add in-built checks if the event loop is blocked. This would be configurable with loop.slow_callback_duration.

This option is less flexible (e.g. I don't know if we could modify the logged message) and enabling debug mode may come with other side effects, so I would prefer the other option for maintainability.

PR / Feedback

I could work on a PR, but first wanted to ask if this is desired and if there is more I should consider / what I should verify to prevent false positives warnings.

Operating System

Linux

Operating System Details

Ubuntu 24

Python Version

3.13.5

Additional Context

No response

[amzker]

Checkout loopsentry

It does exactly this and you can have env var based monitoring start stop.

from contextlib import asynccontextmanager
from fastapi import FastAPI
from loopsentry import LoopSentry

@asynccontextmanager
async def lifespan(app: FastAPI):
# use ir in lifespan so multiple workers can get their own loopsentry monitor thread
    sentry = LoopSentry(
        threshold=0.05,  # seconds
        base_dir="sentry_logs", 
        async_threshold=1.0, # seconds 
        capture_args=True, 
        detect_async_bottlenecks=True,
    )
    sentry.start() # this you can disable on prod based on env var
    yield

app = FastAPI(lifespan=lifespan)

And after testing and running the system llgs will be generated on sentry_logs dir where you can run

loopsentry analyze -d sentry_logs/ --html

Or even cli view. Right now html needs update but it should do the jov