557 lines
22 KiB
Python
557 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import multiprocessing
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
import traceback
|
|
import uuid
|
|
from multiprocessing.connection import Connection
|
|
from typing import Any, Callable
|
|
|
|
|
|
MAX_NEW_TOKENS = 384
|
|
DEFAULT_SYNTHESIS_TIMEOUT_SECONDS = 120.0
|
|
DEFAULT_STARTUP_TIMEOUT_SECONDS = 900.0
|
|
# 启动失败后再次拉起 worker 的最小间隔：避免"启动超时→立即重启→再超时"的死循环
|
|
# 在系统高负载时持续加载 torch/CUDA，进一步加剧卡顿。
|
|
STARTUP_FAILURE_BACKOFF_SECONDS = 60.0
|
|
DEFAULT_CPU_THREADS = 4
|
|
DEFAULT_CPU_AFFINITY_COUNT = 8
|
|
DEFAULT_PROCESS_PRIORITY = "below_normal"
|
|
STARTUP_PROGRESS_POLL_SECONDS = 1.0
|
|
|
|
|
|
class FasterQwenWorkerError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class FasterQwenWorkerTimeout(FasterQwenWorkerError):
|
|
pass
|
|
|
|
|
|
def _generation_kwargs(settings: dict[str, Any], text: str) -> dict[str, Any]:
|
|
return {
|
|
"text": text,
|
|
"language": str(settings.get("language") or "Chinese"),
|
|
"non_streaming_mode": bool(settings.get("non_streaming_mode", True)),
|
|
"max_new_tokens": MAX_NEW_TOKENS,
|
|
}
|
|
|
|
|
|
def _bounded_int(value: Any, default: int, *, minimum: int, maximum: int) -> int:
|
|
try:
|
|
parsed = int(value)
|
|
except (TypeError, ValueError):
|
|
parsed = default
|
|
return max(minimum, min(maximum, parsed))
|
|
|
|
|
|
def _configure_worker_environment(settings: dict[str, Any]) -> int:
|
|
cpu_threads = _bounded_int(
|
|
settings.get("cpu_threads"),
|
|
DEFAULT_CPU_THREADS,
|
|
minimum=1,
|
|
maximum=8,
|
|
)
|
|
thread_value = str(cpu_threads)
|
|
for name in (
|
|
"OMP_NUM_THREADS",
|
|
"MKL_NUM_THREADS",
|
|
"OPENBLAS_NUM_THREADS",
|
|
"NUMEXPR_NUM_THREADS",
|
|
"VECLIB_MAXIMUM_THREADS",
|
|
"BLIS_NUM_THREADS",
|
|
):
|
|
os.environ[name] = thread_value
|
|
os.environ["TOKENIZERS_PARALLELISM"] = "false"
|
|
return cpu_threads
|
|
|
|
|
|
def _configure_torch_threads(torch_module: Any, cpu_threads: int) -> None:
|
|
torch_module.set_num_threads(cpu_threads)
|
|
try:
|
|
torch_module.set_num_interop_threads(1)
|
|
except RuntimeError:
|
|
# PyTorch only allows setting interop threads before parallel work starts.
|
|
pass
|
|
|
|
|
|
def _apply_worker_process_limits(settings: dict[str, Any]) -> dict[str, Any]:
|
|
cpu_count = max(1, int(os.cpu_count() or 1))
|
|
affinity_count = _bounded_int(
|
|
settings.get("cpu_affinity_count"),
|
|
DEFAULT_CPU_AFFINITY_COUNT,
|
|
minimum=0,
|
|
maximum=min(cpu_count, 63),
|
|
)
|
|
priority = str(settings.get("process_priority") or DEFAULT_PROCESS_PRIORITY).strip().lower()
|
|
applied_affinity = 0
|
|
applied_priority = "default"
|
|
|
|
if os.name == "nt":
|
|
try:
|
|
import ctypes
|
|
from ctypes import wintypes
|
|
|
|
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
|
kernel32.GetCurrentProcess.restype = wintypes.HANDLE
|
|
kernel32.SetPriorityClass.argtypes = [wintypes.HANDLE, wintypes.DWORD]
|
|
kernel32.SetPriorityClass.restype = wintypes.BOOL
|
|
kernel32.SetProcessAffinityMask.argtypes = [wintypes.HANDLE, ctypes.c_size_t]
|
|
kernel32.SetProcessAffinityMask.restype = wintypes.BOOL
|
|
process_handle = kernel32.GetCurrentProcess()
|
|
priority_classes = {
|
|
"idle": 0x00000040,
|
|
"below_normal": 0x00004000,
|
|
"normal": 0x00000020,
|
|
}
|
|
priority_class = priority_classes.get(priority, priority_classes[DEFAULT_PROCESS_PRIORITY])
|
|
if kernel32.SetPriorityClass(process_handle, priority_class):
|
|
applied_priority = priority if priority in priority_classes else DEFAULT_PROCESS_PRIORITY
|
|
if affinity_count > 0:
|
|
affinity_mask = (1 << affinity_count) - 1
|
|
if kernel32.SetProcessAffinityMask(process_handle, affinity_mask):
|
|
applied_affinity = affinity_count
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"cpu_affinity_count": applied_affinity,
|
|
"process_priority": applied_priority,
|
|
}
|
|
|
|
|
|
def _load_runtime(
|
|
settings: dict[str, Any],
|
|
progress: Callable[[str], None] | None = None,
|
|
) -> dict[str, Any]:
|
|
def report(stage: str) -> None:
|
|
if progress:
|
|
progress(stage)
|
|
|
|
device = str(settings.get("device") or "cuda")
|
|
if device == "cpu" and "CUDA_VISIBLE_DEVICES" not in os.environ:
|
|
os.environ["CUDA_VISIBLE_DEVICES"] = ""
|
|
|
|
cpu_threads = _configure_worker_environment(settings)
|
|
report("torch_import_started")
|
|
import torch
|
|
report("torch_import_complete")
|
|
_configure_torch_threads(torch, cpu_threads)
|
|
report("tts_library_import_started")
|
|
from faster_qwen3_tts import FasterQwen3TTS
|
|
report("tts_library_import_complete")
|
|
|
|
if device == "cpu":
|
|
torch.cuda.is_available = lambda: False
|
|
|
|
load_kwargs: dict[str, Any] = {}
|
|
if device == "cpu":
|
|
load_kwargs["device"] = "cpu"
|
|
report("from_pretrained_started")
|
|
model = FasterQwen3TTS.from_pretrained(
|
|
str(settings.get("model_name_or_path") or "Qwen/Qwen3-TTS-12Hz-0.6B-Base"),
|
|
**load_kwargs,
|
|
)
|
|
report("from_pretrained_complete")
|
|
|
|
# voice_clone_prompt 不再预计算：当前 faster_qwen3_tts 的 FasterQwen3TTS 没有
|
|
# create_voice_clone_prompt 方法，预计算只会失败。改为每次合成时在
|
|
# _synthesize_wav 里直接传 ref_audio/ref_text/xvec_only 参数。
|
|
voice_clone_prompt = None
|
|
|
|
return {
|
|
"model": model,
|
|
"torch": torch,
|
|
"voice_clone_prompt": voice_clone_prompt,
|
|
"settings": settings,
|
|
"cpu_threads": cpu_threads,
|
|
}
|
|
|
|
|
|
def _synthesize_wav(runtime: dict[str, Any], text: str) -> bytes:
|
|
import soundfile as sf
|
|
|
|
model = runtime["model"]
|
|
torch = runtime["torch"]
|
|
settings = runtime["settings"]
|
|
safe_text = str(text or "").strip()[:80] or "欢迎来到直播间。"
|
|
kwargs = _generation_kwargs(settings, safe_text)
|
|
voice_clone_prompt = runtime.get("voice_clone_prompt")
|
|
if voice_clone_prompt is not None:
|
|
kwargs["voice_clone_prompt"] = voice_clone_prompt
|
|
else:
|
|
ref_audio = str(settings.get("ref_audio") or "")
|
|
if not ref_audio:
|
|
raise RuntimeError("Faster-Qwen3-TTS requires ref_audio")
|
|
kwargs.update({
|
|
"ref_audio": ref_audio,
|
|
"ref_text": str(settings.get("ref_text") or "") or None,
|
|
"xvec_only": bool(settings.get("xvec_only", True)),
|
|
"append_silence": bool(settings.get("append_silence", True)),
|
|
})
|
|
|
|
if not torch.cuda.is_available():
|
|
torch.backends.cudnn.enabled = False
|
|
with torch.inference_mode():
|
|
wavs, sample_rate = model.generate_voice_clone(**kwargs)
|
|
|
|
output = io.BytesIO()
|
|
audio = wavs[0]
|
|
if isinstance(audio, torch.Tensor):
|
|
audio = audio.cpu().numpy()
|
|
sf.write(output, audio, sample_rate, format="WAV")
|
|
return output.getvalue()
|
|
|
|
|
|
def faster_qwen_worker_main(connection: Connection, settings: dict[str, Any]) -> None:
|
|
def report(stage: str, **details: Any) -> None:
|
|
try:
|
|
connection.send({
|
|
"type": "progress",
|
|
"stage": stage,
|
|
"pid": os.getpid(),
|
|
**details,
|
|
})
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
report("process_started", executable=sys.executable, prefix=sys.prefix)
|
|
cpu_threads = _configure_worker_environment(settings)
|
|
report("environment_ready", cpu_threads=cpu_threads)
|
|
load_started = time.monotonic()
|
|
report("model_loading", model_name=str(settings.get("model_name_or_path") or ""))
|
|
runtime = _load_runtime(settings, progress=report)
|
|
load_ms = int((time.monotonic() - load_started) * 1000)
|
|
report("model_loaded", load_ms=load_ms)
|
|
|
|
warmup_started = time.monotonic()
|
|
graph_warmup_ms = 0
|
|
report("warmup_started")
|
|
warmup = getattr(runtime["model"], "warmup", None)
|
|
if callable(warmup):
|
|
graph_warmup_started = time.monotonic()
|
|
warmup()
|
|
graph_warmup_ms = int((time.monotonic() - graph_warmup_started) * 1000)
|
|
report("graph_warmup_complete", graph_warmup_ms=graph_warmup_ms)
|
|
|
|
# model.warmup() 只捕获模型内部 CUDA Graph，不会初始化参考音频和
|
|
# voice-clone 生成路径。若把这一步留到首条播报，后台会在近一分钟内
|
|
# 看起来毫无响应，因此在降低进程优先级之前完成一次真实短句合成。
|
|
voice_clone_warmup_started = time.monotonic()
|
|
report("voice_clone_warmup_started")
|
|
_synthesize_wav(runtime, "系统启动")
|
|
voice_clone_warmup_ms = int((time.monotonic() - voice_clone_warmup_started) * 1000)
|
|
report("voice_clone_warmup_complete", voice_clone_warmup_ms=voice_clone_warmup_ms)
|
|
warmup_ms = int((time.monotonic() - warmup_started) * 1000)
|
|
report(
|
|
"warmup_complete",
|
|
warmup_ms=warmup_ms,
|
|
graph_warmup_ms=graph_warmup_ms,
|
|
voice_clone_warmup_ms=voice_clone_warmup_ms,
|
|
)
|
|
|
|
# 加载和 CUDA Graph 捕获需要完整 CPU 调度能力。仅在预热完成后降低
|
|
# worker 优先级和亲和性，避免繁忙直播环境下启动时间被放大到超时。
|
|
process_limits = _apply_worker_process_limits(settings)
|
|
connection.send({
|
|
"type": "ready",
|
|
"pid": os.getpid(),
|
|
"load_ms": load_ms,
|
|
"warmup_ms": warmup_ms,
|
|
"graph_warmup_ms": graph_warmup_ms,
|
|
"voice_clone_warmup_ms": voice_clone_warmup_ms,
|
|
"max_new_tokens": MAX_NEW_TOKENS,
|
|
"cpu_threads": cpu_threads,
|
|
**process_limits,
|
|
})
|
|
except BaseException as exc:
|
|
try:
|
|
connection.send({
|
|
"type": "startup_error",
|
|
"error_type": type(exc).__name__,
|
|
"error": str(exc),
|
|
"traceback": traceback.format_exc(),
|
|
})
|
|
except Exception:
|
|
pass
|
|
connection.close()
|
|
return
|
|
|
|
while True:
|
|
try:
|
|
message = connection.recv()
|
|
except (EOFError, OSError):
|
|
break
|
|
if not isinstance(message, dict):
|
|
continue
|
|
command = str(message.get("command") or "")
|
|
if command == "stop":
|
|
break
|
|
if command != "synthesize":
|
|
continue
|
|
|
|
request_id = str(message.get("request_id") or "")
|
|
started = time.monotonic()
|
|
try:
|
|
audio = _synthesize_wav(runtime, str(message.get("text") or ""))
|
|
connection.send({
|
|
"type": "result",
|
|
"request_id": request_id,
|
|
"audio": audio,
|
|
"duration_ms": int((time.monotonic() - started) * 1000),
|
|
"bytes": len(audio),
|
|
"max_new_tokens": MAX_NEW_TOKENS,
|
|
})
|
|
except BaseException as exc:
|
|
try:
|
|
connection.send({
|
|
"type": "error",
|
|
"request_id": request_id,
|
|
"error_type": type(exc).__name__,
|
|
"error": str(exc),
|
|
"traceback": traceback.format_exc(),
|
|
})
|
|
except Exception:
|
|
break
|
|
connection.close()
|
|
|
|
|
|
class FasterQwenWorkerClient:
|
|
def __init__(
|
|
self,
|
|
settings: dict[str, Any],
|
|
logger=None,
|
|
*,
|
|
synthesis_timeout_seconds: float = DEFAULT_SYNTHESIS_TIMEOUT_SECONDS,
|
|
startup_timeout_seconds: float = DEFAULT_STARTUP_TIMEOUT_SECONDS,
|
|
context=None,
|
|
process_target: Callable[..., None] | None = None,
|
|
):
|
|
self.settings = dict(settings)
|
|
self.logger = logger
|
|
self.synthesis_timeout_seconds = max(1.0, float(synthesis_timeout_seconds))
|
|
self.startup_timeout_seconds = max(0.1, float(startup_timeout_seconds))
|
|
self._context = context or multiprocessing.get_context("spawn")
|
|
self._process_target = process_target or faster_qwen_worker_main
|
|
self._lock = threading.RLock()
|
|
self._process = None
|
|
self._connection = None
|
|
self._worker_pid = 0
|
|
self._next_start_after = 0.0
|
|
|
|
@property
|
|
def worker_pid(self) -> int:
|
|
return int(self._worker_pid or 0)
|
|
|
|
def _log(self, level: str, message: str, *args) -> None:
|
|
if self.logger:
|
|
getattr(self.logger, level)(message, *args)
|
|
|
|
def _is_alive_locked(self) -> bool:
|
|
return bool(self._process is not None and self._process.is_alive())
|
|
|
|
def ensure_ready(self) -> dict[str, Any]:
|
|
with self._lock:
|
|
if self._is_alive_locked() and self._connection is not None:
|
|
return {"pid": self.worker_pid, "reused": True}
|
|
return self._start_worker_locked()
|
|
|
|
def _start_worker_locked(self) -> dict[str, Any]:
|
|
now = time.monotonic()
|
|
if now < self._next_start_after:
|
|
wait = int(self._next_start_after - now)
|
|
raise FasterQwenWorkerError(
|
|
f"TTS worker 启动退避中，距上次启动失败不足 {int(STARTUP_FAILURE_BACKOFF_SECONDS)} 秒，"
|
|
f"约 {wait} 秒后可重试"
|
|
)
|
|
self._terminate_worker_locked("replace_stale_worker", graceful=False)
|
|
parent_connection, child_connection = self._context.Pipe(duplex=True)
|
|
process = self._context.Process(
|
|
target=self._process_target,
|
|
args=(child_connection, self.settings),
|
|
name="FasterQwen3TTSWorker",
|
|
daemon=True,
|
|
)
|
|
process.start()
|
|
try:
|
|
child_connection.close()
|
|
except Exception:
|
|
pass
|
|
self._process = process
|
|
self._connection = parent_connection
|
|
self._worker_pid = int(getattr(process, "pid", 0) or 0)
|
|
self._log("info", "[FasterQwenTTS] worker 已启动, pid=%s，正在加载和预热", self.worker_pid)
|
|
|
|
started = time.monotonic()
|
|
deadline = started + self.startup_timeout_seconds
|
|
last_stage = "process_spawned"
|
|
message: Any = None
|
|
while True:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
self._next_start_after = time.monotonic() + STARTUP_FAILURE_BACKOFF_SECONDS
|
|
self._terminate_worker_locked("startup_timeout", graceful=False)
|
|
raise FasterQwenWorkerError(
|
|
f"Faster-Qwen3-TTS worker startup exceeded {self.startup_timeout_seconds:.0f}s "
|
|
f"(last_stage={last_stage})"
|
|
)
|
|
if not parent_connection.poll(min(STARTUP_PROGRESS_POLL_SECONDS, remaining)):
|
|
if not process.is_alive():
|
|
exit_code = getattr(process, "exitcode", None)
|
|
self._next_start_after = time.monotonic() + STARTUP_FAILURE_BACKOFF_SECONDS
|
|
self._terminate_worker_locked("startup_process_exited", graceful=False)
|
|
raise FasterQwenWorkerError(
|
|
f"Faster-Qwen3-TTS worker exited during startup "
|
|
f"(exit_code={exit_code}, last_stage={last_stage})"
|
|
)
|
|
continue
|
|
try:
|
|
message = parent_connection.recv()
|
|
except (EOFError, OSError) as exc:
|
|
self._next_start_after = time.monotonic() + STARTUP_FAILURE_BACKOFF_SECONDS
|
|
self._terminate_worker_locked("startup_connection_closed", graceful=False)
|
|
raise FasterQwenWorkerError(
|
|
f"Faster-Qwen3-TTS worker connection closed during startup "
|
|
f"(last_stage={last_stage})"
|
|
) from exc
|
|
if isinstance(message, dict) and message.get("type") == "progress":
|
|
last_stage = str(message.get("stage") or "unknown")
|
|
elapsed_ms = int((time.monotonic() - started) * 1000)
|
|
self._log(
|
|
"info",
|
|
"[FasterQwenTTS] worker 启动进度: %s, elapsed=%sms",
|
|
last_stage,
|
|
elapsed_ms,
|
|
)
|
|
continue
|
|
break
|
|
if not isinstance(message, dict) or message.get("type") != "ready":
|
|
if isinstance(message, dict):
|
|
error = str(message.get("error") or message.get("error_type") or "unknown startup error")
|
|
worker_traceback = str(message.get("traceback") or "").strip()
|
|
if worker_traceback:
|
|
self._log("error", "[FasterQwenTTS] worker 启动异常:\n%s", worker_traceback)
|
|
else:
|
|
error = "invalid startup response"
|
|
self._next_start_after = time.monotonic() + STARTUP_FAILURE_BACKOFF_SECONDS
|
|
self._terminate_worker_locked("startup_error", graceful=False)
|
|
raise FasterQwenWorkerError(error)
|
|
self._next_start_after = 0.0
|
|
self._log(
|
|
"info",
|
|
"[FasterQwenTTS] worker 预热完成, pid=%s load=%sms warmup=%sms "
|
|
"graph=%sms voice_clone=%sms max_new_tokens=%s",
|
|
self.worker_pid,
|
|
message.get("load_ms"),
|
|
message.get("warmup_ms"),
|
|
message.get("graph_warmup_ms"),
|
|
message.get("voice_clone_warmup_ms"),
|
|
message.get("max_new_tokens"),
|
|
)
|
|
self._log(
|
|
"info",
|
|
"[FasterQwenTTS] worker 资源限制: cpu_threads=%s affinity=%s priority=%s",
|
|
message.get("cpu_threads"),
|
|
message.get("cpu_affinity_count"),
|
|
message.get("process_priority"),
|
|
)
|
|
return message
|
|
|
|
def synthesize(self, text: str) -> tuple[bytes, dict[str, Any]]:
|
|
with self._lock:
|
|
self.ensure_ready()
|
|
request_id = uuid.uuid4().hex
|
|
connection = self._connection
|
|
try:
|
|
connection.send({
|
|
"command": "synthesize",
|
|
"request_id": request_id,
|
|
"text": str(text or ""),
|
|
})
|
|
except (BrokenPipeError, EOFError, OSError) as exc:
|
|
self._restart_after_failure_locked("send_failed")
|
|
raise FasterQwenWorkerError("Faster-Qwen3-TTS worker connection failed") from exc
|
|
|
|
if not connection.poll(self.synthesis_timeout_seconds):
|
|
self._log(
|
|
"error",
|
|
"[FasterQwenTTS] 单次合成超过 %.0f 秒，强制终止 worker pid=%s",
|
|
self.synthesis_timeout_seconds,
|
|
self.worker_pid,
|
|
)
|
|
restart_error = self._restart_after_failure_locked("synthesis_timeout")
|
|
suffix = f"; restart failed: {restart_error}" if restart_error else ""
|
|
raise FasterQwenWorkerTimeout(
|
|
f"Faster-Qwen3-TTS synthesis exceeded {self.synthesis_timeout_seconds:.0f}s{suffix}"
|
|
)
|
|
try:
|
|
message = connection.recv()
|
|
except (EOFError, OSError) as exc:
|
|
self._restart_after_failure_locked("worker_exited")
|
|
raise FasterQwenWorkerError("Faster-Qwen3-TTS worker exited during synthesis") from exc
|
|
|
|
if not isinstance(message, dict) or message.get("request_id") != request_id:
|
|
self._restart_after_failure_locked("invalid_response")
|
|
raise FasterQwenWorkerError("Faster-Qwen3-TTS worker returned an invalid response")
|
|
if message.get("type") == "error":
|
|
error = str(message.get("error") or message.get("error_type") or "synthesis failed")
|
|
lowered = error.lower()
|
|
if "cuda" in lowered or "out of memory" in lowered or "device-side" in lowered:
|
|
self._restart_after_failure_locked("cuda_error")
|
|
raise FasterQwenWorkerError(error)
|
|
if message.get("type") != "result" or not isinstance(message.get("audio"), bytes):
|
|
self._restart_after_failure_locked("invalid_result")
|
|
raise FasterQwenWorkerError("Faster-Qwen3-TTS worker returned no audio")
|
|
return message["audio"], message
|
|
|
|
def _restart_after_failure_locked(self, reason: str) -> str:
|
|
self._terminate_worker_locked(reason, graceful=False)
|
|
try:
|
|
self._start_worker_locked()
|
|
return ""
|
|
except Exception as exc:
|
|
self._log("error", "[FasterQwenTTS] worker 自动重启失败: %s", exc)
|
|
return str(exc)
|
|
|
|
def _terminate_worker_locked(self, reason: str, *, graceful: bool) -> None:
|
|
process = self._process
|
|
connection = self._connection
|
|
self._process = None
|
|
self._connection = None
|
|
self._worker_pid = 0
|
|
if process is None:
|
|
if connection is not None:
|
|
try:
|
|
connection.close()
|
|
except Exception:
|
|
pass
|
|
return
|
|
|
|
if graceful and process.is_alive() and connection is not None:
|
|
try:
|
|
connection.send({"command": "stop"})
|
|
process.join(timeout=3.0)
|
|
except Exception:
|
|
pass
|
|
if process.is_alive():
|
|
self._log("warning", "[FasterQwenTTS] 终止 worker, reason=%s pid=%s", reason, process.pid)
|
|
process.terminate()
|
|
process.join(timeout=10.0)
|
|
if process.is_alive():
|
|
process.kill()
|
|
process.join(timeout=5.0)
|
|
if connection is not None:
|
|
try:
|
|
connection.close()
|
|
except Exception:
|
|
pass
|
|
|
|
def close(self) -> None:
|
|
with self._lock:
|
|
self._terminate_worker_locked("shutdown", graceful=True)
|