优化客户端连接状态检测，减少不必要的请求处理，增强异常处理逻辑

Author: CJackHwang

api_utils/queue_worker.py

@@ -100,16 +100,26 @@ async def queue_worker():
             request_data = request_item["request_data"]
             http_request = request_item["http_request"]
             result_future = request_item["result_future"]
-            
+
             if request_item.get("cancelled", False):
                 logger.info(f"[{req_id}] (Worker) 请求已取消，跳过。")
                 if not result_future.done():
                     result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] 请求已被用户取消"))
                 request_queue.task_done()
                 continue
-            
+
             is_streaming_request = request_data.stream
             logger.info(f"[{req_id}] (Worker) 取出请求。模式: {'流式' if is_streaming_request else '非流式'}")
+
+            # 优化：在开始处理前主动检测客户端连接状态，避免不必要的处理
+            from api_utils.request_processor import _test_client_connection
+            is_connected = await _test_client_connection(req_id, http_request)
+            if not is_connected:
+                logger.info(f"[{req_id}] (Worker) ✅ 主动检测到客户端已断开，跳过处理节省资源")
+                if not result_future.done():
+                    result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] 客户端在处理前已断开连接"))
+                request_queue.task_done()
+                continue
 
             # 流式请求间隔控制
             current_time = time.time()
@@ -118,8 +128,10 @@ async def queue_worker():
                 logger.info(f"[{req_id}] (Worker) 连续流式请求，添加 {delay_time:.2f}s 延迟...")
                 await asyncio.sleep(delay_time)
 
-            if await http_request.is_disconnected():
-                logger.info(f"[{req_id}] (Worker) 客户端在等待锁时断开。取消。")
+            # 等待锁前再次主动检测客户端连接
+            is_connected = await _test_client_connection(req_id, http_request)
+            if not is_connected:
+                logger.info(f"[{req_id}] (Worker) ✅ 等待锁时检测到客户端断开，取消处理")
                 if not result_future.done():
                     result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] 客户端关闭了请求"))
                 request_queue.task_done()
@@ -129,8 +141,10 @@ async def queue_worker():
             async with processing_lock:
                 logger.info(f"[{req_id}] (Worker) 已获取处理锁。开始核心处理...")
 
-                if await http_request.is_disconnected():
-                    logger.info(f"[{req_id}] (Worker) 客户端在获取锁后断开。取消。")
+                # 获取锁后最终主动检测客户端连接
+                is_connected = await _test_client_connection(req_id, http_request)
+                if not is_connected:
+                    logger.info(f"[{req_id}] (Worker) ✅ 获取锁后检测到客户端断开，取消处理")
                     if not result_future.done():
                         result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] 客户端关闭了请求"))
                 elif result_future.done():
@@ -172,15 +186,16 @@ async def enhanced_disconnect_monitor():
                                 nonlocal client_disconnected_early
                                 while not completion_event.is_set():
                                     try:
-                                        # 检查客户端是否断开连接
-                                        if await http_request.is_disconnected():
-                                            logger.info(f"[{req_id}] (Worker) 检测到客户端断开连接，提前触发done信号")
+                                        # 主动检查客户端是否断开连接
+                                        is_connected = await _test_client_connection(req_id, http_request)
+                                        if not is_connected:
+                                            logger.info(f"[{req_id}] (Worker) ✅ 流式处理中检测到客户端断开，提前触发done信号")
                                             client_disconnected_early = True
                                             # 立即设置completion_event以提前结束等待
                                             if not completion_event.is_set():
                                                 completion_event.set()
                                             break
-                                        await asyncio.sleep(0.5)  # 更频繁的检查间隔
+                                        await asyncio.sleep(0.3)  # 更频繁的检查间隔
                                     except Exception as e:
                                         logger.error(f"[{req_id}] (Worker) 增强断开检测器错误: {e}")
                                         break

api_utils/request_processor.py

@@ -96,6 +96,44 @@ async def _analyze_model_requirements(req_id: str, context: dict, request: ChatC
     return context
 
 
+async def _test_client_connection(req_id: str, http_request: Request) -> bool:
+    """通过发送测试数据包来主动检测客户端连接状态"""
+    try:
+        # 尝试发送一个小的测试数据包
+        test_chunk = "data: {\"type\":\"ping\"}\n\n"
+
+        # 获取底层的响应对象
+        if hasattr(http_request, '_receive'):
+            # 检查接收通道是否还活跃
+            try:
+                # 尝试非阻塞地检查是否有断开消息
+                import asyncio
+                receive_task = asyncio.create_task(http_request._receive())
+                done, pending = await asyncio.wait([receive_task], timeout=0.01)
+
+                if done:
+                    message = receive_task.result()
+                    if message.get("type") == "http.disconnect":
+                        return False
+                else:
+                    # 取消未完成的任务
+                    receive_task.cancel()
+                    try:
+                        await receive_task
+                    except asyncio.CancelledError:
+                        pass
+
+            except Exception:
+                # 如果检查过程中出现异常，可能表示连接有问题
+                return False
+
+        # 如果上述检查都通过，认为连接正常
+        return True
+
+    except Exception as e:
+        # 任何异常都认为连接已断开
+        return False
+
 async def _setup_disconnect_monitoring(req_id: str, http_request: Request, result_future: Future) -> Tuple[Event, asyncio.Task, Callable]:
     """设置客户端断开连接监控"""
     from server import logger
@@ -105,13 +143,24 @@ async def _setup_disconnect_monitoring(req_id: str, http_request: Request, resul
     async def check_disconnect_periodically():
         while not client_disconnected_event.is_set():
             try:
+                # 使用主动检测方法
+                is_connected = await _test_client_connection(req_id, http_request)
+                if not is_connected:
+                    logger.info(f"[{req_id}] 主动检测到客户端断开连接。")
+                    client_disconnected_event.set()
+                    if not result_future.done():
+                        result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] 客户端关闭了请求"))
+                    break
+
+                # 备用检查：使用原有的is_disconnected方法
                 if await http_request.is_disconnected():
-                    logger.info(f"[{req_id}] 客户端断开，设置事件。")
+                    logger.info(f"[{req_id}] 备用检测到客户端断开连接。")
                     client_disconnected_event.set()
                     if not result_future.done():
                         result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] 客户端关闭了请求"))
                     break
-                await asyncio.sleep(0.5)  # 更频繁的检查间隔，从1.0秒改为0.5秒
+
+                await asyncio.sleep(0.3)  # 更频繁的检查间隔，从0.5秒改为0.3秒
             except asyncio.CancelledError:
                 break
             except Exception as e:
@@ -755,7 +804,16 @@ async def _process_request_refactored(
     result_future: Future
 ) -> Optional[Tuple[Event, Locator, Callable[[str], bool]]]:
     """核心请求处理函数 - 重构版本"""
-    
+
+    # 优化：在开始任何处理前主动检测客户端连接状态
+    is_connected = await _test_client_connection(req_id, http_request)
+    if not is_connected:
+        from server import logger
+        logger.info(f"[{req_id}] ✅ 核心处理前检测到客户端断开，提前退出节省资源")
+        if not result_future.done():
+            result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] 客户端在处理开始前已断开连接"))
+        return None
+
     context = await _initialize_request_context(req_id, request)
     context = await _analyze_model_requirements(req_id, context, request)
 
@@ -788,7 +846,10 @@ async def _process_request_refactored(
             context['parsed_model_list'],
             check_client_disconnected
         )
-        
+
+        # 优化：在提交提示前再次检查客户端连接，避免不必要的后台请求
+        check_client_disconnected("提交提示前最终检查")
+
         await page_controller.submit_prompt(prepared_prompt, check_client_disconnected)
 
         # 响应处理仍然需要在这里，因为它决定了是流式还是非流式，并设置future

api_utils/routes.py

@@ -202,6 +202,13 @@ async def chat_completions(
         raise HTTPException(status_code=504, detail=f"[{req_id}] 请求处理超时。")
     except asyncio.CancelledError:
         raise HTTPException(status_code=499, detail=f"[{req_id}] 请求被客户端取消。")
+    except HTTPException as http_exc:
+        # 对于客户端断开连接的情况，使用更友好的日志级别
+        if http_exc.status_code == 499:
+            logger.info(f"[{req_id}] 客户端断开连接: {http_exc.detail}")
+        else:
+            logger.warning(f"[{req_id}] HTTP异常: {http_exc.detail}")
+        raise http_exc
     except Exception as e:
         logger.exception(f"[{req_id}] 等待Worker响应时出错")
         raise HTTPException(status_code=500, detail=f"[{req_id}] 服务器内部错误: {e}")