diff --git a/jobengine.py b/jobengine.py index fd78c98..43c7ea9 100644 --- a/jobengine.py +++ b/jobengine.py @@ -14,9 +14,56 @@ import time import uuid JOBS: dict[str, dict] = {} +_PROCS: dict[str, subprocess.Popen] = {} # laufende Prozesse je Job-ID (fuer Abbruch) _LOG_CAP = 400 +def _append_log(job: dict, line: str) -> None: + job["log"].append(line) + if len(job["log"]) > _LOG_CAP: + del job["log"][0] + + +def _pump_output(job: dict, stream) -> None: + """Liest die Ausgabe byteweise und behandelt `\\r` (Fortschrittsbalken wie bei + `hf download`/tqdm) als Ueberschreiben der letzten Zeile statt als neue Zeile. + So erscheint der Download-Fortschritt live, ohne das Log mit tausenden Frames zu + fluten. Bewusst BINAER gelesen: im Text-Modus wuerde Python ein einzelnes `\\r` + zu `\\n` uebersetzen (universal newlines) und das Ueberschreiben unmoeglich machen.""" + buf = b"" + overwrite = False # soll die naechste committete Zeile die letzte ueberschreiben? + pending_cr = False # haben wir gerade ein `\r` gesehen und warten auf das naechste Byte? + + def commit(): + line = buf.decode("utf-8", "replace") + if overwrite and job["log"]: + job["log"][-1] = line + else: + _append_log(job, line) + + while True: + ch = stream.read(1) + if not ch: + break + if pending_cr: + pending_cr = False + if ch == b"\n": # `\r\n` zusammen = echte neue Zeile + commit(); overwrite = False; buf = b"" + continue + commit(); overwrite = True; buf = b"" # einzelnes `\r` = Fortschritt (ueberschreiben) + # ch faellt unten normal durch + if ch == b"\r": + pending_cr = True + elif ch == b"\n": + commit(); overwrite = False; buf = b"" + else: + buf += ch + if pending_cr: + commit(); overwrite = True; buf = b"" + if buf: + commit() + + def _run_job(job_id: str, args: list[str], env: dict | None = None, stdin_data: str | None = None): job = JOBS[job_id] job["state"] = "running" @@ -26,31 +73,56 @@ def _run_job(job_id: str, args: list[str], env: dict | None = None, stdin_data: stdin=subprocess.PIPE if stdin_data is not None else None, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, - bufsize=1, + bufsize=0, # binaer + ungepuffert -> Fortschritt (\r) kommt live an env={**os.environ, **(env or {})}, ) + _PROCS[job_id] = proc if stdin_data is not None: # Secret (z.B. sudo-Passwort) ueber stdin fuettern — landet NICHT im Log/ps. try: - proc.stdin.write(stdin_data) # type: ignore[union-attr] + proc.stdin.write(stdin_data.encode()) # type: ignore[union-attr] proc.stdin.close() # type: ignore[union-attr] except Exception: # noqa: BLE001 pass - for line in proc.stdout: # type: ignore[union-attr] - job["log"].append(line.rstrip("\n")) - if len(job["log"]) > _LOG_CAP: - del job["log"][0] + _pump_output(job, proc.stdout) proc.wait() - job["returncode"] = proc.returncode - job["state"] = "done" if proc.returncode == 0 else "failed" + if job.get("canceled"): + job["state"] = "canceled" + job["returncode"] = proc.returncode + _append_log(job, "[mission-control] Vorgang abgebrochen.") + else: + job["returncode"] = proc.returncode + job["state"] = "done" if proc.returncode == 0 else "failed" except Exception as exc: # noqa: BLE001 - job["log"].append(f"[mission-control] Fehler: {exc}") + _append_log(job, f"[mission-control] Fehler: {exc}") job["state"] = "failed" job["returncode"] = -1 + finally: + _PROCS.pop(job_id, None) job["finished_at"] = time.time() +def cancel_job(job_id: str) -> bool: + """Laufenden Job abbrechen: Prozess terminieren. Liefert False, wenn der Job + nicht (mehr) laeuft oder unbekannt ist.""" + job = JOBS.get(job_id) + if not job or job["state"] in ("done", "failed", "canceled"): + return False + job["canceled"] = True + _append_log(job, "[mission-control] Abbruch angefordert…") + proc = _PROCS.get(job_id) + if proc is not None: + try: + proc.terminate() + except Exception: # noqa: BLE001 + pass + else: + # Prozess noch nicht gestartet -> direkt als abgebrochen markieren. + job["state"] = "canceled" + job["finished_at"] = time.time() + return True + + def start_job(args: list[str], label: str, env: dict | None = None, stdin_data: str | None = None, log_cmd: str | None = None) -> str: job_id = uuid.uuid4().hex[:12] diff --git a/routers/jobs.py b/routers/jobs.py index fb4f0a9..f922c10 100644 --- a/routers/jobs.py +++ b/routers/jobs.py @@ -6,7 +6,7 @@ Liefert die Daten fuer das Aktivitaets-Panel mit Live-Log. from fastapi import APIRouter, Depends, HTTPException from auth import auth -from jobengine import JOBS +from jobengine import JOBS, cancel_job router = APIRouter(prefix="/api", dependencies=[Depends(auth)]) @@ -19,6 +19,13 @@ def job_status(job_id: str): return job +@router.post("/jobs/{job_id}/cancel") +def job_cancel(job_id: str): + if not cancel_job(job_id): + raise HTTPException(409, "Job laeuft nicht (mehr) oder ist unbekannt.") + return {"ok": True} + + @router.get("/jobs") def jobs_list(): return sorted(JOBS.values(), key=lambda j: j["started_at"], reverse=True)[:20] diff --git a/static/js/panels/jobs.js b/static/js/panels/jobs.js index 3773b48..457d70b 100644 --- a/static/js/panels/jobs.js +++ b/static/js/panels/jobs.js @@ -1,7 +1,8 @@ // jobs.js — Aktivität (v3): Live-System-Metriken + Hintergrund-Jobs mit Log. // Exportiert track(id), damit andere Panels einen Job auto-aufklappen. -import { $, esc, fmtBytes } from "../core/ui.js"; +import { api } from "../core/api.js"; +import { $, esc, fmtBytes, toast } from "../core/ui.js"; const tracked = new Set(); let JOBS = []; @@ -12,8 +13,21 @@ export function track(id) { tracked.add(id); renderJobs(); } const hist = { cpu: [], ram: [], gpu: [] }; const MAX_HIST = 60; -function statusBadge(s) { return s === "done" ? 'fertig' : s === "failed" ? 'fehler' : 'läuft…'; } -function dotClass(s) { return s === "done" ? "on" : s === "failed" ? "" : "load"; } +function statusBadge(s) { return s === "done" ? 'fertig' : s === "failed" ? 'fehler' : s === "canceled" ? 'abgebrochen' : 'läuft…'; } +function dotClass(s) { return s === "done" ? "on" : (s === "failed" || s === "canceled") ? "" : "load"; } + +// Fortschritt in % aus der letzten Log-Zeile ziehen (tqdm schreibt z.B. " 45%|…"). +function jobPct(j) { + if (j.state !== "running") return null; + const last = (j.log || [])[j.log.length - 1] || ""; + const m = last.match(/(\d+(?:\.\d+)?)\s*%/); + return m ? Math.min(100, parseFloat(m[1])) : null; +} + +async function cancelJob(id) { + try { await api(`/api/jobs/${id}/cancel`, { method: "POST" }); toast("Abbruch angefordert…"); renderJobs(); } + catch (e) { toast(e.message, true); } +} function tile(label, value, sub) { return `