feat(prep/monitor): 打包每 10% 打一行进度 + 监控按实测算进度/ETA;prep 估时 0.5→2.0 h

- build_dataset.cjs:每 10% 打 `[prep] shard_xxxx a/b`(只加日志,不改计算)。新增广
  (40%/3°/12,每个增广样本 6 次碰撞检查)让这批打包从上次的 25 分钟涨到约 1 小时,
  而原来整片完成前一行输出都没有,看起来像卡死。
- monitor.py prep 分支:解析这些行,按 (已完成分片 + 进行中分片的任务比例)/总分片 算 frac,
  按实测速率外推 eta;首批完成前显示"首波处理中,第一批完成后显示预估"。
- pstate/monitor.html:prep 估时 0.5 h → 2.0 h(超时 1.5 h → 6 h),与实际耗时一致。
This commit is contained in:
cbb
2026-10-19 00:33:11 +08:00
parent 0d522dd782
commit 0b47905a1c
4 changed files with 51 additions and 17 deletions
+16 -3
View File
@@ -252,11 +252,24 @@ def stage_progress(st, s):
'eta_s': (N - n) / rate if rate > 0 else None}
if st == 'prep':
nsh = len(glob.glob(os.path.join(ROOT, DATA, 'shard_*.jsonl')))
done = len(newer(f'{DATA}/samples/*.X.npy', t0))
fin = newer(f'{DATA}/samples/*.X.npy', t0)
done = len(fin)
if os.path.exists(os.path.join(ROOT, DATA, 'r2.json')) and newer(f'{DATA}/r2.json', t0): return {'frac': 1, 'label': '打包完成,R2 检查完成', 'eta_s': 0}
# 2026-10-19:build_dataset 每 10% 打一行 [prep] shard_xxxx a/b;用它把"整片完成"之间的进度和 ETA 算出来
fin_n = {os.path.basename(p).split('.')[0] for p in fin}
part = {}
for ln in console(0)['text'].splitlines()[-800:]:
m = re.search(r'\[prep\] (shard_\d+) (\d+)/(\d+)', ln)
if m: part[m.group(1)] = int(m.group(2)) / max(1, int(m.group(3)))
run = {k: v for k, v in part.items() if k not in fin_n}
frac = (done + sum(run.values())) / nsh if nsh else None
el = time.time() - t0
eta = (nsh - done) / (done / el) if done > 0 and el > 1 and nsh > done else None
return {'frac': min(0.95, done / nsh) if nsh else None, 'label': f'已打包 {done} / {nsh} 个分片' + (',R2 检查中' if nsh and done >= nsh else ''), 'eta_s': eta}
eta = el * (1 - frac) / frac if frac and 0.02 < frac < 1 and el > 5 else None
lab = f'已打包 {done} / {nsh} 个分片'
if run: lab += f' · 进行中 {len(run)} 个(平均 {100 * sum(run.values()) / len(run):.0f}%)'
elif done == 0: lab += ' · 首波处理中,第一批完成后显示预估'
elif nsh and done >= nsh: lab += ',R2 检查中'
return {'frac': min(0.95, frac) if frac is not None else None, 'label': lab, 'eta_s': eta}
if st == 'train':
tag, bs = info.get('tag', 'base'), info.get('bench_steps', 2000)
b, m = last_rec(f'bench_{tag}') if newer(f'{RUNS}/bench_{tag}/log.jsonl', t0) else None, last_rec(tag) if newer(f'{RUNS}/{tag}/log.jsonl', t0) else None