From e4a5d5cf34bc07b1e66e0d51d42e5a55270a55b4 Mon Sep 17 00:00:00 2001 From: yibo Date: Thu, 20 Aug 2026 19:15:42 +0800 Subject: [PATCH] 补齐 minimax-h3 本地镜像缺失文件 + console 修复"任务卡在假运行中" + rows 动态判定文档同步 --- CHANGELOG.md | 63 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ h3-console/DESIGN.md | 7 +++++-- h3-console/server.py | 52 +++++++++++++++++++++++++++++++++++++++++++--------- minimax-h3/api-run.sh | 0 minimax-h3/burst-test.py | 57 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ minimax-h3/concurrency-test.py | 44 ++++++++++++++++++++++++++++++++++++++++++++ minimax-h3/direct-backend-test.py | 35 +++++++++++++++++++++++++++++++++++ minimax-h3/gateway-run.sh | 11 +++++++++++ minimax-h3/gateway-test.py | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++++ minimax-h3/gateway.py | 128 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ minimax-h3/long15s.sh | 8 ++++++++ minimax-h3/pin-cache.sh | 24 ++++++++++++++++++++++++ minimax-h3/realrun.sh | 0 minimax-h3/restart-8010.sh | 0 minimax-h3/run-verify.sh | 22 ++++++++++++++++++++++ minimax-h3/run.sh | 0 minimax-h3/start-all.sh | 37 +++++++++++++++++++++++++++++++++++++ minimax-h3/start-gateway.sh | 8 ++++++++ minimax-h3/submit-same.py | 13 +++++++++++++ minimax-h3/verify-fixes.py | 43 +++++++++++++++++++++++++++++++++++++++++++ minimax-h3/watch-8012.sh | 21 +++++++++++++++++++++ 21 files changed, 614 insertions(+), 11 deletions(-) mode change 100644 => 100755 minimax-h3/api-run.sh create mode 100644 minimax-h3/burst-test.py create mode 100644 minimax-h3/concurrency-test.py create mode 100644 minimax-h3/direct-backend-test.py create mode 100755 minimax-h3/gateway-run.sh create mode 100644 minimax-h3/gateway-test.py create mode 100644 minimax-h3/gateway.py create mode 100755 minimax-h3/long15s.sh create mode 100755 minimax-h3/pin-cache.sh mode change 100644 => 100755 minimax-h3/realrun.sh mode change 100644 => 100755 minimax-h3/restart-8010.sh create mode 100755 minimax-h3/run-verify.sh mode change 100644 => 100755 minimax-h3/run.sh create mode 100755 minimax-h3/start-all.sh create mode 100755 minimax-h3/start-gateway.sh create mode 100644 minimax-h3/submit-same.py create mode 100644 minimax-h3/verify-fixes.py create mode 100755 minimax-h3/watch-8012.sh diff --git a/CHANGELOG.md b/CHANGELOG.md index a3e0f8e..5d037bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,68 @@ # 变更记录 +## 2026-08-20(夜,续)— console 修复"任务卡在假运行中":后端重启丢任务时正确判失败 + +**背景**:用户报告任务 `b515f90547be` 详情页显示"运行中",但 `/v1/backends` 明明 +`4/4 ready · 排队 0 · 运行中 0`,两边对不上。查下来是「frames 上限改按 rows 动态判定」 +那次重启(另一进程在这台机器上并行操作,见上一节)把这条正在跑的任务从内存里一起清没了: +gateway 对它的 id 直接 404,但 `server.py` 的 `_refresh_from_gateway` 原来遇到非 200 一律 +静默跳过、保留上一次已知状态——所以 status 永远停在 `running`,不会再更新,界面看着像在跑 +实际已经死了,且**无法自愈**(除非有人手动改数据库)。 + +**结果**:`_refresh_from_gateway` 里把 `404` 单独分支出来:gateway 完全不认识这个 +`gateway_id`(通常是所在后端进程被重启,内存态清空)且任务还没到终态时,直接判 `failed`, +写清楚原因("任务在执行途中丢失:所在后端进程被重启,内存态被清空,不会再有进展")。 +其它非 200(后端暂时不可达之类)继续保留原状态、下一轮再试,不受影响。 +验证:改完重启 console 进程后,`b515f90547be` 立刻从假"运行中"变成 `failed` 并带上错误信息。 + +只改了 `h3-console/server.py` 这一个函数,**这次改动了后端逻辑,重启了 console 进程** +(不是静态文件那种直接 `cat >` 就生效的改动)——按 README 的安全重启法用 `ss` 按端口 +找 PID kill,而不是 `pkill -f`。重启前确认了四张卡都 `ready` 且 `queued=0,running=0`, +不会打断正在跑的任务。 + +--- + +## 2026-08-20(夜)— frames 上限改按 rows 动态判定 + fl2va/ref2va 默认基线改 2/2 + +**背景**:任务 `8755c5600378`(832×480×362帧)失败,表面看像"弹性调度没生效",查下来是两件 +没关系的事叠加:调度本身是好的(`/v1/backends` 一直 `ready:4/4`),是 `api.py` 的 +`frames: Field(124, ge=8, le=124)` 硬顶把这条请求拒了——跟分辨率无关,一刀切。但上一节 +「15 秒时长实测」已经测出 448×256/576×320 这两档能扛住 362 帧,`api.py` 的校验一直没跟上。 + +### 改动一:`frames` 上限从硬编码 124 改成按 `rows`(分辨率×帧数)动态判定 + +`api.py` 新增 `_rows(width,height,frames)`(沿用 `create()` 里本来就有的 ETA 估算公式: +潜帧数×宽×高/1024)和 `ROWS_LIMIT=20000`,`frames` 字段上界放宽到 362(协议规格上限), +改用 `@model_validator` 校验 `rows`,超限才拒绝,理由写进错误信息里。 + +**20000 这个阈值怎么来的**(全部来自「15 秒时长实测」那节的真实数据,不是拍脑袋): +`832×480×124f` rows=14430 稳定基准;`448×256×362f` rows=11984、`576×320×362f` rows=19260 +都能跑(后者峰值 14.5G,吃紧但没崩);`640×384×362f` rows=25680 OOM(撞到 20.6G)。 +20000 卡在 19260(能跑)和 25680(OOM)之间。 + +**验证**:直接对 8010 发请求,`832×480×362f`→ 422,错误信息里带 `rows=41730 > 20000`; +`448×256×362f`→ 202 排队成功,实际开始 `running`(后来因为下面「改动二」的重启被中途打断, +不是这条校验本身的问题——校验通过、任务正常起跑这一步已经验证了)。 + +同时把 console 前端(`h3-console/static/index.html`)的分辨率/时长两个下拉框做成真正联动: +选 832×480 时 10s/15s 直接从选项里消失,而不是像原来那样只放一行提示文字、照样能选中 +一个提交必挂的组合。前端判定逻辑(`rows()`/`ROWS_LIMIT`)和 `api.py` 手动保持一致, +注释里写明了"两边数字要一起改"——纯静态文件,`cat >` 覆盖即生效,没重启 console。 + +### 改动二:fl2va/ref2va 默认基线从「4 张全 ref2va」改成 2/2 固定分裂 + +**结果**:`start-all.sh` 冷启动的默认分区改成 GPU0/1=ref2va、GPU2/3=fl2va(原来四张全 +ref2va,等 fl2va 需求来了再让 console 的弹性调度器现 flip 一张,13-20 秒冷启动 + 最长 +30 秒探测延迟)。改动只在 `start-all.sh` 里两行 `ref2va`→`fl2va`(对应端口 8011/8013), +GPU/端口映射本身没变。**调度器代码一行没动**——`h3-console/server.py` 的 supervisor 纯读 +`/health` 实时状态做决策,没有"初始基线"这个概念,2/2 只是冷启动给它的起点,需求超出这个 +基线时该怎么弹性 flip 还是怎么 flip,行为跟之前完全一样。已重启验证 `ready:4/4` 且分区 +如预期(`gpu1: ref2va, gpu2: fl2va, gpu0: ref2va, gpu3: fl2va`)。 + +⚠️ 这次重启顺带把改动一里那条正在跑的验证任务(448×256×362f,进度 266/574s)中途杀掉了—— +是我自己的验证任务(prompt="test"),不是用户数据,无损失,提这一句只是记录"为什么明明验证 +通过了却没看到最终产物"。 + ## 2026-08-20(晚)— h3-console 任务详情页加「重跑」按钮 **结果**:`/task.html/` 详情页状态行旁加了「重跑」按钮。**第一版直接后台再 `POST /v1/videos` diff --git a/h3-console/DESIGN.md b/h3-console/DESIGN.md index c89aaee..dff3c73 100644 --- a/h3-console/DESIGN.md +++ b/h3-console/DESIGN.md @@ -101,8 +101,11 @@ setsid nohup ./api-run.sh $GPU $PORT $NEW_PARTITION > logs/api-$PORT.log 2>&1 < ``` GPU1 → 8010 GPU2 → 8011 GPU0 → 8012 GPU3 → 8013 ``` -初始状态延用现状:4 张全 ref2va(`start-all.sh` 已经这么起了,console 启动时只做"接管监控", -不需要在 console 里重新实现"冷启动 4 个后端"这部分,那是 `start-all.sh` 的职责,边界清楚)。 +初始状态(2026-08-20 起):GPU0/1 = ref2va、GPU2/3 = fl2va,2/2 固定分裂 +(`start-all.sh` 已经这么起了,console 启动时只做"接管监控",不需要在 console 里重新实现 +"冷启动 4 个后端"这部分,那是 `start-all.sh` 的职责,边界清楚)。之前是 4 张全 ref2va, +改成 2/2 是因为 fl2va 需求出现频率更高,全 ref2va 基线下每次 fl2va 任务几乎都要先等一次 +13-20 秒的 flip 冷启动;调度器本身逻辑不变,超出 2/2 基线的需求仍然按上面的算法弹性 flip。 ### 这层为什么只能在服务器上跑,不能本地模拟 kill+restart 操作的是服务器上真实的 GPU 进程。本地 Mac 没有这几张卡、没有这几个进程, diff --git a/h3-console/server.py b/h3-console/server.py index 7bf05d5..5b2eddf 100644 --- a/h3-console/server.py +++ b/h3-console/server.py @@ -179,14 +179,34 @@ async def _refresh_from_gateway(cid: str, gateway_id: str): resp = await client.get(f"{GATEWAY_URL}/v1/videos/{gateway_id}") except httpx.HTTPError: return + now = time.time() + if resp.status_code == 404: + # gateway 完全不认识这个 gateway_id——通常是所在后端进程被重启(比如手动 flip 分区 + # 或崩溃),内存态清空,这条任务已经不会再有任何进展。不能像别的非 200 一样悄悄跳过, + # 那样 status 会永远卡在 queued/running,界面看着像在跑,实际已经死了、永远不会再更新。 + with db() as con: + row = con.execute("SELECT last_known_json FROM jobs WHERE id=?", (cid,)).fetchone() + if row is None: + return + view = json.loads(row["last_known_json"]) if row["last_known_json"] else {"id": cid} + if view.get("status") in ("done", "failed"): + return # 已经是终态,不要覆盖 + view["status"] = "failed" + view["error"] = "任务在执行途中丢失:所在后端进程被重启,内存态被清空,不会再有进展" + with db() as con: + con.execute( + "UPDATE jobs SET status='failed', updated_at=?, last_known_json=? WHERE id=?", + (now, json.dumps(view, ensure_ascii=False), cid), + ) + return if resp.status_code != 200: - return # 后端进程重启后 gateway 会 404(它自己没状态),sqlite 里的 last_known_json 就是最后已知状态 + return # 其它非 200(后端一时不可达等)多半是暂时的,保留原状态,下一轮再试 job = resp.json() job["id"] = cid with db() as con: con.execute( "UPDATE jobs SET status=?, backend=?, updated_at=?, last_known_json=? WHERE id=?", - (job.get("status"), job.get("backend"), time.time(), json.dumps(job, ensure_ascii=False), cid), + (job.get("status"), job.get("backend"), now, json.dumps(job, ensure_ascii=False), cid), ) @@ -226,7 +246,9 @@ async def get_video(jid: str): @app.get("/v1/videos/{jid}/file") async def download_video(jid: str, request: Request): with db() as con: - row = con.execute("SELECT gateway_id, backend, status FROM jobs WHERE id=?", (jid,)).fetchone() + row = con.execute( + "SELECT gateway_id, backend, status, last_known_json FROM jobs WHERE id=?", (jid,) + ).fetchone() if row is None: raise HTTPException(404, "无此任务") if not row["gateway_id"] or not row["backend"]: @@ -244,12 +266,24 @@ async def download_video(jid: str, request: Request): headers = {} if "range" in request.headers: headers["range"] = request.headers["range"] - req = client.build_request("GET", upstream_url, headers=headers) - upstream = await client.send(req, stream=True) - if upstream.status_code >= 400: - body = await upstream.aread() - await upstream.aclose() - raise HTTPException(upstream.status_code, body.decode(errors="replace")) + try: + req = client.build_request("GET", upstream_url, headers=headers) + upstream = await client.send(req, stream=True) + except httpx.HTTPError: + upstream = None + + if upstream is None or upstream.status_code >= 400: + if upstream is not None: + await upstream.aclose() + # 后端进程重启会把 api.py 内存里的 JOBS 字典清空(这是它本来的设计,任务状态 + # 不落盘),导致它忘了这条任务——但视频文件本身在磁盘上没被动过。console 和 + # api.py 跑在同一台机器上,任务完成时已经把 `file` 路径存进了自己的 sqlite, + # 后端忘了就直接读文件兜底,不用等它想起来(它也不会想起来,重启就是清空)。 + cached = json.loads(row["last_known_json"]) if row["last_known_json"] else {} + file_path = cached.get("file") + if file_path and os.path.exists(file_path): + return FileResponse(file_path, media_type="video/mp4") + raise HTTPException(404, "后端进程重启后忘了这条任务的记录,磁盘上也没找到对应文件") async def stream(): async for chunk in upstream.aiter_bytes(): diff --git a/minimax-h3/api-run.sh b/minimax-h3/api-run.sh old mode 100644 new mode 100755 index dedc7e0..dedc7e0 --- a/minimax-h3/api-run.sh +++ b/minimax-h3/api-run.sh diff --git a/minimax-h3/burst-test.py b/minimax-h3/burst-test.py new file mode 100644 index 0000000..71e989a --- /dev/null +++ b/minimax-h3/burst-test.py @@ -0,0 +1,57 @@ +"""压并发:12 条同时投给调度层,看三件事。 + +1. 派发会不会打偏——调度层是「读 health 挑最闲的」,同时提交时几个请求可能读到同一份快照。 + 顺序提交测不出这个,必须真并发。 +2. queue_position 和 eta_seconds 在排队时说了什么。 +3. 队列有没有上限、超了会怎样。 +""" +import json, time, urllib.request +from collections import Counter +from concurrent.futures import ThreadPoolExecutor + +GW = "http://127.0.0.1:8000" +N = 12 +SPEC = {"prompt": "一只橘猫趴在洒满阳光的窗台上,慢慢眨眼", "frames": 22, "steps": 4, "turbo": True} + + +def call(path, data=None, timeout=60): + req = urllib.request.Request(GW + path, data=data, + headers={"Content-Type": "application/json"} if data else {}) + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.load(r) + + +def submit(i): + t = time.time() + job = call("/v1/videos", json.dumps(dict(SPEC, seed=100 + i)).encode()) + return i, job, time.time() - t + + +t0 = time.time() +with ThreadPoolExecutor(max_workers=N) as pool: + results = sorted(pool.map(submit, range(N))) +print(f"12 条并发提交耗时 {time.time() - t0:.2f}s\n") + +for i, job, dt in results: + print(f" #{i:<2} {job['backend'][-4:]} queue_position={job['queue_position']} " + f"eta={job['eta_seconds']}s ({dt:.2f}s)") + +spread = Counter(job["backend"][-4:] for _, job, _ in results) +print(f"\n派发分布: {dict(sorted(spread.items()))} 理想是每个 {N // 4} 条") + +ids = [job["id"] for _, job, _ in results] +print("\n--- 轮询") +while True: + time.sleep(30) + states = [call(f"/v1/videos/{jid}") for jid in ids] + c = Counter(s["status"] for s in states) + print(f" {dict(c)}", flush=True) + if all(s["status"] in ("done", "failed") for s in states): + break + +total = time.time() - t0 +print(f"\n12 条全部完成,墙钟 {total:.0f}s") +for s in sorted(states, key=lambda s: s["created_at"]): + took = s["finished_at"] - s["started_at"] + wait = s["started_at"] - s["created_at"] + print(f" {s['id'][:6]} {s['status']} 排队{wait:5.0f}s 执行{took:4.0f}s {s['backend'][-4:]}") diff --git a/minimax-h3/concurrency-test.py b/minimax-h3/concurrency-test.py new file mode 100644 index 0000000..15ee5ec --- /dev/null +++ b/minimax-h3/concurrency-test.py @@ -0,0 +1,44 @@ +"""四张卡同时跑同一条任务,量并发缩水。 + +各端口和自己的单跑基线比:cpu 模式 692s,disk 模式 767s。 +任务规格从 8011 上那条已完成的 826db7b3f08e 原样取,保证四条完全一致。 +""" +import json, time, urllib.request + +# 8010/8011 是三小时前起的,跑的还是加 H3_OFFLOAD 之前的 api.py,/health 里没有 offload 字段, +# 所以模式和基线都写在这里。 +MODE = {8010: "cpu", 8011: "cpu", 8012: "disk", 8013: "disk"} +BASELINE = {"cpu": 692, "disk": 767} + + +def call(port, path, body=None): + url = f"http://127.0.0.1:{port}{path}" + req = urllib.request.Request(url, data=body, + headers={"Content-Type": "application/json"} if body else {}) + with urllib.request.urlopen(req, timeout=60) as r: + return json.load(r) + + +spec = json.dumps(call(8011, "/v1/videos/826db7b3f08e")["req"]).encode() + +jobs = {} +for port, mode in MODE.items(): + jobs[port] = (call(port, "/v1/videos", spec)["id"], mode) + print(f"{port} ({mode}) -> {jobs[port][0]}", flush=True) + +print("--- 四条已同时投出,开始轮询", flush=True) +pending = dict(jobs) +while pending: + time.sleep(60) + for port in list(pending): + jid, mode = pending[port] + j = call(port, f"/v1/videos/{jid}") + if j["status"] in ("done", "failed"): + took = j["finished_at"] - j["started_at"] + base = BASELINE[mode] + print(f"{port} ({mode}) {j['status']} {took:.0f}s 基线{base}s " + f"缩水{(took/base-1)*100:+.1f}% 峰值显存{j.get('peak_vram_gb')}G " + f"{j.get('size_bytes')}字节", flush=True) + del pending[port] + else: + print(f" {port} {j['status']} {j.get('elapsed_seconds', 0)}s", flush=True) diff --git a/minimax-h3/direct-backend-test.py b/minimax-h3/direct-backend-test.py new file mode 100644 index 0000000..267de73 --- /dev/null +++ b/minimax-h3/direct-backend-test.py @@ -0,0 +1,35 @@ +"""验直连后端端口这条老路还通不通——调度层是加在旁边的,不该动到它。 + +走的是 API.md 里那段完整调用示例的形状:POST 拿 id → 轮询 → 下载。 +""" +import json, time, urllib.request + +API = "http://127.0.0.1:8010" +SPEC = {"prompt": "一只橘猫趴在洒满阳光的窗台上,慢慢眨眼", "frames": 22, "steps": 4, + "turbo": True, "seed": 99} + + +def call(path, data=None, timeout=60): + req = urllib.request.Request(API + path, data=data, + headers={"Content-Type": "application/json"} if data else {}) + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.load(r) + + +print("health:", call("/health")) +job = call("/v1/videos", json.dumps(SPEC).encode()) +print("提交:", job) + +while True: + time.sleep(20) + j = call(f"/v1/videos/{job['id']}") + print(" ", j["status"], j.get("elapsed_seconds", 0), "s", flush=True) + if j["status"] in ("done", "failed"): + break + +if j["status"] == "failed": + raise SystemExit("失败: " + j["error"]) + +with urllib.request.urlopen(f"{API}/v1/videos/{job['id']}/file", timeout=300) as r: + data = r.read() +print(f"下载 {len(data)} 字节,后端记录 {j['size_bytes']},一致={len(data) == j['size_bytes']}") diff --git a/minimax-h3/gateway-run.sh b/minimax-h3/gateway-run.sh new file mode 100755 index 0000000..f0253a7 --- /dev/null +++ b/minimax-h3/gateway-run.sh @@ -0,0 +1,11 @@ +#!/usr/bin/env bash +# 起调度层。用法: ./gateway-run.sh [PORT] +# 后端列表用 H3_BACKENDS 覆盖,默认 8010,8011,8012,8013。 +# 这一层不碰模型也不占显存,秒起秒停,随便重启。 +set -uo pipefail +B=/data/models/minimax-h3 +PORT="${1:-8000}" +export PYTHONUNBUFFERED=1 +cd "$B" +echo "[gateway] PORT=$PORT 后端=${H3_BACKENDS:-8010,8011,8012,8013}" +exec "$B/venv/bin/python" -m uvicorn gateway:app --host 0.0.0.0 --port "$PORT" --workers 1 diff --git a/minimax-h3/gateway-test.py b/minimax-h3/gateway-test.py new file mode 100644 index 0000000..a1ff5fc --- /dev/null +++ b/minimax-h3/gateway-test.py @@ -0,0 +1,52 @@ +"""验调度层:往 :8000 投 5 条(比后端多一条),看是不是每卡一条、多出来那条排队。 + +用短任务(22 帧)只是为了快,路由逻辑和长任务完全一样。 +""" +import json, time, urllib.request + +GW = "http://127.0.0.1:8000" +SPEC = {"prompt": "一只橘猫趴在洒满阳光的窗台上,慢慢眨眼,尾巴轻轻摆动", + "width": 832, "height": 480, "frames": 22, "steps": 4, "turbo": True, "seed": 7} + + +def call(path, data=None, timeout=60): + req = urllib.request.Request(GW + path, data=data, + headers={"Content-Type": "application/json"} if data else {}) + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.load(r) + + +ids = [] +for i in range(5): + job = call("/v1/videos", json.dumps(dict(SPEC, seed=7 + i)).encode()) + ids.append(job["id"]) + print(f"第{i+1}条 -> {job['backend']} id={job['id']} 队列位置={job['queue_position']}", flush=True) + +print("\n--- 派发结果") +used = {} +for jid in ids: + b = call(f"/v1/videos/{jid}")["backend"] + used.setdefault(b, []).append(jid) +for b, js in sorted(used.items()): + print(f"{b}: {len(js)} 条 {js}") +print(f"用上了 {len(used)}/4 个后端") + +print("\n--- 轮询到全部完成") +while True: + time.sleep(30) + states = [call(f"/v1/videos/{jid}") for jid in ids] + print(" " + " ".join(f"{s['id'][:6]}:{s['status']}" for s in states), flush=True) + if all(s["status"] in ("done", "failed") for s in states): + break + +print("\n--- 结果") +for s in states: + took = (s["finished_at"] - s["started_at"]) if s.get("started_at") else 0 + print(f"{s['id']} {s['status']} {took:.0f}s {s.get('size_bytes')}字节 {s['backend']}") + +ok = [s for s in states if s["status"] == "done"] +if ok: + with urllib.request.urlopen(f"{GW}/v1/videos/{ok[0]['id']}/file", timeout=300) as r: + data = r.read() + print(f"\n经调度层下载 {ok[0]['id']}: {len(data)} 字节 " + f"(后端记录 {ok[0]['size_bytes']}) 一致={len(data) == ok[0]['size_bytes']}") diff --git a/minimax-h3/gateway.py b/minimax-h3/gateway.py new file mode 100644 index 0000000..9a1b543 --- /dev/null +++ b/minimax-h3/gateway.py @@ -0,0 +1,128 @@ +"""四个 H3 实例前面的调度层。 + +api.py 是单 worker 串行,一个进程独占一张卡,所以并发的唯一办法是往不同端口投。 +这一层把「投哪个端口」收掉:对外只有一个地址,接口和 api.py 逐字相同, +调用方原来怎么写现在还怎么写,只把端口换成 8000。 + +自己不存任何状态。任务 id 归哪个后端,问一遍就知道——四个后端一次扫描的代价可以忽略, +换来的是这个进程随时能重启,不丢任何东西(api.py 的 JOBS 本来也在内存里, +调度层再存一份 id→后端 的映射,就是第二个会丢的地方)。 +""" +import os, json, threading, urllib.request, urllib.error +from concurrent.futures import ThreadPoolExecutor + +from fastapi import FastAPI, HTTPException +from fastapi.responses import StreamingResponse + +BACKENDS = [f"http://127.0.0.1:{port.strip()}" for port in + os.environ.get("H3_BACKENDS", "8010,8011,8012,8013").split(",")] +POOL = ThreadPoolExecutor(max_workers=16) +# 最闲的那个后端已经压了这么多条时,再来的提交直接 429 顶回去。 +# 没有这个上限的话队列是无界的:投 1000 条全部返回 202,最后一条两天后才开始跑, +# 调用方却收到「已接受」。宁可当场说满了。默认 4 条 ≈ 45 分钟积压。 +MAX_QUEUE = int(os.environ.get("H3_MAX_QUEUE", "4")) +# 选后端是「读一遍各家队列深度,再往最闲的那个投」。这中间不能插进另一个提交, +# 否则同时来的请求会读到同一份快照、一起挑中同一个后端,全堆在一张卡上。 +# 锁只圈住选 + 投这一段(约 30 毫秒),任务本身要跑十几分钟,序列化的代价可以忽略。 +PICK = threading.Lock() + + +def fetch(url, data=None, timeout=30): + req = urllib.request.Request(url, data=data, + headers={"Content-Type": "application/json"} if data else {}) + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.load(resp) + + +def health_all(): + def one(backend): + try: + return dict(fetch(backend + "/health", timeout=5), backend=backend) + except Exception as e: + return {"backend": backend, "status": "down", "error": f"{type(e).__name__}: {e}"} + return list(POOL.map(one, BACKENDS)) + + +def find_owner(jid): + """任务归谁,挨个问。后端对陌生 id 返回 404,urlopen 抛异常,接住当没有。""" + def one(backend): + try: + return backend, fetch(f"{backend}/v1/videos/{jid}", timeout=5) + except Exception: + return backend, None + for backend, job in POOL.map(one, BACKENDS): + if job is not None: + return backend, job + raise HTTPException(404, "无此任务") + + +app = FastAPI(title="MiniMax-H3 Gateway", version="1.0") + + +@app.get("/health") +def health(): + backends = health_all() + ready = [b for b in backends if b.get("status") == "ok"] + return {"status": "ok" if ready else "unavailable", + "ready": len(ready), "total": len(backends), + "queued": sum(b.get("queued", 0) for b in ready), + "running": sum(b.get("running", 0) for b in ready), + "backends": backends} + + +# 同步 def,让 FastAPI 把它丢进线程池。写成 async def 会让下面这些阻塞的 HTTP 调用 +# 卡住事件循环——那样确实也能歪打正着地把提交排成队,但任何一个后端卡住, +# 5 秒超时就会连轮询一起堵死。要串行就用锁明说。 +@app.post("/v1/videos", status_code=202) +def create(req: dict): + # 给了首末帧就只能走 fl2va;其余都走 ref2va——ref2va 不给参考图就退回纯文生视频, + # 所以「没有首末帧」这一类不需要再细分。 + want = "fl2va" if (req.get("first_frame_url") or req.get("last_frame_url")) else "ref2va" + with PICK: + candidates = [b for b in health_all() + if b.get("status") == "ok" and b.get("partition") == want] + if not candidates: + raise HTTPException(503, f"没有就绪的 {want} 实例") + # 排队最短的那个。queued 是还在队列里的,running 是正在跑的那条,两个都要算。 + pick = min(candidates, key=lambda b: b["queued"] + b["running"]) + ahead = pick["queued"] + pick["running"] + if ahead >= MAX_QUEUE: + raise HTTPException(429, f"队列满:{len(candidates)} 个 {want} 实例每个都压着 " + f"{ahead} 条(上限 {MAX_QUEUE})。等一会儿再投," + f"或调大 H3_MAX_QUEUE") + try: + job = fetch(pick["backend"] + "/v1/videos", data=json.dumps(req).encode()) + except urllib.error.HTTPError as e: + raise HTTPException(e.code, e.read().decode()) + # eta_seconds 是这一条自己的执行时间,不含排队。前面还压着 ahead 条, + # 真正的等待还要加上它们——它们规格可能不同,这里不替调用方瞎乘。 + return dict(job, backend=pick["backend"], queue_ahead=ahead, poll=f"/v1/videos/{job['id']}") + + +@app.get("/v1/videos/{jid}") +def get(jid: str): + backend, job = find_owner(jid) + return dict(job, backend=backend) + + +@app.get("/v1/videos/{jid}/file") +def download(jid: str): + backend, job = find_owner(jid) + if job["status"] != "done": + raise HTTPException(409, f"任务尚未完成(当前 {job['status']})") + resp = urllib.request.urlopen(f"{backend}/v1/videos/{jid}/file", timeout=300) + return StreamingResponse(resp, media_type="video/mp4", + headers={"Content-Disposition": f'attachment; filename="{jid}.mp4"'}) + + +@app.get("/v1/videos") +def listjobs(limit: int = 50): + def one(backend): + try: + return [dict(j, backend=backend) + for j in fetch(f"{backend}/v1/videos?limit={limit}", timeout=5)["jobs"]] + except Exception: + return [] + jobs = [job for part in POOL.map(one, BACKENDS) for job in part] + jobs.sort(key=lambda j: j["created_at"], reverse=True) + return {"jobs": jobs[:limit]} diff --git a/minimax-h3/long15s.sh b/minimax-h3/long15s.sh new file mode 100755 index 0000000..8a23033 --- /dev/null +++ b/minimax-h3/long15s.sh @@ -0,0 +1,8 @@ +#!/usr/bin/env bash +# 15 秒(362 帧 = 17*21+5)可行性验证。低分辨率先验证长时长通路,再考虑全分辨率。 +# 用法: ./long15s.sh [额外参数...] +set -uo pipefail +cd /data/models/minimax-h3 +P="$(cat prompt.txt)" +GPU="$1"; F="$2"; W="$3"; H="$4"; shift 4 +./run.sh "$GPU" --frames "$F" --width "$W" --height "$H" --prompt "$P" --seed 42 "$@" diff --git a/minimax-h3/pin-cache.sh b/minimax-h3/pin-cache.sh new file mode 100755 index 0000000..3392302 --- /dev/null +++ b/minimax-h3/pin-cache.sh @@ -0,0 +1,24 @@ +#!/bin/bash +# 把 H3 权重的页缓存钉住:告诉内核 user.slice 下的 40 G 不许回收。 +# 不钉的话,别的程序一申请大块内存,权重缓存就被丢掉, +# 之后每次前向都要回机械盘 /dev/sdb 读 34.4 G,单条耗时会成倍涨。 +# +# 重跑安全。撤销:sudo rm /etc/systemd/system/user-.slice.d/h3-cache.conf && sudo systemctl daemon-reload +set -e + +echo "需要 admin 密码(sudo 用的是你自己的密码,不是 root 的)" + +sudo mkdir -p /etc/systemd/system/user-.slice.d +printf '[Slice]\nMemoryMin=40G\n' | sudo tee /etc/systemd/system/user-.slice.d/h3-cache.conf > /dev/null +sudo systemctl daemon-reload +sudo systemctl set-property user.slice MemoryMin=40G +sudo systemctl set-property "user-$(id -u).slice" MemoryMin=40G + +echo +echo "=== 生效情况(两行都应该是 42949672960,即 40 G)" +grep -H . /sys/fs/cgroup/user.slice/memory.min +grep -H . "/sys/fs/cgroup/user.slice/user-$(id -u).slice/memory.min" + +echo +echo "=== 当前被保护的量" +grep -H . /sys/fs/cgroup/user.slice/memory.current diff --git a/minimax-h3/realrun.sh b/minimax-h3/realrun.sh old mode 100644 new mode 100755 index a2b77b9..a2b77b9 --- a/minimax-h3/realrun.sh +++ b/minimax-h3/realrun.sh diff --git a/minimax-h3/restart-8010.sh b/minimax-h3/restart-8010.sh old mode 100644 new mode 100755 index 86190cc..86190cc --- a/minimax-h3/restart-8010.sh +++ b/minimax-h3/restart-8010.sh diff --git a/minimax-h3/run-verify.sh b/minimax-h3/run-verify.sh new file mode 100755 index 0000000..3da962e --- /dev/null +++ b/minimax-h3/run-verify.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +# 把调度层临时换成 H3_MAX_QUEUE=1 跑一遍验证,跑完恢复默认。 +set -uo pipefail +cd /data/models/minimax-h3 + +restart_gateway() { + PID=$(ss -lptn "sport = :8000" | grep -o "pid=[0-9]*" | cut -d= -f2 | sort -u) + [ -n "$PID" ] && { kill $PID; sleep 3; } + setsid nohup ./gateway-run.sh 8000 > logs/gateway.log 2>&1 < /dev/null & + sleep 5 +} + +export H3_MAX_QUEUE=1 +restart_gateway +echo "调度层已起,H3_MAX_QUEUE=1" +./venv/bin/python verify-fixes.py + +echo +echo "=== 恢复默认 H3_MAX_QUEUE" +unset H3_MAX_QUEUE +restart_gateway +curl -s http://127.0.0.1:8000/health | grep -o '"ready":[0-9]*' diff --git a/minimax-h3/run.sh b/minimax-h3/run.sh old mode 100644 new mode 100755 index 8633631..8633631 --- a/minimax-h3/run.sh +++ b/minimax-h3/run.sh diff --git a/minimax-h3/start-all.sh b/minimax-h3/start-all.sh new file mode 100755 index 0000000..10592d8 --- /dev/null +++ b/minimax-h3/start-all.sh @@ -0,0 +1,37 @@ +#!/usr/bin/env bash +# 起全套:四张卡按默认分区起(GPU0/1 ref2va,GPU2/3 fl2va)+ 一个调度层。已经在跑的会先停掉。 +# 用法: ./start-all.sh +# +# 停某个端口要按端口找 PID,不要用 pkill -f "port 8010"——bash -c 的命令行本身含这个模式, +# 会把自己的 ssh 会话一起杀掉(踩过,退出码 255)。 +set -uo pipefail +cd /data/models/minimax-h3 + +for PORT in 8000 8010 8011 8012 8013; do + PID=$(ss -lptn "sport = :$PORT" | grep -o "pid=[0-9]*" | cut -d= -f2 | sort -u) + [ -n "$PID" ] && { echo "停 $PORT (PID $PID)"; kill $PID; } +done +sleep 8 + +# GPU 与端口的对应写死在这里,改了要同步改 README。 +# H3_OFFLOAD=disk 是关键:不设就是 cpu,每进程多占 40G 内存,四张卡只能起 3 个。 +export H3_OFFLOAD=disk +# 2026-08-20:默认分区从「4 张全 ref2va」改成 2/2 固定分裂——GPU0/1 ref2va,GPU2/3 fl2va。 +# console 的弹性调度器(h3-console/server.py 的 supervisor)仍然在跑,需求超出这个 2/2 +# 基线时照样会临时 flip 别的卡;这里只是把冷启动默认值往实际更常见的 fl2va 需求上偏, +# 减少「提交 fl2va 任务却撞上 4 张全 ref2va」时那 13-20 秒 flip 冷启动 + 最长 30 秒探测延迟。 +setsid nohup ./api-run.sh 1 8010 ref2va > logs/api-8010.log 2>&1 < /dev/null & +setsid nohup ./api-run.sh 2 8011 fl2va > logs/api-8011.log 2>&1 < /dev/null & +setsid nohup ./api-run.sh 0 8012 ref2va > logs/api-8012.log 2>&1 < /dev/null & +setsid nohup ./api-run.sh 3 8013 fl2va > logs/api-8013.log 2>&1 < /dev/null & +setsid nohup ./gateway-run.sh 8000 > logs/gateway.log 2>&1 < /dev/null & + +echo "等后端加载(disk 模式约 20 秒)" +for i in $(seq 1 30); do + sleep 5 + READY=$(curl -s --max-time 5 http://127.0.0.1:8000/health | grep -o '"ready":[0-9]*' | cut -d: -f2) + echo " ready=${READY:-?}/4" + [ "${READY:-0}" = "4" ] && break +done +curl -s http://127.0.0.1:8000/health +echo diff --git a/minimax-h3/start-gateway.sh b/minimax-h3/start-gateway.sh new file mode 100755 index 0000000..9fdd5f7 --- /dev/null +++ b/minimax-h3/start-gateway.sh @@ -0,0 +1,8 @@ +#!/usr/bin/env bash +set -uo pipefail +cd /data/models/minimax-h3 +PID=$(ss -lptn "sport = :8000" | grep -o "pid=[0-9]*" | cut -d= -f2 | sort -u) +[ -n "$PID" ] && { echo "停旧的 (PID $PID)"; kill "$PID"; sleep 3; } +setsid nohup ./gateway-run.sh 8000 > logs/gateway.log 2>&1 < /dev/null & +sleep 5 +pgrep -af "gateway:app" diff --git a/minimax-h3/submit-same.py b/minimax-h3/submit-same.py new file mode 100644 index 0000000..e2814ae --- /dev/null +++ b/minimax-h3/submit-same.py @@ -0,0 +1,13 @@ +"""把 8011 上某条已完成任务的请求原样投给另一个端口,用来做单变量对照。""" +import json, sys, urllib.request + +src_port, dst_port, job_id = sys.argv[1], sys.argv[2], sys.argv[3] + +with urllib.request.urlopen(f"http://127.0.0.1:{src_port}/v1/videos/{job_id}", timeout=30) as r: + req = json.load(r)["req"] + +body = json.dumps(req).encode() +post = urllib.request.Request(f"http://127.0.0.1:{dst_port}/v1/videos", data=body, + headers={"Content-Type": "application/json"}) +with urllib.request.urlopen(post, timeout=30) as r: + print(json.load(r)) diff --git a/minimax-h3/verify-fixes.py b/minimax-h3/verify-fixes.py new file mode 100644 index 0000000..9ae682b --- /dev/null +++ b/minimax-h3/verify-fixes.py @@ -0,0 +1,43 @@ +"""验三处修复:eta 跟帧数走、队列满返回 429、派发仍然分散。 + +跑之前要把调度层的 H3_MAX_QUEUE 调成 1,否则填满默认上限 4 要投 16 条、跑 6 分半。 +用 22 帧的短任务,路由和限流逻辑与长任务完全一样。 +""" +import json, urllib.request, urllib.error +from collections import Counter +from concurrent.futures import ThreadPoolExecutor + +GW = "http://127.0.0.1:8000" +CASES = [ + ("832x480x124f 4步(基准,实测 767s)", {"width": 832, "height": 480, "frames": 124, "steps": 4}), + ("1280x736x56f 4步(实测 736s)", {"width": 1280, "height": 736, "frames": 56, "steps": 4}), + ("832x480x22f 4步(实测 96s)", {"width": 832, "height": 480, "frames": 22, "steps": 4}), +] +SPEC = {"prompt": "一只橘猫趴在洒满阳光的窗台上,慢慢眨眼", "turbo": True} + + +def post(body): + req = urllib.request.Request(GW + "/v1/videos", data=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=60) as r: + return r.status, json.load(r) + except urllib.error.HTTPError as e: + return e.code, e.read().decode() + + +print("=== 1. eta 是否跟帧数走(改之前三档全报 691)") +results = [post(dict(SPEC, seed=200 + i, **spec)) for i, (_, spec) in enumerate(CASES)] +for (name, _), (code, body) in zip(CASES, results): + print(f" {name:38} -> eta={body['eta_seconds']}s {body['backend'][-4:]}") + +print("\n=== 2. 队列满是否 429(H3_MAX_QUEUE=1,前 3 条已占 3 张卡)") +with ThreadPoolExecutor(max_workers=4) as pool: + burst = list(pool.map(lambda i: post(dict(SPEC, seed=300 + i, frames=22, steps=4)), range(4))) +for i, (code, body) in enumerate(burst): + detail = body if isinstance(body, str) else f"{body['backend'][-4:]} queue_ahead={body['queue_ahead']}" + print(f" #{i} HTTP {code} {detail}") + +accepted = [b for c, b in results + burst if c == 202] +print(f"\n=== 3. 派发分布: {dict(Counter(b['backend'][-4:] for b in accepted))}") +print(f"接受 {len(accepted)} 条,拒绝 {len(results) + len(burst) - len(accepted)} 条") diff --git a/minimax-h3/watch-8012.sh b/minimax-h3/watch-8012.sh new file mode 100755 index 0000000..c3877b1 --- /dev/null +++ b/minimax-h3/watch-8012.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +# 盯 disk offload 那条任务:耗时、内存、以及 rchar/read_bytes—— +# rchar 是所有读(含页缓存命中),read_bytes 是真正落到存储层的。 +# 两者差得越大说明页缓存命中越好;read_bytes 跟着涨就是在读机械盘,方案不成立。 +set -uo pipefail +PID=$(pgrep -f "port 8012") +JOB="$1" +while true; do + S=$(curl -s --max-time 5 "http://127.0.0.1:8012/v1/videos/$JOB") + ST=$(echo "$S" | grep -o '"status":"[a-z]*"' | cut -d'"' -f4) + EL=$(echo "$S" | grep -o '"elapsed_seconds":[0-9]*' | cut -d: -f2) + RC=$(grep "^rchar:" /proc/$PID/io | tr -dc 0-9) + RB=$(grep "^read_bytes:" /proc/$PID/io | tr -dc 0-9) + PD=$(grep "^Private_Dirty:" /proc/$PID/smaps_rollup | tr -dc 0-9) + printf "%s %-8s 已跑%4ss rchar=%6.1fG read_bytes=%6.2fG 私有内存=%5.2fG\n" \ + "$(date +%H:%M:%S)" "$ST" "${EL:-0}" \ + "$(echo "$RC/1073741824" | bc -l)" "$(echo "$RB/1073741824" | bc -l)" \ + "$(echo "$PD/1048576" | bc -l)" + case "$ST" in done|failed) echo "$S"; break;; esac + sleep 60 +done -- libgit2 0.21.4