From 3a7e7fc69fc0e8f2ade3d803dc3d50d6acdfda15 Mon Sep 17 00:00:00 2001 From: SolitaryThinker Date: Thu, 6 Aug 2026 22:40:39 -0700 Subject: [PATCH] studio: single resident generator + engine log tail; fix ray PG leak MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reload-after-unload failed because RayDistributedExecutor.shutdown() killed the actors but never removed the placement group — the GPUs stayed reserved forever. remove it on shutdown. studio generator management simplified to a single slot: one VideoGenerator instance at a time, one load in flight at a time (409 otherwise), loading a new config always releases the old instance, unload shuts it down and deletes it. jobs wait out an in-flight preload and reuse it on config match. GET /api/engine/logs: incremental tail of the server's stdout/stderr (ray relays worker output to the driver, so this captures every rank). --- apps/fastvideo_studio/job_runner.py | 294 +++++++++--------- apps/fastvideo_studio/server.py | 91 +++++- .../tests/test_generator_cache.py | 218 +++++++++---- fastvideo/worker/ray_distributed_executor.py | 10 + 4 files changed, 398 insertions(+), 215 deletions(-) diff --git a/apps/fastvideo_studio/job_runner.py b/apps/fastvideo_studio/job_runner.py index 4ebaece1b..667abcfc9 100644 --- a/apps/fastvideo_studio/job_runner.py +++ b/apps/fastvideo_studio/job_runner.py @@ -259,12 +259,14 @@ class JobRunner: self._jobs_lock = threading.Lock() self._load_jobs() - # Cache of loaded generators keyed by model config so that we only pay - # the model-loading cost once per model configuration. - self._generators: dict[tuple, Any] = {} - self._generators_lock = threading.Lock() - # Explicit preloads in flight/failed, keyed like _generators. - self._preloads: dict[tuple, dict[str, Any]] = {} + # Exactly one generator lives in memory at a time. Loading a new + # config always releases the old instance first (shutdown + placement + # group teardown); unload deletes it outright. + self._generator: Any | None = None + self._generator_config: dict[str, Any] | None = None + self._generator_state: str = "empty" # empty | loading | ready | failed + self._generator_error: str | None = None + self._generator_lock = threading.Lock() # Shared Manager for log queues (avoids spawning a new process per job) self._mp_manager = get_mp_context().Manager() @@ -620,7 +622,7 @@ class JobRunner: } @staticmethod - def _generator_cache_key( + def _generator_config_dict( model_id: str, workload_type: str, num_gpus: int, @@ -633,88 +635,146 @@ class JobRunner: vsa_sparsity: float = 0.0, tp_size: int = -1, sp_size: int = -1, - ) -> tuple: - return ( - model_id, - workload_type, - num_gpus, - dit_cpu_offload, - text_encoder_cpu_offload, - vae_cpu_offload, - image_encoder_cpu_offload, - use_fsdp_inference, - enable_torch_compile, - vsa_sparsity, - tp_size, - sp_size, - ) + ) -> dict[str, Any]: + """Canonical engine-config dict; equality here == same generator.""" + return { + "model_id": model_id, + "workload_type": workload_type, + "num_gpus": num_gpus, + "dit_cpu_offload": dit_cpu_offload, + "text_encoder_cpu_offload": text_encoder_cpu_offload, + "vae_cpu_offload": vae_cpu_offload, + "image_encoder_cpu_offload": image_encoder_cpu_offload, + "use_fsdp_inference": use_fsdp_inference, + "enable_torch_compile": enable_torch_compile, + "vsa_sparsity": vsa_sparsity, + "tp_size": tp_size, + "sp_size": sp_size, + } - @staticmethod - def _generator_key_dict(key: tuple) -> dict[str, Any]: - names = ("model_id", "workload_type", "num_gpus", "dit_cpu_offload", - "text_encoder_cpu_offload", "vae_cpu_offload", - "image_encoder_cpu_offload", "use_fsdp_inference", - "enable_torch_compile", "vsa_sparsity", "tp_size", "sp_size") - return dict(zip(names, key, strict=True)) + def _slot_entry(self) -> dict[str, Any]: + return { + "state": self._generator_state, + "error": self._generator_error, + **(self._generator_config or {}), + } + + def _running_inference_jobs(self) -> list[str]: + with self._jobs_lock: + return [j.id for j in self._jobs.values() + if j.status == JobStatus.RUNNING and j.job_type == "inference"] + + def _release_generator(self) -> None: + """Shut down and delete the resident generator (caller holds no lock).""" + with self._generator_lock: + gen = self._generator + self._generator = None + self._generator_state = "empty" + self._generator_config = None + self._generator_error = None + if gen is not None: + logger.info("Releasing resident generator") + gen.shutdown() + del gen def preload_generator(self, **params: Any) -> dict[str, Any]: - """Load a generator into the cache without running a job, so the first - generation doesn't pay the model load. Returns the preload state dict - (state: loading | ready | failed).""" - key = self._generator_cache_key(**params) - with self._generators_lock: - if key in self._generators: - return {"state": "ready", **self._generator_key_dict(key)} - entry = self._preloads.get(key) - if entry is not None and entry["state"] == "loading": - return entry - entry = { - "state": "loading", - "started_at": time.time(), - "error": None, - **self._generator_key_dict(key), - } - self._preloads[key] = entry + """Load a model into memory ahead of time. One load at a time; loading + a different config always releases the current instance first.""" + config = self._generator_config_dict(**params) + with self._generator_lock: + if self._generator_state == "loading": + raise RuntimeError("a model load is already in progress") + if self._generator_state == "ready" and self._generator_config == config: + return self._slot_entry() + if self._running_inference_jobs(): + raise RuntimeError("cannot swap models while inference jobs are running") + self._generator_state = "loading" + self._generator_config = config + self._generator_error = None + entry = self._slot_entry() def _load() -> None: try: - self._get_or_create_generator(**params) - entry["state"] = "ready" + self._create_generator_into_slot(config) except Exception as exc: # noqa: BLE001 -- surfaced via the API - logger.exception("Preload failed for %s", params.get("model_id")) - entry["state"] = "failed" - entry["error"] = str(exc) + logger.exception("Preload failed for %s", config["model_id"]) + with self._generator_lock: + self._generator_state = "failed" + self._generator_error = str(exc) threading.Thread(target=_load, daemon=True, name="generator-preload").start() return entry - def list_generators(self) -> list[dict[str, Any]]: - """Resident generators plus preloads in flight or failed.""" - out: list[dict[str, Any]] = [] - with self._generators_lock: - resident = set(self._generators) - for key in resident: - out.append({"state": "ready", **self._generator_key_dict(key)}) - for key, entry in self._preloads.items(): - if key not in resident and entry["state"] != "ready": - out.append(entry) - return out + def _create_generator_into_slot(self, config: dict[str, Any]) -> Any: + """Release whatever is resident, load ``config``, make it THE instance.""" + with self._generator_lock: + gen = self._generator + self._generator = None + if gen is not None: + logger.info("Releasing resident generator before loading a new one") + gen.shutdown() + del gen - def unload_generator(self, **params: Any) -> bool: - """Shut down a cached generator and free its GPU memory. Refuses while - any inference job is running (the job may be using it).""" - with self._jobs_lock: - running = [j.id for j in self._jobs.values() - if j.status == JobStatus.RUNNING and j.job_type == "inference"] + # Import lazily so starting the server is fast even without a GPU. + from fastvideo import VideoGenerator + + # Deployment-level knobs (set where the API server is launched): + # FASTVIDEO_STUDIO_MODEL_PATHS="id=/local/dir,..." serves a registered + # model id from local weights instead of the HF hub; + # FASTVIDEO_STUDIO_EXECUTOR_BACKEND=ray runs workers on an existing + # Ray cluster (the multi-node path — "mp" spawns local processes only). + model_path = config["model_id"] + for pair in os.environ.get("FASTVIDEO_STUDIO_MODEL_PATHS", "").split(","): + mid, sep, path = pair.partition("=") + if sep and mid.strip() == config["model_id"]: + model_path = path.strip() + executor_kwargs: dict[str, Any] = {} + backend = os.environ.get("FASTVIDEO_STUDIO_EXECUTOR_BACKEND") + if backend: + executor_kwargs["distributed_executor_backend"] = backend + + logger.info("Loading model %s (%s)", config["model_id"], + ", ".join(f"{k}={v}" for k, v in config.items() if k != "model_id")) + new_gen = VideoGenerator.from_pretrained( + model_path, + workload_type=config["workload_type"], + num_gpus=config["num_gpus"], + dit_cpu_offload=config["dit_cpu_offload"], + text_encoder_cpu_offload=config["text_encoder_cpu_offload"], + vae_cpu_offload=config["vae_cpu_offload"], + image_encoder_cpu_offload=config["image_encoder_cpu_offload"], + use_fsdp_inference=config["use_fsdp_inference"], + enable_torch_compile=config["enable_torch_compile"], + VSA_sparsity=config["vsa_sparsity"], + tp_size=config["tp_size"], + sp_size=config["sp_size"], + **executor_kwargs, + ) + with self._generator_lock: + self._generator = new_gen + self._generator_config = config + self._generator_state = "ready" + self._generator_error = None + return new_gen + + def list_generators(self) -> list[dict[str, Any]]: + """The resident slot, or empty when nothing is loaded.""" + with self._generator_lock: + if self._generator_state == "empty": + return [] + return [self._slot_entry()] + + def unload_generator(self, **_ignored: Any) -> bool: + """Shut down and delete the resident generator, freeing GPU memory.""" + with self._generator_lock: + if self._generator_state == "loading": + raise RuntimeError("cannot unload while a model load is in progress") + if self._generator_state == "empty": + return False + running = self._running_inference_jobs() if running: raise RuntimeError(f"cannot unload while inference jobs are running: {running}") - key = self._generator_cache_key(**params) - with self._generators_lock: - gen = self._generators.pop(key, None) - self._preloads.pop(key, None) - if gen is None: - return False - gen.shutdown() + self._release_generator() return True def _get_or_create_generator( @@ -733,7 +793,10 @@ class JobRunner: sp_size: int = -1, log_queue: mp.Queue | None = None, ) -> Any: - cache_key = self._generator_cache_key( + """Return the resident generator if the config matches; otherwise + replace the slot (waiting out an in-flight preload first).""" + del log_queue # single-slot generators log via the engine tee + config = self._generator_config_dict( model_id=model_id, workload_type=workload_type, num_gpus=num_gpus, @@ -747,72 +810,21 @@ class JobRunner: tp_size=tp_size, sp_size=sp_size, ) + # A preload may be in flight (possibly for exactly this config): wait + # for it rather than racing it for the GPUs. + deadline = time.time() + 3600 + while time.time() < deadline: + with self._generator_lock: + state = self._generator_state + if state != "loading": + if state == "ready" and self._generator_config == config: + return self._generator + break + time.sleep(1.0) + else: + raise RuntimeError("timed out waiting for the in-flight model load") - # Generators are cached by model_id and configuration parameters - with self._generators_lock: - if cache_key in self._generators: - return self._generators[cache_key] - - # Import lazily so starting the server is fast even without a GPU. - from fastvideo import VideoGenerator - - # Deployment-level knobs (set where the API server is launched): - # FASTVIDEO_STUDIO_MODEL_PATHS="id=/local/dir,..." serves a registered - # model id from local weights instead of the HF hub; - # FASTVIDEO_STUDIO_EXECUTOR_BACKEND=ray runs workers on an existing - # Ray cluster (the multi-node path — "mp" spawns local processes only). - model_path = model_id - for pair in os.environ.get("FASTVIDEO_STUDIO_MODEL_PATHS", "").split(","): - mid, sep, path = pair.partition("=") - if sep and mid.strip() == model_id: - model_path = path.strip() - executor_kwargs: dict[str, Any] = {} - backend = os.environ.get("FASTVIDEO_STUDIO_EXECUTOR_BACKEND") - if backend: - executor_kwargs["distributed_executor_backend"] = backend - - logger.info( - "Loading model %s (workload=%s, num_gpus=%d, offloads: " - "dit=%s text_encoder=%s vae=%s image_encoder=%s, fsdp=%s, " - "torch_compile=%s, vsa_sparsity=%.2f, tp=%d sp=%d) …", - model_id, - workload_type, - num_gpus, - dit_cpu_offload, - text_encoder_cpu_offload, - vae_cpu_offload, - image_encoder_cpu_offload, - use_fsdp_inference, - enable_torch_compile, - vsa_sparsity, - tp_size, - sp_size, - ) - - gen = VideoGenerator.from_pretrained( - model_path, - workload_type=workload_type, - num_gpus=num_gpus, - dit_cpu_offload=dit_cpu_offload, - text_encoder_cpu_offload=text_encoder_cpu_offload, - vae_cpu_offload=vae_cpu_offload, - image_encoder_cpu_offload=image_encoder_cpu_offload, - use_fsdp_inference=use_fsdp_inference, - enable_torch_compile=enable_torch_compile, - VSA_sparsity=vsa_sparsity, - tp_size=tp_size, - sp_size=sp_size, - log_queue=log_queue, - **executor_kwargs, - ) - - with self._generators_lock: - if cache_key not in self._generators: - self._generators[cache_key] = gen - else: # Another thread may have created it while we were loading. - gen.shutdown() - gen = self._generators[cache_key] - return gen + return self._create_generator_into_slot(config) def _run_job(self, job: Job): if job.job_type == "inference": diff --git a/apps/fastvideo_studio/server.py b/apps/fastvideo_studio/server.py index 0fb51a1a2..f8a1baf8f 100644 --- a/apps/fastvideo_studio/server.py +++ b/apps/fastvideo_studio/server.py @@ -34,8 +34,8 @@ from fastvideo.registry import (get_registered_model_paths, get_registered_model from fastvideo_studio.database import Database, _get_db_path from fastvideo_studio.gpu import get_gpu_snapshot from fastvideo_studio.job_runner import JobRunner, JobStatus -from fastvideo_studio.models import (CreateDatasetRequest, CreateJobRequest, GeneratorRequest, SettingsUpdate, UpdateCaptionRequest, - model_label) +from fastvideo_studio.models import (CreateDatasetRequest, CreateJobRequest, GeneratorRequest, SettingsUpdate, + UpdateCaptionRequest, model_label) logging.basicConfig( level=logging.INFO, @@ -45,6 +45,60 @@ logger = logging.getLogger("fastvideo.studio.api") DEFAULT_OUTPUT_DIR = os.path.join(os.path.dirname(__file__), "..", "outputs", "ui_jobs") +class _EngineLogBuffer: + """Thread-safe ring buffer over the server process's stdout/stderr. + + Because ray relays worker output to the driver (log_to_driver), teeing the + server's own streams captures engine output from every rank on every node; + under the mp executor, worker logs arrive via the logging handlers which + also write to stderr. + """ + + def __init__(self, maxlen: int = 5000) -> None: + import collections + import threading + self._lines: collections.deque[str] = collections.deque(maxlen=maxlen) + self._dropped = 0 + self._lock = threading.Lock() + self._partial = "" + + def write(self, text: str) -> None: + with self._lock: + buf = self._partial + text + *complete, self._partial = buf.split("\n") + for line in complete: + if len(self._lines) == self._lines.maxlen: + self._dropped += 1 + self._lines.append(line) + + def get_lines(self, after: int = 0) -> tuple[list[str], int]: + with self._lock: + total = self._dropped + len(self._lines) + start = max(0, after - self._dropped) + return list(self._lines)[start:], total + + +class _Tee: + """File-like that forwards to the original stream and the ring buffer.""" + + def __init__(self, orig: Any, buffer: _EngineLogBuffer) -> None: + self._orig = orig + self._buffer = buffer + + def write(self, text: str) -> int: + self._buffer.write(text) + return self._orig.write(text) + + def flush(self) -> None: + self._orig.flush() + + def __getattr__(self, name: str) -> Any: + return getattr(self._orig, name) + + +engine_log = _EngineLogBuffer() + + _available_models: list[dict[str, str]] = [{ "id": path, "label": model_label(path) @@ -261,8 +315,8 @@ def list_generators() -> list[dict[str, Any]]: @app.post("/api/generators/preload", status_code=202) def preload_generator(req: GeneratorRequest) -> dict[str, Any]: - """Load a model into memory ahead of time so the first generation with - this configuration skips the model-load wait. Idempotent.""" + """Load a model into memory ahead of time (replacing whatever is + resident). One load at a time — 409 while another load is in flight.""" valid_ids = {m["id"] for m in _available_models} if req.model_id not in valid_ids and not os.path.isdir(req.model_id): raise HTTPException( @@ -270,21 +324,32 @@ def preload_generator(req: GeneratorRequest) -> dict[str, Any]: detail=(f"Unknown model_id '{req.model_id}'. " f"Valid options: {sorted(valid_ids)}"), ) - return job_runner.preload_generator(**req.model_dump()) + try: + return job_runner.preload_generator(**req.model_dump()) + except RuntimeError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc @app.post("/api/generators/unload") -def unload_generator(req: GeneratorRequest) -> dict[str, Any]: - """Shut down a resident generator and free its GPU memory.""" +def unload_generator() -> dict[str, Any]: + """Shut down and delete the resident generator, freeing GPU memory.""" try: - unloaded = job_runner.unload_generator(**req.model_dump()) + unloaded = job_runner.unload_generator() except RuntimeError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc if not unloaded: - raise HTTPException(status_code=404, detail="No such resident generator") + raise HTTPException(status_code=404, detail="No model is loaded") return {"unloaded": True} +@app.get("/api/engine/logs") +def engine_logs(after: int = 0) -> dict[str, Any]: + """Incremental tail of the engine's stdout/stderr (driver + relayed + worker output). Poll with ?after=.""" + lines, total = engine_log.get_lines(after=after) + return {"lines": lines, "total": total} + + @app.post("/api/jobs", status_code=201) def create_job(req: CreateJobRequest) -> dict[str, Any]: """Create a new job (does **not** start it automatically).""" @@ -656,6 +721,14 @@ def create_local_env(host: str, port: int) -> None: def main() -> None: + import sys + sys.stdout = _Tee(sys.stdout, engine_log) + sys.stderr = _Tee(sys.stderr, engine_log) + # handlers created before the tee (module-level basicConfig) hold the + # original stream objects — re-point them or their output bypasses the buffer + for h in logging.getLogger().handlers: + if isinstance(h, logging.StreamHandler) and h.stream in (sys.__stderr__, sys.__stdout__): + h.setStream(sys.stderr) global job_runner, database, upload_dir, verbose, datasets_upload_dir # noqa: PLW0603 # Set up signal handlers to prevent worker crashes from killing the server diff --git a/apps/fastvideo_studio/tests/test_generator_cache.py b/apps/fastvideo_studio/tests/test_generator_cache.py index b4a6d73f4..38c510ad3 100644 --- a/apps/fastvideo_studio/tests/test_generator_cache.py +++ b/apps/fastvideo_studio/tests/test_generator_cache.py @@ -1,9 +1,9 @@ # SPDX-License-Identifier: Apache-2.0 -"""Tests for the resident-generator API: preload, list, unload. +"""Tests for the single-slot resident generator: preload, list, unload. -The generator itself is faked at the JobRunner boundary — these tests pin the -cache/lifecycle contract (idempotent preload, state transitions, unload -refusal while a job runs), not model loading. +Exactly one VideoGenerator lives in memory. One load at a time; loading a new +config always releases the old instance; unload deletes it. The generator is +faked at the ``_create_generator_into_slot``/VideoGenerator boundary. """ from __future__ import annotations @@ -26,99 +26,187 @@ class _FakeGenerator: @pytest.fixture() -def runner(tmp_path, monkeypatch): +def runner(): r = JobRunner.__new__(JobRunner) # skip __init__: no DB/Manager needed r._jobs = {} r._jobs_lock = threading.Lock() - r._generators = {} - r._generators_lock = threading.Lock() - r._preloads = {} + r._generator = None + r._generator_config = None + r._generator_state = "empty" + r._generator_error = None + r._generator_lock = threading.Lock() return r -def _wait_state(runner, model_id, state, timeout=5.0): +def _install_fake_loader(runner, monkeypatch, made: list | None = None, + gate: threading.Event | None = None, + fail: str | None = None): + """Replace the VideoGenerator load inside _create_generator_into_slot.""" + import fastvideo_studio.job_runner as jr + + class _FakeVG: + @staticmethod + def from_pretrained(model_path, **kwargs): + if gate is not None: + assert gate.wait(5), "test gate never opened" + if fail is not None: + raise RuntimeError(fail) + gen = _FakeGenerator() + if made is not None: + made.append((model_path, gen)) + return gen + + monkeypatch.setitem(__import__("sys").modules, "fastvideo", + SimpleNamespace(VideoGenerator=_FakeVG)) + return jr + + +def _wait_state(runner, state, timeout=5.0): deadline = time.time() + timeout while time.time() < deadline: - entries = [g for g in runner.list_generators() if g["model_id"] == model_id] - if entries and entries[0]["state"] == state: - return entries[0] + with runner._generator_lock: + if runner._generator_state == state: + return runner._slot_entry() time.sleep(0.02) - raise AssertionError(f"{model_id} never reached state {state}: {runner.list_generators()}") + raise AssertionError(f"never reached state {state}: {runner._generator_state}") -def test_preload_loads_once_and_lists_ready(runner, monkeypatch): - created = [] +def test_preload_then_ready_then_idempotent(runner, monkeypatch): + made = [] + _install_fake_loader(runner, monkeypatch, made=made) - def fake_create(self, **params): - gen = _FakeGenerator() - created.append(params["model_id"]) - key = self._generator_cache_key(**params) - with self._generators_lock: - self._generators[key] = gen - return gen + entry = runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8) + assert entry["state"] in ("loading", "ready") + _wait_state(runner, "ready") - monkeypatch.setattr(JobRunner, "_get_or_create_generator", fake_create) - - first = runner.preload_generator(model_id="org/model-a", workload_type="t2v", num_gpus=8) - # the loader thread may finish before we return; either state is legal here - assert first["state"] in ("loading", "ready") - ready = _wait_state(runner, "org/model-a", "ready") - assert ready["num_gpus"] == 8 - - # idempotent: preloading a resident config is an immediate ready, no reload - again = runner.preload_generator(model_id="org/model-a", workload_type="t2v", num_gpus=8) + again = runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8) assert again["state"] == "ready" - assert created == ["org/model-a"] + assert len(made) == 1 # same config never reloads -def test_preload_failure_is_reported_not_raised(runner, monkeypatch): - def fake_create(self, **params): - raise RuntimeError("no CUDA on this box") +def test_only_one_load_at_a_time(runner, monkeypatch): + gate = threading.Event() + _install_fake_loader(runner, monkeypatch, gate=gate) - monkeypatch.setattr(JobRunner, "_get_or_create_generator", fake_create) - entry = runner.preload_generator(model_id="org/broken", workload_type="t2v", num_gpus=1) - assert entry["state"] in ("loading", "failed") - failed = _wait_state(runner, "org/broken", "failed") + runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8) + with pytest.raises(RuntimeError, match="already in progress"): + runner.preload_generator(model_id="org/b", workload_type="t2v", num_gpus=8) + gate.set() + _wait_state(runner, "ready") + + +def test_new_config_always_releases_old_instance(runner, monkeypatch): + made = [] + _install_fake_loader(runner, monkeypatch, made=made) + + runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8) + _wait_state(runner, "ready") + first = made[0][1] + + runner.preload_generator(model_id="org/b", workload_type="t2v", num_gpus=8) + _wait_state(runner, "ready") + assert first.shutdown_calls == 1 # old instance released, not stacked + assert [m[0] for m in made] == ["org/a", "org/b"] + assert len(runner.list_generators()) == 1 + assert runner.list_generators()[0]["model_id"] == "org/b" + + +def test_failed_load_reports_error_and_allows_retry(runner, monkeypatch): + _install_fake_loader(runner, monkeypatch, fail="no CUDA on this box") + runner.preload_generator(model_id="org/broken", workload_type="t2v", num_gpus=1) + failed = _wait_state(runner, "failed") assert "no CUDA" in failed["error"] - # a failed preload must not leave a resident entry - assert all(g["state"] != "ready" for g in runner.list_generators()) + + # retry after failure must be accepted (this was the reported bug) + made = [] + _install_fake_loader(runner, monkeypatch, made=made) + runner.preload_generator(model_id="org/broken", workload_type="t2v", num_gpus=1) + _wait_state(runner, "ready") + assert len(made) == 1 -def test_unload_shuts_down_and_removes(runner): - gen = _FakeGenerator() - key = runner._generator_cache_key(model_id="org/model-a", workload_type="t2v", num_gpus=8) - runner._generators[key] = gen +def test_unload_deletes_instance(runner, monkeypatch): + made = [] + _install_fake_loader(runner, monkeypatch, made=made) + runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8) + _wait_state(runner, "ready") - assert runner.unload_generator(model_id="org/model-a", workload_type="t2v", num_gpus=8) is True - assert gen.shutdown_calls == 1 + assert runner.unload_generator() is True + assert made[0][1].shutdown_calls == 1 assert runner.list_generators() == [] - # second unload: nothing resident - assert runner.unload_generator(model_id="org/model-a", workload_type="t2v", num_gpus=8) is False + assert runner._generator is None + assert runner.unload_generator() is False # nothing resident -def test_unload_refuses_while_inference_job_runs(runner): - key = runner._generator_cache_key(model_id="org/model-a", workload_type="t2v", num_gpus=8) - gen = _FakeGenerator() - runner._generators[key] = gen +def test_unload_refuses_while_inference_job_runs(runner, monkeypatch): + made = [] + _install_fake_loader(runner, monkeypatch, made=made) + runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8) + _wait_state(runner, "ready") runner._jobs["j1"] = SimpleNamespace(id="j1", status=JobStatus.RUNNING, job_type="inference") with pytest.raises(RuntimeError, match="j1"): - runner.unload_generator(model_id="org/model-a", workload_type="t2v", num_gpus=8) - assert gen.shutdown_calls == 0 # still resident + runner.unload_generator() + assert made[0][1].shutdown_calls == 0 - # training jobs don't block unload - runner._jobs["j1"].job_type = "finetune" - assert runner.unload_generator(model_id="org/model-a", workload_type="t2v", num_gpus=8) is True + runner._jobs["j1"].job_type = "finetune" # training doesn't block + assert runner.unload_generator() is True -def test_cache_key_matches_job_lookup_defaults(runner): - """A preload with GeneratorRequest defaults must hit the same cache slot a - job with CreateJobRequest defaults resolves to — else preloading is a lie.""" +def test_job_waits_for_matching_preload(runner, monkeypatch): + gate = threading.Event() + made = [] + _install_fake_loader(runner, monkeypatch, made=made, gate=gate) + + runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8) + got = [] + t = threading.Thread( + target=lambda: got.append( + runner._get_or_create_generator("org/a", "t2v", 8)), + daemon=True) + t.start() + time.sleep(0.2) + assert not got # job blocked on the in-flight load + gate.set() + t.join(5) + assert got and got[0] is made[0][1] + assert len(made) == 1 # the job reused the preloaded instance + + +def test_job_with_different_config_replaces_slot(runner, monkeypatch): + made = [] + _install_fake_loader(runner, monkeypatch, made=made) + runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8) + _wait_state(runner, "ready") + + gen = runner._get_or_create_generator("org/b", "t2v", 4) + assert gen is made[1][1] + assert made[0][1].shutdown_calls == 1 # old released before new load + assert runner.list_generators()[0]["model_id"] == "org/b" + + +def test_config_dict_matches_request_defaults(runner): + """GeneratorRequest defaults and CreateJobRequest defaults must resolve to + the same slot config — else preloading never matches the job.""" from fastvideo_studio.models import CreateJobRequest, GeneratorRequest job = CreateJobRequest(model_id="m", prompt="p").model_dump() pre = GeneratorRequest(model_id="m").model_dump() - job_key = runner._generator_cache_key( + assert runner._generator_config_dict(**pre) == runner._generator_config_dict( **{k: job[k] for k in pre}) - pre_key = runner._generator_cache_key(**pre) - assert job_key == pre_key + + +def test_engine_log_buffer_incremental_tail(): + from fastvideo_studio.server import _EngineLogBuffer + + buf = _EngineLogBuffer(maxlen=3) + buf.write("a\nb\n") + buf.write("c") # partial line: not visible yet + lines, total = buf.get_lines(0) + assert lines == ["a", "b"] and total == 2 + buf.write("!\nd\ne\n") # completes "c!", then overflows the ring + lines, total = buf.get_lines(total) + assert lines == ["c!", "d", "e"] and total == 5 + # reader far behind: dropped lines are skipped, no crash + lines, _ = buf.get_lines(0) + assert lines == ["c!", "d", "e"] diff --git a/fastvideo/worker/ray_distributed_executor.py b/fastvideo/worker/ray_distributed_executor.py index 562a697a4..527b20528 100644 --- a/fastvideo/worker/ray_distributed_executor.py +++ b/fastvideo/worker/ray_distributed_executor.py @@ -385,6 +385,16 @@ class RayDistributedExecutor(Executor): ray.kill(worker) self.workers = [] + # Killing the actors does not release the placement group — without + # this, the GPUs stay reserved and no new engine can ever schedule + # in the same ray cluster. + pg = getattr(self.fastvideo_args, "ray_placement_group", None) + if pg is not None: + try: + ray.util.remove_placement_group(pg) + except Exception: # noqa: BLE001 -- already-removed / cluster gone + logger.warning("Failed to remove ray placement group", exc_info=True) + self.fastvideo_args.ray_placement_group = None def __del__(self): self.shutdown()