cbb/Robot_Arm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(pstate): 改 TARGET_SAMPLES 不再作废 gen 的已有分片——只重跑 gen 补量（run_parallel 跳过写完的 shard、按差额补量），prep 及之后仍作废；此前把它当"从 gen 作废"会让 5 小时生成的 5 万条分片被整体移进 archive，用户改目标样本量就得从头跑

Browse Source
This commit is contained in:
cbb
2026-10-18 13:50:12 +08:00
parent c128288c1b
commit 241333c213
1 changed files with 10 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ml/pstate.py
+10 -3
View File
@@ -34,7 +34,10 @@ TIMEOUT_H = {k: max(1.0, 3 * v) for k, v in EST_H.items()}
TIMEOUT_H['m5'] = 3 * (EST_H['train'] + 2 * EST_H['m5']) TIMEOUT_H['m5'] = 3 * (EST_H['train'] + 2 * EST_H['m5'])
# 参数 → 受它影响的阶段（这些阶段即使上次完成也要重跑，其产物先挪进 ml/archive/）。只改挖错轮数不会让 5 小时的生成作废。 # 参数 → 受它影响的阶段（这些阶段即使上次完成也要重跑，其产物先挪进 ml/archive/）。只改挖错轮数不会让 5 小时的生成作废。
_from = lambda st: ORDER[ORDER.index(st):] _from = lambda st: ORDER[ORDER.index(st):]
AFFECTS = {**{k: _from('gen') for k in ('N_TASKS', 'NPROC', 'TARGET_SAMPLES', 'CONSENSUS', 'MIN_GAP_MM')}, AFFECTS = {**{k: _from('gen') for k in ('N_TASKS', 'NPROC', 'CONSENSUS', 'MIN_GAP_MM')},
# TARGET_SAMPLES 只决定"补量到多少"：不作废 gen 的已有分片（重跑 gen 会跳过写完的 shard、按差额补量），
# 但样本集会变，所以 prep 及之后作废。2026-10-18 实测：把它当"从 gen 作废"会让 5 小时的 5 万条分片白跑。
'TARGET_SAMPLES': ['gen'] + _from('prep'),
'CFG': ['r4'] + _from('train'), **{k: _from('mine') for k in ('ROUNDS', 'APG_STEPS', 'MINE_N')}} 'CFG': ['r4'] + _from('train'), **{k: _from('mine') for k in ('ROUNDS', 'APG_STEPS', 'MINE_N')}}
PARAM_KEYS = list(AFFECTS) PARAM_KEYS = list(AFFECTS)
# 各阶段产物（相对仓库根，可含通配）；作废 / 从头重跑时整体移到 ml/archive/<时间>/，不删除 # 各阶段产物（相对仓库根，可含通配）；作废 / 从头重跑时整体移到 ml/archive/<时间>/，不删除
@@ -179,6 +182,7 @@ def main():
if cmd == 'reset': if cmd == 'reset':
plan = rest[0].split(',') if rest and rest[0] else [] plan = rest[0].split(',') if rest and rest[0] else []
pn = params_now(); prev = d.get('params') or {} pn = params_now(); prev = d.get('params') or {}
changed = []
if '--fresh' in rest: bad = set(ORDER) if '--fresh' in rest: bad = set(ORDER)
else: else:
changed = [k for k in PARAM_KEYS if prev and prev.get(k) != pn[k]] changed = [k for k in PARAM_KEYS if prev and prev.get(k) != pn[k]]
@@ -186,8 +190,11 @@ def main():
if changed: print('参数与上次不同：' + json.dumps({k: (prev.get(k), pn[k]) for k in changed}, ensure_ascii=False) if changed: print('参数与上次不同：' + json.dumps({k: (prev.get(k), pn[k]) for k in changed}, ensure_ascii=False)
+ '；受影响的阶段重跑：' + ' '.join(st for st in ORDER if st in bad), file=sys.stderr) + '；受影响的阶段重跑：' + ' '.join(st for st in ORDER if st in bad), file=sys.stderr)
keep = {k: v for k, v in d.get('stages', {}).items() if v.get('rc') == 0 and k not in bad} keep = {k: v for k, v in d.get('stages', {}).items() if v.get('rc') == 0 and k not in bad}
# 受影响 / 未完成的阶段：旧产物挪走，免得续训、跳过 shard 时把旧参数的东西接上 # 受影响 / 未完成的阶段：旧产物挪走，免得续训、跳过 shard 时把旧参数的东西接上。
arc = archive([st for st in ORDER if st in bad]) if bad else None # 例外：只改了目标样本量时，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
if arc: print(f'旧产物已移到 {arc}/（未删除）', file=sys.stderr) if arc: print(f'旧产物已移到 {arc}/（未删除）', file=sys.stderr)
d = {'started': now, 'plan': plan, 'stages': keep, 'current': None, 'params': pn} d = {'started': now, 'plan': plan, 'stages': keep, 'current': None, 'params': pn}
done = [s for s in plan if s in keep] done = [s for s in plan if s in keep]