[Fix] Drop deprecated multimodal processor residency state (#33308)
Co-authored-by: Mick <mickjagger19@icloud.com>
This commit is contained in:
@@ -337,6 +337,10 @@ class MultimodalDataItem:
|
||||
def set(self, key: str, value: Any):
|
||||
self.__setitem__(key, value)
|
||||
|
||||
def set_hash(self, hash_value: int) -> None:
|
||||
self.hash = hash_value
|
||||
self.pad_value = _compute_pad_value(hash_value)
|
||||
|
||||
@staticmethod
|
||||
def is_empty_list(l):
|
||||
if l is None:
|
||||
|
||||
@@ -157,6 +157,23 @@ _REQUEST_STATE_WAIT_TIMEOUT = envs.SGLANG_REQUEST_STATE_WAIT_TIMEOUT.get()
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _reject_missing_dispatched_encoder_embedding(server_args, request_obj, mm_inputs):
|
||||
"""Do not silently turn a failed EPD request into local vision work."""
|
||||
if (
|
||||
mm_inputs is None
|
||||
and server_args.language_only
|
||||
and server_args.encoder_transfer_backend == "zmq_to_tokenizer"
|
||||
and request_obj.need_wait_for_mm_inputs
|
||||
):
|
||||
raise fastapi.HTTPException(
|
||||
status_code=HTTPStatus.SERVICE_UNAVAILABLE,
|
||||
detail=(
|
||||
"The encoder did not return multimodal embeddings. "
|
||||
"The request was not run locally in language-only mode."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def _ragged_verify_cap_accept() -> bool:
|
||||
# The mode env is fixed at server launch; cache to keep it off the
|
||||
@@ -983,6 +1000,11 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
|
||||
self._validate_mm_limits(obj)
|
||||
|
||||
mm_inputs = None
|
||||
mm_processor_input = (
|
||||
input_ids
|
||||
if self.mm_processor.prefer_tokenized_input and input_ids is not None
|
||||
else (input_text or input_ids)
|
||||
)
|
||||
|
||||
if (
|
||||
not self.server_args.language_only
|
||||
@@ -992,9 +1014,12 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
|
||||
mm_inputs = await self.mm_receiver.recv_mm_data(
|
||||
request_obj=obj,
|
||||
mm_processor=self.mm_processor,
|
||||
prompt=(input_text or input_ids),
|
||||
prompt=mm_processor_input,
|
||||
need_wait_for_mm_inputs=obj.need_wait_for_mm_inputs,
|
||||
)
|
||||
_reject_missing_dispatched_encoder_embedding(
|
||||
self.server_args, obj, mm_inputs
|
||||
)
|
||||
if mm_inputs is None:
|
||||
if self.server_args.language_only:
|
||||
logger.warning(
|
||||
@@ -1004,7 +1029,7 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
|
||||
mm_inputs = await self.mm_processor.process_mm_data_async(
|
||||
image_data=obj.image_data,
|
||||
audio_data=obj.audio_data,
|
||||
input_text=(input_text or input_ids),
|
||||
input_text=mm_processor_input,
|
||||
request_obj=obj,
|
||||
max_req_input_len=self.max_req_input_len,
|
||||
)
|
||||
@@ -1019,7 +1044,7 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
|
||||
mm_inputs = await self.mm_processor.process_mm_data_async(
|
||||
image_data=obj.image_data,
|
||||
audio_data=obj.audio_data,
|
||||
input_text=(input_text or input_ids),
|
||||
input_text=mm_processor_input,
|
||||
request_obj=obj,
|
||||
max_req_input_len=self.max_req_input_len,
|
||||
)
|
||||
@@ -1054,7 +1079,7 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
|
||||
if not isinstance(item, MultimodalDataItem):
|
||||
continue
|
||||
try:
|
||||
item.hash = int(hex_hash, 16)
|
||||
item.set_hash(int(hex_hash, 16))
|
||||
except (TypeError, ValueError):
|
||||
logger.warning(
|
||||
"Ignoring malformed mm_hashes entry %r; "
|
||||
|
||||
@@ -44,8 +44,6 @@ _is_cpu = is_cpu()
|
||||
_is_npu = is_npu()
|
||||
_is_xpu = is_xpu()
|
||||
|
||||
_IPC_POOL_HANDLE_CACHE = envs.SGLANG_USE_IPC_POOL_HANDLE_CACHE.get()
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class BaseMultiModalProcessorOutput:
|
||||
@@ -182,6 +180,8 @@ class MultimodalSpecialTokens:
|
||||
class BaseMultimodalProcessor(ABC):
|
||||
models = []
|
||||
gpu_image_decode = True # Enable GPU decoding by default
|
||||
prefer_tokenized_input = False
|
||||
precompute_hash_before_cpu_transfer = False
|
||||
auto_mm_processor_worker_num = 1
|
||||
auto_mm_io_worker_num = 4
|
||||
supports_mm_processor_concurrency = False
|
||||
@@ -193,7 +193,6 @@ class BaseMultimodalProcessor(ABC):
|
||||
self._processor = _processor
|
||||
self.server_args = server_args
|
||||
self.transport_mode = transport_mode
|
||||
self.keep_mm_feature_on_device = server_args.keep_mm_feature_on_device
|
||||
configured_mm_feature_transport = getattr(
|
||||
server_args, "mm_feature_transport", "cpu"
|
||||
)
|
||||
@@ -203,6 +202,9 @@ class BaseMultimodalProcessor(ABC):
|
||||
else "cpu"
|
||||
)
|
||||
self.use_cuda_ipc = self.mm_feature_transport == "cuda_ipc"
|
||||
self.use_ipc_pool_handle_cache = (
|
||||
self.use_cuda_ipc and envs.SGLANG_USE_IPC_POOL_HANDLE_CACHE.get()
|
||||
)
|
||||
self.disable_fast_image_processor = server_args.disable_fast_image_processor
|
||||
self.skip_tokenizer_init = server_args.skip_tokenizer_init
|
||||
|
||||
@@ -573,16 +575,15 @@ class BaseMultimodalProcessor(ABC):
|
||||
return_tensors="pt",
|
||||
**kwargs,
|
||||
)
|
||||
if not self.keep_mm_feature_on_device:
|
||||
# Deferred: the hash is computed on the GPU tensor first, and
|
||||
# _precompute_hashes_before_cpu_transfer moves it down afterwards.
|
||||
if not self.use_cuda_ipc and not self.precompute_hash_before_cpu_transfer:
|
||||
# move feature tensors to cpu
|
||||
for feature_name in self.FEATURE_NAMES:
|
||||
if self.use_cuda_ipc:
|
||||
pass
|
||||
else:
|
||||
if feature_name in result and isinstance(
|
||||
result[feature_name], torch.Tensor
|
||||
):
|
||||
result[feature_name] = result[feature_name].to("cpu")
|
||||
if feature_name in result and isinstance(
|
||||
result[feature_name], torch.Tensor
|
||||
):
|
||||
result[feature_name] = result[feature_name].to("cpu")
|
||||
|
||||
return result
|
||||
|
||||
@@ -1019,13 +1020,17 @@ class BaseMultimodalProcessor(ABC):
|
||||
for modality, idx, future in futures:
|
||||
try:
|
||||
result = await asyncio.wrap_future(future)
|
||||
except ValueError:
|
||||
logger.exception(
|
||||
"[load_mm_data(simple)] error loading %s data at index=%d",
|
||||
except ValueError as e:
|
||||
logger.info(
|
||||
"[load_mm_data(simple)] invalid %s data at index=%d: %s",
|
||||
modality.name,
|
||||
idx,
|
||||
e,
|
||||
)
|
||||
raise
|
||||
raise ValueError(
|
||||
f"An exception occurred while loading {modality.name} data "
|
||||
f"at index {idx}: {e}"
|
||||
) from e
|
||||
except Exception as e:
|
||||
logger.exception(
|
||||
"[load_mm_data(simple)] error loading %s data at index=%d",
|
||||
@@ -1167,6 +1172,10 @@ class BaseMultimodalProcessor(ABC):
|
||||
raise RuntimeError(
|
||||
f"An exception occurred while loading multimodal data: {e}"
|
||||
)
|
||||
except ValueError as e:
|
||||
raise ValueError(
|
||||
f"An exception occurred while loading multimodal data: {e}"
|
||||
) from e
|
||||
except Exception as e:
|
||||
raise RuntimeError(
|
||||
f"An exception occurred while loading multimodal data: {e}"
|
||||
@@ -1349,16 +1358,38 @@ class BaseMultimodalProcessor(ABC):
|
||||
sync_buffer_meta=sync_flag,
|
||||
pool_ipc_handle=(
|
||||
self.cudaipc_mmfeature_pool._pool_ipc_handle
|
||||
if _IPC_POOL_HANDLE_CACHE
|
||||
if self.use_ipc_pool_handle_cache
|
||||
else None
|
||||
),
|
||||
pool_byte_offset=byte_offset,
|
||||
pool_device_index=self.cudaipc_mmfeature_pool._pool_device_index,
|
||||
)
|
||||
if self.keep_mm_feature_on_device:
|
||||
return tensor
|
||||
return tensor.cpu()
|
||||
|
||||
@staticmethod
|
||||
def _move_feature_to_cpu(value):
|
||||
if isinstance(value, torch.Tensor):
|
||||
return value.cpu()
|
||||
if isinstance(value, list):
|
||||
return [BaseMultimodalProcessor._move_feature_to_cpu(v) for v in value]
|
||||
if isinstance(value, tuple):
|
||||
return tuple(BaseMultimodalProcessor._move_feature_to_cpu(v) for v in value)
|
||||
return value
|
||||
|
||||
def _precompute_hashes_before_cpu_transfer(
|
||||
self, mm_items: List[MultimodalDataItem]
|
||||
) -> None:
|
||||
if not self.precompute_hash_before_cpu_transfer:
|
||||
return
|
||||
|
||||
for item in mm_items:
|
||||
item.set_pad_value()
|
||||
if not self.use_cuda_ipc:
|
||||
item.feature = self._move_feature_to_cpu(item.feature)
|
||||
item.precomputed_embeddings = self._move_feature_to_cpu(
|
||||
item.precomputed_embeddings
|
||||
)
|
||||
|
||||
def resolve_image_token_counts(self, images: List) -> List[int]:
|
||||
"""Per-image expanded token counts, computed without re-tokenizing.
|
||||
|
||||
@@ -1577,14 +1608,10 @@ class BaseMultimodalProcessor(ABC):
|
||||
):
|
||||
item.set_pad_value()
|
||||
|
||||
"""
|
||||
solution for cuda-ipc memory-leak:
|
||||
1. memory-pool: each time get a slice from memory-pool and use it as transport-data (with async lock guard)
|
||||
2. if can not get a slice , transport normal tensor
|
||||
3. copy tensor in scheduler and release it (use position mark)
|
||||
4. copy
|
||||
"""
|
||||
self._precompute_hashes_before_cpu_transfer(all_collected_items)
|
||||
|
||||
# Wrap GPU features in the bounded IPC pool; pool misses fall back to a
|
||||
# plain CPU tensor. The scheduler copies out and releases each slice.
|
||||
if self.use_cuda_ipc:
|
||||
# post-process, prepare for cuda-ipc transfer
|
||||
for item in all_collected_items:
|
||||
|
||||
@@ -346,16 +346,13 @@ class Ernie4_5_VLImageProcessor(SGLangBaseProcessor):
|
||||
if result["pixel_values_videos"].numel() == 0:
|
||||
del result["pixel_values_videos"]
|
||||
|
||||
if not self.keep_mm_feature_on_device:
|
||||
if not self.use_cuda_ipc:
|
||||
# move feature tensors to cpu
|
||||
for feature_name in self.FEATURE_NAMES:
|
||||
if self.use_cuda_ipc:
|
||||
pass
|
||||
else:
|
||||
if feature_name in result and isinstance(
|
||||
result[feature_name], torch.Tensor
|
||||
):
|
||||
result[feature_name] = result[feature_name].to("cpu")
|
||||
if feature_name in result and isinstance(
|
||||
result[feature_name], torch.Tensor
|
||||
):
|
||||
result[feature_name] = result[feature_name].to("cpu")
|
||||
|
||||
return result
|
||||
|
||||
|
||||
@@ -416,6 +416,8 @@ class KimiGPUProcessorWrapper:
|
||||
class KimiK2_5VLImageProcessor(KimiGridMMDataMixin, SGLangBaseProcessor):
|
||||
models = [KimiK25ForConditionalGeneration]
|
||||
gpu_image_decode = True # nvJPEG for JPEG, PIL fallback for others
|
||||
prefer_tokenized_input = True
|
||||
precompute_hash_before_cpu_transfer = True
|
||||
|
||||
def __init__(self, hf_config, server_args, _processor, *args, **kwargs):
|
||||
super().__init__(hf_config, server_args, _processor, *args, **kwargs)
|
||||
|
||||
@@ -70,7 +70,7 @@ class MiDashengLMMultimodalProcessor(BaseMultimodalProcessor):
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
if not self.keep_mm_feature_on_device and not self.use_cuda_ipc:
|
||||
if not self.use_cuda_ipc:
|
||||
for feature_name in ["input_values"]:
|
||||
if feature_name in result:
|
||||
result[feature_name] = result[feature_name].cpu()
|
||||
|
||||
Reference in New Issue
Block a user