File size: 2,613 Bytes
c607869 d8ae160 c607869 d8ae160 c607869 d8ae160 c607869 d8ae160 c607869 d8ae160 c607869 d8ae160 c607869 d8ae160 c607869 d8ae160 c607869 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | from __future__ import annotations
import importlib
import logging
import threading
import time
from collections.abc import Iterable
from typing import Any
logger = logging.getLogger(__name__)
_started: set[tuple[str, ...]] = set()
_lock = threading.Lock()
def _warm_imports(
modules: tuple[str, ...],
functions: tuple[str, ...],
calls: tuple[tuple[str, tuple[Any, ...]], ...],
delay_seconds: float,
) -> None:
if delay_seconds > 0:
time.sleep(delay_seconds)
for module in modules:
try:
importlib.import_module(module)
except Exception:
logger.debug("Background preload failed for %s", module, exc_info=True)
for function_path in functions:
try:
module_name, function_name = function_path.split(":", 1)
function = getattr(importlib.import_module(module_name), function_name)
function()
except Exception:
logger.debug(
"Background preload failed for %s", function_path, exc_info=True
)
for function_path, args in calls:
try:
module_name, function_name = function_path.split(":", 1)
function = getattr(importlib.import_module(module_name), function_name)
function(*args)
except Exception:
logger.debug(
"Background preload failed for %s%r",
function_path,
args,
exc_info=True,
)
def preload_once(
name: str,
*,
modules: Iterable[str] = (),
functions: Iterable[str] = (),
calls: Iterable[tuple[str, tuple[Any, ...]]] = (),
delay_seconds: float = 0.25,
) -> None:
"""Warm small predictable costs on a daemon thread after the visible render.
Keep this limited to imports and tiny metadata. Avoid model construction
and full tensor loads because those can steal enough CPU or I/O to make the
visible page feel slower.
"""
module_tuple = tuple(dict.fromkeys(modules))
function_tuple = tuple(dict.fromkeys(functions))
call_tuple = tuple((path, tuple(args)) for path, args in calls)
if not module_tuple and not function_tuple and not call_tuple:
return
key = (name, *module_tuple, *function_tuple, repr(call_tuple))
with _lock:
if key in _started:
return
_started.add(key)
thread = threading.Thread(
target=_warm_imports,
args=(module_tuple, function_tuple, call_tuple, delay_seconds),
name=f"persona-ui-preload-{name}",
daemon=True,
)
thread.start()
|