worker

Per-GPU worker subprocess entry for :class:SubprocessGpuPool.

The pool manages binding, lifecycle, and message dispatch in the parent process. The worker constructs its :class:VideoGenerator from a typed :class:GeneratorConfig, runs the two-segment warmup so both initial-segment and continuation-branch compile graphs are hot, and then loops on the job queue.

fastvideo.entrypoints.streaming.worker.GeneratorConfig

GeneratorConfig(model_path: str, revision: str | None = None, trust_remote_code: bool = False, engine: EngineConfig = EngineConfig(), pipeline: PipelineSelection = PipelineSelection())

fastvideo.entrypoints.streaming.worker.GenerationRequest

GenerationRequest(prompt: str | list[str] | None = None, negative_prompt: str | None = None, inputs: InputConfig = InputConfig(), sampling: SamplingConfig = SamplingConfig(), runtime: RequestRuntimeConfig = RequestRuntimeConfig(), output: OutputConfig = OutputConfig(), stage_overrides: dict[str, Any] = dict(), state: ContinuationState | None = None, plan: GenerationPlan | None = None, extensions: dict[str, Any] = dict())

fastvideo.entrypoints.streaming.worker.InputConfig

InputConfig(prompt_path: str | None = None, image_path: str | list[str] | None = None, video_path: str | list[str] | None = None, pose_video_path: str | list[str] | None = None, face_video_path: str | list[str] | None = None, background_video_path: str | list[str] | None = None, mask_video_path: str | list[str] | None = None, animate_mode: str | None = None, audio_path: str | list[str] | None = None, pil_image: Any | None = None, last_image: Any | None = None, references: list[Any] | None = None, latents: Any | None = None, audio_latents: Any | None = None, pose: str | None = None, mouse_cond: Any | None = None, keyboard_cond: Any | None = None, grid_sizes: Any | None = None, c2ws_plucker_emb: Any | None = None, action_path: str | None = None)

fastvideo.entrypoints.streaming.worker.OutputConfig

OutputConfig(output_path: str = 'outputs/', output_video_name: str | None = None, save_video: bool = True, return_frames: bool = True, return_state: bool = False, *, return_samples: bool = False)

fastvideo.entrypoints.streaming.worker.SamplingConfig

SamplingConfig(num_videos_per_prompt: int = 1, seed: int = 1024, max_sequence_length: int | None = None, num_frames: int = 125, height: int = 720, width: int = 1280, height_sr: int = 1072, width_sr: int = 1920, fps: int = 24, audio_start_in_s: float | None = None, audio_end_in_s: float | None = None, num_inference_steps: int = 50, num_inference_steps_sr: int = 50, guidance_scale: float = 1.0, batch_cfg: bool = False, guidance_scale_2: float | None = None, guidance_rescale: float = 0.0, true_cfg_scale: float | None = None, use_embedded_guidance: bool | None = None, boundary_ratio: float | None = None, sigmas: list[float] | None = None, pyramid_num_inference_steps_list: list[int] | None = None, history_sizes: list[int] | None = None, num_latent_frames_per_chunk: int = 9, keep_first_frame: bool = True, is_skip_first_chunk: bool = False, use_zero_init: bool = True, zero_steps: int = 1, is_amplify_first_chunk: bool = False)

fastvideo.entrypoints.streaming.worker.WarmupConfig

WarmupConfig(enabled: bool = True, prompt: str = 'A cinematic drone shot over coastal cliffs at sunrise, golden light, gentle ocean waves, ultra detailed', timeout_seconds: int = 2400)

fastvideo.entrypoints.streaming.worker.init_logger

init_logger(name: str) -> _FastvideoLogger

The main purpose of this function is to ensure that loggers are retrieved in such a way that we can be sure the root fastvideo logger has already been configured.

fastvideo.entrypoints.streaming.worker.worker_main

worker_main(*, gpu_id: int, worker_id: str, generator_config: GeneratorConfig, warmup_config: WarmupConfig, job_queue: mp.Queue, result_queue: mp.Queue, shutdown_event: Any) -> None

Per-worker subprocess entry.

Runs inside the child spawned by SubprocessGpuPool. Blocking VideoGenerator construction + generation happens here, not in the parent's event loop.

def worker_main(
    *,
    gpu_id: int,
    worker_id: str,
    generator_config: GeneratorConfig,
    warmup_config: WarmupConfig,
    job_queue: mp.Queue,
    result_queue: mp.Queue,
    shutdown_event: Any,
) -> None:  # pragma: no cover - exercised via integration only
    """Per-worker subprocess entry.
Runs inside the child spawned by ``SubprocessGpuPool``. Blocking
``VideoGenerator`` construction + generation happens here, not in
the parent's event loop.
"""
import fastvideo.envs as envs

envs.set_external("CUDA_VISIBLE_DEVICES", str(gpu_id))
try:
    from fastvideo import VideoGenerator
    from fastvideo.worker.executor import reject_external_launcher

    reject_external_launcher(
        generator_config.engine.execution_backend,
        entrypoint="the streaming worker",
    )

    generator = VideoGenerator.from_pretrained(config=generator_config)
    if warmup_config.enabled:
        _warmup_worker(generator, warmup_config)
    result_queue.put({"kind": "ready", "worker_id": worker_id})
except Exception as exc:
    result_queue.put({"kind": "error", "error": repr(exc)})
    return

while not shutdown_event.is_set():
    try:
        item = job_queue.get(timeout=0.5)
    except queue.Empty:
        continue
    if item is None:
        break
    job_id = item["job_id"]
    request = item["request"]
    try:
        result = generator.generate(request)
        result_queue.put({
            "kind": "result",
            "job_id": job_id,
            "result": result,
        })
    except Exception as exc:
        result_queue.put({
            "kind": "error",
            "job_id": job_id,
            "error": repr(exc),
        &#125;)</code></pre></details>