feat(monitor): 每个阶段都按实时数据算剩余时间,用时精确到秒——打包按已打包分片数、预训练用脚本自报 eta_h、APG 用 it/min、挖错用本轮 APG + 已完成轮次平均时长、头门禁用已完成那次评估、一致性门禁读日志里的 k/300 行(只读日志尾部,不占 CPU);修 mine 的当前轮判断(原来 len(cur)>k 在第二轮起永远判不到在跑)与 APG 漏加的 eta;总用时/本步用时/步骤条用时改为带秒,预估仍只到分钟
This commit is contained in:
+51
-15
@@ -117,7 +117,8 @@ def mahm():
|
||||
psutil 在 Windows 没有 sensors_temperatures;ACPI 热区只有个 27.9 °C 的板温;Power Meter 计数器没有实例。
|
||||
条目布局实测:头 32 字节(签名 MAHM / 版本 / 头大小 / 条目数 / 条目大小 / 时间戳),每条 1324 字节,
|
||||
名字在前 260 字节、单位在其后 260、数值是偏移 1300 的 float(用 GPU1 power 对 nvidia-smi 校准过)。
|
||||
读数比 10 秒旧就当没读到:Afterburner 关了或停了就别显示一个冻住的数。"""
|
||||
读数比 30 秒旧才当没读到:Afterburner 的硬件轮询周期实测 10~13 秒(不是 1 秒),阈值卡 10 秒会
|
||||
正好落在两次刷新之间、把还在用的读数丢掉(表现为温度和功耗一闪一闪地消失)。"""
|
||||
now = time.time()
|
||||
if now - _MAHM['t'] < 1.0: return _MAHM['v']
|
||||
_MAHM['t'] = now
|
||||
@@ -135,7 +136,7 @@ def mahm():
|
||||
hd = ctypes.string_at(p, 32)
|
||||
num, esize, hsize = (int.from_bytes(hd[i * 4:i * 4 + 4], 'little') for i in (3, 4, 2))
|
||||
upd = int.from_bytes(hd[20:24], 'little')
|
||||
if 0 < num < 5000 and 32 < esize < 65536 and now - upd <= 10:
|
||||
if 0 < num < 5000 and 32 < esize < 65536 and now - upd <= 30:
|
||||
buf = ctypes.string_at(p, hsize + num * esize)
|
||||
core = {}
|
||||
for i in range(num):
|
||||
@@ -221,7 +222,7 @@ def newer(pattern, t0):
|
||||
|
||||
|
||||
def stage_progress(st, s):
|
||||
"""当前阶段的进度:返回 {'frac': 0..1 或 None(无法计量), 'label': 文字}。"""
|
||||
"""当前阶段的进度:返回 {'frac': 0..1 或 None(无法计量), 'label': 文字, 'eta_s': 按实测速率外推的剩余秒数或 None}。"""
|
||||
t0, info = s['start'], s.get('info') or {}
|
||||
if st == 'gen':
|
||||
# 分子算全部(含之前轮次已完成的分片),分母是累计目标:补量时显示 50,008/79,688 而不是 0/79,688。
|
||||
@@ -230,37 +231,72 @@ def stage_progress(st, s):
|
||||
n = sum(LC.count(f) for f in allf)
|
||||
tgt = info.get('target_tasks') or 0
|
||||
new = sum(LC.count(f) for f in newer(f'{DATA}/shard_*.jsonl', t0) + newer(f'{DATA}/shard_*.jsonl.part', t0))
|
||||
lab = f'已生成 {n:,} / {tgt:,} 条任务' + (f'(本轮补 {new:,} 条)' if 0 < new < n else '')
|
||||
return {'frac': min(1, n / tgt) if tgt else None, 'label': lab}
|
||||
# 剩余时间按"本轮已写条数 ÷ 本轮已用时"外推(条数是实时数出来的,速率就是实测的),不再用静态估时
|
||||
el = max(1.0, time.time() - t0)
|
||||
rate = new / el if new > 0 else 0
|
||||
eta = ((tgt - n) / rate if rate > 0 else None) if tgt > n else (0.0 if rate > 0 else None)
|
||||
lab = f'已生成 {n:,} / {tgt:,} 条任务' + (f'(本轮补 {new:,} 条 · {rate * 60:.0f} 条/分)' if 0 < new < n else '')
|
||||
return {'frac': min(1, n / tgt) if tgt else None, 'label': lab, 'eta_s': eta}
|
||||
if st == 'gate1':
|
||||
# 一致性门禁每 10 个任务往 stderr 打一行 "k/300"(consistency.cjs:41)。进度只读控制台日志最后 64 KB,
|
||||
# 不跑任何计算、不占 CPU —— 正在生成的节点不受影响。日志里可能还留着上一次门禁的进度行,
|
||||
# 所以只数最后一个 "consistency.cjs" 命令行之后的部分。
|
||||
txt = console(0).get('text') or ''
|
||||
cut = txt.rfind('consistency.cjs')
|
||||
m = re.findall(r'^(\d+)/(\d+)\s*$', txt[cut:] if cut >= 0 else '', re.M)
|
||||
n, N = (int(m[-1][0]), int(m[-1][1])) if m else (0, 300)
|
||||
el = time.time() - t0
|
||||
rate = n / el if n > 0 and el > 2 else 0
|
||||
return {'frac': n / N if N else None,
|
||||
'label': f'门禁 {n} / {N} 个任务' + (f' · {rate * 60:.0f} 个/分' if rate else '') if n else '准备中(日志里还没有进度行)',
|
||||
'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))
|
||||
if os.path.exists(os.path.join(ROOT, DATA, 'r2.json')) and newer(f'{DATA}/r2.json', t0): return {'frac': 1, 'label': '打包完成,R2 检查完成'}
|
||||
return {'frac': min(0.95, done / nsh) if nsh else None, 'label': f'已打包 {done} / {nsh} 个分片' + (',R2 检查中' if nsh and done >= nsh else '')}
|
||||
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}
|
||||
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}
|
||||
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
|
||||
if m and m.get('total'):
|
||||
return {'frac': (bs + m['step']) / (bs + m['total']), 'label': f'正式训练 {m["step"]:,} / {m["total"]:,} 步', 'run': tag}
|
||||
return {'frac': (bs + m['step']) / (bs + m['total']), 'label': f'正式训练 {m["step"]:,} / {m["total"]:,} 步', 'run': tag,
|
||||
'eta_s': m['eta_h'] * 3600 if m.get('eta_h') else None} # 训练脚本自己按实测样本/s 算的
|
||||
if b: return {'frac': b['step'] / (bs + 60000), 'label': f'测速 {b["step"]:,} / {bs:,} 步', 'run': f'bench_{tag}'}
|
||||
return {'frac': 0, 'label': '加载数据中'}
|
||||
if st == 'apg':
|
||||
r = last_rec('apg') if newer(f'{RUNS}/apg/log.jsonl', t0) else None
|
||||
return {'frac': r['it'] / r['total'] if r and r.get('total') else 0, 'label': f'APG {r["it"]:,} / {r["total"]:,} 步' if r else '准备中', 'run': 'apg'}
|
||||
eta = None
|
||||
if r and r.get('total'):
|
||||
it, mn = r.get('it') or 0, r.get('min') or 0
|
||||
eta = (r['total'] - it) / (it / mn) * 60 if it > 0 and mn > 0.1 else None # 步/分实测外推
|
||||
return {'frac': r['it'] / r['total'] if r and r.get('total') else 0, 'label': f'APG {r["it"]:,} / {r["total"]:,} 步' if r else '准备中', 'run': 'apg', 'eta_s': eta}
|
||||
if st == 'mine':
|
||||
R = info.get('rounds', 6)
|
||||
k = len(newer(f'{RUNS}/mine/r*/gate.json', t0))
|
||||
cur = sorted(newer(f'{RUNS}/apg_r*/log.jsonl', t0), key=os.path.getmtime)
|
||||
frac_in, lab = 0, f'第 {k + 1} 轮:挖错出题'
|
||||
if cur and len(cur) > k:
|
||||
name = os.path.basename(os.path.dirname(cur[-1])); r = last_rec(name)
|
||||
if r and r.get('total'): frac_in = 0.3 + 0.6 * r['it'] / r['total']; lab = f'第 {k + 1} 轮:APG {r["it"]:,} / {r["total"]:,} 步'
|
||||
return {'frac': min(1, (k + frac_in) / R), 'label': f'{lab}(共最多 {R} 轮,提升不足会提前停)', 'run': name}
|
||||
lab = f'第 {k + 1} 轮:挖错出题'
|
||||
if cur:
|
||||
name = os.path.basename(os.path.dirname(cur[-1])) # ml/runs/apg_rN
|
||||
rn = re.sub(r'\D', '', name) or '1'
|
||||
gate = os.path.join(ROOT, RUNS, 'mine', f'r{rn}', 'gate.json')
|
||||
r = last_rec(name)
|
||||
if r and r.get('total') and not os.path.exists(gate): # 这一轮还在跑(还没出 gate)
|
||||
frac_in = 0.3 + 0.6 * r['it'] / r['total']
|
||||
lab = f'第 {k + 1} 轮:APG {r["it"]:,} / {r["total"]:,} 步'
|
||||
it, mn = r.get('it') or 0, r.get('min') or 0
|
||||
eta = (r['total'] - it) / (it / mn) * 60 if it > 0 and mn > 0.1 else None
|
||||
if eta is not None and k >= 1: # 剩余轮次按已完成轮次的平均时长外推
|
||||
eta += max(0, R - k - 1) * (time.time() - t0) / k
|
||||
return {'frac': min(1, (k + frac_in) / R), 'label': f'{lab}(共最多 {R} 轮,提升不足会提前停)', 'run': name, 'eta_s': eta}
|
||||
return {'frac': min(1, k / R), 'label': f'{lab}(共最多 {R} 轮,提升不足会提前停)'}
|
||||
if st == 'm5':
|
||||
tag = info.get('tag', 'base')
|
||||
n = len(newer(f'{RUNS}/m5_{tag}_policy.json', t0)) + len(newer(f'{RUNS}/m5_{tag}_safety.json', t0))
|
||||
return {'frac': n / 2, 'label': ['纯策略评估中', '加安全层评估中', '汇总'][min(n, 2)]}
|
||||
el = time.time() - t0
|
||||
eta = (2 - n) * (el / n) if n == 1 else (0.0 if n >= 2 else None) # 两次评估(纯策略 / 加安全层)按已完成的平均耗时外推
|
||||
return {'frac': n / 2, 'label': ['纯策略评估中', '加安全层评估中', '汇总'][min(n, 2)], 'eta_s': eta}
|
||||
return {'frac': None, 'label': '进行中'}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user