Module livekit.agents.utils.aio.debug

Functions

def hook_slow_callbacks(slow_duration: float) ‑> None

Expand source code

def hook_slow_callbacks(slow_duration: float) -> None:
    _run = asyncio.events.Handle._run

    def instrumented(self: Any):
        start = time.monotonic()
        val = _run(self)
        dt = time.monotonic() - start
        if dt >= slow_duration:
            logger.warning(
                "Running %s took too long: %.2f seconds",
                _format_handle(self),  # type: ignore
                dt,
            )
        return val

    asyncio.events.Handle._run = instrumented  # type: ignore