diff --git a/service/index.html b/service/index.html
index 372698d..3d3de0e 100644
--- a/service/index.html
+++ b/service/index.html
@@ -451,6 +451,12 @@
border: 1px solid rgba(21, 32, 27, 0.08);
}
+ button.danger {
+ background: linear-gradient(135deg, #c73528 0%, #8f1f18 100%);
+ color: #fff;
+ border: 1px solid rgba(143, 31, 24, 0.22);
+ }
+
.pill {
display: inline-flex;
align-items: center;
@@ -2093,8 +2099,11 @@ A+B
strand_sequence_design_placeholder: "a b ~c 或 N10A5",
run_btn: "开始分析",
run_design_btn: "开始设计",
- run_waiting_btn: "等待结果返回...",
- submission_locked: "已有任务在等待结果返回,请等待完成;如需强制重新提交请刷新页面。",
+ run_waiting_btn: "终止任务",
+ status_canceling: "正在终止任务...",
+ status_canceled: "任务已终止",
+ cancel_job_failed: "终止任务失败",
+ submission_locked: "已有任务在等待结果返回,请终止或等待。",
example_btn: "载入示例",
history_title: "最近任务",
clear_history: "清空历史",
@@ -2401,8 +2410,11 @@ A+B
strand_sequence_design_placeholder: "a b ~c or N10A5",
run_btn: "Run Analysis",
run_design_btn: "Run Design",
- run_waiting_btn: "Waiting for result...",
- submission_locked: "A job is already waiting for results. Wait for it to finish, or refresh the page to submit again.",
+ run_waiting_btn: "Terminate Job",
+ status_canceling: "Terminating job...",
+ status_canceled: "Job canceled",
+ cancel_job_failed: "Cancel job failed",
+ submission_locked: "A job is already waiting for results. Cancel or wait.",
example_btn: "Load Example",
history_title: "Recent Jobs",
clear_history: "Clear History",
@@ -2660,6 +2672,7 @@ A+B
let lastResult = null;
let currentJobId = null;
let activeSubmission = false;
+ let cancellationRequested = false;
let probabilitySvgCache = new Map();
let styledSvgCache = new Map();
let viewerScale = 1;
@@ -2686,8 +2699,11 @@ A+B
}
function updateRunButtonState() {
- runBtn.disabled = activeSubmission;
+ runBtn.disabled = activeSubmission && (!currentJobId || cancellationRequested);
runBtn.setAttribute("aria-busy", activeSubmission ? "true" : "false");
+ runBtn.classList.toggle("primary", !activeSubmission);
+ runBtn.classList.toggle("secondary", false);
+ runBtn.classList.toggle("danger", activeSubmission);
runBtn.textContent = activeSubmission ? t("run_waiting_btn") : runButtonBaseLabel();
}
@@ -2696,6 +2712,27 @@ A+B
updateRunButtonState();
}
+ async function cancelCurrentJob() {
+ if (!currentJobId || cancellationRequested) return;
+ cancellationRequested = true;
+ setStatus(formatStatus(t("status_canceling")));
+ updateRunButtonState();
+ try {
+ const { response, data } = await fetchJsonOrThrow(`/api/jobs/${encodeURIComponent(currentJobId)}/cancel`, {
+ method: "POST",
+ });
+ if (!response.ok || data.status !== "success") {
+ throw new Error(data?.error || `Cancel failed with HTTP ${response.status}`);
+ }
+ refreshHealth();
+ } catch (error) {
+ cancellationRequested = false;
+ setStatus(`${t("cancel_job_failed")}: ${error.message}`, true);
+ } finally {
+ updateRunButtonState();
+ }
+ }
+
function setQueueMetric(node, value) {
node.textContent = value === undefined || value === null ? "-" : String(value);
}
@@ -5370,7 +5407,6 @@ A+B
}
async function waitForJob(jobId) {
- const startedAt = Date.now();
for (;;) {
const { response, data: job } = await fetchJsonOrThrow(`/api/jobs/${jobId}`);
if (!response.ok) {
@@ -5380,6 +5416,11 @@ A+B
setStatus(formatStatus(t("status_queued")));
} else if (job.status === "running") {
setStatus(formatStatus(t("status_polling")));
+ } else if (job.status === "cancel_requested") {
+ setStatus(formatStatus(t("status_canceling")));
+ } else if (job.status === "canceled") {
+ refreshHealth();
+ throw new Error(t("status_canceled"));
} else if (job.status === "success") {
refreshHealth();
if (job.result && typeof job.elapsed_seconds === "number") {
@@ -5390,9 +5431,6 @@ A+B
refreshHealth();
throw new Error(job.error?.message || "Job failed");
}
- if (Date.now() - startedAt > 10 * 60 * 1000) {
- throw new Error(t("request_timeout"));
- }
await new Promise((resolve) => setTimeout(resolve, 1200));
}
}
@@ -5412,8 +5450,9 @@ A+B
return;
}
- setSubmissionLocked(true);
currentJobId = null;
+ cancellationRequested = false;
+ setSubmissionLocked(true);
setStatus(formatStatus(t("status_running")));
results.innerHTML = "";
@@ -5428,6 +5467,7 @@ A+B
throw new Error(data?.error || `Analysis failed with HTTP ${response.status}`);
}
currentJobId = data.job_id;
+ updateRunButtonState();
setStatus(formatStatus(t("status_waiting_job")));
const result = await waitForJob(data.job_id);
result.job_id = data.job_id;
@@ -5438,11 +5478,19 @@ A+B
renderHistory();
setStatus(formatStatus(t("status_done")));
} catch (error) {
- results.innerHTML = `
${t("request_failed")}
${String(error)} `;
- setStatus(t("status_error"), true);
+ if (cancellationRequested) {
+ results.innerHTML = `${t("status_canceled")}
${String(error)} `;
+ setStatus(formatStatus(t("status_canceled")));
+ } else {
+ results.innerHTML = `${t("request_failed")}
${String(error)} `;
+ setStatus(t("status_error"), true);
+ }
refreshHealth();
} finally {
setSubmissionLocked(false);
+ currentJobId = null;
+ cancellationRequested = false;
+ updateRunButtonState();
}
}
@@ -5459,8 +5507,9 @@ A+B
return;
}
- setSubmissionLocked(true);
currentJobId = null;
+ cancellationRequested = false;
+ setSubmissionLocked(true);
setStatus(formatStatus(t("status_running")));
results.innerHTML = "";
@@ -5475,6 +5524,7 @@ A+B
throw new Error(data?.error || `Analysis failed with HTTP ${response.status}`);
}
currentJobId = data.job_id;
+ updateRunButtonState();
setStatus(formatStatus(t("status_waiting_job")));
const result = await waitForJob(data.job_id);
result.job_id = data.job_id;
@@ -5485,11 +5535,19 @@ A+B
renderHistory();
setStatus(formatStatus(t("status_done")));
} catch (error) {
- results.innerHTML = `${t("request_failed")}
${String(error)} `;
- setStatus(t("status_error"), true);
+ if (cancellationRequested) {
+ results.innerHTML = `${t("status_canceled")}
${String(error)} `;
+ setStatus(formatStatus(t("status_canceled")));
+ } else {
+ results.innerHTML = `${t("request_failed")}
${String(error)} `;
+ setStatus(t("status_error"), true);
+ }
refreshHealth();
} finally {
setSubmissionLocked(false);
+ currentJobId = null;
+ cancellationRequested = false;
+ updateRunButtonState();
}
}
@@ -5727,7 +5785,13 @@ A+B
historyImportInput.value = "";
}
});
- runBtn.addEventListener("click", runAnalysis);
+ runBtn.addEventListener("click", () => {
+ if (activeSubmission) {
+ cancelCurrentJob();
+ return;
+ }
+ runAnalysis();
+ });
document.getElementById("exampleBtn").addEventListener("click", loadExample);
workflowSelect.addEventListener("change", () => {
syncWorkflow();
diff --git a/service/server.py b/service/server.py
index 8f0d065..771724e 100644
--- a/service/server.py
+++ b/service/server.py
@@ -1,6 +1,8 @@
import json
import mimetypes
+import multiprocessing
import os
+import queue
import re
import threading
import time
@@ -80,6 +82,9 @@ IUPAC_CODES = "ACGTUWSMKRYBDHVN"
IUPAC_CONSTRAINT_TOKEN = re.compile(rf"[{IUPAC_CODES}](?:\d+)?")
VALID_COMPUTE = {"pfunc", "pairs", "mfe", "sample", "subopt", "ensemble_size"}
+CANCEL_REQUESTED_STATUS = "cancel_requested"
+CANCELED_STATUS = "canceled"
+TERMINAL_JOB_STATUSES = {"success", "error", CANCELED_STATUS}
JOB_STORE = {}
JOB_LOCK = threading.Lock()
JOB_TTL_SECONDS = int(os.environ.get("NP_JOB_TTL_SECONDS", "3600"))
@@ -1321,12 +1326,14 @@ def run_job_payload(payload):
if workflow == "design":
design_options = parse_design_options(payload)
- design_job_options = DesignOptions(
- f_stop=design_options["stop_condition"],
- seed=design_options["seed"],
- wobble_mutations=design_options["wobble_mutations"],
- max_time=design_options["max_time_seconds"],
- )
+ design_job_option_kwargs = {
+ "f_stop": design_options["stop_condition"],
+ "seed": design_options["seed"],
+ "wobble_mutations": design_options["wobble_mutations"],
+ }
+ if design_options["max_time_seconds"] > 0:
+ design_job_option_kwargs["max_time"] = design_options["max_time_seconds"]
+ design_job_options = DesignOptions(**design_job_option_kwargs)
design_domain_map, design_domains = build_design_domains(payload.get("design_domains") or [])
target_strand_map, design_strands = build_design_strands(
payload.get("strands") or [],
@@ -1562,21 +1569,64 @@ def set_job_data(job):
JOB_STORE[job["job_id"]] = dict(job)
+def get_job_data(job_id, include_payload=False):
+ if redis_enabled():
+ raw = redis_client().get(job_key(job_id))
+ if raw is None:
+ return None
+ job = json.loads(raw)
+ if not include_payload:
+ job.pop("payload", None)
+ return job
+ with JOB_LOCK:
+ prune_jobs()
+ job = JOB_STORE.get(job_id)
+ if job is None:
+ return None
+ output = dict(job)
+ if not include_payload:
+ output.pop("payload", None)
+ return output
+
+
+def should_keep_cancel_state(current, updates):
+ current_status = current.get("status") if current else None
+ next_status = updates.get("status")
+ if current_status in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
+ return next_status not in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}
+ return False
+
+
def update_job_data(job_id, **updates):
if redis_enabled():
client = redis_client()
- current = get_job(job_id)
- if current is None:
- current = {"job_id": job_id, "created_at": time.time()}
- current.update(updates)
- current["updated_at"] = time.time()
- client.setex(job_key(job_id), JOB_TTL_SECONDS, json.dumps(current, ensure_ascii=False))
- return current
+ key = job_key(job_id)
+ while True:
+ pipe = client.pipeline()
+ try:
+ pipe.watch(key)
+ raw = pipe.get(key)
+ current = json.loads(raw) if raw is not None else {"job_id": job_id, "created_at": time.time()}
+ if should_keep_cancel_state(current, updates):
+ pipe.unwatch()
+ return current
+ current.update(updates)
+ current["updated_at"] = time.time()
+ pipe.multi()
+ pipe.setex(key, JOB_TTL_SECONDS, json.dumps(current, ensure_ascii=False))
+ pipe.execute()
+ return current
+ except redis.WatchError:
+ continue
+ finally:
+ pipe.reset()
with JOB_LOCK:
current = JOB_STORE.get(job_id)
if current is None:
current = {"job_id": job_id, "created_at": time.time()}
+ if should_keep_cancel_state(current, updates):
+ return dict(current)
current.update(updates)
current["updated_at"] = time.time()
prune_jobs(current["updated_at"])
@@ -1607,52 +1657,161 @@ def create_job(payload):
return job_id
+def _job_process_entry(payload, result_queue):
+ apply_thread_limits()
+ try:
+ result_queue.put({"status": "success", "result": run_job_payload(payload)})
+ except BaseException as exc:
+ result_queue.put(
+ {
+ "status": "error",
+ "error": {
+ "message": str(exc),
+ "traceback": traceback.format_exc(),
+ },
+ }
+ )
+
+
+def terminate_process(process):
+ if not process.is_alive():
+ return
+ if hasattr(process, "kill"):
+ process.kill()
+ else:
+ process.terminate()
+ process.join(timeout=2)
+ if process.is_alive():
+ process.terminate()
+ process.join(timeout=2)
+
+
def _run_job(job_id, payload):
started_at = time.time()
+ current = get_job_data(job_id)
+ if current and current.get("status") in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
+ elapsed = round(time.time() - started_at, 3)
+ update_job_data(job_id, status=CANCELED_STATUS, payload=None, elapsed_seconds=elapsed)
+ return
+
log_event(f"running job_id={job_id}")
update_job_data(job_id, status="running")
if redis_enabled():
redis_client().sadd(JOB_RUNNING_KEY, job_id)
+
+ ctx = multiprocessing.get_context("spawn")
+ result_queue = ctx.Queue(maxsize=1)
+ process = ctx.Process(target=_job_process_entry, args=(payload, result_queue), daemon=True)
+ message = None
+
try:
- result = run_job_payload(payload)
+ process.start()
+ while process.is_alive():
+ current = get_job_data(job_id)
+ if current and current.get("status") in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
+ terminate_process(process)
+ elapsed = round(time.time() - started_at, 3)
+ update_job_data(
+ job_id,
+ status=CANCELED_STATUS,
+ error={"message": "Job canceled by user."},
+ result=None,
+ payload=None,
+ elapsed_seconds=elapsed,
+ )
+ log_event(f"canceled job_id={job_id} elapsed={elapsed}s")
+ return
+ try:
+ message = result_queue.get_nowait()
+ break
+ except queue.Empty:
+ time.sleep(0.25)
+
+ process.join(timeout=2)
+ if message is None:
+ try:
+ message = result_queue.get_nowait()
+ except queue.Empty:
+ message = None
+
+ current = get_job_data(job_id)
+ if current and current.get("status") in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
+ elapsed = round(time.time() - started_at, 3)
+ update_job_data(
+ job_id,
+ status=CANCELED_STATUS,
+ error={"message": "Job canceled by user."},
+ result=None,
+ payload=None,
+ elapsed_seconds=elapsed,
+ )
+ log_event(f"canceled job_id={job_id} elapsed={elapsed}s")
+ return
+
elapsed = round(time.time() - started_at, 3)
- update_job_data(job_id, status="success", result=result, payload=None, elapsed_seconds=elapsed)
- log_event(f"success job_id={job_id} elapsed={elapsed}s")
- except Exception as exc:
- elapsed = round(time.time() - started_at, 3)
- update_job_data(
- job_id,
- status="error",
- error={
- "message": str(exc),
- "traceback": traceback.format_exc(),
- },
- payload=None,
- elapsed_seconds=elapsed,
- )
- log_event(f"error job_id={job_id} elapsed={elapsed}s message={exc}")
+ if message and message.get("status") == "success":
+ update_job_data(job_id, status="success", result=message["result"], payload=None, elapsed_seconds=elapsed)
+ log_event(f"success job_id={job_id} elapsed={elapsed}s")
+ elif message and message.get("status") == "error":
+ update_job_data(
+ job_id,
+ status="error",
+ error=message.get("error") or {"message": "Job failed."},
+ payload=None,
+ elapsed_seconds=elapsed,
+ )
+ log_event(f"error job_id={job_id} elapsed={elapsed}s message={message.get('error', {}).get('message')}")
+ else:
+ update_job_data(
+ job_id,
+ status="error",
+ error={"message": f"Job process exited with code {process.exitcode}."},
+ payload=None,
+ elapsed_seconds=elapsed,
+ )
+ log_event(f"error job_id={job_id} elapsed={elapsed}s exitcode={process.exitcode}")
finally:
+ terminate_process(process)
+ result_queue.close()
+ result_queue.join_thread()
if redis_enabled():
redis_client().srem(JOB_RUNNING_KEY, job_id)
-def get_job(job_id):
- if redis_enabled():
- raw = redis_client().get(job_key(job_id))
- if raw is None:
- return None
- job = json.loads(raw)
- job.pop("payload", None)
- return job
- with JOB_LOCK:
- prune_jobs()
- job = JOB_STORE.get(job_id)
- if job is None:
- return None
- output = dict(job)
+def cancel_job(job_id):
+ current = get_job_data(job_id, include_payload=True)
+ if current is None:
+ return None
+ status = current.get("status")
+ if status in TERMINAL_JOB_STATUSES:
+ output = dict(current)
output.pop("payload", None)
return output
+ if redis_enabled() and status == "queued":
+ redis_client().lrem(JOB_QUEUE_KEY, 0, job_id)
+
+ if status == "queued":
+ updated = update_job_data(
+ job_id,
+ status=CANCELED_STATUS,
+ error={"message": "Job canceled by user."},
+ result=None,
+ payload=None,
+ )
+ else:
+ updated = update_job_data(
+ job_id,
+ status=CANCEL_REQUESTED_STATUS,
+ error={"message": "Cancellation requested."},
+ )
+ updated.pop("payload", None)
+ return updated
+
+
+def get_job(job_id):
+ return get_job_data(job_id, include_payload=False)
+
def prune_shares(now=None):
if SHARE_MAX_COUNT < 1:
@@ -1759,6 +1918,8 @@ def run_worker_loop():
if raw is None:
continue
job = json.loads(raw)
+ if job.get("status") in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
+ continue
payload = job.get("payload")
if payload is None:
continue
@@ -1975,6 +2136,14 @@ class AppHandler(BaseHTTPRequestHandler):
parsed = urlparse(self.path)
if parsed.path != "/api/analyze":
+ if parsed.path.startswith("/api/jobs/") and parsed.path.endswith("/cancel"):
+ job_id = parsed.path.split("/")[-2]
+ job = cancel_job(job_id)
+ if job is None:
+ self._respond(*json_bytes({"error": "Job not found"}, status=HTTPStatus.NOT_FOUND))
+ return
+ self._respond(*json_bytes({"status": "success", "job": job}))
+ return
if parsed.path == "/api/jobs":
try:
length = int(self.headers.get("Content-Length", "0"))