# main.py
uvicorn.run("src.app:app", host="127.0.0.1", port=8000, workers=3)

# src/test_report/service.py
_executor = concurrent.futures.ThreadPoolExecutor(max_workers=4, thread_name_prefix="calculate_worker")

进程 1 (Worker 1)
├── 请求处理线程 (处理导入请求)
├── 后台任务线程 1 (计算 SPC)
├── 后台任务线程 2 (计算 SPC)
├── 后台任务线程 3 (计算 SPC)
└── 后台任务线程 4 (计算 SPC)

进程 2 (Worker 2)
├── 请求处理线程 (处理导入请求)
├── 后台任务线程 1 (计算 SPC)
├── 后台任务线程 2 (计算 SPC)
├── 后台任务线程 3 (计算 SPC)
└── 后台任务线程 4 (计算 SPC)

进程 3 (Worker 3)
├── 请求处理线程 (处理导入请求)
├── 后台任务线程 1 (计算 SPC)
├── 后台任务线程 2 (计算 SPC)
├── 后台任务线程 3 (计算 SPC)
└── 后台任务线程 4 (计算 SPC)

async def calculate_entry(db: AsyncSessionDep, entry_id: int):
    # 计算x_ucl、x_lcl、cpk、fixed_x_cl、fixed_sigma
    # ...
    agent_x_bar_r = getAgent(df, entry, spc_data_x_bar_r)  # CPU 密集型
    agent_cpk = getAgent(df, entry, spc_data_cpk)  # CPU 密集型
    fixed_x_cl, x_ucl, x_lcl = agent_x_bar_r.get_x_control_limits()  # CPU 密集型
    fixed_sigma = agent_x_bar_r.get_sigma()  # CPU 密集型
    cpk = agent_cpk.get_cpk()  # CPU 密集型

# 导入完成后立即触发后台任务
if successful_report_ids:
    unique_report_ids = list(dict.fromkeys(successful_report_ids))
    loop = asyncio.get_running_loop()
    calculate_reports_entries_in_thread_pool(unique_report_ids, loop)  # 立即提交到线程池

# 线程池执行
def _calculate_reports_entries_sync(report_ids: list[int]):
    asyncio.run(calculate_reports_entries(report_ids))  # 创建新的事件循环
    # 遍历所有 entries，每个都执行 CPU 密集型计算

@router.post("/import")
async def import_test_report(db: AsyncSessionDep, files: list[UploadFile]):
    # FastAPI 自动调用 python-multipart 解析上传的文件
    # files: list[UploadFile] 中的每个 UploadFile 对象已经包含了解析后的文件流

场景：上传 10 个 Excel 文件，每个 5MB

时间线：
T1: 开始接收第 1 个文件（multipart 解析开始，占用 CPU）
T2: 解析第 1 个文件完成（耗时 200ms，期间事件循环可能被阻塞）
T3: 开始接收第 2 个文件（继续解析，占用 CPU）
...
T10: 所有文件解析完成（总耗时约 2 秒）

问题：
- 解析过程中，CPU 被占用
- 如果同时有其他请求到达，响应会变慢
- 事件循环可能被阻塞，导致其他接口卡慢

1. 接收 HTTP 请求（I/O）
2. 解析 multipart 数据（需要 CPU）
3. 提取 Excel 内容（CPU + I/O）
4. 导入数据到数据库（I/O）
5. 提交事务（I/O）
6. 触发后台任务（CPU 密集型）

时间线：
T1: 进程 1 处理导入请求 A，完成后触发后台任务
T2: 进程 1 的线程池开始执行后台任务（占用 CPU）
T3: 新请求 B 被路由到进程 1
T4: 进程 1 的 CPU 被后台任务占用，请求 B 处理变慢

# src/test_report/service.py
_executor = concurrent.futures.ThreadPoolExecutor(max_workers=1, thread_name_prefix="calculate_worker")

# src/test_report/service.py
import multiprocessing
import sys

# 使用进程池而不是线程池
_mp_context = multiprocessing.get_context("spawn" if sys.platform == "win32" else "fork")
_executor = concurrent.futures.ProcessPoolExecutor(
    max_workers=2,  # 每个进程 2 个 worker
    mp_context=_mp_context
)

def _calculate_reports_entries_sync(report_ids: list[int]):
    """同步包装函数：在独立进程中运行异步的 calculate_reports_entries"""
    try:
        import asyncio
        # 在 spawn 模式下，子进程会重新导入整个模块，所以原来的函数可以直接使用
        asyncio.run(calculate_reports_entries(report_ids))
    except Exception as e:
        import logging
        logger = logging.getLogger(__name__)
        logger.error(f"后台计算任务失败 (report_ids={report_ids}): {e}", exc_info=True)

def calculate_reports_entries_in_process_pool(report_ids: list[int], loop: asyncio.AbstractEventLoop):
    """将计算任务提交到进程池执行，不等待结果"""
    executor = _get_process_pool_executor()
    loop.run_in_executor(executor, _calculate_reports_entries_sync, report_ids)

# src/test_report/service.py
import asyncio

# 全局信号量，限制并发任务数
_calculate_semaphore = asyncio.Semaphore(1)  # 最多 1 个并发任务

async def calculate_reports_entries_with_limit(report_ids: list[int]):
    """带并发限制的计算任务"""
    async with _calculate_semaphore:
        await calculate_reports_entries(report_ids)

def _calculate_reports_entries_sync(report_ids: list[int]):
    """同步包装函数"""
    try:
        asyncio.run(calculate_reports_entries_with_limit(report_ids))
    except Exception as e:
        logger.error(f"后台计算任务失败 (report_ids={report_ids}): {e}", exc_info=True)

def calculate_reports_entries_in_thread_pool(report_ids: list[int], loop: asyncio.AbstractEventLoop):
    """将计算任务提交到线程池执行"""
    loop.run_in_executor(_executor, _calculate_reports_entries_sync, report_ids)

# src/test_report/service.py
import asyncio
from collections import deque
from threading import Lock
from typing import Optional

# 任务队列和限流
_task_queue: deque[list[int]] = deque()
_queue_lock = Lock()
_max_concurrent_tasks = 1  # 限制并发任务数
_current_running_tasks = 0
_task_event = asyncio.Event()

async def _task_processor():
    """后台任务处理器"""
    global _current_running_tasks
    
    while True:
        # 等待有任务或事件触发
        await _task_event.wait()
        _task_event.clear()
        
        while True:
            # 检查是否可以执行新任务
            with _queue_lock:
                if _current_running_tasks >= _max_concurrent_tasks:
                    break
                
                if not _task_queue:
                    break
                
                report_ids = _task_queue.popleft()
                _current_running_tasks += 1
            
            # 执行任务
            loop = asyncio.get_running_loop()
            try:
                await loop.run_in_executor(_executor, _calculate_reports_entries_sync, report_ids)
            finally:
                with _queue_lock:
                    _current_running_tasks -= 1
                    # 通知可以处理下一个任务
                    _task_event.set()

# 在应用启动时启动任务处理器
_task_processor_task: Optional[asyncio.Task] = None

def start_task_processor():
    """启动任务处理器"""
    global _task_processor_task
    if _task_processor_task is None:
        loop = asyncio.get_event_loop()
        _task_processor_task = loop.create_task(_task_processor())

def submit_calculate_task(report_ids: list[int]):
    """提交计算任务到队列"""
    with _queue_lock:
        _task_queue.append(report_ids)
        _task_event.set()

# src/test_report/service.py
async def calculate_entries(entries: list[ReportEntry]):
    """分批处理 entries，每批之间 yield 控制权"""
    BATCH_SIZE = 10  # 每批处理 10 个 entries
    
    for i in range(0, len(entries), BATCH_SIZE):
        batch = entries[i:i + BATCH_SIZE]
        
        # 处理当前批次
        for entry in batch:
            async with async_session_maker() as db:
                await calculate_entry(db, entry.id)
                await db.commit()
        
        # 每批处理后 yield，让其他协程有机会执行
        await asyncio.sleep(0)  # 让出控制权

# 使用 Redis 作为消息队列
import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def submit_calculate_task(report_ids: list[int]):
    """提交任务到 Redis 队列"""
    redis_client.lpush('calculate_queue', json.dumps(report_ids))

# 独立的 worker 进程（单独运行）
def worker_process():
    """独立的 worker 进程"""
    while True:
        task = redis_client.brpop('calculate_queue', timeout=1)
        if task:
            report_ids = json.loads(task[1])
            _calculate_reports_entries_sync(report_ids)

# src/test_report/service.py
_executor = concurrent.futures.ThreadPoolExecutor(max_workers=1, thread_name_prefix="calculate_worker")

# 原代码：CPU 密集型计算在事务内
async def calculate_entries(entries: list[ReportEntry]):
    for entry in entries:
        async with async_session_maker() as db:  # 事务开始
            await calculate_entry(db, entry.id)  # 包含 CPU 密集型计算（几秒）
            await db.commit()  # 事务结束（几秒后）

async def calculate_entry(db: AsyncSessionDep, entry_id: int):
    # T1: 开始事务（DEFERRED 模式，不获取锁）
    entry = await db.get(...)  # T2: 第一次查询，获取 SHARED 锁
    rows = await db.execute(...)  # T3: 继续查询，仍持有 SHARED 锁
    
    # T4-T8: CPU 密集型计算（几秒，仍持有 SHARED 锁）
    agent_x_bar_r = getAgent(...)  # 耗时操作
    cpk = agent_cpk.get_cpk()  # 耗时操作
    
    # T9: 修改对象属性（SQLAlchemy 标记为 dirty，但锁仍为 SHARED）
    entry.cpk = cpk  # 只是标记 dirty，还没有执行 SQL UPDATE
    
    # T10: commit() 时，SQLAlchemy flush pending changes
    # 这时才执行 SQL UPDATE，锁从 SHARED 升级到 RESERVED → EXCLUSIVE
    await db.commit()  # T11: 提交，释放所有锁

进程 1：
T1: BEGIN（不获取锁）
T2: SELECT（获取 SHARED 锁）
T3-T7: CPU 计算（几秒，仍持有 SHARED 锁）
T8: entry.cpk = cpk（标记 dirty，锁仍为 SHARED）
T9: commit() → flush → UPDATE（锁升级：SHARED → RESERVED → EXCLUSIVE）
T10: 提交完成，释放锁

进程 2（同时执行）：
T1: BEGIN（不获取锁）
T2: SELECT（获取 SHARED 锁，可以与进程 1 并行）
T3-T7: CPU 计算（几秒，仍持有 SHARED 锁，可以与进程 1 并行）
T8: entry.cpk = cpk（标记 dirty，锁仍为 SHARED）
T9: commit() → flush → UPDATE（尝试获取 RESERVED 锁）
    - 如果进程 1 还在 T9，需要等待进程 1 完成
    - 如果进程 1 已完成，可以立即获取锁
T10: 提交完成，释放锁

async def calculate_entries(entries: list[ReportEntry]):
    for entry in entries:
        # 1. 查询数据（只读事务，快速完成）
        async with async_session_maker() as db:
            entry_obj = await db.get(ReportEntry, entry.id)
            values_result = await db.execute(...)
            rows = values_result.scalars().all()
            spc_data_x_bar_r = ...
            spc_data_cpk = ...
        
        # 2. CPU 密集型计算（在事务外，不持有数据库锁）
        target_variable = spc_data_x_bar_r.params["target_variable"]
        df = pd.DataFrame([...])
        agent_x_bar_r = getAgent(df, entry_obj, spc_data_x_bar_r)
        agent_cpk = getAgent(df, entry_obj, spc_data_cpk)
        fixed_x_cl, x_ucl, x_lcl = agent_x_bar_r.get_x_control_limits()
        fixed_sigma = agent_x_bar_r.get_sigma()
        cpk = agent_cpk.get_cpk()
        
        # 3. 更新数据库（独立写入事务，快速提交）
        async with async_session_maker() as db:
            entry_to_update = await db.get(ReportEntry, entry.id)
            entry_to_update.x_ucl = list(dict.fromkeys(x_ucl))
            entry_to_update.x_lcl = list(dict.fromkeys(x_lcl))
            entry_to_update.cpk = float(cpk) if cpk is not None else None
            entry_to_update.fixed_x_cl = float(fixed_x_cl) if fixed_x_cl is not None else None
            entry_to_update.fixed_sigma = float(fixed_sigma) if fixed_sigma is not None else None
            await db.commit()  # 快速提交，释放锁

进程 1：
T1: BEGIN（不获取锁）
T2: SELECT（获取 SHARED 锁，几毫秒）
T3: commit()（释放 SHARED 锁，事务结束）
T4-T8: CPU 计算（几秒，不持有任何锁！）
T9: BEGIN（不获取锁）
T10: SELECT（获取 SHARED 锁，几毫秒）
T11: entry.cpk = cpk（标记 dirty）
T12: commit() → UPDATE（锁升级：SHARED → RESERVED → EXCLUSIVE，几毫秒）
T13: 提交完成，释放锁

进程 2（同时执行）：
T1: BEGIN（不获取锁）
T2: SELECT（获取 SHARED 锁，可以与进程 1 的 T2 并行）
T3: commit()（释放 SHARED 锁）
T4-T8: CPU 计算（几秒，不持有任何锁，可以与进程 1 的 T4-T8 完全并行！）
T9: BEGIN（不获取锁）
T10: SELECT（获取 SHARED 锁）
T11: entry.cpk = cpk（标记 dirty）
T12: commit() → UPDATE（如果进程 1 在 T12，需要等待；否则立即获取锁）
T13: 提交完成，释放锁