vllm.multimodal.media.decode

Classes:

• MediaDecodeJob –

  A submitted decode and its location in the original request.

Functions:

• collect_media_decodes –

  Drain every decode before reporting the first failure in request order.

• collect_media_decodes_async –

  Drain decodes through caller cancellation without blocking the event loop.

• submit_media_decodes –

  Append submitted work immediately so partial submission can be drained.

MediaDecodeJob

Bases: NamedTuple

A submitted decode and its location in the original request.

Source code in vllm/multimodal/media/decode.py

class MediaDecodeJob(NamedTuple):
    """A submitted decode and its location in the original request."""

    modality: str
    original_index: int
    future: Future[Any]

collect_media_decodes(jobs, error_factory)

Drain every decode before reporting the first failure in request order.

Source code in vllm/multimodal/media/decode.py

def collect_media_decodes(
    jobs: Sequence[MediaDecodeJob], error_factory: DecodeError
) -> None:
    """Drain every decode before reporting the first failure in request order."""
    first_error: tuple[MediaDecodeJob, Exception] | None = None
    for job in jobs:
        try:
            job.future.result()
        except Exception as error:
            if first_error is None:
                first_error = job, error
    if first_error is not None:
        job, cause = first_error
        raise error_factory(job.modality, job.original_index, cause) from cause

collect_media_decodes_async(jobs, error_factory) async

Drain decodes through caller cancellation without blocking the event loop.

Source code in vllm/multimodal/media/decode.py

async def collect_media_decodes_async(
    jobs: Sequence[MediaDecodeJob], error_factory: DecodeError
) -> None:
    """Drain decodes through caller cancellation without blocking the event loop."""
    if not jobs:
        return
    pending = asyncio.gather(
        *(asyncio.wrap_future(job.future) for job in jobs), return_exceptions=True
    )
    results = await await_with_cancellation_drain(pending)
    for job, result in zip(jobs, results):
        if isinstance(result, Exception):
            raise error_factory(job.modality, job.original_index, result) from result
        if isinstance(result, BaseException):
            raise result

submit_media_decodes(items, executor, jobs)

Append submitted work immediately so partial submission can be drained.

Source code in vllm/multimodal/media/decode.py

def submit_media_decodes(
    items: Iterable[tuple[str, int, MediaRef[Any]]],
    executor: Executor,
    jobs: list[MediaDecodeJob],
) -> None:
    """Append submitted work immediately so partial submission can be drained."""
    for modality, index, ref in items:
        if not ref.is_decoded:
            jobs.append(MediaDecodeJob(modality, index, executor.submit(ref.decode)))