INNER CODE UNIT · Python

ensure_job_active

ChenShuo2004/cs-board · webapp/server.py:508

def ensure_job_active(job_id: str) -> None:
    if is_job_cancelled(job_id):
        raise JobCancelled("任务已取消")


def terminate_running_process(job_id: str) -> None:
    with RUNNING_PROCESSES_LOCK:
        process = RUNNING_PROCESSES.get(job_id)
    if process is None or process.poll() is not None:
        return
    try:
        if os.name == "nt":
            subprocess.run(
                ["taskkill", "/PID", str(process.pid), "/T", "/F"],
                capture_output=True,
                text=True,
                timeout=8,
                check=False,

View source record →