cat-shark
b16e0e9f3c
调度与状态机： - 修复 PAUSED 任务被拾起后复活执行（点击暂停反而开始任务）：next_queued_run 只取 QUEUED，execute_run 以 PAUSED 进入直接返回，暂停必须显式 resume - 重启恢复：启动时 recover_interrupted_runs 把遗留 RUNNING 置 QUEUED（保留产物） - 暂停信号 paused.flag：暂停接口写、继续/重试清除，OCR 逐帧检查秒级中断， 节点内被暂停保持 PAUSED 不误报 FAILED - 调度轮询容错：_loop 异常不杀死线程（曾致任务永久停留 QUEUED） subtitle-ocr 节点级断点： - ocr_partial.jsonl 逐帧存档，重启/暂停后只处理未处理帧，产物与一次跑完一致 - 进度日志携带窗口平均耗时与线程数；取消后抑制进度日志井喷 llm-filter 过滤质量与限流自适应： - 上下文净化：喂给 LLM 的是过滤后的字幕（规则层垃圾从上下文剔除） - 正则确定性过滤：裸网址域名、HTML/水印模式直接删除 - 429/5xx 指数退避重试；worker 限流错误 report_failure 内存临时降最大线程数 并缩容（无错误窗口回升），失败条目降并发后重试一轮 - 保留长文本保护（noise 不删 ≥min_keep_len 文本，LLM 判定不稳的必要兜底） 前端： - 工作流编排页支持选择工作流编辑（加载最新/历史版本）、版本历史面板、 新建/编辑双模式；管理后台编辑跳转 workflow.html?edit=<id> 工作流：ocr-subtitle v7（filter pool_max_workers=20、pool_fast_threshold=1）
246 lines
11 KiB
Python
246 lines
11 KiB
Python
"""自适应线程池。
|
||
|
||
用于对耗时的独立子任务（如逐帧 VLM OCR）做弹性并发加速：
|
||
- 以滚动时间窗口统计已完成任务的平均响应时间；
|
||
- 窗口内平均响应 < fast_threshold（默认 0.3s）→ 增加 1 个工作线程（上限 max_workers）；
|
||
- 窗口内平均响应 > slow_threshold（默认 1.0s）→ 减少 1 个工作线程（下限 min_workers）。
|
||
|
||
线程数从 min_workers（默认 1）起步，按实测负载自适应：服务端空闲（响应快）
|
||
就加大并发，服务端变慢就退避，避免盲目并发压垮上游（如本地 Ollama）。
|
||
|
||
线程安全说明：worker 会在多个线程中并发调用，调用方需保证 worker 无共享
|
||
可变状态（registry 处理器是纯函数，符合要求）；结果按输入顺序返回。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import queue
|
||
import threading
|
||
import time
|
||
from typing import Callable
|
||
|
||
# 停止哨兵：压入队列让空闲工作线程退出（用于缩容）。
|
||
_POISON = object()
|
||
|
||
|
||
def decide(
|
||
current: int,
|
||
avg: float,
|
||
min_workers: int,
|
||
max_workers: int,
|
||
fast_threshold: float,
|
||
slow_threshold: float,
|
||
) -> int:
|
||
"""根据窗口平均响应时间返回调整后的目标线程数（纯决策函数）。
|
||
|
||
响应快（avg < fast_threshold）且未达上限 → 加 1；响应慢
|
||
（avg > slow_threshold）且未达下限 → 减 1；其余情况保持不变。
|
||
"""
|
||
if avg < fast_threshold and current < max_workers:
|
||
return current + 1
|
||
if avg > slow_threshold and current > min_workers:
|
||
return current - 1
|
||
return current
|
||
|
||
|
||
class AdaptiveThreadPool:
|
||
"""自适应线程池：单次 map 按输入顺序返回全部结果。"""
|
||
|
||
def __init__(
|
||
self,
|
||
worker: Callable,
|
||
min_workers: int = 1,
|
||
max_workers: int = 16,
|
||
window_seconds: float = 10.0,
|
||
fast_threshold: float = 0.3,
|
||
slow_threshold: float = 1.0,
|
||
clock=time.monotonic,
|
||
on_progress: Callable[[int, int, float, float, int], None] | None = None,
|
||
) -> None:
|
||
"""初始化；clock 可注入便于测试；on_progress(done,total,rate) 每次完成回调。"""
|
||
self._worker = worker
|
||
self.min_workers = max(1, min_workers)
|
||
self.max_workers = max(self.min_workers, max_workers)
|
||
self.window_seconds = window_seconds
|
||
self.fast_threshold = fast_threshold
|
||
self.slow_threshold = slow_threshold
|
||
self._clock = clock
|
||
self._queue: queue.Queue = queue.Queue()
|
||
# 并发目标线程数：决策/缩容的权威依据（线程退出是异步的，不能用
|
||
# len(_threads) 判断，否则并发缩容会重复放哨兵把全部线程毒死）。
|
||
self._target_workers = 0
|
||
self._threads: list[threading.Thread] = []
|
||
self._results: list = []
|
||
self._lock = threading.Lock()
|
||
self._stop = threading.Event()
|
||
# 取消标记：worker 检测到取消（如暂停信号）后设置，后续完成的任务
|
||
# 不再触发进度回调——暂停时队列中剩余大量任务会快速退出，若仍逐项
|
||
# 打印进度会在数秒内打出上万行日志。
|
||
self._cancel_event = threading.Event()
|
||
# 有效最大线程数：初始等于 max_workers；消费错误（如 API 限流）时
|
||
# report_failure 临时收紧，连续无错误窗口后逐步回升——并发自适应配额。
|
||
self._effective_max_workers = max_workers
|
||
# 当前窗口内消费错误计数：窗口评估时无错误才允许恢复有效上限。
|
||
self._window_failures = 0
|
||
# 滚动窗口起点与已记录的单次耗时。
|
||
self._window_start = clock()
|
||
# 观测到的最大并发线程数（供测试与监控）。
|
||
self.max_concurrency = 0
|
||
# 进度回调与计数：on_progress(已完成数, 总数, 平均速度/秒)。
|
||
self._on_progress = on_progress
|
||
self._completed = 0
|
||
self._total = 0
|
||
self._started_at = 0.0
|
||
self._window_times: list[float] = []
|
||
# 最近一次窗口评估的平均单任务耗时（秒）：供进度回调诊断使用，
|
||
# 与扩缩容决策共用同一依据；窗口尚未评估时为 None（回退累计平均）。
|
||
self._window_avg_time: float | None = None
|
||
|
||
def _run(self) -> None:
|
||
"""工作线程主循环：取任务 → 执行 → 记录耗时并自适应评估。"""
|
||
try:
|
||
while not self._stop.is_set():
|
||
try:
|
||
seq, item = self._queue.get(timeout=0.2)
|
||
except queue.Empty:
|
||
continue
|
||
if item is _POISON:
|
||
# 缩容哨兵：处理完即可退出（队列计数照常）。
|
||
self._queue.task_done()
|
||
break
|
||
start = self._clock()
|
||
try:
|
||
result = self._worker(item)
|
||
except Exception as exc:
|
||
# 单任务异常不拖垮整体：以异常对象作为结果，由调用方判定。
|
||
result = exc
|
||
finally:
|
||
elapsed = self._clock() - start
|
||
self._results.append((seq, result))
|
||
# 进度回调：已完成数、总数与平均处理速度（条/秒）。
|
||
self._completed += 1
|
||
if self._on_progress is not None and not self._cancel_event.is_set():
|
||
elapsed_total = max(self._clock() - self._started_at, 1e-9)
|
||
with self._lock:
|
||
workers = self._target_workers
|
||
self._on_progress(
|
||
self._completed,
|
||
self._total,
|
||
self._completed / elapsed_total,
|
||
self._current_avg_time(elapsed_total),
|
||
workers,
|
||
)
|
||
self._tick(elapsed)
|
||
self._queue.task_done()
|
||
finally:
|
||
# 无论何种退出路径都从线程列表移除，保证线程数统计准确。
|
||
with self._lock:
|
||
if threading.current_thread() in self._threads:
|
||
self._threads.remove(threading.current_thread())
|
||
|
||
def _tick(self, elapsed: float) -> None:
|
||
"""记录一次完成耗时；窗口满时按平均响应时间调整线程数。"""
|
||
self._window_times.append(elapsed)
|
||
if self._clock() - self._window_start < self.window_seconds:
|
||
return
|
||
avg = sum(self._window_times) / len(self._window_times)
|
||
self._window_start = self._clock()
|
||
self._window_times.clear()
|
||
# 记录本次窗口平均耗时：进度回调据此展示"当前扩缩容依据"。
|
||
self._window_avg_time = avg
|
||
with self._lock:
|
||
current = self._target_workers
|
||
# 窗口内无消费错误 → 有效上限逐步回升（错误降下来的并发慢慢恢复）。
|
||
if (
|
||
self._window_failures == 0
|
||
and self._effective_max_workers < self.max_workers
|
||
):
|
||
self._effective_max_workers += 1
|
||
# 重置窗口错误计数，进入下一窗口。
|
||
self._window_failures = 0
|
||
# 扩容上限用有效最大线程数：错误窗口内即使响应快也不超过收紧后的上限。
|
||
self._resize(
|
||
decide(
|
||
current, avg, self.min_workers, self._effective_max_workers,
|
||
self.fast_threshold, self.slow_threshold,
|
||
)
|
||
)
|
||
|
||
def _current_avg_time(self, elapsed_total: float) -> float:
|
||
"""返回供进度回调展示的平均单任务耗时（秒）。
|
||
|
||
优先使用最近一次窗口评估的平均耗时（与扩缩容决策同一依据）；
|
||
窗口尚未评估过时回退为启动至今的累计平均，避免无数据可看。
|
||
"""
|
||
if self._window_avg_time is not None:
|
||
return self._window_avg_time
|
||
return elapsed_total / max(self._completed, 1)
|
||
|
||
def _resize(self, target: int) -> None:
|
||
"""调整并发目标：扩容启动新线程；缩容压入等量停止哨兵（幂等）。
|
||
|
||
以 _target_workers 为当前值：重复调用同一 target 不会重复放哨兵，
|
||
避免并发缩容把所有线程毒死导致队列任务无人处理而挂起。
|
||
"""
|
||
with self._lock:
|
||
current = self._target_workers
|
||
if target > current:
|
||
self.max_concurrency = max(self.max_concurrency, target)
|
||
for _ in range(target - current):
|
||
thread = threading.Thread(target=self._run, daemon=True)
|
||
thread.start()
|
||
self._threads.append(thread)
|
||
self._target_workers = target
|
||
elif target < current:
|
||
for _ in range(current - target):
|
||
self._queue.put((None, _POISON))
|
||
self._target_workers = target
|
||
|
||
def cancel(self) -> None:
|
||
"""请求取消本批任务：后续完成的任务不再触发进度回调。
|
||
|
||
供调用方在工作线程内检测到外部信号（如暂停）时调用，抑制暂停后
|
||
队列中剩余任务快速退出导致的进度日志井喷；下一批 map 自动重置。
|
||
"""
|
||
self._cancel_event.set()
|
||
|
||
def report_failure(self) -> None:
|
||
"""通知一次消费错误（如 API 限流 429）：临时降低有效最大线程数并缩容。
|
||
|
||
供工作线程捕获可退避错误（限流/服务端 5xx）后调用：并发立即收紧到
|
||
新上限，后续请求减少从而避开持续限流；连续无错误窗口后有效上限
|
||
逐步回升到 max_workers（见 _tick 的恢复逻辑）。
|
||
"""
|
||
with self._lock:
|
||
self._window_failures += 1
|
||
if self._effective_max_workers > self.min_workers:
|
||
self._effective_max_workers -= 1
|
||
# 缩容到新上限（幂等：目标低于当前才放停止哨兵）。
|
||
self._resize(self._effective_max_workers)
|
||
|
||
def map(self, items) -> list:
|
||
"""按输入顺序返回每个 item 经 worker 处理后的结果列表。"""
|
||
self._results = []
|
||
self._completed = 0
|
||
self._total = len(items)
|
||
self._started_at = self._clock()
|
||
self._stop.clear()
|
||
# 每批任务开始时重置取消状态：上一批的取消不延续到下一批。
|
||
self._cancel_event.clear()
|
||
# 上一批任务结束后工作线程已全部退出（_stop 停止）但 _target_workers
|
||
# 仍记旧值，_resize 不会重新启动线程——实际无线程时归零后重建。
|
||
with self._lock:
|
||
if not self._threads:
|
||
self._target_workers = 0
|
||
self._resize(self.min_workers)
|
||
for seq, item in enumerate(items):
|
||
self._queue.put((seq, item))
|
||
self._queue.join()
|
||
self._stop.set()
|
||
with self._lock:
|
||
threads = list(self._threads)
|
||
for thread in threads:
|
||
thread.join(1.0)
|
||
self._results.sort(key=lambda pair: pair[0])
|
||
return [result for _, result in self._results]
|