Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion components/src/dynamo/common/multimodal/audio_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import numpy as np

from dynamo.common.http import HttpStatusError, fetch_bytes
from dynamo.common.http.media_reference import max_media_bytes
from dynamo.common.http.url_validator import (
UrlValidationError,
UrlValidationPolicy,
Expand Down Expand Up @@ -128,7 +129,10 @@ async def _load_audio_with_vllm(self, audio_url: str) -> tuple[np.ndarray, float
# data: and file:// never touch the network, so vLLM can handle them.
if urlparse(normalized_url).scheme in ("http", "https"):
content = await fetch_bytes(
normalized_url, self._http_timeout, policy=self._url_policy
normalized_url,
self._http_timeout,
policy=self._url_policy,
max_bytes=max_media_bytes(),
Comment thread
host452b marked this conversation as resolved.
Resolved
)
return await asyncio.to_thread(media_io.load_bytes, content)

Expand Down
15 changes: 14 additions & 1 deletion components/src/dynamo/common/multimodal/image_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
HttpTimeoutError,
fetch_bytes,
)
from ..http.media_reference import max_media_bytes
from ..http.url_validator import (
UrlValidationError,
UrlValidationPolicy,
Expand Down Expand Up @@ -71,6 +72,7 @@ def __init__(
http_timeout: float = 30.0,
enable_frontend_decoding: bool = False,
url_policy: UrlValidationPolicy | None = None,
max_bytes: int | None = None,
):
"""
Initialize the ImageLoader with caching, HTTP settings, and optional NIXL config for
Expand All @@ -85,9 +87,12 @@ def __init__(
decoded images directly from frontend memory, bypassing standard
network transport. Defaults to False.
url_policy: Policy for validating URLs. Defaults to UrlValidationPolicy.from_env().
max_bytes: Maximum remote image size in bytes. When omitted, resolve
DYN_MM_MAX_FILE_SIZE_MB for each request.
"""
self._http_timeout = http_timeout
self._cache_size = cache_size
self._configured_max_bytes = max_bytes
self._image_cache: OrderedDict[str, Image.Image] = OrderedDict()
self._inflight: dict[str, asyncio.Task[Image.Image]] = {}
self._enable_frontend_decoding = enable_frontend_decoding
Expand All @@ -100,6 +105,11 @@ def __init__(
self._nixl_connector.initialize
) # Synchronously wait for async init

def _max_bytes(self) -> int:
if self._configured_max_bytes is not None:
return self._configured_max_bytes
return max_media_bytes()

@staticmethod
def _open_image_sync(image_data: BytesIO) -> Image.Image:
"""Open, validate, and decode an image from raw bytes. Runs in a thread."""
Expand Down Expand Up @@ -135,7 +145,10 @@ async def _fetch_and_process(self, image_url: str) -> Image.Image:
try:
with _nvtx.annotate("mm:img:http_fetch", color="lime"):
content = await fetch_bytes(
image_url, self._http_timeout, policy=self._url_policy
image_url,
self._http_timeout,
policy=self._url_policy,
max_bytes=self._max_bytes(),
)
if not content:
raise ValueError("Empty response content from image URL")
Expand Down
6 changes: 5 additions & 1 deletion components/src/dynamo/common/multimodal/video_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import numpy as np

from dynamo.common.http import HttpStatusError, fetch_bytes
from dynamo.common.http.media_reference import max_media_bytes
from dynamo.common.http.url_validator import (
UrlValidationError,
UrlValidationPolicy,
Expand Down Expand Up @@ -171,7 +172,10 @@ async def _load_video_with_vllm(
# data: and file:// never touch the network, so vLLM can handle them.
if urlparse(normalized_url).scheme in ("http", "https"):
content = await fetch_bytes(
normalized_url, self._http_timeout, policy=self._url_policy
normalized_url,
self._http_timeout,
policy=self._url_policy,
max_bytes=max_media_bytes(),
Comment thread
host452b marked this conversation as resolved.
Resolved
Comment thread
host452b marked this conversation as resolved.
Resolved
)
return await self._decode_video_bytes(content, media_io)

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import pytest

from dynamo.common.http import HttpStatusError
from dynamo.common.http.media_reference import DYN_MM_MAX_FILE_SIZE_MB
from dynamo.common.http.url_validator import UrlValidationError, UrlValidationPolicy
from dynamo.common.multimodal import audio_loader as audio_loader_module
from dynamo.common.multimodal.audio_loader import AudioLoader
Expand Down Expand Up @@ -94,6 +95,25 @@ async def test_load_audio_rejects_empty_waveform():
await loader.load_audio("https://example.com/empty.wav")


@pytest.mark.asyncio
async def test_http_fetch_honors_configured_media_limit(monkeypatch):
monkeypatch.setenv(DYN_MM_MAX_FILE_SIZE_MB, "1")
loader = AudioLoader(url_policy=_permissive_http_policy())
waveform = np.zeros(8, dtype=np.float32)

class _MediaIO:
def load_bytes(self, content):
return waveform, 16000.0

mock_fetch = AsyncMock(return_value=b"audio")
monkeypatch.setattr(audio_loader_module, "fetch_bytes", mock_fetch)
monkeypatch.setattr(loader, "_create_vllm_audio_io", lambda: _MediaIO())

await loader._load_audio_with_vllm("https://example.com/limited.wav")

assert mock_fetch.await_args.kwargs["max_bytes"] == 1024 * 1024


@pytest.mark.asyncio
async def test_load_audio_batch_uses_url_loader():
loader = AudioLoader()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
from PIL import Image

from dynamo.common.http import HttpConnectionError, HttpStatusError, HttpTimeoutError
from dynamo.common.http.media_reference import DYN_MM_MAX_FILE_SIZE_MB
from dynamo.common.http.url_validator import UrlValidationError, UrlValidationPolicy
from dynamo.common.multimodal.image_loader import URL_VARIANT_KEY, ImageLoader

Expand Down Expand Up @@ -73,7 +74,7 @@ def _mock_fetch_bytes(
side_effect: If set, the mock raises this exception instead of returning.
"""

async def _fetch(url, timeout, *, policy=None):
async def _fetch(url, timeout, *, policy=None, max_bytes=None):
if delay > 0:
await asyncio.sleep(delay)
if side_effect is not None:
Expand Down Expand Up @@ -207,6 +208,34 @@ async def test_http_timeout_raises_408(loader: ImageLoader) -> None:
assert "https://example.com/img.png" in exc_info.value.url


async def test_http_fetch_honors_configured_media_limit(
loader: ImageLoader, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv(DYN_MM_MAX_FILE_SIZE_MB, "1")
mock_fetch = _mock_fetch_bytes()

with patch(_FETCH_BYTES_PATH, mock_fetch):
await loader.load_image("https://example.com/limited.png")

assert mock_fetch.await_args.kwargs["max_bytes"] == 1024 * 1024


async def test_explicit_media_limit_overrides_environment(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv(DYN_MM_MAX_FILE_SIZE_MB, "1")
loader = ImageLoader(
max_bytes=2 * 1024 * 1024,
url_policy=_permissive_policy(),
)
mock_fetch = _mock_fetch_bytes()

with patch(_FETCH_BYTES_PATH, mock_fetch):
await loader.load_image("https://example.com/explicit-limit.png")

assert mock_fetch.await_args.kwargs["max_bytes"] == 2 * 1024 * 1024


async def test_http_connection_error_raises_400(loader: ImageLoader) -> None:
"""An unreachable user-supplied URL is a client error, so the
frontend must surface 400 instead of an opaque 500."""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import pytest

from dynamo.common.http import HttpStatusError
from dynamo.common.http.media_reference import DYN_MM_MAX_FILE_SIZE_MB
from dynamo.common.http.url_validator import UrlValidationError, UrlValidationPolicy
from dynamo.common.multimodal import codec_errors
from dynamo.common.multimodal import video_loader as video_loader_module
Expand Down Expand Up @@ -75,6 +76,25 @@ async def test_load_video_uses_vllm_media_connector():
assert loaded_metadata == metadata


@pytest.mark.asyncio
async def test_http_fetch_honors_configured_media_limit(monkeypatch):
monkeypatch.setenv(DYN_MM_MAX_FILE_SIZE_MB, "1")
loader = VideoLoader(
url_policy=UrlValidationPolicy(allow_http=True, allow_private_ips=True)
)
frames = np.zeros((1, 2, 2, 3), dtype=np.uint8)
metadata = {"fps": 1.0, "frames_indices": [0], "total_num_frames": 1}
mock_fetch = AsyncMock(return_value=b"video")
mock_decode = AsyncMock(return_value=(frames, metadata))
monkeypatch.setattr(video_loader_module, "fetch_bytes", mock_fetch)
monkeypatch.setattr(loader, "_create_vllm_video_io", lambda kwargs=None: object())
monkeypatch.setattr(loader, "_decode_video_bytes", mock_decode)

await loader._load_video_with_vllm("https://example.com/limited.mp4")

assert mock_fetch.await_args.kwargs["max_bytes"] == 1024 * 1024


@pytest.mark.asyncio
async def test_load_video_batch_uses_url_loader():
loader = VideoLoader()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@

from dynamo._core import Client, Context
from dynamo.common.http import fetch_bytes
from dynamo.common.http.media_reference import max_media_bytes
from dynamo.common.http.url_validator import UrlValidationPolicy, validate_media_url
from dynamo.common.memory.multimodal_embedding_cache_manager import (
CachedEmbedding,
Expand Down Expand Up @@ -614,7 +615,12 @@ async def _maybe_nvdec_decoder(self, url: str) -> Optional[Any]:
normalized = await validate_media_url(url, self._url_policy)
scheme = urlparse(normalized).scheme
if scheme in ("http", "https"):
content = await fetch_bytes(normalized, 30.0, policy=self._url_policy)
content = await fetch_bytes(
normalized,
30.0,
policy=self._url_policy,
max_bytes=max_media_bytes(),
Comment thread
host452b marked this conversation as resolved.
Resolved
)
elif is_local_media_url(normalized):
content = await read_local_media_bytes(normalized, self._url_policy)
else:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import json
from contextlib import asynccontextmanager
from types import SimpleNamespace
from unittest.mock import AsyncMock

import numpy as np
import pytest
Expand Down Expand Up @@ -659,7 +660,7 @@ async def test_vp9_without_software_decoder_is_actionable(monkeypatch):
async def fake_validate(url, policy):
return url

async def fake_fetch(url, timeout, policy=None):
async def fake_fetch(url, timeout, policy=None, max_bytes=None):
return b"vp9-bytes"

monkeypatch.setattr(ewh, "validate_media_url", fake_validate)
Expand Down Expand Up @@ -687,7 +688,7 @@ async def test_vp9_with_software_decoder_passes_bytes_through(monkeypatch):
async def fake_validate(url, policy):
return url

async def fake_fetch(url, timeout, policy=None):
async def fake_fetch(url, timeout, policy=None, max_bytes=None):
return b"vp9-bytes"

monkeypatch.setattr(ewh, "validate_media_url", fake_validate)
Expand All @@ -702,6 +703,24 @@ async def fake_fetch(url, timeout, policy=None):
assert result == b"vp9-bytes"


@pytest.mark.asyncio
async def test_remote_video_fetch_honors_configured_media_limit(monkeypatch):
import dynamo.sglang.request_handlers.multimodal.encode_worker_handler as ewh
from dynamo.common.http.media_reference import DYN_MM_MAX_FILE_SIZE_MB

monkeypatch.setenv(DYN_MM_MAX_FILE_SIZE_MB, "1")
fetch = AsyncMock(return_value=b"vp9-bytes")
monkeypatch.setattr(ewh, "fetch_bytes", fetch)
monkeypatch.setattr(ewh, "probe_video_codec", lambda b: "vp9")
monkeypatch.setattr(ewh, "should_use_nvdec", lambda c: False)
monkeypatch.setattr(ewh.importlib, "import_module", _selective_import({"decord"}))

handler = _bare_handler()
await handler._maybe_nvdec_decoder("https://example.com/clip.webm")

assert fetch.await_args.kwargs["max_bytes"] == 1024 * 1024


@pytest.mark.asyncio
async def test_multimodal_decode_rejects_parallel_sampling_before_prefill_handoff():
"""Reject multimodal n greater than one before contacting prefill."""
Expand Down
8 changes: 6 additions & 2 deletions components/src/dynamo/trtllm/multimodal_processor.py
Original file line number Diff line number Diff line change
Expand Up @@ -163,7 +163,8 @@ def __init__(
self.tokenizer = tokenizer_factory(model_dir)

self.image_loader = ImageLoader(
enable_frontend_decoding=enable_frontend_decoding
enable_frontend_decoding=enable_frontend_decoding,
max_bytes=self.max_file_size_bytes,
)

# Reuse the shared default so this preprocessor and the vLLM/SGLang
Expand Down Expand Up @@ -588,7 +589,10 @@ async def process_openai_request(
normalized_url = await validate_media_url(url, self._url_policy)
if urlparse(normalized_url).scheme in ("http", "https"):
content = await fetch_bytes(
normalized_url, 30.0, policy=self._url_policy
normalized_url,
30.0,
policy=self._url_policy,
max_bytes=self.max_file_size_bytes,
)
# Dual decode path: H.264/H.265 via NVDEC (hardware); other
# codecs via the vendor cv2 loader. NVDEC failure falls back.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,23 @@
# sequential GPU stage so TensorRT-LLM initialization is shared.


def test_image_loader_uses_trtllm_configured_limit(monkeypatch) -> None:
image_loader = MagicMock()
monkeypatch.setattr(mmp, "ImageLoader", image_loader)

MultimodalRequestProcessor(
model_type="multimodal",
model_dir="unused",
max_file_size_mb=200,
tokenizer=MagicMock(),
)

image_loader.assert_called_once_with(
enable_frontend_decoding=False,
max_bytes=200 * 1024 * 1024,
)


@pytest.mark.asyncio
@pytest.mark.parametrize(
"error",
Expand Down Expand Up @@ -101,7 +118,12 @@ async def test_internal_video_uses_dynamo_fetcher_when_allowed(monkeypatch) -> N
ep_disaggregated_params=None,
)

fetch.assert_awaited_once_with(url, 30.0, policy=processor._url_policy)
fetch.assert_awaited_once_with(
url,
30.0,
policy=processor._url_policy,
max_bytes=processor.max_file_size_bytes,
)
assert load_video.await_args.args[0] != url


Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -206,6 +206,7 @@ NVDEC as above needs no additional change. Encode performance does not depend on
| `DYN_DISABLE_NVDEC` | unset | Set to `1` to skip hardware decode. In a shipped image that leaves video input with no decoder at all, so it is a debugging switch rather than a fallback. Read as a boolean: `1`/`true`/`yes` disable, anything else does not. |
| `DYN_NVDEC_GPU_ID` | `0` | GPU ordinal used for decode. |
| `DYN_MM_VIDEO_NUM_FRAMES` | `32` | Frames sampled uniformly from each clip. |
| `DYN_MM_MAX_FILE_SIZE_MB` | `64` | Maximum size in MiB for each remote image, audio, or video download. TensorRT-LLM uses its backend-specific `--max-file-size-mb` option (`DYN_TRTLLM_MAX_FILE_SIZE_MB`) instead. |
Comment thread
rmccorm4 marked this conversation as resolved.
Resolved

## Sources

Expand Down
Loading