From 4ee085ecded0db4366c3757d9b1ab731f48385cf Mon Sep 17 00:00:00 2001 From: fmq Date: Mon, 27 Jul 2026 10:40:18 +0800 Subject: [PATCH] 3 --- cno_grid/DIST.md | 223 ++++++++++---- cno_grid/dist_config.yaml | 31 +- cno_grid/src/_check.py | 125 ++++++++ cno_grid/src/dist_master.py | 576 ++++++++++++++++++++++++++---------- cno_grid/src/qiniu_store.py | 263 ---------------- cno_grid/src/run_grid.py | 94 +++--- cno_grid/src/run_one.py | 6 +- 7 files changed, 764 insertions(+), 554 deletions(-) create mode 100644 cno_grid/src/_check.py delete mode 100644 cno_grid/src/qiniu_store.py diff --git a/cno_grid/DIST.md b/cno_grid/DIST.md index 0225ad1..d3588ca 100644 --- a/cno_grid/DIST.md +++ b/cno_grid/DIST.md @@ -12,8 +12,8 @@ master(WSL2) 异地 worker │ --teff X ... &" 跑完写本地 conv.json + .7 ├ 取状态: ssh worker "python3 dist_check ◀─ 被动提供结果 │ .py --results results" - ├ 取种子: scp worker:.7 master(收敛点) ◀─ - └ 七牛云(种子库): master 代传 .7 ─▶ 七牛 + └ 取种子: scp worker:.7 master(收敛点) ◀─ 汇聚到 master results/ + 供后续点的 seed_step 查找 ``` **核心原则**： @@ -21,7 +21,9 @@ master(WSL2) 异地 worker - **任务队列只在 master 本地**（`fcntl.flock` 强一致，零并发风险） - **worker 无常驻进程**：master 直接 SSH 执行 `run_one.py`/`seed_step.py` 跑单点 - **冷启动零依赖**（seed=None，LTE grey 自建大气）→ 绝大多数点完美分布式 -- **种子库放七牛云**（仅 `.7` 大气 393KB/点）→ 供 seed_step 回退 +- **种子库就是 master 本地 `results/`**：收敛点的 `.7` 大气（393KB/点）由 master + scp 取回汇聚于此，`run_grid.find_seed` 扫描该目录给后续冷启动失败的点做 + seed_step 回退。无需任何外部对象存储。 - **大产物留 worker 本地**（`.spec` 等 13MB）→ 不跨网传 ## 2. 并发安全（重点） @@ -47,63 +49,141 @@ pip3 install pyyaml 2>/dev/null || echo "用内置解析器，无需 pip" ```bash sudo apt install python3 openssh-server ``` -> **worker 不需要 curl、不需要 pip、不需要七牛配置、不需要 SSH 回 master。** +> **worker 不需要 curl、不需要 pip、不需要 SSH 回 master。** + +### 3.3 SSH 免密配置（master → worker，单向即可） + +```bash +# 1. 生成密钥对（已有可跳过） +ls ~/.ssh/id_ed25519.pub || ssh-keygen -t ed25519 -N "" -f ~/.ssh/id_ed25519 + +# 2. 推公钥到每台 worker（这一步会要求输一次密码） +ssh-copy-id user@worker1 +ssh-copy-id user@worker2 + +# 3. 验证免密（BatchMode 不弹交互提示，dist_master 用的就是这个模式） +ssh -o BatchMode=yes user@worker1 "echo ok" +``` + +**localhost 也要配**（如果 workers 里包含 localhost）： +```bash +# WSL 默认没跑 sshd，先启动 +sudo apt install openssh-server -y +sudo service ssh start + +# localhost 的 host key 首次连接需要确认，BatchMode 下会报 +# "Host key verification failed"，必须预先加入 known_hosts： +ssh-keyscan -H localhost >> ~/.ssh/known_hosts 2>/dev/null + +# 自己 SSH 自己也要公钥： +ssh-copy-id localhost + +# 验证 +ssh -o BatchMode=yes localhost "echo ok" +``` + +> **踩坑**：WSL 重启后 sshd 不会自启。加到 `~/.bashrc`： +> ```bash +> sudo service ssh start 2>/dev/null +> ``` +> 或者每次手动 `sudo service ssh start`。 + +**worker 不需要 SSH 回 master。** + +### 3.4 分发 TLUSTY 树到每台 worker（首次，约 2GB） -### 3.3 分发 TLUSTY 树到每台 worker（首次，约 2GB） 在 master 上对每台 worker： ```bash -sudo apt install rsync # master 端装一次 TARGET=user@worker1 -REMOTE_ROOT=~/tlusty/tl208-s54 # 记到 dist_config.yaml 的 tlusty_root -rsync -avz --exclude='cno_grid/results' --exclude='*.log' \ +REMOTE_ROOT=~/DataSheel/tlusty # 记到 dist_config.yaml 的 tlusty_root + +# 优先用 rsync（支持 --exclude） +rsync -avz --exclude='cno_grid/results' --exclude='*.log' --exclude='__pycache__' \ /home/dckj/program/tlusty/tl208-s54/ $TARGET:$REMOTE_ROOT/ ``` -### 3.4 分发 cno_grid 代码到每台 worker（每次代码更新） +**没有 rsync 时用 scp 替代**（注意 scp 不支持 --exclude）： +```bash +# 只推必要子目录，不要推整个树（避免跟踪符号链接） +ssh $TARGET "mkdir -p $REMOTE_ROOT/{tlusty,synspec,cno_grid}" +scp /home/dckj/program/tlusty/tl208-s54/tlusty/tlusty.exe $TARGET:$REMOTE_ROOT/tlusty/ +scp /home/dckj/program/tlusty/tl208-s54/synspec/synspec.exe $TARGET:$REMOTE_ROOT/synspec/ +scp -r /home/dckj/program/tlusty/tl208-s54/data $TARGET:$REMOTE_ROOT/ +scp -r /home/dckj/program/tlusty/tl208-s54/cno_grid/src \ + /home/dckj/program/tlusty/tl208-s54/cno_grid/templates \ + /home/dckj/program/tlusty/tl208-s54/cno_grid/seeds \ + /home/dckj/program/tlusty/tl208-s54/cno_grid/config_dense.yaml \ + $TARGET:$REMOTE_ROOT/cno_grid/ +``` + +> **踩坑：scp -r 会跟踪符号链接！** +> 本地 `cno_grid/results -> /mnt/e/fmq/grid`（几百 MB 历史结果）， +> `scp -r cno_grid/` 会把链接指向的实际内容全推过去。 +> **worker 不需要 master 的 results**（它自己建），所以： +> - 用 rsync + `--exclude='results'`，或 +> - 用 scp 时只推 `src/ templates/ seeds/ config*.yaml`，不推整个目录 + +### 3.5 分发 cno_grid 代码到每台 worker（每次代码更新） ```bash rsync -avz --exclude='results' --exclude='*.log' --exclude='__pycache__' \ + --exclude='tasks.json' \ /home/dckj/program/tlusty/tl208-s54/cno_grid/ \ $TARGET:$REMOTE_ROOT/cno_grid/ ``` - -### 3.5 配置 SSH 免密（master → worker，单向即可） +无 rsync 时： ```bash -ssh-copy-id user@worker1 -ssh-copy-id user@worker2 -# ... +scp -r src/ templates/ seeds/ config_dense.yaml $TARGET:$REMOTE_ROOT/cno_grid/ ``` -验证：`ssh user@worker1 echo ok` 应直接输出 `ok`，不问密码。 -**worker 不需要 SSH 回 master。** -### 3.6 七牛云（可选；不配也能跑，只是没跨机种子共享） -1. 七牛云创建 bucket，拿 AK/SK，绑域名 -2. master 设环境变量： - ```bash - export QINIU_ACCESS_KEY="你的AK" - export QINIU_SECRET_KEY="你的SK" - ``` -> 不配七牛时，seed_step 只能用同台 worker 上已收敛的本地种子。 +### 3.6 验证 worker 部署完整性 +```bash +ssh $TARGET "ls $REMOTE_ROOT/tlusty/tlusty.exe \ + $REMOTE_ROOT/synspec/synspec.exe \ + $REMOTE_ROOT/data/gfVIS99.dat \ + $REMOTE_ROOT/cno_grid/src/run_one.py \ + $REMOTE_ROOT/cno_grid/templates/cno_atmos.5.tpl" +``` +全部列出无报错即可。 + +### 3.7 种子源（master 本地 results/） + +种子（收敛点的 `.7` 大气）只存在 **master 本地的 `results/` 目录**： +- 每个收敛模型在 `results//.7`，附带 `conv.json` +- master 回收异地 worker 的收敛结果时，用 scp 把 `.7` 取回到这个目录 +- 冷启动失败的点做 seed_step 回退时，`run_grid.find_seed` 扫描该目录， + 按同 family (Teff/logg/logHe) 的 CNO 最近邻 → 全局最近邻选取种子 + +**无需任何外部对象存储（七牛/OSS 等）**。worker 端完全不碰种子库。 + +### 3.8 从单机切分布式时（已有 results 里的收敛点） + +如果之前用 `run_grid.py` 跑过一批，已收敛的 `.7` 已经在 `results//.7` +里——这正是分布式 master 查找种子的位置，**无需任何额外操作**。 + +确认种子可用： +```bash +RESULTS=<你的 results 实际路径> # 如 /mnt/e/fmq/grid +echo "收敛种子数: $(find $RESULTS -maxdepth 2 -name '*.7' | wc -l)" +``` + +> 注意：旧的 `.seeds_remote/` 目录（七牛时代的本地缓存）已不再使用， +> 可以删除。master 现在直接扫 `results/`。 ## 4. 配置 dist_config.yaml ```yaml master: workdir: /home/dckj/program/tlusty/tl208-s54/cno_grid - grid_config: /home/dckj/program/tlusty/tl208-s54/cno_grid/config.yaml + grid_config: /home/dckj/program/tlusty/tl208-s54/cno_grid/config_dense.yaml results: results -qiniu: - bucket: your-bucket - domain: http://xxx # 留空 = 纯本地模式 - upload_by: master # master 代传种子（worker 不碰七牛） - workers: - - host: localhost # 本机也算 worker（可选） + - host: localhost # 本机也算 worker tlusty_root: /home/dckj/program/tlusty/tl208-s54 - nproc: 8 - - host: user@worker1 # 异地机器 - tlusty_root: ~/tlusty/tl208-s54 - nproc: 24 + nproc: 22 + - host: dckj@192.168.7.102 # 异地机器 + tlusty_root: ~/DataSheel/tlusty + nproc: 35 # ... 每台机器一项 ``` @@ -112,14 +192,12 @@ workers: ### 5.1 预览（dry-run） ```bash cd /home/dckj/program/tlusty/tl208-s54/cno_grid -export QINIU_ACCESS_KEY=... QINIU_SECRET_KEY=... python3 src/dist_master.py dist_config.yaml --dry-run # 输出：待计算 N 点，总并发 X 核，预计 ~Y 小时 + 部署检查清单 ``` ### 5.2 全量跑 ```bash -export QINIU_ACCESS_KEY=... QINIU_SECRET_KEY=... nohup python3 src/dist_master.py dist_config.yaml > results/dist_run.log 2>&1 & ``` @@ -137,25 +215,32 @@ ssh user@worker1 "tail ~/tlusty/tl208-s54/cno_grid/worker_jobs/<模型名>.log" ## 6. 工作流程（master 内部） ``` -1. init tasks.json（432 点 pending；本地已 converged 标 done） +1. init tasks.json（5120 点；本地已 converged 标 done，其余 pending） + - 按难度排序（CNO 总量升序 → Teff 升序 → logg 降序 → logHe 升序） + - 每个 task 标注 wave 编号（同 CNO 总量归同一 wave） 2. 每台 worker 一个管理线程，并发跑： 循环： a. SSH 查 worker 正在跑的进程数 → 空闲槽 = nproc - running - b. 每个空闲槽：本地 flock 领一个 pending→running，SSH 派 run_one.py（nohup） + b. 每个空闲槽：本地 flock 领一个 pending→running（wave 门控： + 只从最小未完成 wave 取任务，低金属 wave 全终态后才开放下一 wave） c. SSH 调 dist_check.py 查 worker 已完成的点： - - 收敛 → scp .7 回 master → (代传七牛) → 标 done - - 未收敛 → 找种子（本地缓存>七牛）→ scp 种子到 worker → SSH 派 seed_step.py 重试 + - 收敛 → scp .7 回 master results/ → 标 done + - 未收敛 → 扫 master 本地 results/ 找邻居种子 → scp 种子到 worker → SSH 派 seed_step.py 重试 - seed_step 仍失败 → 标 failed（物理极限） 3. 全部终态 → 写 grid_status.json ``` +**Wave 门控的意义**：低金属点易收敛，先算完产出 .7 种子；高金属点冷启动 +困难，需要种子做 seed_step。如果无序派发，高金属点在种子还没产出时就被派出 +去冷启动 → 失败 → 无种子可救 → 标 failed。wave 串行保证种子链从低金属向 +高金属递进。实测：无序提交时 60K 收敛率 25%，wave 调度后期望 >50%。 + ## 7. 文件清单 | 文件 | 角色 | 运行在哪 | |------|------|---------| | `src/dist_master.py` | master 调度器（推/拉/seed_step/汇总） | master | | `src/dist_check.py` | worker 状态查询（被 SSH 调用） | 每台 worker | -| `src/qiniu_store.py` | 七牛云种子库（零 SDK） | master | | `src/claim_task.py` / `report_task.py` | 任务原子操作（master 本地调） | master | | `src/run_one.py` | 单点冷启动执行器（被 SSH 调用） | 每台 worker | | `src/seed_step.py` | 单点种子步进执行器（被 SSH 调用） | 每台 worker | @@ -175,20 +260,50 @@ ssh user@worker1 "tail ~/tlusty/tl208-s54/cno_grid/worker_jobs/<模型名>.log" | seed_step 种子推送 | 393KB/点 | 仅失败点，少数 | | **跨网传输占比** | **<5%** | 相比计算可忽略 | -432 点 / 5台×24核 ≈ **3-4 小时**（vs 单机 15-21 小时）。 +5120 点 / 2台×57核 ≈ **24 小时**（vs 单机 100+ 小时）。 -## 9. 故障排查 +## 9. 从单机 run_grid.py 迁移到分布式 -| 现象 | 排查 | -|------|------| -| master 派不出去 | `ssh user@workerN echo ok` 是否通；worker 上 `run_one.py` 路径对不对 | -| worker 算了但 master 没回收 | 看 `worker_jobs/.log`；`dist_check.py` 是否能列出 | -| 任务一直 running | 可能 worker 进程崩了；master 每 2 小时自动重派（stale_sec） | -| 种子找不到（seed_step 失败） | 同 family 无收敛点 → 物理极限，正常；或配七牛扩种子来源 | -| 单机模式仍可用 | `python3 src/run_grid.py config.yaml`（完全不受影响） | +已有 `run_grid.py` 跑了一部分，想切换到 `dist_master.py` 续算： -## 10. 安全提示 +```bash +# 1. 停掉正在跑的 run_grid.py +ps aux | grep run_grid +kill -- **AK/SK 只放 master 环境变量**，不写进 dist_config.yaml（会被提交仓库） -- worker 默认不碰七牛（`upload_by: master`）；如需 worker 直传再配域名 +# 2. 确保 dist_config.yaml 的 grid_config 指向同一个网格配置 +# （如 config_dense.yaml），results 指向同一个结果目录 + +# 3. 预上传种子（见 3.8 节），否则高金属点 seed_step 无种子可用 + +# 4. 启动分布式（--force-init 重建 tasks.json，已收敛的自动标 done 跳过） +python3 src/dist_master.py dist_config.yaml --force-init +``` + +两边用同一个 `results/` 目录、同一套 `gen_input5.model_name()` 命名， +`init_tasks` 会扫 `results//conv.json` 里 `converged=true` 的标为 +done，完全兼容。 + +> **注意**：切换后不要再用 `run_grid.py` 跑同一个 results 目录， +> 否则两边同时写同一个模型目录会冲突。 + +## 10. 故障排查 + +| 现象 | 原因 | 解决 | +|------|------|------| +| `ssh: connect to host localhost port 22: Connection refused` | WSL 没跑 sshd | `sudo service ssh start` | +| `Host key verification failed` | BatchMode 下首次连接不弹确认 | `ssh-keyscan -H localhost >> ~/.ssh/known_hosts` | +| scp 把 results 全推到 worker 了 | `scp -r` 跟踪符号链接 | 用 rsync --exclude 或只推 src/templates/seeds | +| `FileNotFoundError: '~/xxx/tlusty.exe'` | `tlusty_root` 用了 `~/`，Python 不展开 `~` | dist_config.yaml 改成绝对路径 `/home/user/...` | +| worker 收到派发但 run_one 立刻退出 | 路径错 / 缺库 / 二进制架构不符 | 手动 SSH 跑一遍 `run_one.py` 看报错 | +| master 派不出去 | SSH 免密没配好 | `ssh -o BatchMode=yes user@host "echo ok"` | +| worker 算了但 master 没回收 | 看 `worker_jobs/.log` | `dist_check.py` 是否能列出 | +| tasks.json 读到半截 JSON 崩溃 | 并发写时读方未加锁（已修复） | 升级 dist_master.py（用 `_read_tasks` + tmpfile rename） | +| 任务一直 running | worker 进程崩了 | master 每 2 小时自动重派（stale_sec） | +| 种子找不到（seed_step 失败） | master 本地 results/ 里还没有收敛的邻居点 | 等低金属 wave 先算完产出种子；或先单机跑一批易收敛点 | +| 单机模式仍可用 | — | `python3 src/run_grid.py config_dense.yaml`（完全不受影响） | + +## 11. 安全提示 + +- 本架构无需任何外部云服务凭证（不再使用七牛/OSS），无密钥泄露风险 - SSH 用密钥免密，不用密码 diff --git a/cno_grid/dist_config.yaml b/cno_grid/dist_config.yaml index d5a924a..0577494 100644 --- a/cno_grid/dist_config.yaml +++ b/cno_grid/dist_config.yaml @@ -1,6 +1,7 @@ # 分布式网格计算配置（dist_master.py 用）。 # 架构：master 单向 SSH 推/拉（只需 master 能 SSH 进 worker，worker 不回连 master）。 -# 网格轴本身仍在 config.yaml 定义，这里只配 master/qiniu/workers。 +# 网格轴本身仍在 config.yaml 定义，这里只配 master/workers。 +# 种子来源：master 本地 results/（收敛点的 .7 汇聚处，异地结果由 master scp 取回）。 # # 部署：填好本文件后 # python3 src/dist_master.py dist_config.yaml --dry-run # 预览 @@ -10,30 +11,18 @@ master: # master 上 cno_grid 目录（tasks.json 在此，调度在此） workdir: /home/dckj/program/tlusty/tl208-s54/cno_grid # 网格配置（默认 workdir/config.yaml） - grid_config: /home/dckj/program/tlusty/tl208-s54/cno_grid/config.yaml + grid_config: /home/dckj/program/tlusty/tl208-s54/cno_grid/config_dense.yaml # 结果根目录（相对 workdir 或绝对路径；与 config.yaml 的 results 一致） # master 在此汇总 grid_status.json；各 worker 各自有自己的 results/（同名） + # 收敛点的 .7 也汇聚在此（seed_step 找种子时扫描该目录） results: results -qiniu: - # 七牛云 bucket 名（种子库）。填你的实际 bucket。留空则纯本地模式。 - bucket: your-bucket - # AK/SK：仅 master 持有。建议用环境变量，别提交到仓库： - # export QINIU_ACCESS_KEY=... ; export QINIU_SECRET_KEY=... - access_key: "" - secret_key: "" - # 七牛绑定域名（下载/上传种子用）。公开空间填 http(s)://域名。 - domain: "" - seed_prefix: seeds - # 谁上传种子到七牛： - # master（默认）= master 取回 .7 后代传，worker 完全不碰七牛（worker 零配置） - # worker = worker 跑完自己传（需 worker 配 QINIU_DOMAIN） - upload_by: master - # ---- worker 节点（master 只需能 SSH 进这些机器）---- # 每个 worker： # host : user@host（SSH 目标）。localhost 表示本机也算一个 worker。 # tlusty_root: 该机器上 TLUSTY 树根（含 tlusty/tlusty.exe、synspec/、data/） +# ⚠️ 必须用绝对路径！~/xxx 的 ~ 不会被 Python os.path 展开， +# 导致 run_one.py 找不到 tlusty.exe（FileNotFoundError）。 # nproc : 该机贡献的并发进程数（建议 ≤ 物理核数） # # 注意：单向 SSH。只需 master→worker 免密，worker 不需要 SSH 回 master。 @@ -42,10 +31,10 @@ qiniu: workers: - host: localhost tlusty_root: /home/dckj/program/tlusty/tl208-s54 - nproc: 8 - # - host: user@server2 - # tlusty_root: ~/tlusty/tl208-s54 - # nproc: 24 + nproc: 22 + - host: dckj@192.168.7.102 + tlusty_root: /home/dckj/DataSheel/tlusty + nproc: 35 # - host: user@server3 # tlusty_root: ~/tlusty/tl208-s54 # nproc: 24 diff --git a/cno_grid/src/_check.py b/cno_grid/src/_check.py new file mode 100644 index 0000000..b0f4330 --- /dev/null +++ b/cno_grid/src/_check.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python3 +"""可靠进度检查（ps 精确计数 + tasks.json 权威数据）。检查后删除。""" +import subprocess, json, time, os +from collections import Counter + +os.chdir("/home/dckj/program/tlusty/tl208-s54/cno_grid") + +def run(cmd, timeout=20): + try: + r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + return r.stdout + except Exception as e: + return "[err: %s]" % e + +def count_proc_local(name): + out = run(["ps", "-eo", "comm"]) + return sum(1 for line in out.splitlines() if line.strip() == name) + +def ssh_remote(cmd): + return run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10", + "dckj@192.168.7.102", cmd]) + +# 1. master（精确扫描，不用 pgrep） +print("=" * 52) +print("1. master 进程") +print("=" * 52) +masters = [] +out = run(["ps", "-eo", "pid,etimes,args"]) +for line in out.splitlines()[1:]: + if "dist_master.py" in line and "python3" in line and "_check.py" not in line: + parts = line.split(None, 2) + if len(parts) >= 3: + masters.append((parts[0], int(parts[1]), parts[2][:70])) +if not masters: + print(" !! master 不在跑") +else: + for pid, et, cmd in masters: + print(" PID %s 运行 %d小时%d分" % (pid, et//3600, (et%3600)//60)) + if len(masters) > 1: + print(" !! 有 %d 个 master（应只有1个）" % len(masters)) + +# 2. tasks.json（先验证完整性，再统计） +print("\n" + "=" * 52) +print("2. tasks.json 完整性 + 状态") +print("=" * 52) +try: + d = json.load(open("tasks.json")) + print(" tasks.json 完整 ✓") +except Exception as e: + print(" !! tasks.json 损坏:", str(e)[:60]) + print(" master 的加固逻辑应已备份并继续运行") + # 尝试找最近的备份 + baks = sorted([f for f in os.listdir(".") if f.startswith("tasks.json.corrupt")]) + if baks: + print(" 损坏备份:", baks[-1]) + raise SystemExit(0) + +c = Counter(t["status"] for t in d["tasks"].values()) +total = len(d["tasks"]) +done_now = c.get("done", 0) +print(" 总计 %d:" % total, dict(c)) +print(" 进度: %.1f%% done, %.1f%% failed" % ( + done_now/total*100, c.get("failed",0)/total*100)) +print(" done 较上次检查(1599): %+d" % (done_now - 1599)) + +# wave +wc = {} +for n, t in d["tasks"].items(): + wc.setdefault(t.get("wave",0), Counter())[t["status"]] += 1 +print("\n 各 wave:") +for w in sorted(wc): + s = wc[w] + print(" wave %d: done=%d failed=%d running=%d pending=%d" % ( + w, s.get("done",0), s.get("failed",0), s.get("running",0), s.get("pending",0))) + +# 3. running 健康度 +print("\n" + "=" * 52) +print("3. running 任务（%d 个）" % c.get("running", 0)) +print("=" * 52) +now = time.time() +running = [(n, t) for n, t in d["tasks"].items() if t["status"] == "running"] +if running: + by_host = Counter(t.get("host","?") for _, t in running) + print(" 按 host:", dict(by_host)) + ages = [(now - t.get("claimed_at", 0))/60 for _, t in running] + print(" age: 最长 %.0f分 最短 %.0f分 平均 %.0f分" % ( + max(ages), min(ages), sum(ages)/len(ages))) + ss = sum(1 for _, t in running if t.get("seed_step_used")) + print(" seed_step中: %d 冷启动: %d" % (ss, len(running)-ss)) + stale = [(n, a) for (n, t), a in zip(running, ages) if a > 50] + if stale: + print(" !! 超50分钟: %d 个" % len(stale)) + else: + print(" 无超时 ✓") + +# 4. 算力 +print("\n" + "=" * 52) +print("4. worker 算力") +print("=" * 52) +lt = count_proc_local("tlusty.exe") +rout = ssh_remote("ps -eo comm") +rt = sum(1 for l in rout.splitlines() if l.strip() == "tlusty.exe") +print(" 本地 tlusty=%d 异地=%d 总=%d (核57)" % (lt, rt, lt+rt)) + +# 5. 加固触发情况（应都=0） +print("\n" + "=" * 52) +print("5. 健康检查") +print("=" * 52) +log = run(["cat", "results/dist_run.log"]) +print(" 循环异常: %d" % log.count("循环异常")) +print(" monitor临时错误: %d" % log.count("临时错误")) +print(" tasks.json损坏备份: %d" % log.count("tasks.json 损坏")) +fail_notes = Counter((t.get("note") or "?")[:45] for t in d["tasks"].values() + if t["status"] == "failed") +if fail_notes: + print(" failed 原因:") + for note, cnt in fail_notes.most_common(3): + print(" %d × %s" % (cnt, note)) + +# 6. 最近活动 +print("\n" + "=" * 52) +print("6. 日志最近活动") +print("=" * 52) +for line in log.splitlines()[-10:]: + print(" " + line.rstrip()) diff --git a/cno_grid/src/dist_master.py b/cno_grid/src/dist_master.py index baa63c7..b93cd6c 100644 --- a/cno_grid/src/dist_master.py +++ b/cno_grid/src/dist_master.py @@ -7,7 +7,8 @@ - 任务队列 tasks.json 只在 master 本地（fcntl.flock 强一致，零并发风险） - worker 无常驻进程；master 直接 SSH 执行 run_one.py 跑单点（异步 nohup） - master 周期 SSH 进 worker 查状态、取结果（conv.json + .7） - - 七牛云只存种子（master 取回 .7 后代传；worker 不碰七牛） + - 种子（收敛点的 .7）汇聚在 master 本地 results/，seed_step 回退时从这里找 + 最近邻（无需任何外部对象存储；异地 worker 的收敛 .7 由 master scp 取回） 主循环： init tasks.json（断点续算：本地已 converged 标 done） @@ -20,7 +21,7 @@ sleep 轮询 for 每个 worker: SSH 查已完成的（conv.json 存在且进程已退出） - 读结果 → 更新 tasks.json → scp .7（若收敛）→ 上传七牛 + 读结果 → 更新 tasks.json → scp .7（若收敛，汇聚到 master results/） 写 grid_status.json 并发安全： @@ -37,6 +38,7 @@ import fcntl import itertools import json import os +import shutil import subprocess import sys import threading @@ -44,7 +46,7 @@ import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import gen_input5 # noqa: E402 -import qiniu_store # noqa: E402 +from run_grid import find_seed as _find_seed_in_results # noqa: E402 # ---- YAML 加载（优先 pyyaml，否则内置极简解析器，零 pip）---- try: @@ -136,31 +138,110 @@ except ImportError: # ==================== tasks.json 原子操作（本地 flock）==================== def _lock_tasks(tasks_path, fn): - """对 tasks.json 加排他锁，执行 fn(data)，写回。返回 fn 的返回值。""" + """对 tasks.json 加排他锁，执行 fn(data)，写回。返回 fn 的返回值。 + + 原子性保证（防止写入中断导致 tasks.json 损坏）： + 1. 读旧数据；若解析失败，备份损坏文件后抛异常（绝不用空数据覆盖， + 否则所有任务状态丢失——这是之前 master 误判"全终态"退出的根因） + 2. 写到 tmp 文件 + fsync，再 os.replace 原子替换 + 3. 任何异常都不破坏旧 tasks.json（replace 失败时旧文件仍在） + """ fd = open(tasks_path, "r+") try: fcntl.flock(fd.fileno(), fcntl.LOCK_EX) fd.seek(0) + raw = fd.read() try: - data = json.load(fd) + data = json.loads(raw) except ValueError: - data = {"tasks": {}} + # tasks.json 损坏！备份后抛异常，绝不静默用空数据覆盖 + # （否则 fn 在空 tasks 上操作，写回后会丢光所有任务状态） + import traceback + bak = tasks_path + ".corrupt.%d" % int(time.time()) + try: + with open(bak, "w") as bf: + bf.write(raw) + print("[_lock_tasks] tasks.json 损坏！已备份到 {}".format(bak), + file=sys.stderr, flush=True) + except OSError: + pass + raise RuntimeError("tasks.json 损坏，已备份，拒绝用空数据覆盖") result = fn(data) - fd.seek(0) - fd.truncate() - json.dump(data, fd, indent=2) - fd.flush() - os.fsync(fd.fileno()) + # 先写临时文件，再 rename 覆盖（原子操作，避免读方看到半截 JSON） + tmp = tasks_path + ".tmp" + with open(tmp, "w") as wf: + json.dump(data, wf, indent=2) + wf.flush() + os.fsync(wf.fileno()) + os.replace(tmp, tasks_path) return result finally: fcntl.flock(fd.fileno(), fcntl.LOCK_UN) fd.close() -def expand_grid(grid): +def _read_tasks(tasks_path): + """加共享锁安全读取 tasks.json。 + + 损坏时抛 RuntimeError（而非返回空 dict）——返回空会让主循环误判 + "无 pending 无 running → 全终态"而退出，master 误死亡（见之前的故障）。 + 抛异常则被 WorkerManager.run 的 try/except 捕获，下轮重试，不退出。 + """ + fd = open(tasks_path, "r") + try: + fcntl.flock(fd.fileno(), fcntl.LOCK_SH) + fd.seek(0) + try: + return json.load(fd) + except ValueError as e: + raise RuntimeError("tasks.json 损坏: {}".format(e)) + finally: + fcntl.flock(fd.fileno(), fcntl.LOCK_UN) + fd.close() + + +def expand_grid_ordered(grid): + """生成按"难度"排序的网格点列表（与 run_grid.expand_grid_ordered 一致）。 + + 排序键（优先级从高到低）： + 1. CNO 总丰度升序 —— 金属越低越易收敛，且是后续点的种子 + 2. Teff 升序 —— 低温冷启动更稳 + 3. logg 降序 —— 高重力大气更紧凑、更稳 + 4. logHe 升序 + + 配合 wave 门控，保证低金属点先完成、产出种子后高金属点才被派发。 + """ keys = ["teff", "logg", "loghe", "logc", "logn", "logo"] - for combo in itertools.product(*(grid[k] for k in keys)): - yield dict(zip(keys, combo)) + points = [dict(zip(keys, combo)) + for combo in itertools.product(*(grid[k] for k in keys))] + points.sort(key=lambda p: ( + p["logc"] + p["logn"] + p["logo"], # CNO 总量升序（主键） + p["teff"], # Teff 升序 + -p["logg"], # logg 降序（高 g 先） + p["loghe"], # logHe 升序 + )) + return points + + +def _assign_waves(points): + """按 CNO 总量给排序后的点分 wave（与 run_grid._split_waves 逻辑一致）。 + + 返回 {model_name: wave_index} 映射。同 CNO 总量的点归同一 wave。 + """ + waves = {} + current_cno = None + wave_idx = 0 + for pt in points: + cno_sum = pt["logc"] + pt["logn"] + pt["logo"] + if current_cno is None: + current_cno = cno_sum + elif cno_sum != current_cno: + wave_idx += 1 + current_cno = cno_sum + name = gen_input5.model_name(pt["teff"], pt["logg"], pt["loghe"], + pt["logc"], pt["logn"], pt["logo"]) + waves[name] = wave_idx + return waves def model_done_local(results_root, name): @@ -173,18 +254,46 @@ def model_done_local(results_root, name): return False +def _read_conv_summary(results_root, name): + """读 results//conv.json 的关键字段，用于 init_tasks 回填 done 任务。 + + 避免 --force-init 重建 tasks.json 时丢失 max_relc/elapsed/seed_step 等字段 + （否则 grid_status.json 的质量评估全是 None，见 #4）。 + """ + p = os.path.join(results_root, name, "conv.json") + if not os.path.isfile(p): + return {} + try: + m = json.load(open(p)) + except Exception: + return {} + return { + "max_relc": m.get("final_max_relc"), + "elapsed_sec": m.get("elapsed_sec"), + "seed_step_used": bool(m.get("seed_step_used", False)), + } + + def init_tasks(grid_cfg, results_root, tasks_path, grid_id="config.yaml", force=False): if os.path.isfile(tasks_path) and not force: - return json.load(open(tasks_path)) + return _read_tasks(tasks_path) + points = expand_grid_ordered(grid_cfg["grid"]) + wave_map = _assign_waves(points) tasks = {} - for pt in expand_grid(grid_cfg["grid"]): + for pt in points: name = gen_input5.model_name(pt["teff"], pt["logg"], pt["loghe"], pt["logc"], pt["logn"], pt["logo"]) done = model_done_local(results_root, name) - tasks[name] = {"params": pt, "status": "done" if done else "pending", - "worker": None, "host": None, "claimed_at": None, - "finished_at": None, "attempts": 0, "max_relc": None, - "seed_step_used": False, "note": None} + entry = {"params": pt, "status": "done" if done else "pending", + "wave": wave_map[name], + "worker": None, "host": None, "claimed_at": None, + "finished_at": None, "attempts": 0, "max_relc": None, + "seed_step_used": False, "note": None} + # done 任务从 conv.json 回填字段，避免 grid_status 评估信息丢失 + if done: + entry.update(_read_conv_summary(results_root, name)) + entry["note"] = "converged (resumed)" + tasks[name] = entry data = {"grid": grid_id, "created": time.time(), "tasks": tasks} with open(tasks_path, "w") as f: json.dump(data, f, indent=2) @@ -192,13 +301,32 @@ def init_tasks(grid_cfg, results_root, tasks_path, grid_id="config.yaml", force= def claim_task(tasks_path, host, worker_id, phase=None): - """领一个 pending 任务 → running，返回 {name, params} 或 None。""" + """领一个 pending 任务 → running，返回 {name, params} 或 None。 + + Wave 门控：优先从最小未完成 wave 取任务，保证低金属 wave 先产出种子。 + 但不阻塞——当某 wave 只剩 running（pending=0）时，立即开放下一 wave， + 避免"一个慢任务卡住整个网格 56 核闲置"。 + """ def do(data): tasks = data["tasks"] now = time.time() + # 找最小"还有 pending"的 wave（running 不阻塞下一 wave）。 + # 这比旧逻辑（running 也算活跃）更合理：种子是否产出看 done， + # 只要低 metal wave 已有 done，高 metal 点冷启动失败也能找到种子。 + min_pending_wave = None + for t in tasks.values(): + if t["status"] == "pending": + w = t.get("wave", 0) + if min_pending_wave is None or w < min_pending_wave: + min_pending_wave = w + if min_pending_wave is None: + return None # 无 pending 任务 for name, t in tasks.items(): if t["status"] != "pending": continue + # wave 门控：只派当前最小 pending wave 的任务 + if t.get("wave", 0) > min_pending_wave: + continue if phase == "phase2" and t.get("attempts", 0) == 0: continue t["status"] = "running" @@ -249,6 +377,16 @@ def reset_stale(tasks_path, stale_sec): # ==================== SSH 工具（master → worker 单向）==================== +def is_local_target(target): + """判断 SSH target 是否指向本机（localhost / 127.0.0.1 / 本机名）。 + + localhost 这台 worker 与 master 共用文件系统（results 同一个符号链接）， + 对它的查询/取回可走本地文件系统，避免依赖 WSL sshd（常断，见 #2）。 + """ + host = target.split("@")[-1] if "@" in target else target + return host in ("localhost", "127.0.0.1", "0.0.0.0") or host == os.uname()[1] + + def ssh_ok(target, cmd, timeout=30): """SSH 执行命令，返回 (rc, stdout, stderr)。target=user@host 或 host。""" full = ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=15", target, cmd] @@ -260,8 +398,15 @@ def ssh_ok(target, cmd, timeout=30): def count_running_on_worker(target, worker_cwd): - """SSH 查 worker 上正在跑的 run_one 进程数。""" - cmd = "pgrep -fc 'run_one.py' 2>/dev/null || echo 0" + """SSH 查 worker 上正在跑的 run_one 进程数。 + + 注意：pgrep 的命令行匹配会把"执行本次 pgrep 的临时 shell"也算进去（因为 + ssh 远端命令 `bash -c "pgrep ... run_one.py"` 的 argv 含 'run_one.py' 字样）， + 导致计数虚高 1~2，进而 master 少派任务槽。用精确正则 + `python3.*src/run_one.py` 排除这些临时 shell（它们只匹配到 pgrep 自身， + 不会匹配到 python3 解释器）。pgrep -f 全命令行匹配。 + """ + cmd = "pgrep -fc 'python3.*run_one.py' 2>/dev/null || echo 0" rc, out, _ = ssh_ok(target, cmd, timeout=20) if rc != 0 or not out: return 0 @@ -271,29 +416,70 @@ def count_running_on_worker(target, worker_cwd): return 0 +def pids_alive_on_worker(target, pids): + """SSH 进 worker，用 kill -0 批量探活一组 pid，返回仍存活的 pid 集合。 + + pids 是 master 本地记录的活跃 pid（dispatch_task 时 echo $! 拿到的）。 + kill -0 不发信号，只检测进程是否存在与权限。单次 SSH 内用 for 循环 + 逐个检测（kill -0 多 pid 时整体退出码无法区分单个），避免 N 次 SSH 往返。 + 用于精确判定"这台 worker 实际还在跑几个任务"，不受 pgrep 误匹配影响。 + """ + if not pids: + return set() + # 单次 SSH：对每个 pid 输出 "pid:Y" 或 "pid:N" + checks = "; ".join( + "kill -0 {0} >/dev/null 2>&1 && echo '{0}:Y' || echo '{0}:N'".format(p) + for p in pids) + rc, out, _ = ssh_ok(target, checks, timeout=15) + alive = set() + for line in out.splitlines(): + line = line.strip() + if line.endswith(":Y"): + try: + alive.add(int(line[:-2])) + except ValueError: + pass + return alive + + +def _parse_pid(pid_out): + """从 dispatch_task 的 stdout（nohup ... & echo $! 的输出）解析 pid。 + + nohup 成功时 echo $! 输出后台进程 pid；失败/异常时输出为空或错误信息。 + 返回 int pid 或 None。 + """ + if not pid_out: + return None + try: + return int(pid_out.strip().splitlines()[-1]) + except (ValueError, IndexError): + return None + + def dispatch_task(target, worker_cwd, tlusty_root, name, params, results, - seed_path=None, qiniu_domain="", timeout=3600, grid_config="config.yaml"): + seed_path=None, timeout=3600, grid_config="config.yaml"): """SSH 派一个任务到 worker（异步 nohup，立即返回）。 master 本地已 claim；这里只负责在 worker 上启动 run_one.py。 """ - parts = ["{teff:.0f}", "{logg:.1f}", "{loghe:.0f}", - "{logc:.0f}", "{logn:.0f}", "{logo:.0f}"] args = ("--teff {teff} --logg {logg} --loghe {loghe} " "--logc {logc} --logn {logn} --logo {logo}").format(**params) cmd_seed = "--seed '{}'".format(seed_path) if seed_path else "" # run_one.py 在 cno_grid/src 下；results 相对 worker_cwd env = " ".join([ "TLUSTY='{}'".format(tlusty_root), - "QINIU_DOMAIN='{}'".format(qiniu_domain), "DIST_TASK_NAME='{}'".format(name), ]) # 日志按任务名，便于查；结果写 worker 本地 results/ log_file = "{}/worker_jobs/{}.log".format(worker_cwd, name) + # 关键：SSH 远端启动后台进程必须用 setsid 完全脱离 SSH 会话，否则 SSH + # 客户端会一直等后台进程释放 stdio fd，导致 ssh_ok 超时（实测 nohup & echo + # 会 hang 到 30s 超时；setsid bash -c '... & echo $!' {logf} 2>&1 & echo $!" + "setsid bash -c '{env} nohup python3 src/run_one.py {args} {seed} " + "--results {res} --timeout {to} {logf} 2>&1 & echo $!' /；这里指定 results 让它和冷启动同目录 + # 同 dispatch_task：用 setsid 脱离 SSH 会话，否则 SSH 客户端 hang 到超时。 + # 关键：内层 bash -c 必须用单引号包裹（和 dispatch_task 一致），否则 $! 会被 + # 外层 SSH shell 先展开为空（外层无后台进程），导致拿不到 pid → pid 探活失效 + # → seed_step 被重复派发。seed_path 是 .seed_cache/xxx.seed.7，不含单引号，安全。 run_cmd = ( "mkdir -p {cwd}/worker_jobs && cd {cwd} && " - "{env} nohup python3 src/seed_step.py {args} --seed '{seed}' " - "--results {res} > {logf} 2>&1 & echo $!" + "setsid bash -c '{env} nohup python3 src/seed_step.py {args} --seed {seed} " + "--results {res} {logf} 2>&1 & echo $!' /.7： + # 既作为该模型的大气产物，也作为后续 seed_step 的种子源 + # （run_grid.find_seed 扫的就是 results//.7）。 + # localhost 这台 worker 与 master 共用 results（同符号链接）， + # scp 同文件无害；异地 worker 则必须取回才能供后续点做种子。 + local_seed = os.path.join(self.results_root, name, name + ".7") + os.makedirs(os.path.dirname(local_seed), exist_ok=True) + fetch_seed(self.target, self.worker_cwd, self.results, + name, local_seed) report_task(self.tasks_path, name, "done", max_relc=max_relc, elapsed_sec=elapsed, - seed_step_used=seed_step_used, note="converged") + seed_step_used=task_seed_step_used, note="converged") self.log("{} DONE max_relc={}".format(name, max_relc)) else: # 冷启动未收敛：尝试 seed_step 补算（若有种子）。 - # 找种子：master 本地 seed_dir > 七牛云。seed_step 只试一次。 + # 找种子：扫 master 本地 results/（收敛点 .7 汇聚处）。 + # seed_step 只试一次。 if not self.seed_step_on: report_task(self.tasks_path, name, "failed", max_relc=max_relc, elapsed_sec=elapsed, note="unconverged cold-start (seed_step off)") self.log("{} FAILED (seed_step 关闭)".format(name)) continue - # 已尝试过 seed_step 仍失败 → 终态 failed - if seed_step_used or data["tasks"][name].get("attempts", 0) >= 2: + # 已尝试过 seed_step 仍失败 → 终态 failed。 + # 注意：以 tasks.json 的 seed_step_used 为准（master 自己标的）， + # 不能用 conv.json 的（永远是 None，会死循环）。 + if task_seed_step_used or data["tasks"][name].get("attempts", 0) >= 2: report_task(self.tasks_path, name, "failed", max_relc=max_relc, elapsed_sec=elapsed, - seed_step_used=seed_step_used, + seed_step_used=task_seed_step_used, note="unconverged after seed_step (物理极限?)") self.log("{} FAILED (seed_step 后仍发散)".format(name)) continue @@ -522,86 +790,44 @@ class WorkerManager(threading.Thread): note="seed_step: push seed failed") self.log("{} seed_step 种子推送失败".format(name)) continue - # 标记 running（attempts++），重新派 seed_step + # 派 seed_step（dispatch_seed_step 返回 pid_out, err） + pid_out, err = dispatch_seed_step( + self.target, self.worker_cwd, self.tlusty_root, + name, params, self.results, worker_seed, timeout=self.timeout) + ss_pid = _parse_pid(pid_out) + # 标记 running + seed_step_used + 记录 pid。 + # 关键：seed_step_pid 让下一轮 _collect_results 能用 pid 探活判断 + # seed_step 是否还在跑——还在跑就跳过（避免据旧 conv.json 重复派发）。 def mark_retry(d): t = d["tasks"].get(name) if t: t["status"] = "running" t["seed_step_used"] = True + t["seed_step_pid"] = ss_pid t["claimed_at"] = time.time() _lock_tasks(self.tasks_path, mark_retry) - # 备份 worker 上的冷启动失败结果（seed_step 会覆盖同目录） - # seed_step.py 默认写 results/seed_step/，不会覆盖冷启动目录； - # 但为统一回收，这里让它写到 results/ 下同 name 目录会冲突， - # 所以让 seed_step 写到 seed_step/ 子目录，回收时合并。 - pid, err = dispatch_seed_step( - self.target, self.worker_cwd, self.tlusty_root, - name, params, self.results, worker_seed, timeout=self.timeout) - self.log("{} seed_step 重试 (seed={})".format( - name, os.path.basename(worker_seed))) + self.log("{} seed_step 重试 pid={} (seed={})".format( + name, ss_pid, os.path.basename(worker_seed))) def _find_seed_for(self, params): - """为某点找种子 .7：master 本地 seed_dir > 七牛云。返回本地路径或 None。""" - # 本地：扫 seed_dir 找最近邻 - teff, logg, loghe = params["teff"], params["logg"], params["loghe"] - logc, logn, logo = params["logc"], params["logn"], params["logo"] - best = None - best_d = None - if os.path.isdir(self.seed_dir): - for fn in os.listdir(self.seed_dir): - if not fn.endswith(".7"): - continue - nm = fn[:-2] - try: - parts = nm.split("_") - pte = int(parts[0][1:]); plogg = float(parts[1][1:]) - phe = float(parts[2][2:]); pc = float(parts[3][1:]) - pn = float(parts[4][1:]); po = float(parts[5][1:]) - except (IndexError, ValueError): - continue - if pte == teff and plogg == logg and phe == loghe: - d = abs(pc - logc) + abs(pn - logn) + abs(po - logo) - else: - d = (abs(pte - teff) / 5000.0 + abs(plogg - logg) * 2.0 - + abs(phe - loghe) * 0.5) * 10 # 同 family 优先 - if best_d is None or d < best_d: - best_d, best = d, os.path.join(self.seed_dir, fn) - if best is not None and best_d is not None and best_d < 100: - return best - # 七牛云 - if self.store.domain: - try: - names = self.store.list_seeds() - qb, qd = None, None - for nm in names: - try: - parts = nm.split("_") - pte = int(parts[0][1:]); plogg = float(parts[1][1:]) - phe = float(parts[2][2:]); pc = float(parts[3][1:]) - pn = float(parts[4][1:]); po = float(parts[5][1:]) - except (IndexError, ValueError): - continue - if pte == teff and plogg == logg and phe == loghe: - d = abs(pc - logc) + abs(pn - logn) + abs(po - logo) - else: - d = (abs(pte - teff) / 5000.0 + abs(plogg - logg) * 2.0 - + abs(phe - loghe) * 0.5) * 10 - if qd is None or d < qd: - qd, qb = d, nm - if qb is not None and qd is not None and qd < 100: - local = os.path.join(self.seed_dir, qb + ".7") - if not os.path.exists(local): - if self.store.download_seed(qb, local): - return local - else: - return local - except Exception: - pass - return None + """为某点找邻居种子 .7：扫 master 本地 results/。 + + 复用 run_grid.find_seed 的成熟逻辑：遍历 results//， + 校验 conv.json 的 converged 且 atmosphere_has_nan，按"同 family + (Teff/logg/logHe) 的 CNO 最近邻 → 全局最近邻"优先级返回。这样 + 低金属 wave 先算完收敛的 .7 自然成为高金属点 seed_step 的种子。 + + 返回 .7 的本地绝对路径或 None。find_seed 在无可用种子时返回 None + 是安全的（LTE-grey 冷启动可独立工作）。 + """ + return _find_seed_in_results( + self.results_root, + params["teff"], params["logg"], params["loghe"], + params["logc"], params["logn"], params["logo"]) def write_grid_status(tasks_path, results_root, elapsed): - data = json.load(open(tasks_path)) + data = _read_tasks(tasks_path) counts = count_statuses(data) seed_step_count = sum(1 for t in data["tasks"].values() if t.get("seed_step_used")) @@ -627,13 +853,12 @@ def main(): ap.add_argument("dist_config") ap.add_argument("--dry-run", action="store_true") ap.add_argument("--phase", default=None, help="phase1=冷启动；phase2=补算") - ap.add_argument("--stale-sec", type=int, default=7200, - help="running 超此时长无回收 → 重派（默认 2 小时）") + ap.add_argument("--stale-sec", type=int, default=2400, + help="running 超此时长无回收 → 重派（默认 40 分钟；" + "略大于单点典型耗时，配合 pid 探活快速识别崩溃）") ap.add_argument("--force-init", action="store_true") ap.add_argument("--poll-sec", type=int, default=20) ap.add_argument("--timeout", type=int, default=3600, help="单点墙钟上限") - ap.add_argument("--no-qiniu", action="store_true", - help="不传种子到七牛（仅本地）") args = ap.parse_args() dcfg = _load_yaml(args.dist_config) @@ -645,7 +870,6 @@ def main(): results_root = results if os.path.isabs(results) else os.path.join(workdir, results) os.makedirs(results_root, exist_ok=True) tasks_path = os.path.join(workdir, "tasks.json") - seed_dir = os.path.join(results_root, ".seeds_remote") data = init_tasks(grid_cfg, results_root, tasks_path, grid_id=args.dist_config, force=args.force_init) @@ -653,16 +877,6 @@ def main(): print("网格任务：{}".format(counts), flush=True) workers = dcfg.get("workers", []) - qiniu = dcfg.get("qiniu", {}) - store = qiniu_store.QiniuStore( - access_key=qiniu.get("access_key"), - secret_key=qiniu.get("secret_key"), - bucket=qiniu.get("bucket"), - domain=qiniu.get("domain"), - seed_prefix=qiniu.get("seed_prefix", "seeds")) - if args.no_qiniu: - store = qiniu_store.QiniuStore() # 空 store，不传 - qiniu_upload_by = qiniu.get("upload_by", "master") print("worker 节点 {} 个：".format(len(workers)), flush=True) for i, w in enumerate(workers): @@ -675,11 +889,27 @@ def main(): est = to_compute * 1400 / max(total_cores, 1) print("\n待计算 {} 点，总并发 {} 核".format(to_compute, total_cores)) print("预计 ~{:.1f} 小时（单点 1400s 估算）".format(est / 3600)) + # 显示 wave 划分 + wave_stats = {} + for t in data["tasks"].values(): + w = t.get("wave", 0) + if w not in wave_stats: + wave_stats[w] = {"total": 0, "pending": 0, "done": 0} + wave_stats[w]["total"] += 1 + if t["status"] == "pending": + wave_stats[w]["pending"] += 1 + elif t["status"] == "done": + wave_stats[w]["done"] += 1 + print("\nwave 调度（按 CNO 总量分批，低金属先算）：") + for wi in sorted(wave_stats): + ws = wave_stats[wi] + print(" wave {}: {} 点 (pending={}, done={})".format( + wi, ws["total"], ws["pending"], ws["done"])) print("\n部署检查清单：") print(" 1. master 能 SSH 进每台 worker（免密）") print(" 2. 每台 worker 上 {}/src/run_one.py 可执行".format( "/cno_grid")) - print(" 3. 七牛: {}".format("已配置" if store.domain else "未配置(本地模式)")) + print(" 3. 种子源: master 本地 {}（收敛点 .7 汇聚处）".format(results_root)) return if not workers: @@ -693,9 +923,9 @@ def main(): wm = WorkerManager( i, w["host"], w["tlusty_root"], w.get("nproc", 1), os.path.join(w["tlusty_root"], "cno_grid"), - results, tasks_path, store, qiniu_upload_by, + results, tasks_path, os.path.basename(grid_config_path), args.timeout, - args.stale_sec, args.phase, True, seed_dir) + args.stale_sec, args.phase, True, results_root) wm.start() managers.append(wm) @@ -706,17 +936,35 @@ def main(): try: while any(m.is_alive() for m in managers): time.sleep(args.poll_sec) - reset = reset_stale(tasks_path, args.stale_sec) - if reset: - print(" [monitor] 重派 {} 个僵死任务".format(reset), flush=True) - data = json.load(open(tasks_path)) - counts = count_statuses(data) - elapsed = time.time() - t0 - print("[{:.0f}min] {}".format(elapsed / 60, counts), flush=True) - write_grid_status(tasks_path, results_root, elapsed) - if counts.get("pending", 0) == 0 and counts.get("running", 0) == 0: - print("\n所有任务终态。", flush=True) - break + try: + reset = reset_stale(tasks_path, args.stale_sec) + if reset: + print(" [monitor] 重派 {} 个僵死任务".format(reset), flush=True) + data = _read_tasks(tasks_path) + counts = count_statuses(data) + elapsed = time.time() - t0 + # 显示当前活跃 wave + active_wave = None + total_waves = 0 + for t in data["tasks"].values(): + w = t.get("wave", 0) + if w > total_waves: + total_waves = w + if t["status"] in ("pending", "running"): + if active_wave is None or w < active_wave: + active_wave = w + wave_info = " wave={}/{}".format(active_wave, total_waves) \ + if active_wave is not None else " wave=all done" + print("[{:.0f}min] {}{}".format(elapsed / 60, counts, wave_info), + flush=True) + write_grid_status(tasks_path, results_root, elapsed) + if counts.get("pending", 0) == 0 and counts.get("running", 0) == 0: + print("\n所有任务终态。", flush=True) + break + except RuntimeError as e: + # tasks.json 损坏等：打印但继续循环，不要退出（否则 master 误死亡） + print(" [monitor] 临时错误（继续）: {}".format(e), flush=True) + time.sleep(10) except KeyboardInterrupt: print("\n中断。worker 远端仍在跑，重跑本命令可续算。", flush=True) for m in managers: diff --git a/cno_grid/src/qiniu_store.py b/cno_grid/src/qiniu_store.py deleted file mode 100644 index 888af9b..0000000 --- a/cno_grid/src/qiniu_store.py +++ /dev/null @@ -1,263 +0,0 @@ -#!/usr/bin/env python3 -"""七牛云对象存储的轻量封装（零 SDK 依赖）。 - -仅用 Python 标准库（hmac/hashlib/base64/json）+ curl 实现七牛云的上传/ -下载/列举。worker 端无需 pip 装任何东西。 - -七牛上传签名机制： - PutPolicy = {"scope":":", "deadline":} - 编码后用 AccessKey/SecretKey 做 HMAC-SHA1 签名，得到 uploadToken。 - -用途（本分布式网格）： - - 七牛云当"种子库"：每个收敛模型的 .7 大气（393KB）上传为 - /.7，供异地 worker 在 seed_step 回退时拉取。 - - 仅 master 持有 AK/SK 生成 uptoken；worker 只拿 token 上传（无需密钥）。 - - 大产物（.spec 等 13MB）不上传，留各机本地。 - -环境变量： - QINIU_ACCESS_KEY / QINIU_SECRET_KEY / QINIU_BUCKET / QINIU_DOMAIN - （可被 dist_config.yaml 的 qiniu 段覆盖；master 在派发时把 token 传给 worker） -""" -import base64 -import hashlib -import hmac -import json -import os -import subprocess -import time - -try: - import urllib.request as urlreq -except ImportError: # py2 兜底（不会触发） - urlreq = None - - -class QiniuStore: - """七牛云存储客户端。master 与 worker 共用；worker 只用 download/upload(token)。""" - - def __init__(self, access_key=None, secret_key=None, bucket=None, - domain=None, seed_prefix="seeds"): - self.ak = access_key or os.environ.get("QINIU_ACCESS_KEY", "") - self.sk = secret_key or os.environ.get("QINIU_SECRET_KEY", "") - self.bucket = bucket or os.environ.get("QINIU_BUCKET", "") - # 绑定域名（如 http:// 或 https:// + 域名，无尾斜杠）。公开读下载用。 - self.domain = (domain or os.environ.get("QINIU_DOMAIN", "")).rstrip("/") - self.seed_prefix = seed_prefix - - # ---- base64 url-safe（七牛要求）---- - @staticmethod - def _b64u(data): - """bytes -> url-safe base64 字符串（去 padding 也可，七牛两边都接受）。""" - if isinstance(data, str): - data = data.encode("utf-8") - return base64.urlsafe_b64encode(data).decode("ascii") - - @staticmethod - def _b64u_json(obj): - return QiniuStore._b64u(json.dumps(obj, separators=(",", ":"))) - - def _sign(self, data): - """HMAC-SHA1(data, sk) -> url-safe base64。""" - if isinstance(data, str): - data = data.encode("utf-8") - return self._b64u(hmac.new(self.sk.encode("utf-8"), data, - hashlib.sha1).digest()) - - # ---- uptoken 生成（master 持有 AK/SK 时调用）---- - def gen_uptoken(self, key, expires=3600): - """为指定 key 生成上传 token。 - - key: 对象名（不含 bucket 前缀），如 "seeds/t30000_g5.0_he0_c-1_n-1_o-1.7" - 返回 uploadToken 字符串，交给 worker 用 curl 上传。 - """ - scope = "{}:{}".format(self.bucket, key) if key else self.bucket - policy = {"scope": scope, "deadline": int(time.time()) + expires} - encoded = self._b64u_json(policy) - sign = self._sign(encoded) - return "{}:{}".format(self.ak, sign) + ":" + encoded - - def gen_private_url(self, key, expires=3600): - """生成私有空间的临时下载 URL（带 e/token 签名）。 - - 若 bucket 为公开读，直接用 domain/key 即可，不必调这个。 - """ - if not self.domain: - raise ValueError("QINIU_DOMAIN 未配置") - url = "{}/{}".format(self.domain, key) - e = int(time.time()) + expires - to_sign = "{}?e={}".format(url, e) - token = "{}:{}".format(self.ak, self._sign(to_sign)) - return "{}?e={}&token={}".format(url, e, token) - - def public_url(self, key): - """公开空间的下载 URL（无签名）。""" - if not self.domain: - raise ValueError("QINIU_DOMAIN 未配置") - return "{}/{}".format(self.domain, key) - - # ---- 上传 / 下载（通过 curl，worker 无需 pip）---- - def upload(self, local_path, key, uptoken=None): - """上传本地文件到七牛。uptoken 可由 master 预生成传入。 - - 用 curl 表单 POST（七牛上传接口）： - POST http://upload.qiniup.com multipart/form-data - field "token"=uptoken, "key"=key, "file"=@local_path - 返回 (ok:bool, resp_text)。 - """ - if uptoken is None: - uptoken = self.gen_uptoken(key) - # 七牛上传域名：华东 upload.qiniup.com，其它区域用对应域名。 - upload_host = os.environ.get("QINIU_UPLOAD_HOST", "upload.qiniup.com") - url = "http://{}/".format(upload_host) - cmd = ["curl", "-sS", "-m", "300", "-X", "POST", url, - "-F", "token={}".format(uptoken), - "-F", "key={}".format(key), - "-F", "file=@{}".format(local_path)] - try: - out = subprocess.run(cmd, capture_output=True, text=True, - timeout=320) - except subprocess.TimeoutExpired: - return False, "upload timeout (>300s)" - if out.returncode != 0: - return False, out.stderr.strip() or out.stdout.strip() - try: - resp = json.loads(out.stdout) - except ValueError: - return False, out.stdout.strip() - # 成功响应含 "key" 与 "hash"；失败含 "error" - if "error" in resp: - return False, resp["error"] - return True, resp.get("key", "") - - def download(self, key, local_path, timeout=120): - """下载对象到本地。公开空间用 public_url，私有空间用 private_url。 - - 返回 True/False。 - """ - if self.ak and self.sk: - url = self.gen_private_url(key) - else: - url = self.public_url(key) - cmd = ["curl", "-sS", "-m", str(timeout), "-f", "-o", local_path, url] - try: - r = subprocess.run(cmd, capture_output=True, timeout=timeout + 10) - return r.returncode == 0 - except subprocess.TimeoutExpired: - return False - - # ---- 种子库语义封装 ---- - def seed_key(self, model_name): - """规范种子对象名：seeds/.7""" - return "{}/{}.7".format(self.seed_prefix, model_name) - - def upload_seed(self, local_path, model_name, uptoken=None): - """上传某模型的最终大气 .7 作为种子。返回 (ok, msg)。""" - return self.upload(local_path, self.seed_key(model_name), uptoken) - - def download_seed(self, model_name, local_path): - """下载某模型的种子 .7 到 local_path。返回 True/False。""" - return self.download(self.seed_key(model_name), local_path) - - # ---- 列举（用于 find_seed 跨网查最近邻种子）---- - def list_seeds(self): - """返回七牛种子库中所有 .7 的模型名列表。 - - 调用七牛 RS API: GET /list?bucket=&prefix= - （需 AK/SK 签名）。返回 [] 当未配置或失败。 - 结果缓存到本地文件避免每次都拉（见 _seeds_cache_path）。 - """ - if not (self.ak and self.sk and self.bucket): - return [] - cache = self._seeds_cache_path() - # 缓存 5 分钟 - if os.path.isfile(cache) and time.time() - os.path.getmtime(cache) < 300: - try: - return json.load(open(cache)) - except Exception: - pass - names = self._fetch_seed_list() - try: - with open(cache, "w") as f: - json.dump(names, f) - except Exception: - pass - return names - - def _seeds_cache_path(self): - return os.path.join(os.path.dirname(os.path.abspath(__file__)), - "..", ".qiniu_seeds_cache.json") - - def _fetch_seed_list(self): - """分页拉取七牛种子列表，返回 model_name 列表。""" - rs_host = os.environ.get("QINIU_RS_HOST", "http://rsf.qiniuapi.com") - names = [] - marker = "" - prefix = self.seed_prefix + "/" - while True: - path = "/list?bucket={}&prefix={}".format(self.bucket, prefix) - if marker: - path += "&marker={}".format(marker) - # 签名：AccessToken = AK:urlsafe_b64(hmac(sk, path)): - encoded_path = path # 七牛签名用原始 path - sign = self._sign(encoded_path) - access_token = "{}:{}".format(self.ak, sign) + encoded_path - url = rs_host + path - cmd = ["curl", "-sS", "-m", "60", - "-H", "Authorization: QBox " + access_token, url] - try: - r = subprocess.run(cmd, capture_output=True, text=True, - timeout=70) - if r.returncode != 0: - break - data = json.loads(r.stdout) - except (ValueError, subprocess.TimeoutExpired): - break - for item in data.get("items", []): - k = item.get("key", "") - # seeds/.7 -> - if k.startswith(prefix) and k.endswith(".7"): - names.append(k[len(prefix):-2]) - marker = data.get("marker", "") - if not marker: - break - return names - - -def load_from_dist_config(cfg): - """从 dist_config.yaml 的 qiniu 段构造 QiniuStore。""" - q = (cfg or {}).get("qiniu", {}) if isinstance(cfg, dict) else {} - return QiniuStore( - access_key=q.get("access_key"), - secret_key=q.get("secret_key"), - bucket=q.get("bucket"), - domain=q.get("domain"), - seed_prefix=q.get("seed_prefix", "seeds"), - ) - - -if __name__ == "__main__": - # 自检：打印 token（需配置 AK/SK/Bucket） - import argparse - ap = argparse.ArgumentParser(description="qiniu store 自检") - ap.add_argument("--gen-token", help="为某 key 生成 uptoken") - ap.add_argument("--list", action="store_true", help="列出种子") - ap.add_argument("--config", help="dist_config.yaml 路径") - args = ap.parse_args() - if args.config: - try: - import yaml - cfg = yaml.safe_load(open(args.config)) - except Exception: - # 兜底：不装 yaml 时只从环境变量读 - cfg = {"qiniu": {}} - store = load_from_dist_config(cfg) - else: - store = QiniuStore() - if args.gen_token: - print(store.gen_uptoken(args.gen_token)) - elif args.list: - for n in store.list_seeds(): - print(n) - else: - print("AK set:", bool(store.ak), "Bucket:", store.bucket, - "Domain:", store.domain or "(未配置)") diff --git a/cno_grid/src/run_grid.py b/cno_grid/src/run_grid.py index 556ac7a..f817804 100644 --- a/cno_grid/src/run_grid.py +++ b/cno_grid/src/run_grid.py @@ -187,6 +187,18 @@ def model_done(results_root, name): return False +def model_attempted(results_root, name): + """该模型是否已经跑过一轮（存在 conv.json，不论是否收敛）。 + + conv.json 由 run_model 在收敛链 + synspec 都结束后才写 + （run_one.py 末尾），所以它存在即意味着该点曾被完整尝试过—— + 不论最终是 converged / unfinished / error / 假收敛被作废。 + 配合 model_done() 可区分"从未跑过"与"跑过但失败"两种情况， + 供 --skip-failed 使用。 + """ + return os.path.exists(os.path.join(results_root, name, "conv.json")) + + def _atmos_clean(atmos_path): """检查大气文件是否干净（无 NaN 污染）。 @@ -269,62 +281,22 @@ def find_seed(results_root, teff, logg, loghe, logc=None, logn=None, logo=None): local_seed = exact_family or closest if local_seed is not None: return local_seed - # 分布式模式回退：本地无种子时，查七牛云种子库（QINIU_SEED_FALLBACK=1 开启）。 - # 单机本地模式默认关闭，完全不影响原有行为。 - if os.environ.get("QINIU_SEED_FALLBACK") == "1": - try: - import qiniu_store - store = qiniu_store.QiniuStore() - names = store.list_seeds() - if not names: - return None - best = None - best_d = None - best_exact = None - best_exact_d = None - for nm in names: - try: - parts = nm.split("_") - pte = int(parts[0][1:]) - plogg = float(parts[1][1:]) - phe = float(parts[2][2:]) - pc = float(parts[3][1:]) - pn = float(parts[4][1:]) - po = float(parts[5][1:]) - except (IndexError, ValueError): - continue - if pte == teff and plogg == logg and phe == loghe: - dcno = abs(pc - logc) + abs(pn - logn) + abs(po - logo) - if best_exact_d is None or dcno < best_exact_d: - best_exact_d, best_exact = dcno, nm - continue - d_tot = (abs(pte - teff) / 5000.0 - + abs(plogg - logg) * 2.0 - + abs(phe - loghe) * 0.5) - if best_d is None or d_tot < best_d: - best_d, best = d_tot, nm - chosen = best_exact or best - if chosen is None: - return None - cache = os.path.join(results_root, ".seed_cache", chosen + ".7") - os.makedirs(os.path.dirname(cache), exist_ok=True) - if not os.path.exists(cache): - if not store.download_seed(chosen, cache): - return None - return cache - except Exception: - return None return None def _worker(args): """Pool worker：跑一个模型。必须是顶层函数以便 pickle。""" (pt, results_root, template, fort55, linelist, chain, itek_fallback, - niter, timeout, seed_step_fallback) = args + niter, timeout, seed_step_fallback, skip_failed) = args name = gen_input5.model_name(pt["teff"], pt["logg"], pt["loghe"], pt["logc"], pt["logn"], pt["logo"]) if model_done(results_root, name): return {"name": name, "status": "skipped"} + # --skip-failed：跳过已跑过但未收敛的点（只算从没跑过的点）。 + # 用于先把整个网格跑完一遍拿到全局种子库，再用 repair_grid.py 回头 + # 收拾失败点。与 "skipped" 区分以便统计。 + if skip_failed and model_attempted(results_root, name): + return {"name": name, "status": "skipped_failed"} # 把 chain/fallback patch 到模块级，让 run_model 能取到 if chain: run_one.DEFAULT_CHAIN = chain @@ -380,6 +352,9 @@ def main(): help="filter, e.g. teff=35000,logg=5.5 (exact matches)") ap.add_argument("--limit", type=int, default=None, help="compute at most N models (for testing)") + ap.add_argument("--skip-failed", action="store_true", + help="skip points already attempted but not converged " + "(only compute never-attempted points)") args = ap.parse_args() with open(args.config) as f: @@ -401,6 +376,10 @@ def main(): k, v = part.split("=") filt[k.strip()] = float(v) + # 跳过已尝试但未收敛的点（只算从没跑过的点）。CLI 优先，config 兜底。 + # 用于先把整个网格推一遍建立全局种子库，再用 repair_grid.py 回头收拾失败点。 + skip_failed = args.skip_failed or bool(cfg.get("skip_failed", False)) + points = [] for pt in expand_grid_ordered(cfg["grid"]): if all(pt.get(k) == v for k, v in filt.items()): @@ -414,8 +393,21 @@ def main(): gen_input5.model_name( pt["teff"], pt["logg"], pt["loghe"], pt["logc"], pt["logn"], pt["logo"]))) - print("grid: {} points total, {} already done, {} to compute".format( - n_total, n_skip, n_total - n_skip)) + # 跑过但未收敛的点数（仅在 --skip-failed 时影响实际计算量） + n_skip_failed = sum( + 1 for pt in points + if skip_failed + and model_attempted( + results_root, + gen_input5.model_name(pt["teff"], pt["logg"], pt["loghe"], + pt["logc"], pt["logn"], pt["logo"])) + and not model_done( + results_root, + gen_input5.model_name(pt["teff"], pt["logg"], pt["loghe"], + pt["logc"], pt["logn"], pt["logo"]))) + print("grid: {} points total, {} done, {} failed({}skip), {} to compute".format( + n_total, n_skip, n_skip_failed, "will " if skip_failed else "won't ", + n_total - n_skip - n_skip_failed)) os.makedirs(results_root, exist_ok=True) chain = cfg.get("chain") @@ -430,7 +422,8 @@ def main(): # 显示排序后的顺序 + wave 划分 waves = _split_waves( [(pt, results_root, template, fort55, linelist, - chain, itek_fallback, niter, timeout, seed_step_fallback) + chain, itek_fallback, niter, timeout, seed_step_fallback, + skip_failed) for pt in points], nworkers) print("scheduling: {} wave(s) by CNO total abundance".format(len(waves))) for wi, wave in enumerate(waves): @@ -450,7 +443,8 @@ def main(): return worker_args = [(pt, results_root, template, fort55, linelist, - chain, itek_fallback, niter, timeout, seed_step_fallback) + chain, itek_fallback, niter, timeout, seed_step_fallback, + skip_failed) for pt in points] t0 = time.time() diff --git a/cno_grid/src/run_one.py b/cno_grid/src/run_one.py index 7de71dc..4df7e65 100644 --- a/cno_grid/src/run_one.py +++ b/cno_grid/src/run_one.py @@ -353,8 +353,10 @@ def run_model(teff, logg, loghe, logc, logn, logo, if os.path.exists(final_7): try: with open(final_7) as f: - atmos_has_nan = sum( - 1 for l in f if "nan" in l.lower()) > len(open(final_7).readlines()) * 0.1 + lines = f.readlines() + if lines: + nan_count = sum(1 for l in lines if "nan" in l.lower()) + atmos_has_nan = nan_count > len(lines) * 0.1 except Exception: pass summary["atmosphere_has_nan"] = atmos_has_nan