From f3faad0391a267dc04d2e0257ecd9a5890e41305 Mon Sep 17 00:00:00 2001 From: cat <1716967236@qq.com> Date: Fri, 11 Sep 2026 15:59:46 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E8=87=AA=E9=80=82?= =?UTF-8?q?=E5=BA=94=E7=BA=BF=E7=A8=8B=E6=B1=A0=E9=99=90=E6=B5=81=E6=89=A9?= =?UTF-8?q?=E5=AE=B9=E5=8F=8A=E7=BC=A9=E5=AE=B9=E6=BB=9E=E5=90=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 10 ++ docs/代码审查问题跟踪.md | 13 +- nodes/adaptive_pool.py | 278 ++++++++++++++++-------------------- tests/test_adaptive_pool.py | 140 +++++++++++++++--- 4 files changed, 262 insertions(+), 179 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 9ff940b..7c30fcd 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -214,6 +214,16 @@ subtitle-ocr 逐帧调 vlm-ocr 时使用 `nodes/adaptive_pool.py` 的自适应 服务端变慢就退避，避免盲目并发压垮本地 Ollama； - 结果按帧顺序返回，SRT 时间轴不受并发影响；worker 需无共享可变状态 （vlm-ocr 处理器为纯函数，线程安全）。 +**并发实现修复（审查 R02）**：`AdaptiveThreadPool` 使用标准线程执行器复用线程， +由 `map` 控制在途任务数，不一次性把全片任务压入执行器队列。上面的“线程数” +及进度日志中的 N 指目标在途并发额度，不是执行器已创建的线程总数。降低目标后， +已发出的请求允许完成，后续提交立即遵守新额度；不再向积压队列尾部追加退出哨兵。 +`report_failure()` 只保持或降低当前额度，绝不因上限从 20 降到 19 就把当前 1 +并发扩为 19。错误窗口不扩容，干净窗口逐步恢复有效上限；有效上限跨重试 `map` +保留，每批重新统计耗时窗口。进度回调与结果汇总由 map 线程串行处理，窗口未满 +时平均耗时取 worker 实际耗时均值。`cancel()` 保持原约定，仅抑制进度回调， +节点自行检测暂停并返回异常。相关回归见 `tests/test_adaptive_pool.py`。 + ### 前端 OCR 框选 首页选择工作流后，若 DAG 中存在声明 `crop` 参数的节点（frame-extract）， diff --git a/docs/代码审查问题跟踪.md b/docs/代码审查问题跟踪.md index 0faf33e..6413ce2 100644 --- a/docs/代码审查问题跟踪.md +++ b/docs/代码审查问题跟踪.md @@ -9,7 +9,7 @@ | ID | 优先级 | 问题 | 状态 | 验收标准 | | --- | --- | --- | --- | --- | | R01 | P0 | 普通删除接口可能删除批量源视频目录；清理逻辑从输入路径推导删除范围 | 已修复 | 普通接口拒绝单独删除批量 run；手动与自动清理只删除该上传任务的私有目录；视频、旁挂字幕及其他任务文件不受影响 | -| R02 | P1 | 自适应线程池限流后可能扩容；退出标记位于积压队列尾部，缩容不及时 | 待处理 | 限流不增加并发；降低目标后不再超额提交；真实积压队列验证 | +| R02 | P1 | 自适应线程池限流后可能扩容；退出标记位于积压队列尾部，缩容不及时 | 已修复 | 限流不增加并发；降低目标后不再超额提交；真实积压队列验证 | | R03 | P1 | 最后节点执行时暂停再继续会覆盖有效下载 URI；批量缺产物仍清理并完成 | 待处理 | 收尾幂等，恢复后全部必需产物可下载；缺产物不清理工作空间 | | R04 | P1 | 环形 DAG 可通过保存校验，执行失败后仍在 QUEUED 堵塞队列 | 待处理 | 保存/发布拒绝无效 DAG；历史无效任务进入 FAILED，不阻塞后续任务 | | R05 | P1 | 翻译按固定四行解析 SRT；补齐行数不能保证文本与时间轴对应 | 待处理 | 合法多行 cue 正确解析；按稳定 ID 回填译文并校验缺失、重复项 | @@ -48,6 +48,17 @@ - 状态：修复及验证完成，纳入 `fix/review-improvements` 分支；未部署。运行中的旧进程需加载新代码后才能获得保护。 - 后续边界：普通列表仍会显示批量 run，但点击删除会收到批量入口提示；运行中删除的统一取消/状态协调属于 O08，未在本次展开。 +## R02 修复记录 + +- 根因：`report_failure()` 用降低后的最大值直接调用扩缩容，导致当前 1 并发扩为 19；缩容哨兵追加在 FIFO 积压队列尾部，存量任务仍按旧并发执行。 +- 实现：[adaptive_pool.py](../nodes/adaptive_pool.py) 改为标准 `ThreadPoolExecutor` 复用线程，map 按目标额度有界提交并串行收集结果。移除自建工作队列与退出哨兵，限流更新和提交共享锁。 +- 限流语义：只保持或降低当前额度；已发出的请求允许完成，后续提交遵守新额度；错误窗口不扩容，干净窗口逐步恢复上限。有效上限跨重试 map 保留，每批重新统计窗口与真实单任务耗时。 +- 兼容行为：map 按输入顺序返回结果，worker 异常仍作为结果返回；cancel 仍只抑制进度回调，OCR 自行检测暂停。线程执行器在 map 结束时回收。 +- TDD 红：3 个新增复现测试全部失败，分别观测到限流后目标 19、缩容后后续实际并发 4、错误窗口仍增加额度。 +- TDD 绿及调用方回归：`uv run pytest tests/test_adaptive_pool.py tests/test_llm_filter.py tests/test_ocr_flow.py tests/test_subtitle_ocr_order_threading.py -q --tb=short`，85 passed，51.60 秒；包含真实 14236 帧顺序数据、OCR 暂停、过滤限流重试。 +- 补充重试额度用例后：`uv run pytest tests/test_adaptive_pool.py -q`，22 passed，0.27 秒。连续限流后的第二轮 map 保持 1 并发，进度序号重新从 1 开始。`git diff --check` 通过。 +- 状态：修复及验证完成，纳入 `fix/review-improvements` 分支；未部署。尚未对真实服务配额下的吞吐做性能结论；已发出请求的终止依赖节点自身超时。 + ## 审查基线 - 修复前全套测试：`uv run pytest`，369 passed、6 skipped，76.75 秒。 diff --git a/nodes/adaptive_pool.py b/nodes/adaptive_pool.py index 5f163ec..938c702 100644 --- a/nodes/adaptive_pool.py +++ b/nodes/adaptive_pool.py @@ -1,27 +1,23 @@ """自适应线程池。 -用于对耗时的独立子任务（如逐帧 VLM OCR）做弹性并发加速： -- 以滚动时间窗口统计已完成任务的平均响应时间； -- 窗口内平均响应 < fast_threshold（默认 0.3s）→ 增加 1 个工作线程（上限 max_workers）； -- 窗口内平均响应 > slow_threshold（默认 1.0s）→ 减少 1 个工作线程（下限 min_workers）。 +用于逐帧 VLM OCR、逐条 LLM 判定等独立 I/O 子任务： +- 从 min_workers 起步，每个时间窗口按平均单任务耗时增减目标并发； +- 快响应增加 1 个在途任务额度，慢响应减少 1 个，受 min/max_workers 限制； +- 限流降低有效上限，发生错误的窗口禁止扩容，干净窗口逐步恢复上限。 -线程数从 min_workers（默认 1）起步，按实测负载自适应：服务端空闲（响应快） -就加大并发，服务端变慢就退避，避免盲目并发压垮上游（如本地 Ollama）。 - -线程安全说明：worker 会在多个线程中并发调用，调用方需保证 worker 无共享 -可变状态（registry 处理器是纯函数，符合要求）；结果按输入顺序返回。 +执行器按需创建线程并复用；map 只提交目标额度内的任务，不把整批输入压入 +执行器队列。缩容立即限制后续提交，已发出的请求允许完成，不强制中断。 +结果与进度由 map 所在线程统一收集，worker 只负责处理输入；返回结果保持 +输入顺序，worker 异常作为结果交给调用方决定是否重试。 """ from __future__ import annotations -import queue import threading import time +from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait from typing import Callable -# 停止哨兵：压入队列让空闲工作线程退出（用于缩容）。 -_POISON = object() - def decide( current: int, @@ -31,11 +27,7 @@ def decide( fast_threshold: float, slow_threshold: float, ) -> int: - """根据窗口平均响应时间返回调整后的目标线程数（纯决策函数）。 - - 响应快（avg < fast_threshold）且未达上限 → 加 1；响应慢 - （avg > slow_threshold）且未达下限 → 减 1；其余情况保持不变。 - """ + """按平均耗时返回目标并发：快则 +1、慢则 -1，达到上下界后保持。""" if avg < fast_threshold and current < max_workers: return current + 1 if avg > slow_threshold and current > min_workers: @@ -44,7 +36,7 @@ def decide( class AdaptiveThreadPool: - """自适应线程池：单次 map 按输入顺序返回全部结果。""" + """有界自适应执行器；允许顺序重复 map，不允许同一实例并行调用 map。""" def __init__( self, @@ -57,7 +49,7 @@ class AdaptiveThreadPool: clock=time.monotonic, on_progress: Callable[[int, int, float, float, int], None] | None = None, ) -> None: - """初始化；clock 可注入便于测试；on_progress(done,total,rate) 每次完成回调。""" + """保存 worker、窗口策略及进度回调；clock 可在测试中注入。""" self._worker = worker self.min_workers = max(1, min_workers) self.max_workers = max(self.min_workers, max_workers) @@ -65,181 +57,151 @@ class AdaptiveThreadPool: 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._map_lock = threading.Lock() + # 保留 cancel 的调用约定：抑制回调，worker 自行检测暂停并返回异常。 + # OCR 因此仍能为每个输入得到结果，同时不会产生上万条暂停进度日志。 self._cancel_event = threading.Event() - # 有效最大线程数：初始等于 max_workers；消费错误（如 API 限流）时 - # report_failure 临时收紧，连续无错误窗口后逐步回升——并发自适应配额。 - self._effective_max_workers = max_workers - # 当前窗口内消费错误计数：窗口评估时无错误才允许恢复有效上限。 + # 有效上限跨 map 保留：LLM 失败条目重试时继续遵守已收紧的配额。 + self._effective_max_workers = self.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._elapsed_sum = 0.0 self._window_times: list[float] = [] - # 最近一次窗口评估的平均单任务耗时（秒）：供进度回调诊断使用， - # 与扩缩容决策共用同一依据；窗口尚未评估时为 None（回退累计平均）。 self._window_avg_time: float | None = None - def _run(self) -> None: - """工作线程主循环：取任务 → 执行 → 记录耗时并自适应评估。""" + def _run(self, item) -> tuple[object, float]: + """执行一次 worker，保留异常对象并记录真实单任务耗时。""" + start = self._clock() 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()) + result = self._worker(item) + except Exception as exc: + result = exc + return result, self._clock() - start 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 + """收集窗口耗时并调整额度；与 worker 报告限流共用锁，避免决策竞态。""" with self._lock: - current = self._target_workers - # 窗口内无消费错误 → 有效上限逐步回升（错误降下来的并发慢慢恢复）。 - if ( - self._window_failures == 0 - and self._effective_max_workers < self.max_workers - ): + self._window_times.append(elapsed) + now = self._clock() + if now - self._window_start < self.window_seconds: + return + avg = sum(self._window_times) / len(self._window_times) + self._window_start = now + self._window_times.clear() + self._window_avg_time = avg + had_failures = self._window_failures > 0 + # 有错误的窗口禁止恢复上限或增加并发；干净窗口每次只恢复 1。 + if not had_failures 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, + target = decide( + self._target_workers, avg, self.min_workers, + self._effective_max_workers, self.fast_threshold, self.slow_threshold, + ) + if had_failures: + target = min(target, self._target_workers) + self._target_workers = max( + self.min_workers, min(target, self._effective_max_workers) ) - ) def _current_avg_time(self, elapsed_total: float) -> float: - """返回供进度回调展示的平均单任务耗时（秒）。 + """返回最近窗口均值；窗口未满时返回实际单任务耗时均值。 - 优先使用最近一次窗口评估的平均耗时（与扩缩容决策同一依据）； - 窗口尚未评估过时回退为启动至今的累计平均，避免无数据可看。 + elapsed_total 保留旧调用签名；墙钟时间除以任务数会受并发倍数影响， + 因此均值改用 worker 耗时总和计算。 """ if self._window_avg_time is not None: return self._window_avg_time - return elapsed_total / max(self._completed, 1) + return self._elapsed_sum / 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 + self._target_workers = max( + self.min_workers, min(target, self._effective_max_workers) + ) def cancel(self) -> None: - """请求取消本批任务：后续完成的任务不再触发进度回调。 + """抑制本批后续进度回调；暂停与输入结果处理仍交给 worker。 - 供调用方在工作线程内检测到外部信号（如暂停）时调用，抑制暂停后 - 队列中剩余任务快速退出导致的进度日志井喷；下一批 map 自动重置。 + 下一批 map 重置此标记，保持 OCR 暂停后继续及 LLM 重试的调用约定。 """ 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) + self._effective_max_workers = max( + self.min_workers, self._effective_max_workers - 1 + ) + # 上限 20 -> 19 不意味着当前 1 个任务应扩到 19 个。 + self._target_workers = min(self._target_workers, 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] + """有界提交并按输入顺序返回结果；上下文退出时回收执行器线程。""" + if not self._map_lock.acquire(blocking=False): + raise RuntimeError("同一自适应线程池不能同时执行多个 map") + try: + self._completed = 0 + self._total = len(items) + self._started_at = self._clock() + self._elapsed_sum = 0.0 + self._cancel_event.clear() + with self._lock: + # 新批次重新计时，空闲时间不构成快响应窗口；错误上限仍保留。 + self._window_start = self._started_at + self._window_times.clear() + self._window_avg_time = None + self._target_workers = self.min_workers + results = [None] * self._total + pending = {} + iterator = iter(enumerate(items)) + exhausted = False + with ThreadPoolExecutor(max_workers=self.max_workers) as executor: + while True: + # 与 report_failure 共用锁：每次提交都依据最新额度。 + # pending 包含尚未收集的完成任务，限制只会更保守，不会超额。 + with self._lock: + while not exhausted and len(pending) < self._target_workers: + try: + seq, item = next(iterator) + except StopIteration: + exhausted = True + break + future = executor.submit(self._run, item) + pending[future] = seq + self.max_concurrency = max(self.max_concurrency, len(pending)) + if not pending: + break + done, _ = wait(pending, return_when=FIRST_COMPLETED) + for future in sorted(done, key=pending.__getitem__): + seq = pending.pop(future) + result, elapsed = future.result() + results[seq] = result + self._completed += 1 + self._elapsed_sum += elapsed + # 单一收集线程串行回调和统计，不再发生 queue.task_done + # 因回调异常未执行而使整批永久挂起的问题。 + 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) + return results + finally: + self._map_lock.release() diff --git a/tests/test_adaptive_pool.py b/tests/test_adaptive_pool.py index 6716333..757191e 100644 --- a/tests/test_adaptive_pool.py +++ b/tests/test_adaptive_pool.py @@ -5,6 +5,7 @@ （通过注入假时钟做确定性验证）。 """ +import threading import time from nodes.adaptive_pool import AdaptiveThreadPool, decide @@ -76,19 +77,21 @@ def test_pool_progress_reports_window_avg_after_first_window() -> None: """ clock = FakeClock() seen: list[tuple[int, int, float, float, int]] = [] + def progress(done, total, rate, avg_time, workers): + # 推进窗口时钟但不增加 worker 耗时，确定性触发快响应扩容。 + seen.append((done, total, rate, avg_time, workers)) + if done == 1: + clock.advance(11) + pool = AdaptiveThreadPool( worker=lambda item: item, min_workers=1, max_workers=16, window_seconds=10.0, fast_threshold=0.3, - clock=clock, - on_progress=lambda done, total, rate, avg_time, workers: seen.append( - (done, total, rate, avg_time, workers) - ), + clock=clock, on_progress=progress, ) - clock.advance(11) # 首个任务完成即越过窗口 → 触发评估。 pool.map(list(range(5))) - # 第一个完成的任务回调在窗口评估前：回退累计平均（>0）。 - assert seen[0][3] > 0 + # 第一个完成的任务回调在窗口评估前：真实 worker 耗时为 0，均值也为 0。 + assert seen[0][3] == 0.0 # 窗口评估后：回调携带最近窗口平均（≈0.0），且线程已扩容到 2。 assert any(item[3] == 0.0 for item in seen) assert any(item[4] == 2 for item in seen) @@ -165,9 +168,9 @@ def test_pool_grows_when_fast() -> None: min_workers=1, max_workers=16, window_seconds=10.0, fast_threshold=0.3, clock=clock, + on_progress=lambda done, *_: clock.advance(11) if done == 1 else None, ) - # 拨快时钟越过窗口：首个任务完成即触发评估 → 平均响应≈0 < 0.3 → +1 线程。 - clock.advance(11) + # 首个任务完成后越过窗口 → 平均响应≈0 < 0.3 → +1 个在途额度。 pool.map(list(range(4))) assert pool.max_concurrency == 2 @@ -184,11 +187,8 @@ def test_pool_shrink_when_slow() -> None: 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() + # 缩容只调整提交额度；实际积压任务行为由下方并发回归测试验证。 + assert pool._target_workers == 1 def test_resize_shrink_idempotent() -> None: @@ -199,12 +199,8 @@ def test_resize_shrink_idempotent() -> None: 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() + # 重复缩容不会遗留哨兵影响下一批，真实 map 必须返回全部输入。 + assert pool.map(list(range(20))) == list(range(20)) def test_pool_survives_mixed_grow_shrink() -> None: @@ -259,6 +255,110 @@ def test_pool_effective_max_recovers_after_clean_window() -> None: assert pool._effective_max_workers == 16 +def test_failure_at_single_worker_never_expands() -> None: + """真实 map 内报告限流：1 个在途任务不能因上限 20 变 19 而突然扩容。""" + targets = [] + + def worker(item): + if item == 0: + pool.report_failure() + targets.append(pool._target_workers) + return item + + pool = AdaptiveThreadPool(worker=worker, max_workers=20, window_seconds=1000) + assert pool.map(list(range(30))) == list(range(30)) + assert targets == [1] + + +def test_failure_shrink_limits_backlogged_work() -> None: + """积压任务中从 4 降到 1：已开始任务可完成，后续实际并发必须为 1。""" + barrier = threading.Barrier(4, timeout=3) + reduced = threading.Event() + lock = threading.Lock() + active = 0 + subsequent_peaks = [] + + def worker(item): + nonlocal active + with lock: + active += 1 + if item >= 5: + subsequent_peaks.append(active) + try: + if 1 <= item <= 4: + barrier.wait() + if item == 1: + # 连续限流将有效上限压到下限，存量请求不强制中断。 + for _ in range(3): + pool.report_failure() + reduced.set() + assert reduced.wait(3) + elif item >= 5: + # 模拟 I/O 等待，给其他工作线程实际进入任务的机会。 + time.sleep(0.005) + return item + finally: + with lock: + active -= 1 + + def progress(done, *_): + if done == 1: + pool._resize(4) + + pool = AdaptiveThreadPool(worker=worker, max_workers=4, window_seconds=1000, + on_progress=progress) + items = list(range(40)) + assert pool.map(items) == items + assert subsequent_peaks and max(subsequent_peaks) == 1 + + +def test_error_window_does_not_regrow() -> None: + """报告限流的同一窗口即使响应很快，也不能重新加并发。""" + clock = FakeClock() + pool = AdaptiveThreadPool(worker=lambda item: item, max_workers=4, clock=clock) + pool.report_failure() + clock.advance(11) + pool._tick(0.01) + assert pool._target_workers <= 1 + + +def test_retry_map_preserves_reduced_limit() -> None: + """两轮真实 map：首轮连续限流后，重试期间实际并发及进度都遵守新上限。""" + lock = threading.Lock() + active = 0 + peak = 0 + progress_counts = [] + + def worker(item): + nonlocal active, peak + with lock: + active += 1 + peak = max(peak, active) + try: + if item == "limited": + for _ in range(3): + pool.report_failure() + raise RuntimeError("rate limited") + time.sleep(0.002) + return item + finally: + with lock: + active -= 1 + + pool = AdaptiveThreadPool( + worker=worker, max_workers=4, window_seconds=1000, + on_progress=lambda done, *_: progress_counts.append(done), + ) + assert isinstance(pool.map(["limited"])[0], RuntimeError) + assert pool._effective_max_workers == 1 + # 主动申请扩容也受已收紧额度约束，下一轮不能重置有效上限。 + pool._resize(4) + assert pool._target_workers == 1 + assert pool.map(list(range(12))) == list(range(12)) + assert peak == 1 + assert progress_counts == [1] + list(range(1, 13)) + + def test_pool_decide_uses_effective_max() -> None: """扩容上限按有效最大线程数：错误窗口内即使响应快也不超过收紧后的上限。""" clock = FakeClock()