[DO NOT MERGE] perf(multimodal): reduce video decode, Qwen preprocess, and TokenSpeed handoff overhead

Cargo.toml

@@ -30,10 +30,12 @@ axum = { version = "0.8.9" }
 blake3 = "1.8"
 lz4_flex = "0.13"
 bytemuck = { version = "1.25" }
+memmap2 = "0.9"
 chrono = { version = "0.4" }
 dashmap = "6.2.1"
 http = "1.4.2"
 lru = "0.18.0"
+libc = "0.2"
 num-traits = "0.2"
 parking_lot = "0.12.5"
 rand = "0.10.1"

crates/multimodal/Cargo.toml

@@ -28,14 +28,23 @@ image = { version = "0.25.10", default-features = false, features = ["png", "jpe
 libloading = "0.8"
 ndarray = "0.17"
 once_cell = "1.21.4"
+rayon = "1.12"
 opencv = { version = "0.99.0", default-features = false, features = ["clang-runtime", "imgproc", "videoio"], optional = true }
 reqwest = { workspace = true, features = ["stream"] }
 serde = { workspace = true, features = ["derive"] }
 serde_bytes = "0.11"
 serde_json.workspace = true
 tempfile = "3.27"
 thiserror.workspace = true
-tokio = { workspace = true, features = ["sync", "fs", "rt-multi-thread", "process", "time"] }
+tokio = { workspace = true, features = [
+    "sync",
+    "fs",
+    "io-util",
+    "macros",
+    "rt-multi-thread",
+    "process",
+    "time",
+] }
 tracing.workspace = true
 url = "2.5.8"
 
@@ -49,6 +58,10 @@ criterion = { version = "0.8", features = ["html_reports"] }
 npyz = { version = "0.9", features = ["npz"] }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 
+[build-dependencies]
+cc = "1"
+pkg-config = "0.3"
+
 [[bench]]
 name = "image_preprocess"
 harness = false

crates/multimodal/build.rs

@@ -0,0 +1,31 @@
+use std::env;
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    println!("cargo:rerun-if-changed=src/opencv_buffer_capture.cpp");
+    println!("cargo:rerun-if-env-changed=OPENCV_INCLUDE_PATHS");
+    if env::var_os("CARGO_FEATURE_OPENCV_VIDEO").is_none() {
+        return Ok(());
+    }
+
+    let mut build = cc::Build::new();
+    build
+        .cpp(true)
+        .file("src/opencv_buffer_capture.cpp")
+        .flag_if_supported("-std=c++17");
+
+    if let Some(paths) = env::var_os("OPENCV_INCLUDE_PATHS") {
+        for path in env::split_paths(&paths) {
+            build.include(path);
+        }
+    } else {
+        let opencv = pkg_config::Config::new()
+            .cargo_metadata(false)
+            .probe("opencv4")?;
+        for path in opencv.include_paths {
+            build.include(path);
+        }
+    }
+
+    build.compile("smg_opencv_buffer_capture");
+    Ok(())
+}

crates/multimodal/src/error.rs

@@ -24,6 +24,8 @@ pub enum MediaConnectorError {
     DataUrl(String),
     #[error("media decode task failed: {0}")]
     Blocking(#[from] tokio::task::JoinError),
+    #[error("failed to initialize multimodal runtime: {0}")]
+    Runtime(String),
     #[error("image decode error: {0}")]
     Image(#[from] image::ImageError),
     #[error("video decode error: {0}")]

crates/multimodal/src/lib.rs

@@ -3,7 +3,10 @@ pub mod hasher;
 pub mod hub;
 pub mod jpeg_turbo;
 pub mod media;
+#[cfg(feature = "opencv-video")]
+mod opencv_buffer;
 pub mod registry;
+pub mod runtime;
 pub mod tracker;
 pub mod types;
 pub mod vision;
@@ -13,6 +16,7 @@ pub use media::{
     ImageFetchConfig, MediaConnector, MediaConnectorConfig, MediaSource, VideoFetchConfig,
 };
 pub use registry::{ModelMetadata, ModelProcessorSpec, ModelRegistry};
+pub use runtime::MultimodalRuntime;
 pub use tracker::{AsyncMultiModalTracker, TrackerOutput};
 pub use types::{
     FieldLayout, ImageDetail, ImageFrame, ImageSize, ImageSource, MediaContentPart, Modality,
@@ -21,6 +25,8 @@ pub use types::{
 };
 // Re-export vision processing components
 pub use vision::{
-    LlavaNextProcessor, LlavaProcessor, ModelSpecificValue, PreProcessorConfig,
-    PreprocessedEncoderInputs, TransformError, VisionPreProcessor, VisionProcessorRegistry,
+    DeferredNormalizedEncoderInput, EncoderInput, LlavaNextProcessor, LlavaProcessor,
+    ModalityPreProcessor, ModalityProcessorRegistry, ModelSpecificValue, OutputPreference,
+    PreProcessorConfig, PreprocessRequest, PreprocessedEncoderInputs, TransformError, VideoInput,
+    VisionInput, VisionPreProcessor, VisionPreprocessRequest, VisionProcessorRegistry,
 };