mirror of
https://github.com/fawney19/Aether.git
synced 2026-09-02 01:10:23 +08:00
perf: 在 Redis 密集操作前释放 DB 连接,candidate_records 改为异步写入
- 号池排序涉及大量 Redis I/O,在调用前提前释放 DB 连接避免连接池压力 - 新增 create_candidate_records_async,通过 asyncio.to_thread 执行同步 DB 写入 - 同步执行、异步提交、TaskService 三条路径统一改用异步版本
This commit is contained in:
@@ -4,6 +4,7 @@
|
|||||||
负责获取和排序可用的 Provider/Endpoint/Key 组合
|
负责获取和排序可用的 Provider/Endpoint/Key 组合
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
import uuid
|
import uuid
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from typing import Any
|
from typing import Any
|
||||||
@@ -378,6 +379,28 @@ class CandidateResolver:
|
|||||||
|
|
||||||
return candidate_record_map
|
return candidate_record_map
|
||||||
|
|
||||||
|
async def create_candidate_records_async(
|
||||||
|
self,
|
||||||
|
all_candidates: list[ProviderCandidate],
|
||||||
|
request_id: str | None,
|
||||||
|
user_id: str | None,
|
||||||
|
user_api_key: ApiKey | None,
|
||||||
|
required_capabilities: dict[str, bool] | None = None,
|
||||||
|
*,
|
||||||
|
expand_retries: bool = True,
|
||||||
|
) -> dict[tuple[int, int], str]:
|
||||||
|
"""异步版本的 create_candidate_records,将同步 DB 操作放到线程池执行,
|
||||||
|
避免阻塞 asyncio 事件循环。"""
|
||||||
|
return await asyncio.to_thread(
|
||||||
|
self.create_candidate_records,
|
||||||
|
all_candidates,
|
||||||
|
request_id,
|
||||||
|
user_id,
|
||||||
|
user_api_key,
|
||||||
|
required_capabilities,
|
||||||
|
expand_retries=expand_retries,
|
||||||
|
)
|
||||||
|
|
||||||
def get_active_candidates(
|
def get_active_candidates(
|
||||||
self,
|
self,
|
||||||
all_candidates: list[ProviderCandidate],
|
all_candidates: list[ProviderCandidate],
|
||||||
|
|||||||
@@ -568,6 +568,9 @@ class CacheAwareScheduler:
|
|||||||
candidates, db, affinity_key, api_format
|
candidates, db, affinity_key, api_format
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# 排序完成后释放 DB 连接,避免后续 Redis 操作期间占用连接
|
||||||
|
release_db_connection_before_await(db)
|
||||||
|
|
||||||
# 2. 调度模式排序
|
# 2. 调度模式排序
|
||||||
if self.scheduling_mode == self.SCHEDULING_MODE_CACHE_AFFINITY:
|
if self.scheduling_mode == self.SCHEDULING_MODE_CACHE_AFFINITY:
|
||||||
if affinity_key and candidates and global_model_id:
|
if affinity_key and candidates and global_model_id:
|
||||||
|
|||||||
@@ -565,6 +565,10 @@ class CandidateBuilder:
|
|||||||
if pool_cfg is not None:
|
if pool_cfg is not None:
|
||||||
# 号池优化:跳过逐 key 的 _check_key_availability 检查,
|
# 号池优化:跳过逐 key 的 _check_key_availability 检查,
|
||||||
# 直接收集全部 active key,将检查推迟到 PoolManager 排序后分页执行。
|
# 直接收集全部 active key,将检查推迟到 PoolManager 排序后分页执行。
|
||||||
|
# 在此释放 DB 连接,因为后续的 PoolManager 排序涉及大量 Redis 操作,
|
||||||
|
# 避免在 Redis I/O 期间长时间占用 DB 连接池。
|
||||||
|
release_db_connection_before_await(db)
|
||||||
|
|
||||||
pool_keys = list(keys_to_check)
|
pool_keys = list(keys_to_check)
|
||||||
|
|
||||||
if not pool_keys:
|
if not pool_keys:
|
||||||
|
|||||||
@@ -161,12 +161,17 @@ class SyncTaskExecutionService:
|
|||||||
request_body=request_body,
|
request_body=request_body,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# 号池排序涉及大量 Redis 操作,提前释放 DB 连接避免连接池压力
|
||||||
|
from src.services.scheduling.utils import release_db_connection_before_await
|
||||||
|
|
||||||
|
release_db_connection_before_await(self.db)
|
||||||
|
|
||||||
# Account Pool: reorder candidates for claude_code providers.
|
# Account Pool: reorder candidates for claude_code providers.
|
||||||
all_candidates, pool_traces = await self._pool_ops.apply_pool_reorder(
|
all_candidates, pool_traces = await self._pool_ops.apply_pool_reorder(
|
||||||
all_candidates, request_body=request_body
|
all_candidates, request_body=request_body
|
||||||
)
|
)
|
||||||
|
|
||||||
candidate_record_map = candidate_resolver.create_candidate_records(
|
candidate_record_map = await candidate_resolver.create_candidate_records_async(
|
||||||
all_candidates=all_candidates,
|
all_candidates=all_candidates,
|
||||||
request_id=request_id,
|
request_id=request_id,
|
||||||
user_id=user_id,
|
user_id=user_id,
|
||||||
|
|||||||
@@ -390,11 +390,17 @@ class TaskService:
|
|||||||
_logger.warning("创建 pending 使用记录失败: {}", str(exc))
|
_logger.warning("创建 pending 使用记录失败: {}", str(exc))
|
||||||
|
|
||||||
all_candidates = list(candidates)
|
all_candidates = list(candidates)
|
||||||
|
|
||||||
|
# 号池排序涉及大量 Redis 操作,提前释放 DB 连接避免连接池压力
|
||||||
|
from src.services.scheduling.utils import release_db_connection_before_await
|
||||||
|
|
||||||
|
release_db_connection_before_await(self.db)
|
||||||
|
|
||||||
all_candidates, pool_traces = await pool_ops.apply_pool_reorder(
|
all_candidates, pool_traces = await pool_ops.apply_pool_reorder(
|
||||||
all_candidates, request_body=request_body
|
all_candidates, request_body=request_body
|
||||||
)
|
)
|
||||||
|
|
||||||
candidate_record_map = candidate_resolver.create_candidate_records(
|
candidate_record_map = await candidate_resolver.create_candidate_records_async(
|
||||||
all_candidates=all_candidates,
|
all_candidates=all_candidates,
|
||||||
request_id=request_id,
|
request_id=request_id,
|
||||||
user_id=user_id,
|
user_id=user_id,
|
||||||
|
|||||||
@@ -87,6 +87,11 @@ class AsyncSubmitPreparationService:
|
|||||||
last_status_code=None,
|
last_status_code=None,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# 号池排序涉及大量 Redis 操作,提前释放 DB 连接避免连接池压力
|
||||||
|
from src.services.scheduling.utils import release_db_connection_before_await
|
||||||
|
|
||||||
|
release_db_connection_before_await(self.db)
|
||||||
|
|
||||||
# Account Pool: keep internal key failover order/skip behavior
|
# Account Pool: keep internal key failover order/skip behavior
|
||||||
# consistent with the SYNC path.
|
# consistent with the SYNC path.
|
||||||
candidates, _pool_traces = await apply_pool_reorder(
|
candidates, _pool_traces = await apply_pool_reorder(
|
||||||
@@ -103,7 +108,7 @@ class AsyncSubmitPreparationService:
|
|||||||
record_map: dict[tuple[int, int], str] = {}
|
record_map: dict[tuple[int, int], str] = {}
|
||||||
if request_id:
|
if request_id:
|
||||||
try:
|
try:
|
||||||
record_map = resolver.create_candidate_records(
|
record_map = await resolver.create_candidate_records_async(
|
||||||
all_candidates=candidates,
|
all_candidates=candidates,
|
||||||
request_id=request_id,
|
request_id=request_id,
|
||||||
user_id=str(user_api_key.user_id),
|
user_id=str(user_api_key.user_id),
|
||||||
|
|||||||
Reference in New Issue
Block a user