Skip to content

serving_engine

Shared asynchronous execution substrate for OpenAI-compatible routes.

Classes

fastvideo.entrypoints.openai.serving_engine.OpenAIServingEngine 

OpenAIServingEngine(generator: ServingGenerator, video_request_validator: Callable[[VideoGenerationRequest], None] | None = None)

Own generator lifecycle and serialize access to its mutable pipeline.

FastVideo pipelines contain request-mutated sampling state and some LoRA implementations merge weights in place. Running two Python threads through one pipeline is therefore unsafe even if the HTTP layer accepts requests concurrently. This engine gives every OpenAI route one model-agnostic async entrypoint while preserving that invariant. A future scheduler can replace the lock without changing the transport contract.

Source code in fastvideo/entrypoints/openai/serving_engine.py 
36
37
38
39
40
41
42
43
def __init__(self,
             generator: ServingGenerator,
             video_request_validator: Callable[[VideoGenerationRequest], None] | None = None) -> None:
    self._generator = generator
    self._video_request_validator = video_request_validator
    self._generation_lock = asyncio.Lock()
    self._closed = False
    self._unhealthy_reason: str | None = None

Methods:

fastvideo.entrypoints.openai.serving_engine.OpenAIServingEngine.generate async 
generate(request: GenerationRequest, *, on_start: Callable[[], Awaitable[None]] | None = None) -> Any

Generate one typed request without blocking the event loop.

Source code in fastvideo/entrypoints/openai/serving_engine.py 
75
76
77
78
79
80
81
82
async def generate(
    self,
    request: GenerationRequest,
    *,
    on_start: Callable[[], Awaitable[None]] | None = None,
) -> Any:
    """Generate one typed request without blocking the event loop."""
    return await self.run_serialized(self._generator.generate, request, on_start=on_start)
fastvideo.entrypoints.openai.serving_engine.OpenAIServingEngine.run_async_serialized async 
run_async_serialized(function: Callable[[], Awaitable[_T]]) -> _T

Run an async operation under the same pipeline lock.

Source code in fastvideo/entrypoints/openai/serving_engine.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
async def run_async_serialized(self, function: Callable[[], Awaitable[_T]]) -> _T:
    """Run an async operation under the same pipeline lock."""
    if self._closed:
        raise RuntimeError("FastVideo serving engine is shutting down")
    async with self._generation_lock:
        if self._closed:
            raise RuntimeError("FastVideo serving engine is shutting down")
        worker: asyncio.Future[_T] = asyncio.ensure_future(function())
        try:
            return await asyncio.shield(worker)
        except asyncio.CancelledError:
            await self._wait_after_cancellation(worker)
            raise
fastvideo.entrypoints.openai.serving_engine.OpenAIServingEngine.run_serialized async 
run_serialized(function: Callable[..., _T], *args: Any, on_start: Callable[[], Awaitable[None]] | None = None, **kwargs: Any) -> _T

Run a synchronous pipeline operation under the serving lock.

Source code in fastvideo/entrypoints/openai/serving_engine.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
async def run_serialized(
    self,
    function: Callable[..., _T],
    *args: Any,
    on_start: Callable[[], Awaitable[None]] | None = None,
    **kwargs: Any,
) -> _T:
    """Run a synchronous pipeline operation under the serving lock."""
    if self._closed:
        raise RuntimeError("FastVideo serving engine is shutting down")
    async with self._generation_lock:
        if self._closed:
            raise RuntimeError("FastVideo serving engine is shutting down")
        if on_start is not None:
            await on_start()
        worker = asyncio.create_task(asyncio.to_thread(function, *args, **kwargs))
        try:
            return await asyncio.shield(worker)
        except asyncio.CancelledError:
            # Python cannot stop a running worker thread. Keep the lock
            # until the pipeline call really exits so cancellation cannot
            # expose mutable model state to a second request.
            await self._wait_after_cancellation(worker)
            raise
        except (BrokenPipeError, EOFError) as error:
            self._unhealthy_reason = str(error)
            raise
fastvideo.entrypoints.openai.serving_engine.OpenAIServingEngine.shutdown async 
shutdown() -> None

Stop accepting requests and release the generator after in-flight work.

Source code in fastvideo/entrypoints/openai/serving_engine.py 
139
140
141
142
143
async def shutdown(self) -> None:
    """Stop accepting requests and release the generator after in-flight work."""
    self._closed = True
    async with self._generation_lock:
        await asyncio.to_thread(self._generator.shutdown)