fmq/TLUSTY
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

3

Browse Source
This commit is contained in:
fmq 2026-07-27 10:40:18 +08:00
parent 92d9c515c9
commit 4ee085ecde
7 changed files with 764 additions and 554 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

223
cno_grid/DIST.md
View File

@ -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/<model_name>/<model_name>.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/<name>/<name>.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/<name>.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 <pid>
- **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/<name>/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/<name>.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 用密钥免密，不用密码

31
cno_grid/dist_config.yaml
View File

@ -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

125
cno_grid/src/_check.py Normal file
View File

@ -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())

576
cno_grid/src/dist_master.py
View File

@ -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/<name>/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 $!' </dev/null 0.3s 返回）。
# 内层 </dev/null 关闭后台进程 stdin，外层 </dev/null 让 setsid 不读 SSH stdin。
run_cmd = (
"mkdir -p {cwd}/worker_jobs && cd {cwd} && "
"{env} nohup python3 src/run_one.py {args} {seed} "
"--results {res} --timeout {to} > {logf} 2>&1 & echo $!"
"setsid bash -c '{env} nohup python3 src/run_one.py {args} {seed} "
"--results {res} --timeout {to} </dev/null > {logf} 2>&1 & echo $!' </dev/null"
).format(cwd=worker_cwd, env=env, args=args, seed=cmd_seed,
res=results, to=timeout, logf=log_file)
rc, pid, err = ssh_ok(target, run_cmd, timeout=30)
@ -304,8 +490,8 @@ def dispatch_seed_step(target, worker_cwd, tlusty_root, name, params,
results, seed_path, timeout=3600):
"""SSH 派一个 seed_step 补算任务到 worker（异步 nohup）。
冷启动失败后，master 从本地/七牛拉到种子 .7，scp 到 worker，再调
seed_step.py（LTGRAY=F 热启动）重试。
冷启动失败后，master 从本地 results/ 找到邻居种子 .7，scp 到 worker，
再调 seed_step.py（LTGRAY=F 热启动）重试。
"""
args = ("--teff {teff} --logg {logg} --loghe {loghe} "
"--logc {logc} --logn {logn} --logo {logo}").format(**params)
@ -313,10 +499,14 @@ def dispatch_seed_step(target, worker_cwd, tlusty_root, name, params,
env = "TLUSTY='{}'".format(tlusty_root)
log_file = "{}/worker_jobs/{}.seedstep.log".format(worker_cwd, name)
# seed_step.py 把结果写到 results/seed_step/<name>/；这里指定 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} </dev/null > {logf} 2>&1 & echo $!' </dev/null"
).format(cwd=worker_cwd, env=env, args=args, seed=seed_path,
res=results, logf=log_file)
rc, pid, err = ssh_ok(target, run_cmd, timeout=30)
@ -324,7 +514,22 @@ def dispatch_seed_step(target, worker_cwd, tlusty_root, name, params,
def push_seed_to_worker(target, local_seed, remote_seed):
"""scp master 本地的 .7 种子到 worker。返回 worker 上路径或 None。"""
"""把 master 本地的 .7 种子推到 worker。返回 worker 上路径或 None。
localhost 直接 shutil.copy（同文件系统）；异地用 scp。
两种方式都先确保 remote_seed 的父目录存在（scp 不会自动建父目录，
之前 .seed_cache 不存在导致种子推送失败、任务误判 failed）。
"""
remote_dir = os.path.dirname(remote_seed)
if is_local_target(target):
try:
os.makedirs(remote_dir, exist_ok=True)
shutil.copy(local_seed, remote_seed)
return remote_seed
except OSError:
return None
# 异地：先 ssh 建目录，再 scp
ssh_ok(target, "mkdir -p '{}'".format(remote_dir), timeout=15)
try:
r = subprocess.run(["scp", "-o", "BatchMode=yes", "-o", "ConnectTimeout=15",
local_seed, "{}:{}".format(target, remote_seed)],
@ -337,11 +542,20 @@ def push_seed_to_worker(target, local_seed, remote_seed):
def fetch_worker_status(target, worker_cwd, results):
"""SSH 查 worker 上所有已完成任务（有 conv.json 的）。
"""查 worker 上所有已完成任务（有 conv.json 的）。
返回 {model_name: conv_summary_dict}。供 master 决定哪些该回收。
用 dist_check.py（一次列出全部，比逐个 cat 高效）。
localhost 走本地文件系统（直接调 dist_check.scan_results），避免依赖
WSL sshd（常断，导致回收静默失败、任务卡 running，见 #2）。异地 worker
经 SSH 调 dist_check.py。
"""
if is_local_target(target):
local_results = results if os.path.isabs(results) \
else os.path.join(worker_cwd, results)
if not os.path.isdir(local_results):
return {}
import dist_check
return dist_check.scan_results(local_results)
cmd = "cd {cwd} && python3 src/dist_check.py --results {res} 2>/dev/null".format(
cwd=worker_cwd, res=results)
rc, out, err = ssh_ok(target, cmd, timeout=40)
@ -354,8 +568,20 @@ def fetch_worker_status(target, worker_cwd, results):
def fetch_seed(target, worker_cwd, results, name, dest_path):
"""scp 取 worker 上的 .7 大气到 master 本地 dest_path。"""
src = "{}:{}/{}/{}.7".format(target, worker_cwd.rstrip("/"), results, name)
"""取 worker 上的 .7 大气到 master 本地 dest_path。
localhost 走本地 shutil.copy（同文件系统），异地 worker 用 scp。
"""
src_remote = os.path.join(worker_cwd.rstrip("/"), results, name, name + ".7")
if is_local_target(target):
if os.path.isfile(src_remote):
try:
shutil.copy(src_remote, dest_path)
return True
except OSError:
return False
return False
src = "{}:{}/{}/{}/{}.7".format(target, worker_cwd.rstrip("/"), results, name, name)
try:
r = subprocess.run(["scp", "-o", "BatchMode=yes", "-o", "ConnectTimeout=15",
src, dest_path], capture_output=True, timeout=60)
@ -394,8 +620,8 @@ class WorkerManager(threading.Thread):
"""每台 worker 一个线程：填充空闲槽 + 回收已完成。"""
def __init__(self, wid, target, tlusty_root, nproc, worker_cwd,
results, tasks_path, store, qiniu_upload_by, grid_config,
timeout, stale_sec, phase, seed_step_on, seed_dir):
results, tasks_path, grid_config,
timeout, stale_sec, phase, seed_step_on, results_root):
super().__init__(daemon=True)
self.wid = wid
self.target = target
@ -404,21 +630,24 @@ class WorkerManager(threading.Thread):
self.worker_cwd = worker_cwd
self.results = results
self.tasks_path = tasks_path
self.store = store
self.qiniu_upload_by = qiniu_upload_by
self.grid_config = grid_config
self.timeout = timeout
self.stale_sec = stale_sec
self.phase = phase
self.seed_step_on = seed_step_on
self.seed_dir = seed_dir # master 本地存从 worker 取回的 .7
# master 本地 results 根：异地 worker 的收敛 .7 汇聚于此，
# seed_step 找种子时也从这里扫（run_grid.find_seed）。
self.results_root = results_root
# master 本地记录的活跃 pid（dispatch_task 时 echo $! 拿到）。
# 用于精确计算空闲槽，避免 pgrep 误匹配查询命令自身（见 #1）。
self._active_pids = set()
self.stop = False
def log(self, msg):
print(" [w{} {}] {}".format(self.wid, self.target, msg), flush=True)
def run(self):
os.makedirs(self.seed_dir, exist_ok=True)
os.makedirs(self.results_root, exist_ok=True)
while not self.stop:
try:
self._fill_slots()
@ -426,15 +655,28 @@ class WorkerManager(threading.Thread):
except Exception as e:
self.log("循环异常: {}".format(e))
# 检查是否还有 pending
data = json.load(open(self.tasks_path))
data = _read_tasks(self.tasks_path)
counts = count_statuses(data)
if counts.get("pending", 0) == 0 and counts.get("running", 0) == 0:
break
time.sleep(15)
def _fill_slots(self):
"""查 worker 空闲槽，派任务。"""
running = count_running_on_worker(self.target, self.worker_cwd)
"""查 worker 空闲槽，派任务。
空闲槽计算优先用 master 本地记录的活跃 pid（kill -0 探活），精确且
不受 pgrep 误匹配影响（pgrep 会把查询命令自身的 shell 算进去，见 #1）。
pid 集合为空时（如刚启动、或 pid 都已退出）才退回 pgrep 兜底。
派发后把新 pid 加入集合（#6：派发失败/进程未起的会很快被下一轮探活剔除）。
"""
# 先清掉已退出的 pid（探活），得到当前真实活跃数
if self._active_pids:
alive = pids_alive_on_worker(self.target, self._active_pids)
self._active_pids = alive
running = len(self._active_pids)
# pid 集合空时退回 pgrep 兜底（避免冷启动时漏算 worker 已有进程）
if running == 0:
running = count_running_on_worker(self.target, self.worker_cwd)
free = max(0, self.nproc - running)
if free == 0:
return
@ -446,22 +688,42 @@ class WorkerManager(threading.Thread):
name = task["name"]
params = task["params"]
self.log("派发 {} (worker 现有 {})".format(name, running))
pid, err = dispatch_task(
pid_out, err = dispatch_task(
self.target, self.worker_cwd, self.tlusty_root,
name, params, self.results,
seed_path=None, qiniu_domain=self.store.domain,
seed_path=None,
timeout=self.timeout, grid_config=self.grid_config)
if err and "nohup" not in str(err).lower():
self.log("派发 {} 警告: {}".format(name, err[:100]))
# 记录 pid；dispatch_task 返回的是 nohup 的 $!（字符串或空）
pid = _parse_pid(pid_out)
if pid is not None:
self._active_pids.add(pid)
else:
self.log("派发 {} 未能获取 pid（err={}）".format(name, err[:80]))
def _collect_results(self):
"""回收 worker 上已完成的结果。"""
statuses = fetch_worker_status(self.target, self.worker_cwd, self.results)
if not statuses:
return
data = json.load(open(self.tasks_path))
data = _read_tasks(self.tasks_path)
# 先对"已派 seed_step"的任务做 pid 探活：还在跑就跳过（避免据旧的
# 冷启动 conv.json 重复派发 seed_step 的竞态）。只有 pid 死了才回收。
seed_step_running = set()
ss_pids_to_check = {}
for n, t in data["tasks"].items():
if (t.get("host") == self.target and t["status"] == "running"
and t.get("seed_step_used") and t.get("seed_step_pid")):
ss_pids_to_check[n] = t["seed_step_pid"]
if ss_pids_to_check:
alive_pids = pids_alive_on_worker(self.target, set(ss_pids_to_check.values()))
for n, pid in ss_pids_to_check.items():
if pid in alive_pids:
seed_step_running.add(n) # seed_step 还在跑，本轮跳过
claimed_here = {n: t for n, t in data["tasks"].items()
if t.get("host") == self.target and t["status"] == "running"}
if t.get("host") == self.target and t["status"] == "running"
and n not in seed_step_running}
for name in claimed_here:
if name not in statuses:
continue
@ -469,38 +731,44 @@ class WorkerManager(threading.Thread):
converged = conv.get("converged", False)
max_relc = conv.get("final_max_relc")
elapsed = conv.get("elapsed_sec")
seed_step_used = conv.get("seed_step_used", False)
# 注意：seed_step 是否"已派发"必须以 tasks.json 里 master 自己标的为准，
# 不能用 conv.json 的（那是 run_one 写的，run_one 不知道 master 做没做
# seed_step，永远是 None/False）。否则会反复触发 seed_step 死循环。
task_seed_step_used = data["tasks"][name].get("seed_step_used", False)
params = claimed_here[name]["params"]
# 但注意：run_one 的 seed_step 不会自动发生（需 master 触发），
# 这里 converged=False 的点，master 后续做 seed_step 补算。
if converged:
# 取回 .7 种子到 master
local_seed = os.path.join(self.seed_dir, name + ".7")
ok = fetch_seed(self.target, self.worker_cwd, self.results,
name, local_seed)
if ok and self.store.domain:
# 上传七牛（master 代传）
if self.qiniu_upload_by == "master":
up_ok, msg = self.store.upload_seed(local_seed, name)
self.log("{} 收敛，种子上传七牛: {}".format(
name, "ok" if up_ok else msg))
# 取回 .7 到 master 本地 results/<name>/<name>.7：
# 既作为该模型的大气产物，也作为后续 seed_step 的种子源
# （run_grid.find_seed 扫的就是 results/<name>/<name>.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/<name>/，
校验 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(
"<tlusty_root>/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:

263
cno_grid/src/qiniu_store.py
View File

@ -1,263 +0,0 @@
#!/usr/bin/env python3
"""七牛云对象存储的轻量封装（零 SDK 依赖）。
仅用 Python 标准库（hmac/hashlib/base64/json）+ curl 实现七牛云的上传/
下载/列举。worker 端无需 pip 装任何东西。
七牛上传签名机制：
PutPolicy = {"scope":"<bucket>:<key>", "deadline":<unix秒>}
编码后用 AccessKey/SecretKey 做 HMAC-SHA1 签名，得到 uploadToken。
用途（本分布式网格）：
- 七牛云当"种子库"：每个收敛模型的 .7 大气（393KB）上传为
<seed_prefix>/<model_name>.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/<model>.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):
"""返回七牛种子库中所有 <model_name>.7 的模型名列表。
调用七牛 RS API: GET /list?bucket=<b>&prefix=<seed_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)):<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/<model>.7 -> <model>
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 "(未配置)")

94
cno_grid/src/run_grid.py
View File

@ -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()

6
cno_grid/src/run_one.py
View File

@ -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