diff --git a/app/controllers/upload_router.py b/app/controllers/upload_router.py index fb0d159..937a5e6 100644 --- a/app/controllers/upload_router.py +++ b/app/controllers/upload_router.py @@ -6,6 +6,7 @@ complete 成功后创建转写 Task 并交由 scheduler 入队。 from __future__ import annotations from fastapi import APIRouter, Depends, Request +from starlette.concurrency import run_in_threadpool from sqlalchemy.orm import Session from ..database import get_db @@ -48,7 +49,10 @@ async def upload_chunk( request: Request, service: UploadService = Depends(_service), ) -> ChunkUploadResponse: - uploaded = service.write_chunk(upload_id, index, await request.body()) + # write_chunk 做文件 fsync + DB commit(阻塞 I/O),必须放到线程池跑, + # 否则会阻塞 uvicorn 事件循环,导致并发分片上传被串行化、web 请求卡顿。 + body = await request.body() + uploaded = await run_in_threadpool(service.write_chunk, upload_id, index, body) return ChunkUploadResponse(upload_id=upload_id, index=index, uploaded_chunks=uploaded) diff --git a/app/services/upload_service.py b/app/services/upload_service.py index 8132963..22feb45 100644 --- a/app/services/upload_service.py +++ b/app/services/upload_service.py @@ -16,6 +16,7 @@ from __future__ import annotations import logging import os import shutil +import threading import uuid from datetime import datetime, timedelta, timezone from pathlib import Path @@ -36,6 +37,20 @@ from ..schemas.task import ( logger = logging.getLogger("audio2text.upload") +# 按 upload_id 串行化 uploaded_chunks 的读-改-写,避免并发分片写入丢失更新。 +# SQLite 无行锁,JSON 列的 append 操作不是原子的,必须进程内加锁。 +_chunk_locks: dict[str, threading.Lock] = {} +_chunk_locks_guard = threading.Lock() + + +def _get_chunk_lock(upload_id: str) -> threading.Lock: + with _chunk_locks_guard: + lock = _chunk_locks.get(upload_id) + if lock is None: + lock = threading.Lock() + _chunk_locks[upload_id] = lock + return lock + class UploadService: def __init__(self, db: Session) -> None: @@ -93,6 +108,7 @@ class UploadService: session_dir.mkdir(parents=True, exist_ok=True) chunk_path = session_dir / f"{index}.part" + # 先落盘(无锁,IO 可并行),再持锁更新 DB 计数 try: with chunk_path.open("wb") as out: out.write(data) @@ -102,12 +118,16 @@ class UploadService: chunk_path.unlink(missing_ok=True) raise - uploaded = list(session.uploaded_chunks or []) - if index not in uploaded: - uploaded.append(index) - session.uploaded_chunks = uploaded - session.updated_at = datetime.now(timezone.utc) - self.db.commit() + # 持锁重读+更新:避免并发分片各自 read old list → append → write,后者覆盖前者 + lock = _get_chunk_lock(upload_id) + with lock: + self.db.refresh(session) # 拿最新 uploaded_chunks,不被其他并发请求的旧值覆盖 + uploaded = list(session.uploaded_chunks or []) + if index not in uploaded: + uploaded.append(index) + session.uploaded_chunks = uploaded + session.updated_at = datetime.now(timezone.utc) + self.db.commit() return sorted(uploaded) # ---------------- 拼接 + 创建任务 ---------------- @@ -163,6 +183,9 @@ class UploadService: # 清理分片暂存(commit 后,即使清理失败也不影响已建任务) self._cleanup_session_dir(upload_id) + # 清理进程内锁,避免长期运行后 _chunk_locks 无限增长 + with _chunk_locks_guard: + _chunk_locks.pop(upload_id, None) logger.info("上传完成 task_id=%s file=%s size=%d", task.id, session.filename, session.size_bytes) return CompleteResponse( diff --git a/app/views/_shared.py b/app/views/_shared.py index 91bcc40..e6f1109 100644 --- a/app/views/_shared.py +++ b/app/views/_shared.py @@ -10,7 +10,8 @@ from __future__ import annotations # 分片上传参数(home / upload 共用) DEFAULT_CHUNK_SIZE = 4 * 1024 * 1024 -DEFAULT_CONCURRENCY = 3 +DEFAULT_CONCURRENCY = 3 # 单文件内分片并发数 +DEFAULT_FILE_CONCURRENCY = 5 # 同时上传的文件数(不阻塞 web 请求/ffmpeg/gpu) MAX_RETRY = 2 POLL_INTERVAL = 2000 @@ -259,7 +260,8 @@ def render_upload_js(on_complete: str) -> str: """ return f""" const CHUNK_SIZE = {DEFAULT_CHUNK_SIZE}; -const CONCURRENCY = {DEFAULT_CONCURRENCY}; +const CHUNK_CONCURRENCY = {DEFAULT_CONCURRENCY}; // 单文件内分片并发 +const FILE_CONCURRENCY = {DEFAULT_FILE_CONCURRENCY}; // 同时上传的文件数 const MAX_RETRY = {MAX_RETRY}; const UPLOAD_API = "/api/tasks/chunk-uploads"; @@ -315,13 +317,13 @@ function setUploadProgress(t, pct) {{ }} function pump() {{ + // 统计正在上传的文件数,启动等待中的文件直到达到 FILE_CONCURRENCY const active = pending.filter(t => t.state === "running").length; for (const t of pending) {{ - if (active >= CONCURRENCY) break; + if (active >= FILE_CONCURRENCY) break; if (t.state === "pending") {{ - t.state = "running"; + setUploadState(t, "running"); startUpload(t); - active++; }} }} }} @@ -342,7 +344,7 @@ async function startUpload(t) {{ const need = []; for (let i = 0; i < t.totalChunks; i++) if (!t.uploaded.has(i)) need.push(i); - await runPool(need, CONCURRENCY, i => uploadChunk(t, i)); + await runPool(need, CHUNK_CONCURRENCY, i => uploadChunk(t, i)); if (t.uploaded.size < t.totalChunks) throw new Error("部分分片未能上传"); setUploadState(t, "hashing"); @@ -359,6 +361,9 @@ async function startUpload(t) {{ meta.className = "task-meta fail-msg"; meta.textContent = String(e.message || e); t.el.appendChild(meta); + }} finally {{ + // 无论成功还是失败,都触发 pump 让队列中下一个文件开始上传 + pump(); }} }}