|
@@ -13,8 +13,9 @@ One serial worker thread (single GPU). Job state persists to
|
|
|
from __future__ import annotations
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
import json
|
|
import json
|
|
|
-import queue
|
|
|
|
|
-import threading
|
|
|
|
|
|
|
+import os
|
|
|
|
|
+import subprocess
|
|
|
|
|
+import sys
|
|
|
import time
|
|
import time
|
|
|
import uuid
|
|
import uuid
|
|
|
from pathlib import Path
|
|
from pathlib import Path
|
|
@@ -27,7 +28,7 @@ import config
|
|
|
import pipeline
|
|
import pipeline
|
|
|
|
|
|
|
|
app = FastAPI(title="GenRecon", version="1.0")
|
|
app = FastAPI(title="GenRecon", version="1.0")
|
|
|
-_work: "queue.Queue[str]" = queue.Queue()
|
|
|
|
|
|
|
+RUN_JOB = str(Path(__file__).resolve().parent / "run_job.py")
|
|
|
|
|
|
|
|
|
|
|
|
|
def _job_dir(job_id: str) -> Path:
|
|
def _job_dir(job_id: str) -> Path:
|
|
@@ -61,42 +62,51 @@ def _log_line(job_id: str, line: str) -> None:
|
|
|
f.write(line + "\n")
|
|
f.write(line + "\n")
|
|
|
|
|
|
|
|
|
|
|
|
|
-def _worker() -> None:
|
|
|
|
|
- while True:
|
|
|
|
|
- job_id = _work.get()
|
|
|
|
|
- try:
|
|
|
|
|
- _write(job_id, status="running", stage=config.STAGES[0][0], progress=1, message="start", error=None)
|
|
|
|
|
- pipeline.run_pipeline(
|
|
|
|
|
- _job_dir(job_id),
|
|
|
|
|
- log=lambda ln, j=job_id: _log_line(j, ln),
|
|
|
|
|
- set_progress=lambda stage, pct, msg, j=job_id: _write(j, status="running", stage=stage, progress=pct, message=msg),
|
|
|
|
|
- )
|
|
|
|
|
- _write(job_id, status="done", stage="glb", progress=100, message="ready")
|
|
|
|
|
- except Exception as e: # noqa: BLE001 — surface any stage failure to the client
|
|
|
|
|
- _log_line(job_id, f"ERROR: {e}")
|
|
|
|
|
- _write(job_id, status="error", message="failed", error=str(e))
|
|
|
|
|
- finally:
|
|
|
|
|
- _work.task_done()
|
|
|
|
|
|
|
+def _pid_alive(pid: int) -> bool:
|
|
|
|
|
+ try:
|
|
|
|
|
+ os.kill(pid, 0)
|
|
|
|
|
+ return True
|
|
|
|
|
+ except (OSError, ValueError):
|
|
|
|
|
+ return False
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _runner_alive(job_id: str) -> bool:
|
|
|
|
|
+ try:
|
|
|
|
|
+ return _pid_alive(int((_job_dir(job_id) / "worker.pid").read_text().strip()))
|
|
|
|
|
+ except Exception:
|
|
|
|
|
+ return False
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _launch_runner(job_id: str) -> None:
|
|
|
|
|
+ """Run the reconstruction in a DETACHED process that outlives this web service
|
|
|
|
|
+ (KillMode=process on the unit), self-serialized on the GPU by an flock. A
|
|
|
|
|
+ redeploy / restart of genrecon-app therefore never kills an in-flight scan."""
|
|
|
|
|
+ subprocess.Popen(
|
|
|
|
|
+ [sys.executable, RUN_JOB, job_id],
|
|
|
|
|
+ cwd=str(Path(RUN_JOB).parent),
|
|
|
|
|
+ stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
|
|
|
+ start_new_session=True, close_fds=True,
|
|
|
|
|
+ )
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.on_event("startup")
|
|
@app.on_event("startup")
|
|
|
def _startup() -> None:
|
|
def _startup() -> None:
|
|
|
- # Re-queue jobs that were queued, fail jobs that were mid-flight at shutdown.
|
|
|
|
|
|
|
+ # Re-attach: any job that isn't finished gets a live runner. If its detached
|
|
|
|
|
+ # runner survived our restart (KillMode=process) we leave it alone; otherwise
|
|
|
|
|
+ # we (re)launch — so a service restart never strands a reconstruction.
|
|
|
for d in sorted(config.JOBS_DIR.glob("*/")):
|
|
for d in sorted(config.JOBS_DIR.glob("*/")):
|
|
|
st = _read(d.name)
|
|
st = _read(d.name)
|
|
|
- if not st:
|
|
|
|
|
|
|
+ if not st or st.get("status") in ("done", "error"):
|
|
|
continue
|
|
continue
|
|
|
- if st.get("status") == "queued":
|
|
|
|
|
- _work.put(d.name)
|
|
|
|
|
- elif st.get("status") == "running":
|
|
|
|
|
- _write(d.name, status="error", message="interrupted by restart",
|
|
|
|
|
- error="server restarted while this job was running")
|
|
|
|
|
- threading.Thread(target=_worker, daemon=True).start()
|
|
|
|
|
|
|
+ if not _runner_alive(d.name):
|
|
|
|
|
+ _launch_runner(d.name)
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.get("/api/health")
|
|
@app.get("/api/health")
|
|
|
def health() -> dict:
|
|
def health() -> dict:
|
|
|
- return {"ok": True, "mock": config.MOCK, "queue": _work.qsize()}
|
|
|
|
|
|
|
+ active = sum(1 for d in config.JOBS_DIR.glob("*/")
|
|
|
|
|
+ if (_read(d.name) or {}).get("status") in ("queued", "running"))
|
|
|
|
|
+ return {"ok": True, "mock": config.MOCK, "active": active}
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.post("/api/jobs")
|
|
@app.post("/api/jobs")
|
|
@@ -119,7 +129,7 @@ async def create_job(video: UploadFile = File(...)) -> dict:
|
|
|
f.write(chunk)
|
|
f.write(chunk)
|
|
|
_write(job_id, status="queued", stage=None, progress=0, message="queued",
|
|
_write(job_id, status="queued", stage=None, progress=0, message="queued",
|
|
|
created=time.time(), filename=video.filename, error=None)
|
|
created=time.time(), filename=video.filename, error=None)
|
|
|
- _work.put(job_id)
|
|
|
|
|
|
|
+ _launch_runner(job_id)
|
|
|
return {"id": job_id, "status": "queued"}
|
|
return {"id": job_id, "status": "queued"}
|
|
|
|
|
|
|
|
|
|
|