diff --git a/ml/monitor.html b/ml/monitor.html index 5aa53e3..bb33288 100644 --- a/ml/monitor.html +++ b/ml/monitor.html @@ -633,8 +633,8 @@ function renderPipe(s) { ${cur ? `
当前步骤 · ${SM[cur]?.n}${pr.frac != null ? pct(pr.frac) : '无法计量，按用时估'}
` : ''}
-
总用时
${hms((cur ? now : Math.max(...Object.values(P.stages).map(r => r.end || r.start))) - P.started)}
-
本步用时
${cr ? hms(now - cr.start) : '—'}
+
总用时
${hms(plan.reduce((a, k) => a + ((P.elapsed || {})[k] || 0), 0))}
+
本步用时
${cur ? hms((P.elapsed || {})[cur] || 0) : '—'}
本步粗估 / 上限
${cur ? hmsShort((SM[cur]?.w || 0) * 3600) + ' / ' + hmsShort((cr && cr.timeout_h || SM[cur]?.to || 0) * 3600) : '—'}
全流程剩余（粗估）
${allDone ? '—' : '约 ' + hmsShort(remain * 3600)}
${btns}`; @@ -730,10 +730,10 @@ function demoApi() { } let sc, runs, logs, pipe, jobS, con, genAt = 0; // 示例流水线：stages 里的 [阶段, 开始距今小时, 用时小时] - const mkPipe = (done, cur, curStartH) => { const t = now(); const st = {}; let tt = t - curStartH * 3600; - for (let i = done.length - 1; i >= 0; i--) { const [k, dur] = done[i]; st[k] = { start: tt - dur * 3600, end: tt, rc: 0, info: {} }; tt -= dur * 3600; } - if (cur) st[cur] = { start: t - curStartH * 3600, end: null, rc: null, info: {} }; - return { started: tt, plan: ['kin', 'r4', 'gate1', 'gen', 'prep', 'train', 'm5', 'apg', 'mine', 'export', 'bench', 'accept'], stages: st, current: cur }; }; + const mkPipe = (done, cur, curStartH) => { const t = now(); const st = {}, el = {}; let tt = t - curStartH * 3600; + for (let i = done.length - 1; i >= 0; i--) { const [k, dur] = done[i]; st[k] = { start: tt - dur * 3600, end: tt, rc: 0, info: {} }; el[k] = dur * 3600; tt -= dur * 3600; } + if (cur) { st[cur] = { start: t - curStartH * 3600, end: null, rc: null, info: {} }; el[cur] = curStartH * 3600; } + return { started: tt, plan: ['kin', 'r4', 'gate1', 'gen', 'prep', 'train', 'm5', 'apg', 'mine', 'export', 'bench', 'accept'], stages: st, elapsed: el, current: cur }; }; const CON = { train: '$ FROM=kin TO=accept bash ml/pipeline.sh all\n计划：kin r4 gate1 gen prep train m5 apg mine export bench accept\n== node ml/prep/export_kin.cjs\n== node ml/eval/dump_parity.cjs 2000\n== python -m ml.eval.check_parity\n{"PASS": true, "fk_mm": 0.0004, "gap_mm": 0.0011}\n== node ml/gen/consistency.cjs 300 4242\n{"consistentByGateRule": 0.86}\n一致性门禁结果 → CONSENSUS=1\n== bash ml/gen/run_parallel.sh 1786 28 1 --consensus 4\n样本数 1702000 ≥ 目标，完成\n== bash ml/prep/run_prep.sh 28\n== python -m ml.eval.check_r2 --samples ml/data/samples --out ml/data/r2.json\n{"pairs": 250, "label_std_mean": 0.071, "pass": true}\n== python -m ml.train --samples ml/data/samples --out ml/runs/bench_base --steps 2000 --no-eval\nstep 2000 loss 1.912 6150 samples/s\n== python -m ml.train --samples ml/data/samples --out ml/runs/base\n', gen: '$ FROM=kin TO=accept bash ml/pipeline.sh all\n计划：kin r4 gate1 gen prep train m5 apg mine export bench accept\n== node ml/prep/export_kin.cjs\n== python -m ml.eval.check_parity\n{"PASS": true}\n== node ml/gen/consistency.cjs 300 4242\n{"consistentByGateRule": 0.86}\n一致性门禁结果 → CONSENSUS=1\n== bash ml/gen/run_parallel.sh 1786 28 1 --consensus 4\n', diff --git a/ml/monitor.py b/ml/monitor.py index c7ff04d..6d33af9 100644 --- a/ml/monitor.py +++ b/ml/monitor.py @@ -216,9 +216,27 @@ def stage_progress(st, s): return {'frac': None, 'label': '进行中'} +def stage_elapsed(pl, now=None): + """每个阶段的累计用时（秒）：pipeline.json 的 time{阶段: 秒} 是不含当前段的累计，运行中的阶段 + 再把"这一段"实时加上。老记录（没有 time 字段）用 start/end 的差值兜底。 + 网页的「总用时」＝各阶段之和，「本步用时」＝当前阶段这条 —— 确定/续跑都不归零。""" + now = now or time.time() + acc = pl.get('time') or {} + cur = pl.get('current') + out = {} + for k, r in (pl.get('stages') or {}).items(): + t = acc.get(k) + if t is None and r.get('start') and r.get('end'): t = r['end'] - r['start'] + t = t or 0 + if k == cur and r.get('start') and r.get('end') is None: t += max(0.0, now - r['start']) + out[k] = round(t, 1) + return out + + def pipeline(): d = jload(os.path.join(ROOT, RUNS, 'pipeline.json')) if not d: return None + d['elapsed'] = stage_elapsed(d) cur = d.get('current') if cur and cur in d['stages']: try: d['progress'] = stage_progress(cur, d['stages'][cur]) diff --git a/ml/pstate.py b/ml/pstate.py index 883dd39..548d0fa 100644 --- a/ml/pstate.py +++ b/ml/pstate.py @@ -13,7 +13,9 @@ pipeline.sh 自动调用，一般不需要手动运行： 心跳超过 LOCK_STALE_S 未刷新视为陈旧（进程被强杀），可接管。 - 续跑（缺口 1）：reset 默认保留已完成阶段；参数变了只作废受它影响的阶段（AFFECTS），其产物移到 ml/archive/，避免新旧参数混用。 - 阶段表 STAGES 是估时 / 超时 / 网页进度权重的唯一来源（审查 N-14：网页权重与耗时表同源）。 - - 停止也走本模块（审查 N-12：此前 monitor 直写 pipeline.json，与子进程竞态）。""" + - 停止也走本模块（审查 N-12：此前 monitor 直写 pipeline.json，与子进程竞态）。 + - 累计用时（2026-10-18）：每个阶段跑过的时间并进顶部 time{阶段: 秒}，跨「停止 → 继续训练」累加， + 网页的「总用时 / 本步用时」据此显示；产物被归档（要重做）的阶段清零，其余保留。""" import glob, json, os, shutil, socket, sys, time ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) @@ -69,7 +71,7 @@ def load(): try: with open(P, encoding='utf-8') as f: return json.load(f) except Exception: - return {'started': time.time(), 'plan': None, 'stages': {}, 'current': None, 'params': {}} + return {'started': time.time(), 'plan': None, 'stages': {}, 'current': None, 'params': {}, 'time': {}} def save(d): @@ -146,11 +148,21 @@ def todo(d=None): return [s for s in (d.get('plan') or []) if (d['stages'].get(s) or {}).get('rc') != 0] +def _acc_time(d, st, now): + """把"这一段"的用时并进累计 time{阶段: 秒}。幂等：stage.end 已写就不再加，停止（mark_stopped） + 与流水线的 end 都会调，重复调用不会双记。""" + s = d['stages'].get(st) + if s and s.get('start') and s.get('end') is None: + t = d.setdefault('time', {}) + t[st] = round(t.get(st, 0) + max(0.0, now - s['start']), 1) + + def mark_stopped(rc=-15): - """网页「停止」后调用：当前阶段记为被停止，释放锁。""" - d = load(); st = d.get('current') + """网页「停止」后调用：当前阶段记为被停止、这段用时并进累计，释放锁。""" + d = load(); st = d.get('current'); now = time.time() if st and st in d['stages'] and d['stages'][st].get('end') is None: - d['stages'][st].update(end=time.time(), rc=rc) + _acc_time(d, st, now) + d['stages'][st].update(end=now, rc=rc) d['current'] = None save(d); lock_release() @@ -194,14 +206,21 @@ def main(): # 例外：只改了目标样本量时，gen 要重跑补量，但已有分片必须留着（run_parallel 跳过写完的 shard、 # 按差额补量），否则 5 小时的生成白跑。 no_archive = {'gen'} if 'TARGET_SAMPLES' in changed else set() - arc = archive([st for st in ORDER if st in bad and st not in no_archive]) if bad else None + arch = [st for st in ORDER if st in bad and st not in no_archive] + arc = archive(arch) if arch else None if arc: print(f'旧产物已移到 {arc}/（未删除）', file=sys.stderr) - d = {'started': now, 'plan': plan, 'stages': keep, 'current': None, 'params': pn} + # 累计用时跨续跑保留；只有产物被挪走的阶段（要从头重做）才清零 + tm = {k: v for k, v in (d.get('time') or {}).items() if k not in arch} + d = {'started': now, 'plan': plan, 'stages': keep, 'current': None, 'params': pn, 'time': tm} done = [s for s in plan if s in keep] if done: print('跳过已完成的 %d 个阶段：%s' % (len(done), ' '.join(done)), file=sys.stderr) save(d); return elif cmd == 'begin': st = rest[0] + prev = d['stages'].get(st) or {} + # 老记录（本次改动之前完成的）没有累计字段：用上一次的整段补上，免得续跑时用时从 0 起 + if prev.get('start') and prev.get('end') and st not in (d.get('time') or {}): + d.setdefault('time', {})[st] = round(prev['end'] - prev['start'], 1) d['stages'][st] = {'start': now, 'end': None, 'rc': None, 'info': kv(rest[1:]), 'timeout_h': TIMEOUT_H.get(st)} d['current'] = st elif cmd == 'set': @@ -210,6 +229,7 @@ def main(): elif cmd == 'end': st, rc = rest[0], int(rest[1]) if len(rest) > 1 else 0 s = d['stages'].setdefault(st, {'start': now, 'info': {}}) + _acc_time(d, st, now) s['end'], s['rc'] = now, rc if d.get('current') == st: d['current'] = None else: