From 2d0f67ca3814b6e43b2d0349335b967ab842a58c Mon Sep 17 00:00:00 2001 From: saas-backend Date: Mon, 21 Sep 2026 17:07:06 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20musetalk=5Fserver?= =?UTF-8?q?=20=E4=B8=89=E4=B8=AA=E5=85=B3=E9=94=AEbug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bug1: 60fps降帧逻辑未生效 - 输入视频>30fps时先用ffmpeg降帧至25fps再进行MuseTalk推理 - 推理完成后使用minterpolate MCI运动补偿插帧回原帧率 - 记录input_fps和inference_fps便于排查 Bug2: 推理超时后线程未终止，锁提前释放 - 引入abort_event机制，超时时set event让推理循环检测退出 - 全局追踪活跃ffmpeg子进程，超时时kill所有子进程 - 推理线程改为daemon线程，超时后等待线程退出再释放锁和清理目录 - /cancel接口同步使用abort_event+kill ffmpeg子进程机制 Bug3: /health接口缺少GFPGAN状态字段 - 新增gfpgan_loaded和gfpgan_load_error字段 - 模型加载时同步更新全局状态变量 Bonus: - 动态超时计算：基础120s + 预估帧数*0.15s，上限1800s - GFPGAN增强异常时log warning而非静默pass --- deploy/gpu_worker/musetalk_server.py | 232 ++++++++++++++++++++++----- 1 file changed, 195 insertions(+), 37 deletions(-) diff --git a/deploy/gpu_worker/musetalk_server.py b/deploy/gpu_worker/musetalk_server.py index 96d39fd7b..69968de59 100644 --- a/deploy/gpu_worker/musetalk_server.py +++ b/deploy/gpu_worker/musetalk_server.py @@ -2,6 +2,15 @@ 部署在 RTX2060 本地，接收 gpu_worker.py 的推理请求，调用 MuseTalk 生成口型同步视频。 +Bug 修复（2026-09-21）： + - Bug1: 60fps降帧逻辑 — 输入视频 >30fps 时先降帧至 25fps 推理，推理后用 + minterpolate MCI 插帧回原帧率，并记录 input_fps/inference_fps + - Bug2: 超时终止机制 — 推理线程改为 daemon + abort_event 机制，超时时 set event + 让推理循环检测退出，同时 kill 所有活跃 ffmpeg 子进程，等线程退出后再释放锁和清理目录 + - Bug3: /health 接口增加 gfpgan_loaded 和 gfpgan_load_error 字段 + - Bonus: 动态超时计算（基础120s + 帧数*0.15s，上限1800s） + - Bonus: GFPGAN 增强异常时 log warning 而非静默跳过 + #2000 关键修复： - 集成真实 MuseTalk 推理（替换原有 stub 代码） - 音频预处理：22050Hz MP3 → 16kHz mono 16bit WAV（MuseTalk 要求） @@ -95,6 +104,15 @@ class Config: inference_lock = threading.Lock() current_task: dict = {"task_id": None, "process": None, "start_time": 0.0} shutdown_event = threading.Event() +# 当前推理线程的终止信号和线程引用 +current_abort_event: Optional[threading.Event] = None +current_thread: Optional[threading.Thread] = None +# 全局追踪正在运行的 ffmpeg 子进程（用于超时终止） +_active_ffmpeg_procs: list[subprocess.Popen] = [] +_active_ffmpeg_lock = threading.Lock() +# GFPGAN 加载状态（供 /health 接口查询） +_gfpgan_loaded = False +_gfpgan_load_error: Optional[str] = None # ── MuseTalk 模型懒加载 ───────────────────────────────────────────── _muse_models = None @@ -374,21 +392,33 @@ def _check_file_size(file, max_mb: int, label: str) -> Optional[str]: def _run_ffmpeg(cmd: list, timeout: float = 120) -> subprocess.CompletedProcess: - """运行 ffmpeg 命令，检查返回码和超时.""" + """运行 ffmpeg 命令，检查返回码和超时。进程会被注册到全局列表以便外部终止.""" + proc = None try: - result = subprocess.run( + proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - timeout=timeout, - check=True, ) - return result - except subprocess.CalledProcessError as exc: - stderr = exc.stderr.decode(errors="ignore") if exc.stderr else "" - raise RuntimeError(f"ffmpeg 失败 (code={exc.returncode}): {stderr[:500]}") from exc - except subprocess.TimeoutExpired as exc: - raise RuntimeError(f"ffmpeg 超时（>{timeout}s）") from exc + with _active_ffmpeg_lock: + _active_ffmpeg_procs.append(proc) + try: + stdout, stderr = proc.communicate(timeout=timeout) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + raise RuntimeError(f"ffmpeg 超时（>{timeout}s）") + if proc.returncode != 0: + stderr_text = stderr.decode(errors="ignore") if stderr else "" + raise RuntimeError(f"ffmpeg 失败 (code={proc.returncode}): {stderr_text[:500]}") + return subprocess.CompletedProcess(cmd, proc.returncode, stdout, stderr) + finally: + if proc is not None: + with _active_ffmpeg_lock: + try: + _active_ffmpeg_procs.remove(proc) + except ValueError: + pass # ── MuseTalk 模型加载 ──────────────────────────────────────────────── @@ -401,7 +431,7 @@ def _load_musetalk_models(): 加载到 GPU 后转为 FP16（如果配置开启）以节省显存。 RTX2060 6G 显存，FP16 大约需要 3-4GB。 """ - global _muse_models, _muse_models_loaded, _muse_load_error + global _muse_models, _muse_models_loaded, _muse_load_error, _gfpgan_loaded, _gfpgan_load_error if _muse_models_loaded: return _muse_models @@ -500,13 +530,21 @@ def _load_musetalk_models(): gfpgan_model = gfpgan_model.half() gfpgan_model = gfpgan_model.to(device) del gfpgan_ckpt + _gfpgan_loaded = True + _gfpgan_load_error = None logger.info("GFPGAN 加载完成 (FP16=%s)", Config.use_float16) else: + _gfpgan_loaded = False + _gfpgan_load_error = f"模型文件不存在: {gfpgan_path}" logger.warning("GFPGAN 模型不存在: %s，跳过人脸增强", gfpgan_path) except Exception as e: + _gfpgan_loaded = False + _gfpgan_load_error = str(e) logger.warning("GFPGAN 加载失败，跳过人脸增强: %s", e) gfpgan_model = None else: + _gfpgan_loaded = False + _gfpgan_load_error = "已通过环境变量禁用 (MUSE_USE_GFPGAN=0)" logger.info("GFPGAN 已禁用 (MUSE_USE_GFPGAN=0)") _muse_models = { @@ -562,6 +600,7 @@ def _run_inference( audio_path: Path, output_path: Path, bbox_shift: int = 0, + abort_event: threading.Event = None, ) -> None: """执行 MuseTalk 真实推理. @@ -678,13 +717,39 @@ def _run_inference( audio_processor.feature2chunks = _types.MethodType(_feature2chunks, audio_processor) fps = _get_video_fps(video_path) + original_fps = fps audio_duration = _get_media_duration(audio_path) video_duration = _get_media_duration(video_path) logger.info( - "MuseTalk 推理开始: video=%.2fs, audio=%.2fs, fps=%.1f, bbox_shift=%d", + "MuseTalk 推理开始: video=%.2fs, audio=%.2fs, input_fps=%.1f, bbox_shift=%d", video_duration, audio_duration, fps, bbox_shift, ) + # ── Step 0: 高帧率视频降帧（>30fps → 25fps 推理，推理后再插帧回原帧率）── + inference_fps = fps + video_downsampled = False + if fps > 30: + inference_fps = 25.0 + downsampled_video = video_path.parent / "input_25fps.mp4" + logger.info("检测到高帧率视频 %.1f fps，降帧至 %.1f fps 进行推理", fps, inference_fps) + try: + _run_ffmpeg([ + "ffmpeg", "-y", "-v", "warning", + "-i", str(video_path), + "-r", str(inference_fps), + "-c:v", "libx264", "-preset", "veryfast", + "-crf", "18", "-pix_fmt", "yuv420p", + str(downsampled_video), + ], timeout=120) + video_path = downsampled_video + video_downsampled = True + # 更新帧数和时长 + total_input_frames_approx = int(audio_duration * inference_fps) + logger.info("降帧完成: input_fps=%.1f → inference_fps=%.1f", original_fps, inference_fps) + except Exception as e: + logger.warning("视频降帧失败，使用原始帧率 %.1f fps 推理: %s", fps, e) + inference_fps = fps + # ── Step 1: 音频预处理（关键修复：22050Hz MP3 → 16kHz mono WAV）── audio_wav_path = video_path.parent / "audio_16k_mono.wav" _preprocess_audio(audio_path, audio_wav_path, target_sr=16000) @@ -703,7 +768,7 @@ def _run_inference( logger.info("人脸检测完成，有效 bbox: %d/%d", sum(1 for c in coord_list if c is not coord_placeholder), total_frames) # 使用 mirror indexing 循环帧和坐标（避免硬切跳变） - num_output_frames = int(audio_duration * fps) + num_output_frames = int(audio_duration * inference_fps) if num_output_frames <= 0: num_output_frames = total_frames @@ -713,7 +778,7 @@ def _run_inference( audio_array, _ = librosa.load(str(audio_wav_path), sr=16000, mono=True) whisper_features = audio_processor.feature2chunks( feature_array=audio_array, - fps=fps, + fps=inference_fps, weight_dtype=(torch.float16 if Config.use_float16 else torch.float32), batch_size=Config.batch_size, ) @@ -769,6 +834,10 @@ def _run_inference( total_batches = (video_num + bs - 1) // bs for bi in tqdm(range(total_batches), desc="MuseTalk 推理"): + # 检查是否收到终止信号 + if abort_event is not None and abort_event.is_set(): + logger.warning("收到终止信号，中止推理 (batch %d/%d)", bi, total_batches) + raise InterruptedError("推理被外部终止") whisper_batch = whisper_features[bi*bs:(bi+1)*bs] if len(whisper_batch) == 0: break @@ -818,7 +887,7 @@ def _run_inference( _ff_cmd = [ "ffmpeg", "-y", "-v", "warning", "-f", "rawvideo", "-pix_fmt", "bgr24", - "-s", f"{frame_w}x{frame_h}", "-r", str(fps), + "-s", f"{frame_w}x{frame_h}", "-r", str(inference_fps), "-i", "-", "-vcodec", "libx264", "-preset", "veryfast", "-vf", "format=yuv420p", "-crf", "18", @@ -826,10 +895,17 @@ def _run_inference( ] import subprocess as _sp _ff_proc = _sp.Popen(_ff_cmd, stdin=_sp.PIPE, stdout=_sp.DEVNULL, stderr=_sp.PIPE) + # 注册 ffmpeg 进程到全局列表，以便超时终止 + with _active_ffmpeg_lock: + _active_ffmpeg_procs.append(_ff_proc) n_out = min(len(res_frame_list), num_output_frames) try: for i in tqdm(range(n_out), desc="合成帧"): + # 检查终止信号 + if abort_event is not None and abort_event.is_set(): + logger.warning("收到终止信号，中止合成 (frame %d/%d)", i, n_out) + raise InterruptedError("推理被外部终止") cyc_i = i % len(coord_cycle) x1, y1, x2, y2 = coord_cycle[cyc_i] ori_frame = copy.deepcopy(frame_list_cycle[cyc_i]) @@ -863,8 +939,8 @@ def _run_inference( res_frame_resized = cv2.resize(_out_bgr, (_fw, _fh), interpolation=cv2.INTER_LANCZOS4) del _face_t, _out, _out_bgr - except Exception: - pass + except Exception as _gfpgan_err: + logger.warning("GFPGAN 增强失败（帧 %d），使用原图: %s", i, _gfpgan_err) # face parsing 融合 try: @@ -882,27 +958,60 @@ def _run_inference( _ff_proc.stdin.close() _ff_ret = _ff_proc.wait(timeout=120) + # 从全局列表中移除已完成的 ffmpeg 进程 + with _active_ffmpeg_lock: + try: + _active_ffmpeg_procs.remove(_ff_proc) + except ValueError: + pass if _ff_ret != 0: _ff_err = _ff_proc.stderr.read().decode(errors="ignore") if _ff_proc.stderr else "" raise RuntimeError(f"ffmpeg编码失败(exit={_ff_ret}): {_ff_err[-300:]}") except Exception: try: _ff_proc.kill() except Exception: pass + with _active_ffmpeg_lock: + try: + _active_ffmpeg_procs.remove(_ff_proc) + except ValueError: + pass raise - shutil.copy2(str(silent_video_path), str(output_path)) + # ── Step 8: 高帧率视频插帧还原（如输入 >30fps，从 25fps 插帧回原帧率）── + final_video = silent_video_path + if video_downsampled: + upscaled_video = video_path.parent / "output_upscaled.mp4" + logger.info("将推理结果从 %.1f fps 插帧还原至 %.1f fps", inference_fps, original_fps) + try: + _run_ffmpeg([ + "ffmpeg", "-y", "-v", "warning", + "-i", str(silent_video_path), + "-vf", f"minterpolate=mi_mode=mci:mc_mode=aobmc:me_mode=bidir:vsbmc=1:fps={int(original_fps)}", + "-c:v", "libx264", "-preset", "veryfast", + "-crf", "18", "-pix_fmt", "yuv420p", + str(upscaled_video), + ], timeout=max(300, int(original_fps * 10))) + final_video = upscaled_video + logger.info("插帧还原完成: %.1f fps", original_fps) + except Exception as e: + logger.warning("插帧还原失败，使用 %.1f fps 结果: %s", inference_fps, e) + final_video = silent_video_path + + shutil.copy2(str(final_video), str(output_path)) try: torch.cuda.empty_cache() if audio_wav_path.exists(): audio_wav_path.unlink() - if silent_video_path.exists() and str(silent_video_path) != str(output_path): - silent_video_path.unlink() + # 清理中间文件 + for _tmp in [silent_video_path] + ([video_path] if video_downsampled else []): + if _tmp.exists() and str(_tmp) != str(output_path): + _tmp.unlink() except Exception as e: logger.warning("清理中间文件失败: %s", e) - logger.info("MuseTalk 推理完成: output=%s, duration=%.2fs", - output_path.name, _get_media_duration(output_path)) + logger.info("MuseTalk 推理完成: output=%s, duration=%.2fs, input_fps=%.1f, inference_fps=%.1f", + output_path.name, _get_media_duration(output_path), original_fps, inference_fps) # ── 路由 ────────────────────────────────────────────────────────────── @@ -924,6 +1033,8 @@ def health(): "current_task": task_info, "musetalk_loaded": _muse_models_loaded, "musetalk_load_error": str(_muse_load_error) if _muse_load_error else None, + "gfpgan_loaded": _gfpgan_loaded, + "gfpgan_load_error": _gfpgan_load_error, "timestamp": time.time(), } ) @@ -982,24 +1093,52 @@ def inference(): current_task["start_time"] = time.time() current_task["process"] = "inference_thread" # 标记为运行中 - # 在线程中运行推理（支持超时） + # 在线程中运行推理（支持超时 + 终止信号） result_container = {"error": None} + abort_event = threading.Event() + global current_abort_event, current_thread + current_abort_event = abort_event def inference_thread(): try: - _run_inference(video_path, audio_path, output_path, bbox_shift=bbox_shift) + _run_inference(video_path, audio_path, output_path, bbox_shift=bbox_shift, abort_event=abort_event) except Exception as exc: logger.exception("推理异常: %s", exc) result_container["error"] = str(exc) - thread = threading.Thread(target=inference_thread) + thread = threading.Thread(target=inference_thread, daemon=True) + current_thread = thread thread.start() - thread.join(timeout=Config.inference_timeout) + + # 动态超时：基础 120s + 预估帧数×0.15s，上限 1800s + video_fps = _get_video_fps(video_path) + video_duration = _get_media_duration(video_path) + estimated_frames = int(video_duration * video_fps) if video_duration > 0 else 500 + dynamic_timeout = min(max(Config.inference_timeout, 120 + estimated_frames * 0.15), 1800) + + thread.join(timeout=dynamic_timeout) if thread.is_alive(): - # 超时，终止 - logger.error("推理超时 (>%ds)，终止任务 %s", Config.inference_timeout, task_id) - return jsonify({"error": f"推理超时（>{Config.inference_timeout}s）", "task_id": task_id}), 504 + # 超时处理：发终止信号 + 杀 ffmpeg 子进程 + 等待线程退出 + logger.error("推理超时 (>%ds)，终止任务 %s", dynamic_timeout, task_id) + abort_event.set() + # 终止所有正在运行的 ffmpeg 子进程 + with _active_ffmpeg_lock: + for proc in _active_ffmpeg_procs[:]: + try: + proc.kill() + proc.wait(timeout=5) + except Exception: + pass + _active_ffmpeg_procs.clear() + # 等待线程退出（daemon 线程会在主进程退出时自动终止，但这里给一定时间让它清理） + thread.join(timeout=10) + # 标记超时错误，由 finally 统一清理 + result_container["error"] = f"推理超时（>{dynamic_timeout:.0f}s）" + result_container["timeout"] = True + + if result_container.get("timeout"): + return jsonify({"error": result_container["error"], "task_id": task_id}), 504 if result_container["error"]: logger.error("推理失败 task_id=%s: %s", task_id, result_container["error"]) @@ -1019,6 +1158,8 @@ def inference(): current_task["task_id"] = None current_task["process"] = None current_task["start_time"] = 0.0 + current_abort_event = None + current_thread = None # 清理临时文件 if video_path and video_path.parent.exists(): @@ -1032,20 +1173,35 @@ def inference(): @app.route("/cancel", methods=["POST"]) def cancel(): """终止当前正在进行的推理任务.""" + global current_abort_event, current_thread + if current_task["task_id"] is None: return jsonify({"message": "当前无正在运行的任务"}) task_id = current_task["task_id"] logger.info("收到取消请求，终止任务 %s", task_id) - # 终止推理进程（如果是 subprocess） - if current_task["process"] and current_task["process"] != "inference_thread": - try: - current_task["process"].terminate() - current_task["process"].wait(timeout=5) - logger.info("已终止推理进程") - except Exception as exc: - logger.warning("终止进程失败: %s", exc) + # 发送终止信号给推理线程 + if current_abort_event is not None: + current_abort_event.set() + logger.info("已发送终止信号给推理线程") + + # 终止所有正在运行的 ffmpeg 子进程 + with _active_ffmpeg_lock: + for proc in _active_ffmpeg_procs[:]: + try: + proc.kill() + proc.wait(timeout=5) + logger.info("已终止 ffmpeg 子进程") + except Exception as exc: + logger.warning("终止 ffmpeg 子进程失败: %s", exc) + _active_ffmpeg_procs.clear() + + # 等待推理线程退出（daemon 线程会在主进程退出时自动终止） + if current_thread is not None and current_thread.is_alive(): + current_thread.join(timeout=10) + if current_thread.is_alive(): + logger.warning("推理线程未能在 10s 内退出，将作为 daemon 线程自动终止") # 清理临时文件 task_dir = Path(Config.temp_dir) / task_id @@ -1060,6 +1216,8 @@ def cancel(): current_task["task_id"] = None current_task["process"] = None current_task["start_time"] = 0.0 + current_abort_event = None + current_thread = None return jsonify({"message": f"已取消任务 {task_id}"}) -- 2.54.0