From 2894979a356e43bfa8bf79ba44f39ac29b3fa587 Mon Sep 17 00:00:00 2001 From: kiyo9w Date: Mon, 28 Sep 2026 12:57:21 +0900 Subject: [PATCH] fix: keep catalog IO off the live event loop --- recognization/code/app.py | 150 +++++++++++++++++++++++--------------- 1 file changed, 93 insertions(+), 57 deletions(-) diff --git a/recognization/code/app.py b/recognization/code/app.py index d3f5882..5c16dbd 100644 --- a/recognization/code/app.py +++ b/recognization/code/app.py @@ -309,42 +309,53 @@ def create_app( # ---------------------------------------------------------------- health @app.get("/v1/health") async def health() -> Dict[str, Any]: - return { - "status": "ok", - "model": { - "version": engine.model_version(), - "device": getattr(engine.settings, "device", None) or "auto", - "imgsz": getattr(engine.settings, "imgsz", 640), - "warm": True, - }, - "catalog": {"version": engine.catalog_version(), "sku_count": engine.sku_count()}, - "identity": { - "backend": getattr(engine.settings, "identity_backend", "matcher"), - "gallery_views": ( - engine.identity_gallery().view_count() - if getattr(engine.settings, "identity_backend", "") == "gallery" and engine.identity_gallery() - else None - ), - }, - "ocr": { - # Without this an operator could not tell a broken OCR from a shelf - # with no readable tags. - "engine": getattr(engine.ocr_engine, "name", "none"), - "available": bool(getattr(engine.ocr_engine, "available", lambda: False)()), - }, - "recording": recorder.stats() if recorder is not None else None, - "sessions_active": len(sessions), - "uptime_s": round(engine.uptime_s(), 1), - } + loop = asyncio.get_running_loop() + + def _health() -> Dict[str, Any]: + gallery = ( + engine.identity_gallery() + if getattr(engine.settings, "identity_backend", "") == "gallery" + else None + ) + return { + "status": "ok", + "model": { + "version": engine.model_version(), + "device": getattr(engine.settings, "device", None) or "auto", + "imgsz": getattr(engine.settings, "imgsz", 640), + "warm": True, + }, + "catalog": {"version": engine.catalog_version(), "sku_count": engine.sku_count()}, + "identity": { + "backend": getattr(engine.settings, "identity_backend", "matcher"), + "gallery_views": gallery.view_count() if gallery is not None else None, + }, + "ocr": { + # Without this an operator could not tell a broken OCR from a shelf + # with no readable tags. + "engine": getattr(engine.ocr_engine, "name", "none"), + "available": bool(getattr(engine.ocr_engine, "available", lambda: False)()), + }, + "sessions_active": sum(session is not None for session in sessions.values()), + "uptime_s": round(engine.uptime_s(), 1), + } + + return await loop.run_in_executor(None, _health) # ---------------------------------------------------------------- catalog @app.get("/v1/catalog", dependencies=[Depends(authorize)]) async def catalog() -> Dict[str, Any]: - return {"catalog_version": engine.catalog_version(), "products": engine.catalog()} + loop = asyncio.get_running_loop() + + def _catalog() -> Dict[str, Any]: + return {"catalog_version": engine.catalog_version(), "products": engine.catalog()} + + return await loop.run_in_executor(None, _catalog) @app.get("/v1/catalog/{sku_id}/image", dependencies=[Depends(authorize)]) async def catalog_image(sku_id: str) -> Response: - path = engine.catalog_image_path(sku_id) + loop = asyncio.get_running_loop() + path = await loop.run_in_executor(None, engine.catalog_image_path, sku_id) if not path: raise _error("NOT_FOUND", f"no catalog image for '{sku_id}'", 404) return FileResponse(path) @@ -353,7 +364,8 @@ def create_app( async def update_stock(sku_id: str, body: StockUpdateBody) -> Dict[str, Any]: if body.delta == 0: raise _error("VALIDATION_ERROR", "delta must not be zero", 422) - product = engine.update_stock(sku_id, body.delta) + loop = asyncio.get_running_loop() + product = await loop.run_in_executor(None, engine.update_stock, sku_id, body.delta) if product is None: raise _error("NOT_FOUND", f"unknown sku '{sku_id}'", 404) return {"status": "success", "product": product} @@ -378,7 +390,7 @@ def create_app( @app.get("/v1/samples/{sample_id}", dependencies=[Depends(authorize)]) async def get_sample(sample_id: str) -> Response: item = samples.get(sample_id) - if not item or not os.path.exists(item["path"]): + if not item or not await asyncio.get_running_loop().run_in_executor(None, os.path.exists, item["path"]): raise _error("NOT_FOUND", f"no sample '{sample_id}'", 404) return FileResponse(item["path"]) @@ -640,11 +652,16 @@ def create_app( if settings.token and header and not bearer_matches(header, settings.token): await websocket.close(code=1008) return + loop = asyncio.get_running_loop() async with sessions_lock: if len(sessions) >= settings.max_sessions: await websocket.close(code=1013) return session_id = str(uuid.uuid4()) + sessions[session_id] = None # reserve before any disk work + session = None + try: + model_version = await loop.run_in_executor(None, engine.model_version) session = LiveSession( session_id, engine, @@ -656,33 +673,46 @@ def create_app( tags=settings.live_tags, ), ) - session.refresh_catalog() + await loop.run_in_executor(None, session.refresh_catalog) + catalog_version = session._catalog_version if settings.live_tags: - session.enable_tags() - sessions[session_id] = session - await websocket.accept() - send_lock = asyncio.Lock() - loop = asyncio.get_running_loop() + await loop.run_in_executor(None, session.enable_tags) + async with sessions_lock: + sessions[session_id] = session + except BaseException: + if session is not None: + session.close() + async with sessions_lock: + sessions.pop(session_id, None) + raise + try: + await websocket.accept() + send_lock = asyncio.Lock() - async def send(payload: Dict[str, Any]) -> None: - async with send_lock: - await websocket.send_text(json.dumps(payload)) + async def send(payload: Dict[str, Any]) -> None: + async with send_lock: + await websocket.send_text(json.dumps(payload)) - await send( - { - "type": "hello", - "protocol": 1, - "session_id": session_id, - "catalog_version": engine.catalog_version(), - "model": { - "version": engine.model_version(), - "device": getattr(engine.settings, "device", None) or "auto", - "imgsz": getattr(engine.settings, "imgsz", 640), - }, - "config": session.config(), - "limits": {"max_frame_bytes": MAX_FRAME_BYTES}, - } - ) + await send( + { + "type": "hello", + "protocol": 1, + "session_id": session_id, + "catalog_version": catalog_version, + "model": { + "version": model_version, + "device": getattr(engine.settings, "device", None) or "auto", + "imgsz": getattr(engine.settings, "imgsz", 640), + }, + "config": session.config(), + "limits": {"max_frame_bytes": MAX_FRAME_BYTES}, + } + ) + except BaseException: + session.close() + async with sessions_lock: + sessions.pop(session_id, None) + raise async def processing_loop() -> None: while not session.closed: @@ -851,7 +881,14 @@ def create_app( # Base64 expands ~4/3. Reject the text before decoding it. if len(raw) > (MAX_FRAME_BYTES * 4) // 3 + 8: raise _error("PAYLOAD_TOO_LARGE", f"frame larger than {MAX_FRAME_BYTES} bytes", 413) - frame_bytes = base64.b64decode(raw) if raw else b"" + loop = asyncio.get_running_loop() + try: + frame_bytes = ( + await loop.run_in_executor(None, lambda: base64.b64decode(raw, validate=True)) + if raw else b"" + ) + except (ValueError, base64.binascii.Error): + raise _error("BAD_REQUEST", "frame_bytes_b64 is not valid base64", 400) if len(frame_bytes) > MAX_FRAME_BYTES: raise _error("PAYLOAD_TOO_LARGE", f"frame larger than {MAX_FRAME_BYTES} bytes", 413) request = FrameRequest( @@ -868,7 +905,6 @@ def create_app( ) if not request.frame_id: raise _error("BAD_REQUEST", "frame_id required", 400) - loop = asyncio.get_running_loop() observation = await loop.run_in_executor(None, scans.observe_frame, request) return {"status": "success", "observation": observation.as_dict(), "session": session.snapshot()} except ObserveError as exc: