feat(monitor): 用时改为按阶段累计——"总用时"=各阶段之和、"本步用时"=该阶段累计,停止后再"继续训练"不再归零(训练跑 5 万步停一次,之前那段的时间会保留);只有产物被归档、要从头重做的阶段才清零;本次改动前完成的老记录用 start/end 兜底;停止与流水线退出两处记账幂等(实测 180 s 不翻倍)
This commit is contained in:
+27
-7
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user