gpu_pool

@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)</code></pre></details>
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.
&quot;&quot;&quot;

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

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

@abstractmethod
async def release(self, session_id: str) -&gt; None:
    ...

@abstractmethod
async def shutdown(self) -&gt; None:
    ...

@abstractmethod
def health(self) -&gt; PoolHealth:
    ...</code></pre></details>
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&#x27;s server uses by default; PR 7.6 adds the real
``SubprocessGpuPool`` alternative but keeps this one for tests and
small deployments.
&quot;&quot;&quot;

def __init__(
    self,
    generator: _GeneratorLike,
    *,
    gpu_id: int = 0,
    session_store: SessionStore | None = None,
) -&gt; None:
    self._generator = generator
    self._gpu_id = gpu_id
    self._worker_id = f&quot;inproc-&#123;uuid.uuid4().hex&#91;:6&#93;&#125;&quot;
    self._session_store = session_store or InMemorySessionStore()
    self._active: dict&#91;str, PoolAssignment&#93; = &#123;&#125;
    self._lock = asyncio.Lock()
    self._gen_lock = asyncio.Lock()

async def acquire(
    self,
    session_id: str,
    *,
    timeout: float | None = None,
) -&gt; 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&#91;session_id&#93; = assignment
        return assignment

async def run(
    self,
    session_id: str,
    request: GenerationRequest,
) -&gt; Any:
    if session_id not in self._active:
        raise RuntimeError(f&quot;session &#123;session_id!r&#125; is not acquired on this pool&quot;)
    # Serialize generator access so one GPU runs one request at a
    # time, matching the internal gpu_pool&#x27;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) -&gt; None:
    async with self._lock:
        self._active.pop(session_id, None)

async def shutdown(self) -&gt; None:
    self._active.clear()

def health(self) -&gt; PoolHealth:
    return PoolHealth(
        total_workers=1,
        available_workers=1 if not self._active else 0,
        active_sessions=len(self._active),
    )</code></pre></details>
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&#x27;s in-process pool remains the
default until nightly runs validate the subprocess path.
&quot;&quot;&quot;

def __init__(
    self,
    generator_config: GeneratorConfig,
    *,
    pool_config: GpuPoolConfig,
    warmup_config: WarmupConfig | None = None,
    session_store: SessionStore | None = None,
    worker_factory: WorkerFactory | None = None,
) -&gt; 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&#91;_WorkerHandle&#93; = &#91;&#93;
    self._available: asyncio.Queue&#91;int&#93; = asyncio.Queue()
    self._assignments: dict&#91;str, PoolAssignment&#93; = &#123;&#125;
    self._worker_by_id: dict&#91;str, _WorkerHandle&#93; = &#123;&#125;
    self._pending: dict&#91;str, _PendingJob&#93; = &#123;&#125;
    self._lock = asyncio.Lock()
    self._result_reader_tasks: list&#91;asyncio.Task&#93; = &#91;&#93;

async def start(self) -&gt; None:
    &quot;&quot;&quot;Spawn worker processes and wait for each to report ready.&quot;&quot;&quot;
    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&#91;handle.worker_id&#93; = handle

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

    # 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(
                &quot;pool: worker %s failed to boot; skipping&quot;,
                handle.worker_id,
            )

async def acquire(
    self,
    session_id: str,
    *,
    timeout: float | None = None,
) -&gt; 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&quot;no worker available after &#123;timeout&#125;s&quot;) from exc
    handle = self._workers&#91;idx&#93;
    assignment = PoolAssignment(gpu_id=handle.gpu_id, worker_id=handle.worker_id)
    async with self._lock:
        self._assignments&#91;session_id&#93; = assignment
    return assignment

async def run(
    self,
    session_id: str,
    request: GenerationRequest,
) -&gt; Any:
    assignment = self._assignments.get(session_id)
    if assignment is None:
        raise RuntimeError(f&quot;session &#123;session_id!r&#125; not acquired on this pool&quot;)
    handle = self._worker_by_id&#91;assignment.worker_id&#93;
    job_id = uuid.uuid4().hex
    future: Future = Future()
    self._pending&#91;job_id&#93; = _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,
            &#123;
                &quot;job_id&quot;: job_id,
                &quot;request&quot;: request
            &#125;,
        )
        return await asyncio.wrap_future(future)
    finally:
        self._pending.pop(job_id, None)
        future.cancel()

async def release(self, session_id: str) -&gt; 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&#x27;t return a dead worker to the pool; otherwise the next
    # acquire will hand a session to a process that can&#x27;t run jobs.
    if not self._workers&#91;idx&#93;.process.is_alive():
        logger.warning(
            &quot;pool: worker %s died; not returning to available queue&quot;,
            self._workers&#91;idx&#93;.worker_id,
        )
        return
    await self._available.put(idx)

async def shutdown(self) -&gt; 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) -&gt; 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) -&gt; 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) -&gt; 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(&quot;pool: worker %s result reader failed&quot;, handle.worker_id)
                return
            if msg is None:
                continue
            job_id = msg.get(&quot;job_id&quot;)
            if job_id is None:
                continue
            pending = self._pending.pop(job_id, None)
            # wrap_future can cancel the source future before run()&#x27;s
            # finally block gets its turn on the event loop, and an
            # abandoned job&#x27;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(&quot;pool: worker %s dropping reply for job %s&quot;, handle.worker_id, job_id)
                continue
            if msg.get(&quot;kind&quot;) == &quot;error&quot;:
                pending.future.set_exception(RuntimeError(msg&#91;&quot;error&quot;&#93;))
            else:
                pending.future.set_result(msg.get(&quot;result&quot;))
    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 &#91;jid for jid, job in self._pending.items() if job.worker_id == handle.worker_id&#93;:
            pending = self._pending.pop(jid, None)
            if pending is not None and not pending.future.done():
                pending.future.set_exception(
                    RuntimeError(f&quot;worker &#123;handle.worker_id&#125; result reader exited &quot;
                                 &quot;with pending jobs&quot;))</code></pre></details>
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&#x27;s ready event in a thread to avoid
# blocking the event loop.
loop = asyncio.get_running_loop()
await asyncio.gather(*&#91;
    loop.run_in_executor(None, handle.ready.wait, self._warmup_config.timeout_seconds)
    for handle in self._workers
&#93;)

# 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(
            &quot;pool: worker %s failed to boot; skipping&quot;,
            handle.worker_id,
        )</code></pre></details>
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) -&gt; 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()</code></pre></details>
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,
) -&gt; _WorkerHandle:
    ...</code></pre></details>