cat-shark
4746e0363f
为视频生成 VR 双眼字幕的单体实现：FastAPI 后端、调度器与全部节点 （提音/转写/翻译/ASS/抽帧/OCR/LLM 过滤）在单进程内运行。 - 节点协议（wov_sdk 数据模型）与分布式版保持一致，预留回退桥梁 - 工作流即数据：DAG 存于 workflows/*.json，模型/链路改动只改数据 - 调度器：拓扑顺序执行、断点续跑（产物重建）、任务暂停/继续 - 抽帧按帧间隔（select 按帧号精确取帧），VLM OCR 与 LLM 过滤使用 自适应线程池弹性并发，并打印数据处理速度进度日志 - 100% 行覆盖率（pytest --cov-fail-under=100）
149 lines
5.1 KiB
Python
149 lines
5.1 KiB
Python
"""自适应线程池测试。
|
|
|
|
覆盖决策函数（增/减/保持/边界）、map 顺序返回、worker 异常隔离，
|
|
以及"10s 窗口内平均响应 < 0.3s 加线程 / > 1.0s 减线程"的弹性行为
|
|
（通过注入假时钟做确定性验证）。
|
|
"""
|
|
|
|
import time
|
|
|
|
from nodes.adaptive_pool import AdaptiveThreadPool, decide
|
|
|
|
|
|
class FakeClock:
|
|
"""可手动拨动的假时钟，用于确定性验证弹性窗口逻辑。"""
|
|
|
|
def __init__(self, now: float = 0.0) -> None:
|
|
self.now = now
|
|
|
|
def __call__(self) -> float:
|
|
return self.now
|
|
|
|
def advance(self, seconds: float) -> None:
|
|
self.now += seconds
|
|
|
|
|
|
def test_decide_increase_when_fast() -> None:
|
|
"""平均响应低于 fast_threshold 且未达上限：线程数 +1。"""
|
|
assert decide(1, 0.1, 1, 16, 0.3, 1.0) == 2
|
|
|
|
|
|
def test_decide_decrease_when_slow() -> None:
|
|
"""平均响应高于 slow_threshold 且高于下限：线程数 -1。"""
|
|
assert decide(3, 2.0, 1, 16, 0.3, 1.0) == 2
|
|
|
|
|
|
def test_decide_keep_when_mid() -> None:
|
|
"""平均响应介于两阈值之间：保持不变。"""
|
|
assert decide(2, 0.5, 1, 16, 0.3, 1.0) == 2
|
|
|
|
|
|
def test_decide_bounds() -> None:
|
|
"""已达上限不再增、已达下限不再减。"""
|
|
assert decide(16, 0.1, 1, 16, 0.3, 1.0) == 16
|
|
assert decide(1, 2.0, 1, 16, 0.3, 1.0) == 1
|
|
|
|
|
|
def test_pool_map_ordered_results() -> None:
|
|
"""map 按输入顺序返回结果，worker 简单映射。"""
|
|
pool = AdaptiveThreadPool(worker=lambda item: item * 2)
|
|
assert pool.map([1, 2, 3, 4]) == [2, 4, 6, 8]
|
|
|
|
|
|
def test_pool_on_progress_callback() -> None:
|
|
"""进度回调：每次完成触发一次，携带已完成数/总数/速度。"""
|
|
progress: list[tuple[int, int, float]] = []
|
|
pool = AdaptiveThreadPool(
|
|
worker=lambda item: item,
|
|
on_progress=lambda done, total, rate: progress.append((done, total, rate)),
|
|
)
|
|
pool.map([10, 20, 30])
|
|
assert [item[0] for item in progress] == [1, 2, 3] # 已完成数递增。
|
|
assert all(item[1] == 3 for item in progress) # 总数固定。
|
|
assert all(item[2] > 0 for item in progress) # 速度为正值。
|
|
|
|
def test_pool_map_empty() -> None:
|
|
"""空输入：不启动任务，直接返回空列表。"""
|
|
pool = AdaptiveThreadPool(worker=lambda item: item)
|
|
assert pool.map([]) == []
|
|
|
|
|
|
def test_pool_worker_exception_isolated() -> None:
|
|
"""worker 抛异常时以异常对象作为结果，不拖垮整体。"""
|
|
def boom(item):
|
|
raise RuntimeError("boom")
|
|
|
|
pool = AdaptiveThreadPool(worker=boom)
|
|
results = pool.map([1, 2])
|
|
assert len(results) == 2
|
|
assert all(isinstance(result, RuntimeError) for result in results)
|
|
|
|
|
|
def test_pool_grows_when_fast() -> None:
|
|
"""10s 窗口内平均响应 < 0.3s：线程数从 1 增至 2（弹性扩容）。"""
|
|
clock = FakeClock()
|
|
pool = AdaptiveThreadPool(
|
|
worker=lambda item: item,
|
|
min_workers=1, max_workers=16,
|
|
window_seconds=10.0, fast_threshold=0.3,
|
|
clock=clock,
|
|
)
|
|
# 拨快时钟越过窗口：首个任务完成即触发评估 → 平均响应≈0 < 0.3 → +1 线程。
|
|
clock.advance(11)
|
|
pool.map(list(range(4)))
|
|
assert pool.max_concurrency == 2
|
|
|
|
|
|
def test_pool_shrink_when_slow() -> None:
|
|
"""窗口平均响应 > 1.0s：线程数从 2 减至 1（弹性退避）。"""
|
|
clock = FakeClock()
|
|
pool = AdaptiveThreadPool(
|
|
worker=lambda item: item,
|
|
min_workers=1, max_workers=16,
|
|
window_seconds=10.0, fast_threshold=0.3, slow_threshold=1.0,
|
|
clock=clock,
|
|
)
|
|
pool._resize(2) # 先扩到 2 个线程。
|
|
clock.advance(11)
|
|
pool._tick(2.0) # 窗口内平均 2.0 > 1.0 → 缩回 1。
|
|
deadline = time.monotonic() + 2
|
|
while len(pool._threads) > 1 and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
assert len(pool._threads) == 1
|
|
pool._stop.set()
|
|
|
|
|
|
def test_resize_shrink_idempotent() -> None:
|
|
"""回归：重复缩容到同一目标不会重复放哨兵（曾因并发缩容毒死全部线程而死锁）。"""
|
|
pool = AdaptiveThreadPool(worker=lambda item: item, min_workers=1, max_workers=8)
|
|
pool._resize(3)
|
|
assert pool._target_workers == 3
|
|
pool._resize(2)
|
|
pool._resize(2) # 目标已是 2：幂等，不再放哨兵。
|
|
assert pool._target_workers == 2
|
|
# 只有 1 个线程被哨兵退出，最终存活 2 个。
|
|
deadline = time.monotonic() + 2
|
|
while len(pool._threads) > 2 and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
assert len(pool._threads) == 2
|
|
pool._stop.set()
|
|
|
|
|
|
def test_pool_survives_mixed_grow_shrink() -> None:
|
|
"""回归：扩容+缩容混合场景 map 必须完成且保序（修复前会死锁挂起）。"""
|
|
clock = FakeClock()
|
|
state = {"count": 0}
|
|
|
|
def worker(item):
|
|
state["count"] += 1
|
|
clock.advance(0.06 if state["count"] <= 20 else 0.6)
|
|
return item
|
|
|
|
pool = AdaptiveThreadPool(
|
|
worker=worker, min_workers=1, max_workers=4,
|
|
window_seconds=0.5, fast_threshold=0.2, slow_threshold=0.4,
|
|
clock=clock,
|
|
)
|
|
out = pool.map(list(range(60)))
|
|
assert out == list(range(60))
|