mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 14:24:15 +08:00
shy3130
5618b4ef1d
线上曾出现并发 LazyFrame.collect 触发 polars streaming 执行器死锁, 叠加 _write_lock 区间内做重活, 放大为全站请求冻结。本次按触发缩小、 爆炸半径收缩、自动恢复三层布防: - polars_guard: BoundedSemaphore 并发闸 (总闸 4 + 后台车道 2, 后台 先拿子闸再拿总闸防死锁); repository 18 处 collect 按交互/后台分级接入 - repository 写锁区间收缩: 分区合并移出锁外, 锁内 (mtime_ns,size) 指纹校验 + 3 次乐观重试, 失败回退锁内合并; 5 处 _write_lock 重构 - watchdog: 周期探测 collect 闸与全局写锁, 连续 2 次失败退出交由 supervisor 拉起 (可配置, 默认开) - polars >=1.44,<1.45 (1.44.1); 附并发压测脚本 scripts/stress_polars_concurrency.py 供复现验证 另: config 新增 polars_collect_permits / watchdog_* / strategy_run_all_workers / strategy_run_all_first_return_s 旋钮 (后两者供后续 run_all 优化提交使用, 默认保持旧行为基准)。
116 lines
4.3 KiB
Python
116 lines
4.3 KiB
Python
"""后端自愈看门狗。
|
||
|
||
2026-09-07 事故形态: polars 并发死锁把线程悬死在 collect 内部 (0 CPU 永久
|
||
挂起), 其中持锁者让 _write_lock 永久被占, 所有请求线程排队冻结, 只能人工
|
||
重启。并发闸 (app.polars_guard) 与写锁瘦身 (repository 乐观并发) 分别削减
|
||
触发概率与扩散半径; 看门狗是最后一层兜底 —— 探测走与事故相同的共享资源
|
||
路径 (collect 闸 + 全局写锁), 连续 N 次超时即判定进程已僵死, 主动退出交由
|
||
supervisor / Docker restart / dev 脚本拉起, 把恢复时间从"人工发现"缩短到
|
||
约一分钟。
|
||
|
||
误伤防护: 探测本身是毫秒级微型 collect + 1s 写锁试探, 阈值要求连续失败
|
||
(默认 2 次 × 15s 超时), 高负载下"慢而未死"不会触发。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import contextlib
|
||
import logging
|
||
import os
|
||
import threading
|
||
from collections.abc import Callable
|
||
|
||
import polars as pl
|
||
|
||
from app.config import settings
|
||
from app.polars_guard import guarded_collect
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def default_probe(write_lock: threading.Lock | None = None) -> None:
|
||
"""探测关键共享资源: polars collect 闸 + 仓库全局写锁。
|
||
|
||
任一被悬死线程占住即超时 — 正是 2026-09-07 冻结事故中被毒化的两条路径。
|
||
"""
|
||
guarded_collect(pl.LazyFrame({"probe": [1]}).sum())
|
||
if write_lock is not None:
|
||
acquired = write_lock.acquire(timeout=1.0)
|
||
if not acquired:
|
||
raise TimeoutError("repository write lock unavailable")
|
||
write_lock.release()
|
||
|
||
|
||
class HealthWatchdog:
|
||
"""周期探测; 连续 failure_threshold 次失败后调用 exit_cb(退出码)。"""
|
||
|
||
def __init__(
|
||
self,
|
||
probe: Callable[[], None],
|
||
*,
|
||
exit_cb: Callable[[int], None],
|
||
interval_s: float | None = None,
|
||
probe_timeout_s: float | None = None,
|
||
failure_threshold: int | None = None,
|
||
) -> None:
|
||
self._probe = probe
|
||
self._exit_cb = exit_cb
|
||
self._interval_s = settings.watchdog_interval_s if interval_s is None else interval_s
|
||
self._probe_timeout_s = (
|
||
settings.watchdog_probe_timeout_s if probe_timeout_s is None else probe_timeout_s
|
||
)
|
||
self._failure_threshold = (
|
||
settings.watchdog_failure_threshold if failure_threshold is None else failure_threshold
|
||
)
|
||
self._consecutive_failures = 0
|
||
self._task: asyncio.Task | None = None
|
||
|
||
async def _loop(self) -> None:
|
||
while True:
|
||
try:
|
||
await asyncio.wait_for(
|
||
asyncio.to_thread(self._probe), timeout=self._probe_timeout_s
|
||
)
|
||
self._consecutive_failures = 0
|
||
except BaseException as exc: # 探测任何异常都算失败 (含 to_thread 超时)
|
||
self._consecutive_failures += 1
|
||
logger.error(
|
||
"watchdog probe failed (%d/%d): %r",
|
||
self._consecutive_failures,
|
||
self._failure_threshold,
|
||
exc,
|
||
)
|
||
if self._consecutive_failures >= self._failure_threshold:
|
||
logger.critical(
|
||
"watchdog: backend wedged (probe failed %d consecutive times); "
|
||
"exiting for supervisor restart",
|
||
self._consecutive_failures,
|
||
)
|
||
self._exit_cb(70)
|
||
return
|
||
await asyncio.sleep(self._interval_s)
|
||
|
||
def start(self) -> None:
|
||
if self._task is None or self._task.done():
|
||
self._task = asyncio.create_task(self._loop(), name="health-watchdog")
|
||
|
||
async def stop(self) -> None:
|
||
if self._task is not None:
|
||
self._task.cancel()
|
||
with contextlib.suppress(asyncio.CancelledError):
|
||
await self._task
|
||
self._task = None
|
||
|
||
|
||
def start_watchdog(app_state, repo) -> HealthWatchdog | None:
|
||
"""lifespan 启动钩子; 返回实例挂到 app.state.watchdog 便于关闭。"""
|
||
if not settings.watchdog_enabled:
|
||
return None
|
||
write_lock = getattr(repo, "_write_lock", None)
|
||
watchdog = HealthWatchdog(
|
||
lambda: default_probe(write_lock),
|
||
exit_cb=lambda code: os._exit(code),
|
||
)
|
||
watchdog.start()
|
||
return watchdog
|