diff --git a/app/api/routes/image_queries.py b/app/api/routes/image_queries.py index 913f86c37..8e8e0f57a 100644 --- a/app/api/routes/image_queries.py +++ b/app/api/routes/image_queries.py @@ -5,6 +5,7 @@ from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, status from groundlight import Groundlight from model import ImageQuery +from starlette.concurrency import run_in_threadpool from app.core.app_state import ( AppState, @@ -114,7 +115,9 @@ async def post_image_query( # noqa: PLR0913, PLR0915, PLR0912 # Ensure that detector_id has correct casing by pulling the detector ID out of detector_metadata # get_detector_metadata returns the correctly-cased, canonical detector ID - detector_metadata = get_detector_metadata(detector_id=detector_id, gl=gl) # NOTE: API call (once, then cached) + detector_metadata = await run_in_threadpool( + get_detector_metadata, detector_id=detector_id, gl=gl + ) # NOTE: API call (once, then cached) # Refresh against the caller-supplied key, since that is the key this cache entry is stored under. background_tasks.add_task(refresh_detector_metadata_if_needed, detector_id, gl) detector_id = detector_metadata.id @@ -173,11 +176,15 @@ async def post_image_query( # noqa: PLR0913, PLR0915, PLR0912 # If human review is required, we should skip edge inference completely logger.debug("Received human_review=ALWAYS. Skipping edge inference.") record_activity_for_metrics(detector_id, activity_type="escalations") - elif app_state.edge_inference_manager.inference_is_available(detector_id=detector_id): + elif await run_in_threadpool(app_state.edge_inference_manager.inference_is_available, detector_id=detector_id): # -- Edge-model Inference -- logger.debug(f"Local inference is available for {detector_id=}. Running inference...") - results = app_state.edge_inference_manager.run_inference( - detector_id=detector_id, image_bytes=image_bytes, content_type=content_type, mode=detector_metadata.mode + results = await run_in_threadpool( + app_state.edge_inference_manager.run_inference, + detector_id=detector_id, + image_bytes=image_bytes, + content_type=content_type, + mode=detector_metadata.mode, ) ml_confidence = results["confidence"] class_index = results["label"] diff --git a/app/core/app_state.py b/app/core/app_state.py index ff4927af0..283731b03 100644 --- a/app/core/app_state.py +++ b/app/core/app_state.py @@ -112,7 +112,7 @@ def __init__(self): @trace_span -def get_app_state(request: Request) -> AppState: +async def get_app_state(request: Request) -> AppState: """FastAPI dependency that returns the singleton AppState attached to the running app.""" if not hasattr(request.app.state, "app_state"): raise RuntimeError("App state is not initialized.") diff --git a/app/core/edge_inference.py b/app/core/edge_inference.py index c3bab3900..84c4199b7 100644 --- a/app/core/edge_inference.py +++ b/app/core/edge_inference.py @@ -38,7 +38,7 @@ def is_edge_inference_ready(inference_client_url: str) -> bool: model_ready_url = f"http://{inference_client_url}/health/ready" try: - response = requests.get(model_ready_url) + response = requests.get(model_ready_url, timeout=2) return response.status_code == status.HTTP_200_OK except requests.exceptions.RequestException as e: logger.warning(f"Failed to connect to {model_ready_url}: {e}") diff --git a/deploy/helm/groundlight-edge-endpoint/files/inference-deployment-template.yaml b/deploy/helm/groundlight-edge-endpoint/files/inference-deployment-template.yaml index a70bd4d51..98bad8297 100644 --- a/deploy/helm/groundlight-edge-endpoint/files/inference-deployment-template.yaml +++ b/deploy/helm/groundlight-edge-endpoint/files/inference-deployment-template.yaml @@ -75,6 +75,17 @@ spec: value: "1" - name: EDGE_COMPILE_ENABLED value: "{{ .Values.edgeCompileEnabled }}" + # Cap PyTorch/OpenMP/MKL/OpenBLAS intra-op threads so that many inference pods + # don't oversubscribe the node's cores. See values.yaml inferenceDeployment.cpuThreads. + - name: OMP_NUM_THREADS + value: "{{ .Values.inferenceDeployment.cpuThreads }}" + - name: MKL_NUM_THREADS + value: "{{ .Values.inferenceDeployment.cpuThreads }}" + - name: OPENBLAS_NUM_THREADS + value: "{{ .Values.inferenceDeployment.cpuThreads }}" + # No CPU requests or limits: inference pods run BestEffort. Threads are bounded by + # cpuThreads and most detectors are idle at any instant, so the scheduler launches + # every detector and the OS time-slices the occasional simultaneous burst. volumeMounts: - name: edge-endpoint-persistent-volume mountPath: *modelRepository diff --git a/deploy/helm/groundlight-edge-endpoint/values.yaml b/deploy/helm/groundlight-edge-endpoint/values.yaml index cd51cffaf..e0f3b71d9 100644 --- a/deploy/helm/groundlight-edge-endpoint/values.yaml +++ b/deploy/helm/groundlight-edge-endpoint/values.yaml @@ -98,6 +98,15 @@ modelUpdater: # inference pod (inferencemodel-*) behavior inferenceDeployment: + # CPU thread cap per inference pod for PyTorch/OpenMP/MKL/OpenBLAS intra-op parallelism. + # Without this each pod defaults to ALL node cores, so N pods spawn N x cores threads + # that thrash under load (observed: per-request CPU cost grew 5x and throughput collapsed + # past ~3 detectors). Total compute threads ~= cpuThreads x (2 x num_detectors); keep + # small (2-4) on dense multi-detector edges. Tune empirically per device. + cpuThreads: 4 + # Note: inference pods set no CPU requests or limits (BestEffort). Threads are bounded by + # cpuThreads above, and most detectors are idle at any instant, so the scheduler launches + # every detector regardless of core count and the OS time-slices simultaneous bursts. startupProbe: # Consecutive 10-second startup-probe failures before kubelet kills the container. # Ceiling = failureThreshold × 10s. Default 540 = 90 min.