Spaces:
Running on Zero
Running on Zero
| """Session-scoped cooperative cancellation, safe across the ZeroGPU worker boundary. | |
| The SDK owns its worker. This app creates no inference thread or GPU generator. | |
| The cancellation marker lives on the shared filesystem and is unlinked only after | |
| the decorated call has returned. A new request cannot reuse the previous marker. | |
| """ | |
| from dataclasses import dataclass | |
| import hashlib | |
| import logging | |
| from pathlib import Path | |
| import secrets | |
| import threading | |
| import time | |
| log = logging.getLogger("zonos.lifecycle") | |
| class Cancelled(Exception): | |
| pass | |
| class AlreadyRunning(Exception): | |
| pass | |
| def check_cancelled(marker): | |
| if Path(marker).exists(): | |
| raise Cancelled() | |
| class Job: | |
| id: str | |
| owner: str | |
| marker: Path | |
| started: float | |
| class Jobs: | |
| def __init__(self, directory): | |
| self.directory = Path(directory) | |
| self.directory.mkdir(parents=True, exist_ok=True) | |
| self._lock = threading.Lock() | |
| self._active = {} | |
| def begin(self, owner): | |
| if not isinstance(owner, str) or not owner: | |
| raise ValueError("Missing session identity") | |
| key = hashlib.sha256(owner.encode()).hexdigest() | |
| with self._lock: | |
| if key in self._active: | |
| raise AlreadyRunning() | |
| identifier = secrets.token_hex(24) | |
| job = Job(identifier, key, self.directory / (identifier + ".cancel"), time.monotonic()) | |
| self._active[key] = job | |
| log.info("request_started request=%s", identifier) | |
| return job | |
| def cancel(self, owner): | |
| key = hashlib.sha256(owner.encode()).hexdigest() | |
| with self._lock: | |
| job = self._active.get(key) | |
| if job is None: | |
| return False | |
| job.marker.touch(exist_ok=True) | |
| log.info("cancellation_requested request=%s", job.id) | |
| return True | |
| def finish(self, job): | |
| with self._lock: | |
| if self._active.get(job.owner) == job: | |
| del self._active[job.owner] | |
| job.marker.unlink(missing_ok=True) | |
| log.info("request_finished request=%s", job.id) | |
| def active(self): | |
| with self._lock: | |
| return len(self._active) | |