Spaces:
Sleeping
Sleeping
File size: 3,723 Bytes
6854aa7 | 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 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """
backend/api/priority.py — Priority job semaphores (S-DUAL-1)
Due classi di job con concorrenza controllata via asyncio.Semaphore:
REALTIME — agent steps interattivi, exec code da UI, terminal commands
Semaphore(6): latency-sensitive, risposta attesa < 30s
BACKGROUND — benchmark, research multi-URL, pip-install headless
Semaphore(2): best-effort, può attendere, non blocca mai REALTIME
I contatori live sono esposti da /api/health/load per il routing adattivo CF.
"""
import asyncio, time, logging
from contextlib import asynccontextmanager
from typing import AsyncGenerator
_logger = logging.getLogger("api.priority")
_boot_time = time.monotonic()
# ── Semaphores ─────────────────────────────────────────────────────────────────
_REALTIME_LIMIT = 6
_BACKGROUND_LIMIT = 2
_realtime_sem = asyncio.Semaphore(_REALTIME_LIMIT)
_background_sem = asyncio.Semaphore(_BACKGROUND_LIMIT)
# Contatori atomici per metriche /api/health/load
_realtime_active = 0
_background_active = 0
@asynccontextmanager
async def realtime_job(timeout_s: float = 300.0) -> AsyncGenerator[None, None]:
"""
Context manager per job REALTIME (agent steps, exec interattivo, terminal).
Acquisisce il semaphore con timeout — rilancia asyncio.TimeoutError
se non ci sono slot liberi entro timeout_s (default 300s = non dovrebbe mai
scadere per richieste UI normali, ma protegge da leak di semaphore).
Uso:
async with realtime_job():
result = await run_subprocess(...)
"""
global _realtime_active
try:
await asyncio.wait_for(_realtime_sem.acquire(), timeout=timeout_s)
except asyncio.TimeoutError:
_logger.warning("[priority] REALTIME semaphore timeout dopo %.0fs", timeout_s)
raise
_realtime_active += 1
try:
yield
finally:
_realtime_active = max(0, _realtime_active - 1)
_realtime_sem.release()
@asynccontextmanager
async def background_job(timeout_s: float = 30.0) -> AsyncGenerator[None, None]:
"""
Context manager per job BACKGROUND (benchmark, research, pip-install).
Timeout più aggressivo (default 30s): se entrambi gli slot BACKGROUND sono
occupati e non si liberano in 30s → 429 Too Many Requests al chiamante.
Garantisce che il benchmark non blocchi mai i job REALTIME.
Uso:
async with background_job(timeout_s=30.0):
result = await run_benchmark(...)
"""
global _background_active
try:
await asyncio.wait_for(_background_sem.acquire(), timeout=timeout_s)
except asyncio.TimeoutError:
_logger.warning("[priority] BACKGROUND semaphore timeout dopo %.0fs — job rifiutato", timeout_s)
raise
_background_active += 1
try:
yield
finally:
_background_active = max(0, _background_active - 1)
_background_sem.release()
def get_load_metrics() -> dict:
"""
Metriche live per /api/health/load.
realtime_waiting: slot REALTIME occupati (Semaphore usa valore interno).
Il valore _sem._value è il numero di slot LIBERI.
"""
return {
"realtime_active": _realtime_active,
"realtime_capacity": _REALTIME_LIMIT,
"realtime_available": _realtime_sem._value, # slot liberi
"background_active": _background_active,
"background_capacity": _BACKGROUND_LIMIT,
"background_available": _background_sem._value, # slot liberi
"uptime_s": int(time.monotonic() - _boot_time),
}
|