--- title: gpu_pool ---

GPU pool manager for the streaming server.

Replaces the single-generator path in PR 7.5 with a typed pool abstraction. Three implementations ship here:

Session-to-GPU binding lives in the pool so continuation state stays on the GPU that generated the previous segment (matching the internal gpu_pool.py's per-GPU cache behavior). Cross-GPU handoff is supported via :class:SessionStore snapshot + hydrate, which serializes the state before the migration and rehydrates it on the new worker.

Typed config: workers start from a :class:GeneratorConfig (no flat LTX-2 kwargs), satisfying the PR 6 + PR 7 contracts that the public surface doesn't reintroduce the legacy kwarg bag.

fastvideo.entrypoints.streaming.gpu_pool.GeneratorConfig

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

fastvideo.entrypoints.streaming.gpu_pool.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.gpu_pool.GpuPoolConfig

GpuPoolConfig(num_workers: int | None = None, enable_audio_reencode: bool = True, conditioning_num_frames: int = 9, conditioning_end_offset: int = 0)

fastvideo.entrypoints.streaming.gpu_pool.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.gpu_pool.InMemorySessionStore

InMemorySessionStore()

Thread-safe in-memory :class:SessionStore.

Default implementation used by single-process deployments; a future Redis-backed store can be dropped in without changes to the server.

No eviction / TTL / bounded capacity — sessions only leave via :meth:drop. The live streaming server (PR 7.5+) is responsible for bounding growth and for dropping any :class:BlobStore blobs referenced by a state when that state is replaced or a session ends; this class does not know about blobs.

fastvideo.entrypoints.streaming.gpu_pool.InMemorySessionStore.store

store(session_id: str, state: ContinuationState) -> None

fastvideo.entrypoints.streaming.gpu_pool.InMemorySessionStore.snapshot

snapshot(session_id: str) -> ContinuationState | None

fastvideo.entrypoints.streaming.gpu_pool.InMemorySessionStore.hydrate

hydrate(state: ContinuationState, *, session_id: str | None = None) -> str

fastvideo.entrypoints.streaming.gpu_pool.InMemorySessionStore.drop

drop(session_id: str) -> None

fastvideo.entrypoints.streaming.gpu_pool.SessionStore

SessionStore()

Keyed store for per-session continuation state.

Implementations own the session-id → state mapping. The streaming server calls :meth:store after each segment and :meth:snapshot when a client explicitly asks for an exportable state handle.

fastvideo.entrypoints.streaming.gpu_pool.SessionStore.store

store(session_id: str, state: ContinuationState) -> None

Persist state for session_id, replacing any prior value.

fastvideo.entrypoints.streaming.gpu_pool.SessionStore.snapshot

snapshot(session_id: str) -> ContinuationState | None

Return the current state for session_id (or None).

fastvideo.entrypoints.streaming.gpu_pool.SessionStore.hydrate

hydrate(state: ContinuationState, *, session_id: str | None = None) -> str

Install state as the starting point for a session.

When session_id is None the store allocates a fresh id (UUID4); when provided the store uses it verbatim, overwriting any prior state at that id.

fastvideo.entrypoints.streaming.gpu_pool.SessionStore.drop

drop(session_id: str) -> None

Forget a session. Missing ids are a no-op.

fastvideo.entrypoints.streaming.gpu_pool.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.

fastvideo.entrypoints.streaming.gpu_pool.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.gpu_pool.PoolAssignment

PoolAssignment(gpu_id: int, worker_id: str, pinned_at: float = time.monotonic())

The worker a session is currently bound to.

Source
@dataclass
class PoolAssignment:
    """The worker a session is currently bound to."""

    gpu_id: int
    worker_id: str
    pinned_at: float = field(default_factory=time.monotonic)

fastvideo.entrypoints.streaming.gpu_pool.GpuPool

GpuPool()

Abstract GPU pool.

acquire binds a session to a worker and holds that binding across segments so continuation state can stay hot. run submits a single GenerationRequest for a bound session.

Acquire / release are independent of run — a session can run many segments on one acquired worker, and must release on disconnect.

Source
class GpuPool(ABC):
    """Abstract GPU pool.

    ``acquire`` binds a session to a worker and holds that binding
    across segments so continuation state can stay hot. ``run`` submits
    a single ``GenerationRequest`` for a bound session.

    Acquire / release are independent of run — a session can run many
    segments on one acquired worker, and must release on disconnect.
    """

    @abstractmethod
    async def acquire(
        self,
        session_id: str,
        *,
        timeout: float | None = None,
    ) -> PoolAssignment:
        ...

    @abstractmethod
    async def run(
        self,
        session_id: str,
        request: GenerationRequest,
    ) -> Any:
        ...

    @abstractmethod
    async def release(self, session_id: str) -> None:
        ...

    @abstractmethod
    async def shutdown(self) -> None:
        ...

    @abstractmethod
    def health(self) -> PoolHealth:
        ...

fastvideo.entrypoints.streaming.gpu_pool.GpuPool.acquire

acquire(session_id: str, *, timeout: float | None = None) -> PoolAssignment
Source
@abstractmethod
async def acquire(
    self,
    session_id: str,
    *,
    timeout: float | None = None,
) -> PoolAssignment:
    ...

fastvideo.entrypoints.streaming.gpu_pool.GpuPool.run

run(session_id: str, request: GenerationRequest) -> Any
Source
@abstractmethod
async def run(
    self,
    session_id: str,
    request: GenerationRequest,
) -> Any:
    ...

fastvideo.entrypoints.streaming.gpu_pool.GpuPool.release

release(session_id: str) -> None
Source
@abstractmethod
async def release(self, session_id: str) -> None:
    ...

fastvideo.entrypoints.streaming.gpu_pool.GpuPool.shutdown

shutdown() -> None
Source
@abstractmethod
async def shutdown(self) -> None:
    ...

fastvideo.entrypoints.streaming.gpu_pool.GpuPool.health

health() -> PoolHealth
Source
@abstractmethod
def health(self) -> PoolHealth:
    ...

fastvideo.entrypoints.streaming.gpu_pool.PoolHealth

PoolHealth(total_workers: int, available_workers: int, active_sessions: int, queued_sessions: int = 0)
Source
@dataclass
class PoolHealth:
    total_workers: int
    available_workers: int
    active_sessions: int
    queued_sessions: int = 0

fastvideo.entrypoints.streaming.gpu_pool.PoolAcquireTimeout

PoolAcquireTimeout()

Raised when acquire times out waiting for a free worker.

Source
class PoolAcquireTimeout(RuntimeError):
    """Raised when ``acquire`` times out waiting for a free worker."""

fastvideo.entrypoints.streaming.gpu_pool.InProcessGpuPool

InProcessGpuPool(generator: _GeneratorLike, *, gpu_id: int = 0, session_store: SessionStore | None = None)

Single-process pool backed by one :class:_GeneratorLike.

This is what PR 7.5's server uses by default; PR 7.6 adds the real SubprocessGpuPool alternative but keeps this one for tests and small deployments.

Source
class InProcessGpuPool(GpuPool):
    """Single-process pool backed by one :class:`_GeneratorLike`.

    This is what PR 7.5's server uses by default; PR 7.6 adds the real
    ``SubprocessGpuPool`` alternative but keeps this one for tests and
    small deployments.
    """

    def __init__(
        self,
        generator: _GeneratorLike,
        *,
        gpu_id: int = 0,
        session_store: SessionStore | None = None,
    ) -> None:
        self._generator = generator
        self._gpu_id = gpu_id
        self._worker_id = f"inproc-{uuid.uuid4().hex[:6]}"
        self._session_store = session_store or InMemorySessionStore()
        self._active: dict[str, PoolAssignment] = {}
        self._lock = asyncio.Lock()
        self._gen_lock = asyncio.Lock()

    async def acquire(
        self,
        session_id: str,
        *,
        timeout: float | None = None,
    ) -> PoolAssignment:
        async with self._lock:
            existing = self._active.get(session_id)
            if existing is not None:
                return existing
            assignment = PoolAssignment(gpu_id=self._gpu_id, worker_id=self._worker_id)
            self._active[session_id] = assignment
            return assignment

    async def run(
        self,
        session_id: str,
        request: GenerationRequest,
    ) -> Any:
        if session_id not in self._active:
            raise RuntimeError(f"session {session_id!r} is not acquired on this pool")
        # Serialize generator access so one GPU runs one request at a
        # time, matching the internal gpu_pool's per-GPU lock.
        async with self._gen_lock:
            loop = asyncio.get_running_loop()
            return await loop.run_in_executor(None, self._generator.generate, request)

    async def release(self, session_id: str) -> None:
        async with self._lock:
            self._active.pop(session_id, None)

    async def shutdown(self) -> None:
        self._active.clear()

    def health(self) -> PoolHealth:
        return PoolHealth(
            total_workers=1,
            available_workers=1 if not self._active else 0,
            active_sessions=len(self._active),
        )

fastvideo.entrypoints.streaming.gpu_pool.InProcessGpuPool.acquire

acquire(session_id: str, *, timeout: float | None = None) -> PoolAssignment
Source
async def acquire(
    self,
    session_id: str,
    *,
    timeout: float | None = None,
) -> PoolAssignment:
    async with self._lock:
        existing = self._active.get(session_id)
        if existing is not None:
            return existing
        assignment = PoolAssignment(gpu_id=self._gpu_id, worker_id=self._worker_id)
        self._active[session_id] = assignment
        return assignment

fastvideo.entrypoints.streaming.gpu_pool.InProcessGpuPool.run

run(session_id: str, request: GenerationRequest) -> Any
Source
async def run(
    self,
    session_id: str,
    request: GenerationRequest,
) -> Any:
    if session_id not in self._active:
        raise RuntimeError(f"session {session_id!r} is not acquired on this pool")
    # Serialize generator access so one GPU runs one request at a
    # time, matching the internal gpu_pool's per-GPU lock.
    async with self._gen_lock:
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(None, self._generator.generate, request)

fastvideo.entrypoints.streaming.gpu_pool.InProcessGpuPool.release

release(session_id: str) -> None
Source
async def release(self, session_id: str) -> None:
    async with self._lock:
        self._active.pop(session_id, None)

fastvideo.entrypoints.streaming.gpu_pool.InProcessGpuPool.shutdown

shutdown() -> None
Source
async def shutdown(self) -> None:
    self._active.clear()

fastvideo.entrypoints.streaming.gpu_pool.InProcessGpuPool.health

health() -> PoolHealth
Source
def health(self) -> PoolHealth:
    return PoolHealth(
        total_workers=1,
        available_workers=1 if not self._active else 0,
        active_sessions=len(self._active),
    )

fastvideo.entrypoints.streaming.gpu_pool.SubprocessGpuPool

SubprocessGpuPool(generator_config: GeneratorConfig, *, pool_config: GpuPoolConfig, warmup_config: WarmupConfig | None = None, session_store: SessionStore | None = None, worker_factory: WorkerFactory | None = None)

One multiprocessing.Process per GPU.

Each worker boots :class:fastvideo.VideoGenerator from a typed :class:GeneratorConfig inside the child process (post- CUDA_VISIBLE_DEVICES setup) and consumes jobs from an mp Queue.

This is the production shape: the parent process stays CPU-only, and GPU state never crosses process boundaries. Continuation state is serialized through :class:SessionStore for cross-GPU handoff.

PR 7.6 ships this as an opt-in; PR 7.5's in-process pool remains the default until nightly runs validate the subprocess path.

Source
class SubprocessGpuPool(GpuPool):
    """One ``multiprocessing.Process`` per GPU.

    Each worker boots :class:`fastvideo.VideoGenerator` from a typed
    :class:`GeneratorConfig` inside the child process (post-
    ``CUDA_VISIBLE_DEVICES`` setup) and consumes jobs from an mp Queue.

    This is the production shape: the parent process stays CPU-only, and
    GPU state never crosses process boundaries. Continuation state is
    serialized through :class:`SessionStore` for cross-GPU handoff.

    PR 7.6 ships this as an opt-in; PR 7.5's in-process pool remains the
    default until nightly runs validate the subprocess path.
    """

    def __init__(
        self,
        generator_config: GeneratorConfig,
        *,
        pool_config: GpuPoolConfig,
        warmup_config: WarmupConfig | None = None,
        session_store: SessionStore | None = None,
        worker_factory: WorkerFactory | None = None,
    ) -> None:
        self._generator_config = generator_config
        self._pool_config = pool_config
        self._warmup_config = warmup_config or WarmupConfig()
        self._session_store = session_store or InMemorySessionStore()
        self._worker_factory = worker_factory or _default_worker_factory
        self._workers: list[_WorkerHandle] = []
        self._available: asyncio.Queue[int] = asyncio.Queue()
        self._assignments: dict[str, PoolAssignment] = {}
        self._worker_by_id: dict[str, _WorkerHandle] = {}
        self._pending: dict[str, _PendingJob] = {}
        self._lock = asyncio.Lock()
        self._result_reader_tasks: list[asyncio.Task] = []

    async def start(self) -> None:
        """Spawn worker processes and wait for each to report ready."""
        num_workers = self._pool_config.num_workers or 1
        for gpu_id in range(num_workers):
            handle = self._worker_factory(
                gpu_id=gpu_id,
                generator_config=self._generator_config,
                warmup_config=self._warmup_config,
            )
            self._workers.append(handle)
            self._worker_by_id[handle.worker_id] = handle

        # Wait for each worker's ready event in a thread to avoid
        # blocking the event loop.
        loop = asyncio.get_running_loop()
        await asyncio.gather(*[
            loop.run_in_executor(None, handle.ready.wait, self._warmup_config.timeout_seconds)
            for handle in self._workers
        ])

        # Start background result readers — one task per worker
        # drains its result queue and resolves futures in _pending.
        for handle in self._workers:
            task = asyncio.create_task(self._drain_results(handle))
            self._result_reader_tasks.append(task)

        # Only admit workers that successfully booted. Anything that
        # failed boot (timeout, crash, error sentinel) stays out of the
        # available queue so we never assign a session to it.
        for idx, handle in enumerate(self._workers):
            if handle.boot_ok.is_set():
                await self._available.put(idx)
            else:
                logger.error(
                    "pool: worker %s failed to boot; skipping",
                    handle.worker_id,
                )

    async def acquire(
        self,
        session_id: str,
        *,
        timeout: float | None = None,
    ) -> PoolAssignment:
        async with self._lock:
            existing = self._assignments.get(session_id)
            if existing is not None:
                return existing
        try:
            idx = await asyncio.wait_for(self._available.get(), timeout=timeout)
        except asyncio.TimeoutError as exc:
            raise PoolAcquireTimeout(f"no worker available after {timeout}s") from exc
        handle = self._workers[idx]
        assignment = PoolAssignment(gpu_id=handle.gpu_id, worker_id=handle.worker_id)
        async with self._lock:
            self._assignments[session_id] = assignment
        return assignment

    async def run(
        self,
        session_id: str,
        request: GenerationRequest,
    ) -> Any:
        assignment = self._assignments.get(session_id)
        if assignment is None:
            raise RuntimeError(f"session {session_id!r} not acquired on this pool")
        handle = self._worker_by_id[assignment.worker_id]
        job_id = uuid.uuid4().hex
        future: Future = Future()
        self._pending[job_id] = _PendingJob(
            job_id=job_id,
            future=future,
            session_id=session_id,
            worker_id=handle.worker_id,
        )
        # mp.Queue.put can block if the underlying pipe buffer is full;
        # offload to a thread so the event loop keeps serving other
        # sessions. A cancelled caller cannot stop the executor thread or
        # generation, so detach its pending entry and discard late replies.
        loop = asyncio.get_running_loop()
        try:
            await loop.run_in_executor(
                None,
                handle.job_queue.put,
                {
                    "job_id": job_id,
                    "request": request
                },
            )
            return await asyncio.wrap_future(future)
        finally:
            self._pending.pop(job_id, None)
            future.cancel()

    async def release(self, session_id: str) -> None:
        async with self._lock:
            assignment = self._assignments.pop(session_id, None)
        if assignment is None:
            return
        idx = next((i for i, h in enumerate(self._workers) if h.worker_id == assignment.worker_id), None)
        if idx is None:
            return
        # Don't return a dead worker to the pool; otherwise the next
        # acquire will hand a session to a process that can't run jobs.
        if not self._workers[idx].process.is_alive():
            logger.warning(
                "pool: worker %s died; not returning to available queue",
                self._workers[idx].worker_id,
            )
            return
        await self._available.put(idx)

    async def shutdown(self) -> None:
        loop = asyncio.get_running_loop()

        # Signal all workers in parallel; .put may block on a full pipe,
        # so off-load it the same way run() does.
        async def _signal(handle: _WorkerHandle) -> None:
            try:
                handle.shutdown_event.set()
                await loop.run_in_executor(None, handle.job_queue.put, None)
            except Exception:  # pragma: no cover - best-effort cleanup
                pass

        await asyncio.gather(*(_signal(h) for h in self._workers))
        # Join in parallel so total shutdown is bounded by the slowest
        # worker, not the sum of all timeouts.
        await asyncio.gather(*(loop.run_in_executor(None, handle.process.join, 5.0) for handle in self._workers))
        for handle in self._workers:
            if handle.process.is_alive():
                handle.process.kill()
        for task in self._result_reader_tasks:
            task.cancel()
        self._result_reader_tasks.clear()
        self._workers.clear()
        self._worker_by_id.clear()

    def health(self) -> PoolHealth:
        return PoolHealth(
            total_workers=len(self._workers),
            available_workers=self._available.qsize(),
            active_sessions=len(self._assignments),
        )

    async def _drain_results(self, handle: _WorkerHandle) -> None:
        loop = asyncio.get_running_loop()
        try:
            while not handle.shutdown_event.is_set():
                try:
                    msg = await loop.run_in_executor(None, _safe_queue_get, handle.result_queue, 0.5)
                except Exception:
                    logger.exception("pool: worker %s result reader failed", handle.worker_id)
                    return
                if msg is None:
                    continue
                job_id = msg.get("job_id")
                if job_id is None:
                    continue
                pending = self._pending.pop(job_id, None)
                # wrap_future can cancel the source future before run()'s
                # finally block gets its turn on the event loop, and an
                # abandoned job's late reply can arrive after run() already
                # popped its entry. Either way there is no live waiter.
                if pending is None or pending.future.done():
                    logger.debug("pool: worker %s dropping reply for job %s", handle.worker_id, job_id)
                    continue
                if msg.get("kind") == "error":
                    pending.future.set_exception(RuntimeError(msg["error"]))
                else:
                    pending.future.set_result(msg.get("result"))
        finally:
            # If we exit for any reason — shutdown, exception, cancel —
            # surface that to any in-flight jobs on this worker so their
            # await never hangs on a future no one will resolve.
            for jid in [jid for jid, job in self._pending.items() if job.worker_id == handle.worker_id]:
                pending = self._pending.pop(jid, None)
                if pending is not None and not pending.future.done():
                    pending.future.set_exception(
                        RuntimeError(f"worker {handle.worker_id} result reader exited "
                                     "with pending jobs"))

fastvideo.entrypoints.streaming.gpu_pool.SubprocessGpuPool.start

start() -> None

Spawn worker processes and wait for each to report ready.

Source
async def start(self) -> None:
    """Spawn worker processes and wait for each to report ready."""
    num_workers = self._pool_config.num_workers or 1
    for gpu_id in range(num_workers):
        handle = self._worker_factory(
            gpu_id=gpu_id,
            generator_config=self._generator_config,
            warmup_config=self._warmup_config,
        )
        self._workers.append(handle)
        self._worker_by_id[handle.worker_id] = handle

    # Wait for each worker's ready event in a thread to avoid
    # blocking the event loop.
    loop = asyncio.get_running_loop()
    await asyncio.gather(*[
        loop.run_in_executor(None, handle.ready.wait, self._warmup_config.timeout_seconds)
        for handle in self._workers
    ])

    # Start background result readers — one task per worker
    # drains its result queue and resolves futures in _pending.
    for handle in self._workers:
        task = asyncio.create_task(self._drain_results(handle))
        self._result_reader_tasks.append(task)

    # Only admit workers that successfully booted. Anything that
    # failed boot (timeout, crash, error sentinel) stays out of the
    # available queue so we never assign a session to it.
    for idx, handle in enumerate(self._workers):
        if handle.boot_ok.is_set():
            await self._available.put(idx)
        else:
            logger.error(
                "pool: worker %s failed to boot; skipping",
                handle.worker_id,
            )

fastvideo.entrypoints.streaming.gpu_pool.SubprocessGpuPool.acquire

acquire(session_id: str, *, timeout: float | None = None) -> PoolAssignment
Source
async def acquire(
    self,
    session_id: str,
    *,
    timeout: float | None = None,
) -> PoolAssignment:
    async with self._lock:
        existing = self._assignments.get(session_id)
        if existing is not None:
            return existing
    try:
        idx = await asyncio.wait_for(self._available.get(), timeout=timeout)
    except asyncio.TimeoutError as exc:
        raise PoolAcquireTimeout(f"no worker available after {timeout}s") from exc
    handle = self._workers[idx]
    assignment = PoolAssignment(gpu_id=handle.gpu_id, worker_id=handle.worker_id)
    async with self._lock:
        self._assignments[session_id] = assignment
    return assignment

fastvideo.entrypoints.streaming.gpu_pool.SubprocessGpuPool.run

run(session_id: str, request: GenerationRequest) -> Any
Source
async def run(
    self,
    session_id: str,
    request: GenerationRequest,
) -> Any:
    assignment = self._assignments.get(session_id)
    if assignment is None:
        raise RuntimeError(f"session {session_id!r} not acquired on this pool")
    handle = self._worker_by_id[assignment.worker_id]
    job_id = uuid.uuid4().hex
    future: Future = Future()
    self._pending[job_id] = _PendingJob(
        job_id=job_id,
        future=future,
        session_id=session_id,
        worker_id=handle.worker_id,
    )
    # mp.Queue.put can block if the underlying pipe buffer is full;
    # offload to a thread so the event loop keeps serving other
    # sessions. A cancelled caller cannot stop the executor thread or
    # generation, so detach its pending entry and discard late replies.
    loop = asyncio.get_running_loop()
    try:
        await loop.run_in_executor(
            None,
            handle.job_queue.put,
            {
                "job_id": job_id,
                "request": request
            },
        )
        return await asyncio.wrap_future(future)
    finally:
        self._pending.pop(job_id, None)
        future.cancel()

fastvideo.entrypoints.streaming.gpu_pool.SubprocessGpuPool.release

release(session_id: str) -> None
Source
async def release(self, session_id: str) -> None:
    async with self._lock:
        assignment = self._assignments.pop(session_id, None)
    if assignment is None:
        return
    idx = next((i for i, h in enumerate(self._workers) if h.worker_id == assignment.worker_id), None)
    if idx is None:
        return
    # Don't return a dead worker to the pool; otherwise the next
    # acquire will hand a session to a process that can't run jobs.
    if not self._workers[idx].process.is_alive():
        logger.warning(
            "pool: worker %s died; not returning to available queue",
            self._workers[idx].worker_id,
        )
        return
    await self._available.put(idx)

fastvideo.entrypoints.streaming.gpu_pool.SubprocessGpuPool.shutdown

shutdown() -> None
Source
async def shutdown(self) -> None:
    loop = asyncio.get_running_loop()

    # Signal all workers in parallel; .put may block on a full pipe,
    # so off-load it the same way run() does.
    async def _signal(handle: _WorkerHandle) -> None:
        try:
            handle.shutdown_event.set()
            await loop.run_in_executor(None, handle.job_queue.put, None)
        except Exception:  # pragma: no cover - best-effort cleanup
            pass

    await asyncio.gather(*(_signal(h) for h in self._workers))
    # Join in parallel so total shutdown is bounded by the slowest
    # worker, not the sum of all timeouts.
    await asyncio.gather(*(loop.run_in_executor(None, handle.process.join, 5.0) for handle in self._workers))
    for handle in self._workers:
        if handle.process.is_alive():
            handle.process.kill()
    for task in self._result_reader_tasks:
        task.cancel()
    self._result_reader_tasks.clear()
    self._workers.clear()
    self._worker_by_id.clear()

fastvideo.entrypoints.streaming.gpu_pool.SubprocessGpuPool.health

health() -> PoolHealth
Source
def health(self) -> PoolHealth:
    return PoolHealth(
        total_workers=len(self._workers),
        available_workers=self._available.qsize(),
        active_sessions=len(self._assignments),
    )

fastvideo.entrypoints.streaming.gpu_pool.WorkerFactory

WorkerFactory()
Source
class WorkerFactory(Protocol):

    def __call__(
        self,
        *,
        gpu_id: int,
        generator_config: GeneratorConfig,
        warmup_config: WarmupConfig,
    ) -> _WorkerHandle:
        ...