INNER CODE UNIT · Python

_preload_models

roboflow/inference · inference_server/inference_server/app.py:109

async def _preload_models(
    state,
    proxy,
    preload: list[tuple[str, str]],
    pinned: list[tuple[str, str]],
) -> None:
    loads = {mid: (api_key, False) for mid, api_key in preload}
    loads.update({mid: (api_key, True) for mid, api_key in pinned})
    semaphore = asyncio.Semaphore(_PRELOAD_CONCURRENCY)

    async def _load_once(mid, api_key, is_pinned, deadline) -> bool:
        async with semaphore:
            timeout_s = _PRELOAD_LOAD_TIMEOUT_S
            if deadline is not None:
                timeout_s = max(0.0, min(timeout_s, deadline - time.monotonic()))

            try:
                result = await proxy.load(

View source record →