Skip to content

cortex.utils.loop

loop

Event loop utilities for Cortex.

Provides uvloop integration for improved async performance on Unix systems.

Functions

run 

run(
    coro: Coroutine[Any, Any, Any], *, debug: bool = False
) -> Any

Run a coroutine, preferring uvloop when available.

Drop-in replacement for :func:asyncio.run. On Unix with uvloop installed, this yields noticeably lower tail latency on high-rate small-message workloads.

Parameters:

Name Type Description Default
coro Coroutine[Any, Any, Any]

The top-level coroutine to run to completion.

 required
debug bool

Pass through to the event loop's debug flag.

 False

Returns:

Type Description
Any

Whatever coro returns.
Source code in src/cortex/utils/loop.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def run(coro: Coroutine[Any, Any, Any], *, debug: bool = False) -> Any:
    """Run a coroutine, preferring ``uvloop`` when available.

    Drop-in replacement for :func:`asyncio.run`. On Unix with ``uvloop``
    installed, this yields noticeably lower tail latency on high-rate
    small-message workloads.

    Args:
        coro: The top-level coroutine to run to completion.
        debug: Pass through to the event loop's ``debug`` flag.

    Returns:
        Whatever ``coro`` returns.
    """
    if _uvloop_available:
        import uvloop

        logger.info("Using uvloop event loop")
        return uvloop.run(coro, debug=debug)

    logger.info("Using asyncio event loop")
    return asyncio.run(coro, debug=debug)