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.

Source

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),
            })