第五轮：锁/续跑/超时/自动分支/R4 探针与 encode-once/控制台加固；阶段内断点续跑（gen 按条、prep 按 shard、训练与 APG 定时原子存档+体面停止）；分歧以实现方意见为准
This commit is contained in:
Hark
+160
-41
@@ -1,16 +1,25 @@
|
||||
"""训练控制台（本地网页，零第三方依赖，离线可用）：一键开始全流程、看进度、可停止。
|
||||
双击 ml/console.bat（Windows）或运行 bash ml/console.sh（Linux/WSL）→ 自动打开 http://localhost:8100
|
||||
也可：python -m ml.monitor [--port 8100] [--host 0.0.0.0] [--open]
|
||||
网页上点「开始训练」→ 选起止阶段与参数 → 后台执行 bash ml/pipeline.sh all（FROM/TO 环境变量），
|
||||
输出写 ml/runs/console.log，网页实时显示；「停止」结束整个进程树。网页关掉不影响训练，重新打开即可继续看。
|
||||
网页上点「开始训练」→ 选参数 → 后台执行 bash ml/pipeline.sh all（已完成的阶段自动跳过 = 续跑；勾「从头重跑」= FRESH=1），
|
||||
输出写 ml/runs/console.log，网页实时显示；「停止」结束整个进程树。网页和控制台窗口关掉都不影响训练，重新打开即可继续看。
|
||||
只读数据来源：
|
||||
ml/runs/pipeline.json 各阶段开始/结束/返回码（pipeline.sh 自动写）→ 全流程进度
|
||||
ml/runs/*/log.jsonl 预训练（base / wta / bench_*）与 APG（apg / apg_r*）逐步日志与闭环评估
|
||||
ml/runs/m5_gate_*.json、mine/r*/report.json + gate.json、policy_bench.json、acceptance.json
|
||||
ml/data/stats.json / consistency.json / r2.json / samples/index.json、nvidia-smi
|
||||
安全：默认只监听 127.0.0.1；--host 0.0.0.0 时其他机器只能看，开始/停止只接受本机请求（--remote-control 放开）。
|
||||
2026-10-18 新增（用户要求：训练全程可视化，并在网页上点按钮一键开始）。"""
|
||||
2026-10-18 新增（用户要求：训练全程可视化，并在网页上点按钮一键开始）。
|
||||
2026-10-18 第五轮（审查第四轮）：
|
||||
N-7 是否在跑看 ml/runs/pipeline.lock（与终端同一把锁），不再靠"日志 120 s 内有更新"去猜；
|
||||
N-8 开始/停止要求自定义头 X-Console + 同源 Origin/Sec-Fetch-Site，外站网页发不进来；
|
||||
N-10 启动自检（训练用 Python 能否 import torch、GPU、node、bash），不过就拒绝「开始训练」并在顶栏标红；
|
||||
N-11 Windows 子进程用 CREATE_NO_WINDOW 自带隐藏控制台：关掉控制台黑窗口训练照跑，重开控制台可接着看、可停止；
|
||||
N-9 生成阶段的行数按文件增量统计（只读新增字节），其他阶段不扫 shard；
|
||||
N-12 停止后的状态改写走 ml.pstate（单一写者、原子替换）；N-14 阶段名/估时/权重由 ml.pstate.STAGES 下发。"""
|
||||
import argparse, glob, json, os, re, shutil, signal, subprocess, sys, threading, time, webbrowser
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
from ml import pstate
|
||||
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
|
||||
from urllib.parse import urlparse, parse_qs
|
||||
|
||||
@@ -88,6 +97,30 @@ def last_rec(name):
|
||||
return None
|
||||
|
||||
|
||||
class LineCount:
|
||||
"""按文件增量数行：记住每个文件已读到的偏移与行数，只读新增字节（审查 N-9：此前每 5 s 全量重读全部 shard）。"""
|
||||
def __init__(self): self.c = {}; self.lock = threading.Lock()
|
||||
|
||||
def count(self, f):
|
||||
with self.lock:
|
||||
try: size = os.path.getsize(f)
|
||||
except OSError: return 0
|
||||
off, n = self.c.get(f, (0, 0))
|
||||
if size < off: off, n = 0, 0
|
||||
if size > off:
|
||||
with open(f, 'rb') as fh:
|
||||
fh.seek(off)
|
||||
while True:
|
||||
b = fh.read(1 << 22)
|
||||
if not b: break
|
||||
n += b.count(b'\n'); off += len(b)
|
||||
self.c[f] = (off, n)
|
||||
return n
|
||||
|
||||
|
||||
LC = LineCount()
|
||||
|
||||
|
||||
def newer(pattern, t0):
|
||||
return [f for f in glob.glob(os.path.join(ROOT, pattern)) if os.path.getmtime(f) >= t0 - 1]
|
||||
|
||||
@@ -96,9 +129,7 @@ def stage_progress(st, s):
|
||||
"""当前阶段的进度：返回 {'frac': 0..1 或 None（无法计量）, 'label': 文字}。"""
|
||||
t0, info = s['start'], s.get('info') or {}
|
||||
if st == 'gen':
|
||||
n = 0
|
||||
for f in newer(f'{DATA}/shard_*.jsonl', t0):
|
||||
with open(f, 'rb') as fh: n += sum(1 for _ in fh)
|
||||
n = sum(LC.count(f) for f in newer(f'{DATA}/shard_*.jsonl', t0) + newer(f'{DATA}/shard_*.jsonl.part', t0))
|
||||
tgt = info.get('target_tasks') or 0
|
||||
return {'frac': min(1, n / tgt) if tgt else None, 'label': f'已生成 {n:,} / {tgt:,} 条任务'}
|
||||
if st == 'prep':
|
||||
@@ -154,15 +185,16 @@ def status():
|
||||
stats = jload(P(DATA, 'stats.json'))
|
||||
if stats: stats = {k: stats.get(k) for k in ('tasks', 'keep', 'gapDrop', 'minGapFilterMm', 'keepRate', 'samplesEvery3', 'samplesPerTask', 'projectedSamplesFor50k', 'consensusDrop', 'vertexMinGapMm', 'execS', 'travelDeg')}
|
||||
shards = glob.glob(P(DATA, 'shard_*.jsonl'))
|
||||
gen_lines = 0
|
||||
for s in shards:
|
||||
try:
|
||||
with open(s, 'rb') as fh: gen_lines += sum(1 for _ in fh)
|
||||
except Exception: pass
|
||||
pl = pipeline()
|
||||
if pl and pl.get('current') == 'gen': # 只在生成阶段数行（增量）；其他阶段用 stats.json 的 tasks
|
||||
gen_lines = sum(LC.count(f) for f in shards + glob.glob(P(DATA, 'shard_*.jsonl.part')))
|
||||
else: gen_lines = (stats or {}).get('tasks') or 0
|
||||
pb = jload(P(RUNS, 'policy_bench.json'))
|
||||
if pb: pb = {k: {kk: v.get(kk) for kk in ('n', 'ok', 'reached', 'collisionFrames', 'brakeRate', 'recoveryRate', 'livelock', 'inferMsP99', 'safetyMsP99')}
|
||||
for k, v in pb.items() if isinstance(v, dict) and 'n' in v}
|
||||
return {'now': time.time(), 'gpu': gpu(), 'pipeline': pipeline(),
|
||||
return {'now': time.time(), 'gpu': gpu(), 'pipeline': pl, 'check': CHECK.get(), 'lock': lock_info(),
|
||||
'stage_meta': [{'id': k, 'name': n, 'est_h': h, 'timeout_h': pstate.TIMEOUT_H[k]} for k, n, h in pstate.STAGES],
|
||||
'r4': jload(P(RUNS, 'r4.json')),
|
||||
'data': {'consistency': (jload(P(DATA, 'consistency.json')) or {}).get('consistentByGateRule'),
|
||||
'gen_tasks_written': gen_lines, 'shards': len(shards), 'stats': stats,
|
||||
'index': jload(P(DATA, 'samples', 'index.json')), 'r2': jload(P(DATA, 'r2.json'))},
|
||||
@@ -171,9 +203,10 @@ def status():
|
||||
'mine': mine, 'bench': pb, 'acceptance': jload(P(RUNS, 'acceptance.json'))}
|
||||
|
||||
|
||||
STAGES = ['setup', 'kin', 'gate1', 'gen', 'prep', 'train', 'm5', 'apg', 'mine', 'export', 'bench', 'accept']
|
||||
ENV_KEYS = {'N_TASKS': r'\d+', 'NPROC': r'\d+', 'TARGET_SAMPLES': r'\d+', 'CFG': r'ml/configs/(base|wta)\.json',
|
||||
'CONSENSUS': r'[01]?', 'ROUNDS': r'\d+', 'APG_STEPS': r'\d+', 'MINE_N': r'\d+'}
|
||||
STAGES = pstate.ORDER
|
||||
ENV_KEYS = {'N_TASKS': r'\d+', 'NPROC': r'\d+', 'TARGET_SAMPLES': r'\d+', 'CFG': r'ml/configs/(base|wta|small)\.json',
|
||||
'CONSENSUS': r'[01]?', 'ROUNDS': r'\d+', 'APG_STEPS': r'\d+', 'MINE_N': r'\d+', 'MIN_GAP_MM': r'\d+(\.\d+)?',
|
||||
'FRESH': r'[01]', 'R2_STRICT': r'[01]', 'AUTO_WTA': r'[01]'}
|
||||
CONSOLE = os.path.join(ROOT, RUNS, 'console.log')
|
||||
|
||||
|
||||
@@ -191,6 +224,56 @@ def find_bash():
|
||||
return shutil.which('bash')
|
||||
|
||||
|
||||
def train_python():
|
||||
"""流水线用的 Python：尊重 PYTHON 环境变量（与 console.sh / console.bat 一致），否则用控制台自己的解释器。"""
|
||||
return os.environ.get('PYTHON') or sys.executable
|
||||
|
||||
|
||||
class SelfCheck:
|
||||
"""启动自检（审查 N-10）：在后台线程跑一次，结果缓存；「开始训练」前不过就拒绝，网页顶栏显示。"""
|
||||
def __init__(self): self.r = {'pending': True}; threading.Thread(target=self.run, daemon=True).start()
|
||||
|
||||
def run(self):
|
||||
py = train_python(); r = {'python': py, 'bash': find_bash(), 'node': shutil.which('node')}
|
||||
try:
|
||||
o = subprocess.run([py, '-c', 'import json,torch;c=torch.cuda.is_available();print(json.dumps({"torch":torch.__version__,"cuda":c,"gpu":torch.cuda.get_device_name(0) if c else None}))'],
|
||||
capture_output=True, text=True, timeout=120)
|
||||
r.update(json.loads(o.stdout.strip().splitlines()[-1]) if o.returncode == 0 else {'torch': None, 'torch_error': (o.stderr or o.stdout).strip()[-300:]})
|
||||
except Exception as e: r.update(torch=None, torch_error=str(e))
|
||||
probs = []
|
||||
if not r.get('torch'): probs.append(f'这个 Python 没有 torch：{py}（先激活训练环境再启动控制台，或设 PYTHON=...）')
|
||||
if not r['bash']: probs.append('找不到 bash：Windows 请安装 Git for Windows（自带 Git Bash）')
|
||||
if not r['node']: probs.append('找不到 node')
|
||||
if r.get('torch') and not r.get('cuda'): r['warn'] = 'torch 看不到 GPU，训练会极慢'
|
||||
r['problems'] = probs; r['ok'] = not probs
|
||||
self.r = r
|
||||
|
||||
def get(self): return self.r
|
||||
|
||||
|
||||
def lock_info():
|
||||
lk = pstate.lock_read()
|
||||
if not lk: return None
|
||||
return dict(lk, alive=pstate.lock_alive(lk), age_s=time.time() - lk.get('beat', 0))
|
||||
|
||||
|
||||
def kill_session(sid):
|
||||
"""POSIX：结束会话 sid 里的所有进程（含 timeout 另起的进程组）。"""
|
||||
pids = []
|
||||
for d in os.listdir('/proc') if os.path.isdir('/proc') else []:
|
||||
if d.isdigit():
|
||||
try:
|
||||
if os.getsid(int(d)) == sid: pids.append(int(d))
|
||||
except Exception: pass
|
||||
if not pids:
|
||||
try: os.killpg(sid, signal.SIGTERM)
|
||||
except Exception: pass
|
||||
return
|
||||
for p in pids:
|
||||
try: os.kill(p, signal.SIGTERM)
|
||||
except Exception: pass
|
||||
|
||||
|
||||
class Job:
|
||||
def __init__(self):
|
||||
self.p = None; self.info = {}; self.lock = threading.Lock()
|
||||
@@ -200,55 +283,80 @@ class Job:
|
||||
run = self.p is not None and self.p.poll() is None
|
||||
if self.p is not None and not run and self.info.get('rc') is None:
|
||||
self.info['rc'] = self.p.returncode; self.info['ended'] = time.time()
|
||||
return {'running': run, **self.info, 'bash': find_bash() is not None}
|
||||
lk = lock_info()
|
||||
ext = bool(lk and lk['alive']) and not run # 终端或上一个控制台窗口起的流水线
|
||||
return {'running': run or ext, 'external': ext, 'lock': lk, **self.info, 'bash': find_bash() is not None}
|
||||
|
||||
def start(self, frm, to, env):
|
||||
with self.lock:
|
||||
if self.p is not None and self.p.poll() is None: return {'error': '已经在运行'}
|
||||
pj = jload(os.path.join(ROOT, RUNS, 'pipeline.json')) or {}
|
||||
cur = pj.get('current')
|
||||
if cur and time.time() - (pj['stages'].get(cur, {}).get('start') or 0) < 7 * 86400 and not self.p:
|
||||
# 终端里另起的流水线还在跑（或上次被强杀没写结束）：看有没有最近更新的日志
|
||||
fresh = [r for r in list_runs() if time.time() - r['mtime'] < 120]
|
||||
if fresh: return {'error': f'检测到终端里正在运行阶段 {cur}（日志 {fresh[0]["name"]} 2 分钟内有更新），请先在终端结束它'}
|
||||
lk = lock_info()
|
||||
if lk and lk['alive']:
|
||||
return {'error': f"已有一条流水线在运行（{lk.get('owner')}，pid {lk.get('pid')}，{lk['age_s']:.0f} s 前心跳）；先停止它"}
|
||||
ck = CHECK.get()
|
||||
if ck.get('pending'): return {'error': '环境自检还没完成，稍等几秒再点'}
|
||||
if not ck.get('ok'): return {'error': '环境自检未通过：' + '；'.join(ck['problems'])}
|
||||
if frm not in STAGES or to not in STAGES or STAGES.index(frm) > STAGES.index(to): return {'error': '起止阶段无效'}
|
||||
bash = find_bash()
|
||||
if not bash: return {'error': '找不到 bash：Windows 请安装 Git for Windows（自带 Git Bash）'}
|
||||
e = dict(os.environ, FROM=frm, TO=to, MONITOR='0', PYTHONUNBUFFERED='1', PYTHONIOENCODING='utf-8')
|
||||
if ' ' not in sys.executable: e['PYTHON'] = sys.executable.replace('\\', '/') # 与控制台同一个 Python
|
||||
e = dict(os.environ, FROM=frm, TO=to, MONITOR='0', PYTHONUNBUFFERED='1', PYTHONIOENCODING='utf-8', PIPE_OWNER='网页')
|
||||
py = train_python()
|
||||
if ' ' not in py: e['PYTHON'] = py.replace('\\', '/')
|
||||
for k, v in (env or {}).items():
|
||||
if k in ENV_KEYS and v not in (None, '') and re.fullmatch(ENV_KEYS[k], str(v)): e[k] = str(v)
|
||||
os.makedirs(os.path.dirname(CONSOLE), exist_ok=True)
|
||||
with open(CONSOLE, 'w', encoding='utf-8') as f: f.write(f'$ FROM={frm} TO={to} bash ml/pipeline.sh all [{time.strftime("%Y-%m-%d %H:%M:%S")}]\n')
|
||||
shown = ' '.join(f'{k}={e[k]}' for k in ('FRESH', 'FROM', 'TO') if k in e)
|
||||
with open(CONSOLE, 'w', encoding='utf-8') as f: f.write(f'$ {shown} bash ml/pipeline.sh all [{time.strftime("%Y-%m-%d %H:%M:%S")}]\n')
|
||||
log = open(CONSOLE, 'a', encoding='utf-8') # 追加模式：网页写入的"已停止"提示不会被子进程覆盖
|
||||
kw = {'creationflags': subprocess.CREATE_NEW_PROCESS_GROUP} if os.name == 'nt' else {'start_new_session': True}
|
||||
# Windows：CREATE_NO_WINDOW 给子进程树一个隐藏的独立控制台 → 关掉控制台黑窗口不会连带杀训练（审查 N-11）
|
||||
kw = {'creationflags': subprocess.CREATE_NEW_PROCESS_GROUP | 0x08000000} if os.name == 'nt' else {'start_new_session': True}
|
||||
self.p = subprocess.Popen([bash, 'ml/pipeline.sh', 'all'], cwd=ROOT, env=e, stdout=log, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, **kw)
|
||||
self.info = {'pid': self.p.pid, 'from': frm, 'to': to, 'started': time.time(), 'rc': None, 'ended': None,
|
||||
'env': {k: e[k] for k in ENV_KEYS if k in e}}
|
||||
return {'ok': True, 'pid': self.p.pid}
|
||||
|
||||
def stop(self):
|
||||
# 2026-10-18 体面停止：训练类阶段先写 ml/runs/STOP，等训练进程自己存 last.pt 并退出（最多 90 s），再硬杀剩下的。
|
||||
# Windows 的 taskkill /F 收不到信号，不这样做会丢掉上次存档之后的进度。生成/打包本身按 shard/条断点续，直接杀。
|
||||
with self.lock:
|
||||
ours = self.p is not None and self.p.poll() is None
|
||||
lk = lock_info()
|
||||
if not ours and not (lk and lk['alive']): return {'error': '没有在运行'}
|
||||
pid = self.p.pid if ours else int(lk.get('pid') or 0)
|
||||
if not pid: return {'error': '锁里没有 pid，无法停止'}
|
||||
cur = (pstate.load().get('current') or '')
|
||||
if cur in ('train', 'm5', 'apg', 'mine'):
|
||||
open(STOPF, 'w').write(time.strftime('%H:%M:%S'))
|
||||
with open(CONSOLE, 'a', encoding='utf-8') as f: f.write(f'\n[停止请求已发出，等待 {cur} 存档 last.pt…]\n')
|
||||
for _ in range(180):
|
||||
time.sleep(0.5)
|
||||
l2 = pstate.lock_read()
|
||||
if (self.p is not None and self.p.poll() is not None) if ours else not (l2 and pstate.lock_alive(l2)): break
|
||||
with self.lock:
|
||||
if self.p is None or self.p.poll() is not None: return {'error': '没有在运行'}
|
||||
PJ = os.path.join(ROOT, RUNS, 'pipeline.json')
|
||||
cur = (jload(PJ) or {}).get('current') # 先记下正在跑的阶段：被杀的子进程可能以返回码 0 写结束
|
||||
if os.name == 'nt':
|
||||
subprocess.run(['taskkill', '/T', '/F', '/PID', str(self.p.pid)], capture_output=True)
|
||||
subprocess.run(['taskkill', '/T', '/F', '/PID', str(pid)], capture_output=True)
|
||||
else:
|
||||
try: os.killpg(self.p.pid, signal.SIGTERM)
|
||||
except ProcessLookupError: pass
|
||||
try: self.p.wait(10)
|
||||
except subprocess.TimeoutExpired: os.killpg(self.p.pid, signal.SIGKILL)
|
||||
if ours:
|
||||
kill_session(pid)
|
||||
try: self.p.wait(15)
|
||||
except subprocess.TimeoutExpired: kill_session(pid); self.p.kill()
|
||||
else: # 只动它自己的会话 / 进程组，绝不按会话杀到用户的终端
|
||||
try:
|
||||
if os.getsid(pid) == pid: kill_session(pid)
|
||||
elif os.getpgid(pid) == pid: os.killpg(pid, signal.SIGTERM)
|
||||
else: os.kill(pid, signal.SIGTERM)
|
||||
except ProcessLookupError: pass
|
||||
self.info['stopped'] = True
|
||||
with open(CONSOLE, 'a', encoding='utf-8') as f: f.write(f'\n[已在网页上停止 {time.strftime("%H:%M:%S")}]\n')
|
||||
time.sleep(0.5); pj = jload(PJ)
|
||||
if pj and cur and cur in pj['stages']: # 该阶段记为"已停止"（返回码 -15）
|
||||
pj['stages'][cur].update(end=time.time(), rc=-15); pj['current'] = None
|
||||
with open(PJ, 'w', encoding='utf-8') as f: json.dump(pj, f, ensure_ascii=False, indent=1)
|
||||
time.sleep(0.5)
|
||||
pstate.mark_stopped(-15) # 当前阶段记为"已停止"、释放锁（单一写者）
|
||||
try: os.remove(STOPF)
|
||||
except FileNotFoundError: pass
|
||||
return {'ok': True}
|
||||
|
||||
|
||||
STOPF = os.path.join(ROOT, 'ml', 'runs', 'STOP') # 与 ml/common/ckpt.py 的 STOP 同一文件
|
||||
JOB = Job()
|
||||
CHECK = None # main() 里创建（自检要等 find_bash 定义完）
|
||||
|
||||
|
||||
def console(offset):
|
||||
@@ -287,9 +395,18 @@ class H(BaseHTTPRequestHandler):
|
||||
u = urlparse(self.path)
|
||||
if not (self.server.remote_control or self.client_address[0] in ('127.0.0.1', '::1')):
|
||||
return self.send(403, json.dumps({'error': '开始/停止只接受本机操作'}), 'application/json')
|
||||
# 审查 N-8：本机任意网页都能向 localhost 发 POST（源地址同为 127.0.0.1）。要求自定义头（跨站请求会触发 CORS 预检，
|
||||
# 本服务不应答 OPTIONS → 浏览器不发），并且 Origin / Sec-Fetch-Site 若存在必须同源。
|
||||
org, sfs, host = self.headers.get('Origin'), self.headers.get('Sec-Fetch-Site'), self.headers.get('Host', '')
|
||||
if self.headers.get('X-Console') != '1' or (sfs and sfs not in ('same-origin', 'none')) or \
|
||||
(org and urlparse(org).netloc not in (host, f'localhost:{self.server.server_port}', f'127.0.0.1:{self.server.server_port}')):
|
||||
return self.send(403, json.dumps({'error': '拒绝：非本控制台页面发起的请求'}), 'application/json')
|
||||
try:
|
||||
n = int(self.headers.get('Content-Length') or 0); body = json.loads(self.rfile.read(n) or b'{}')
|
||||
if u.path == '/api/start': r = JOB.start(body.get('from', 'kin'), body.get('to', 'accept'), body.get('env') or {})
|
||||
if u.path == '/api/start':
|
||||
env = dict(body.get('env') or {})
|
||||
if body.get('fresh'): env['FRESH'] = '1'
|
||||
r = JOB.start(body.get('from', 'kin'), body.get('to', 'accept'), env)
|
||||
elif u.path == '/api/stop': r = JOB.stop()
|
||||
else: return self.send(404, 'not found', 'text/plain')
|
||||
self.send(200 if 'ok' in r else 409, json.dumps(r, ensure_ascii=False), 'application/json')
|
||||
@@ -303,9 +420,11 @@ def main():
|
||||
ap.add_argument('--open', action='store_true', help='启动后自动打开浏览器')
|
||||
ap.add_argument('--remote-control', action='store_true', help='允许其他机器点开始/停止（默认只能看）')
|
||||
a = ap.parse_args()
|
||||
global CHECK
|
||||
CHECK = SelfCheck()
|
||||
srv = ThreadingHTTPServer((a.host, a.port), H); srv.remote_control = a.remote_control
|
||||
url = f'http://localhost:{a.port}/'
|
||||
print(f'训练控制台：{url} （关闭本窗口会停止控制台；已开始的训练会一并结束，先在网页上确认跑完）', flush=True)
|
||||
print(f'训练控制台：{url} （关闭本窗口只关控制台，已开始的训练继续跑；重新打开控制台即可接着看、可停止）', flush=True)
|
||||
if a.open: threading.Timer(0.8, lambda: webbrowser.open(url)).start()
|
||||
try: srv.serve_forever()
|
||||
except KeyboardInterrupt: pass
|
||||
|
||||
Reference in New Issue
Block a user