class_docstr: A wrapper class that wraps the model load function and
  provides the LRU caching functionality.
  
  The model multiplexer is a wrapper class that wraps the model load function
  and provides the LRU caching functionality, and the model load function should
  be a coroutine function that takes the model ID as the first argument and
  returns the user-constructed model object.
  The model multiplexer will also ensure that the number of models on the current
  replica does not exceed the specified limit.
  The model will be unloaded in the LRU order, the model multiplexer will call the
  model's __del__ attribute if it exists to clean up the model resources eagerly.
class_docstr_tok: ["A","wrapper","class","that","wraps","the","model","load","function","and","provides","the","LRU","caching","functionality",".","The","model","multiplexer","is","a","wrapper","class","that","wraps","the","model","load","function","and","provides","the","LRU","caching","functionality",",","and","the","model","load","function","should","be","a","coroutine","function","that","takes","the","model","ID","as","the","first","argument","and","returns","the","user-constructed","model","object",".","The","model","multiplexer","will","also","ensure","that","the","number","of","models","on","the","current","replica","does","not","exceed","the","specified","limit",".","The","model","will","be","unloaded","in","the","LRU","order",",","the","model","multiplexer","will","call","the","model","'s","__del__","attribute","if","it","exists","to","clean","up","the","model","resources","eagerly","."]
docstr: Initialize the model multiplexer.
  Args:
      model_load_func: the model load async function.
      self_arg: self argument when model_load_func is class method.
      max_num_models_per_replica: the maximum number of models to be loaded on the
          current replica. If it is -1, there is no limit for the number of models
          per replica.
docstr_tok: ["Initialize","the","model","multiplexer",".","Args",":","model_load_func",":","the","model","load","async","function",".","self_arg",":","self","argument","when","model_load_func","is","class","method",".","max_num_models_per_replica",":","the","maximum","number","of","models","to","be","loaded","on","the","current","replica",".","If","it","is","-1",",","there","is","no","limit","for","the","number","of","models","per","replica","."]
code: def __init__(
      self,
      model_load_func: Callable[[str], Any],
      self_arg: Any,
      max_num_models_per_replica: int,
  ):
      """Initialize the model multiplexer.
      Args:
          model_load_func: the model load async function.
          self_arg: self argument when model_load_func is class method.
          max_num_models_per_replica: the maximum number of models to be loaded on the
              current replica. If it is -1, there is no limit for the number of models
              per replica.
      """
  
      ServeUsageTag.MULTIPLEXED_API_USED.record("1")
  
      self.models = OrderedDict()
      self._func: Callable = model_load_func
      self.self_arg: Any = self_arg
      self.max_num_models_per_replica: int = max_num_models_per_replica
  
      self.model_load_latency_ms = metrics.Histogram(
          "serve_multiplexed_model_load_latency_ms",
          description="The time it takes to load a model.",
          boundaries=DEFAULT_LATENCY_BUCKET_MS,
      )
      self.model_unload_latency_ms = metrics.Histogram(
          "serve_multiplexed_model_unload_latency_ms",
          description="The time it takes to unload a model.",
          boundaries=DEFAULT_LATENCY_BUCKET_MS,
      )
      self.num_models_gauge = metrics.Gauge(
          "serve_num_multiplexed_models",
          description="The number of models loaded on the current replica.",
      )
  
      self.registered_model_gauge = metrics.Gauge(
          "serve_registered_multiplexed_model_id",
          description="The model id registered on the current replica.",
          tag_keys=("model_id",),
      )
      self.get_model_requests_counter = metrics.Counter(
          "serve_multiplexed_get_model_requests_counter",
          description="The counter for get model requests on the current replica.",
      )
      self.models_unload_counter = metrics.Counter(
          "serve_multiplexed_models_unload_counter",
          description="The counter for unloaded models on the current replica.",
      )
      self.models_load_counter = metrics.Counter(
          "serve_multiplexed_models_load_counter",
          description="The counter for loaded models on the current replica.",
      )
  
      context = _get_internal_replica_context()
      if context is None:
          raise RuntimeError(
              "Fail to retrieve serve replica context, the model multiplexer ",
              "can only be used within `Deployment`.",
          )
      self._app_name: str = context.app_name
      self._deployment_name: str = context.deployment
      self._replica_tag: str = context.replica_tag
  
      # Whether to push the multiplexed replica info to the controller.
      self._push_multiplexed_replica_info: bool = False
  
      # Model cache lock to ensure that only one model is loading/unloading at a time.
      self._model_cache_lock = asyncio.Lock()
      # The set of model IDs that are being loaded. This is used to early push
      # model ids info to the controller. The tasks will be added when there is cache
      # miss, and will be removed when the model is loaded successfully or
      # failed to load.
      self._model_load_tasks: Set[str] = set()
  
      self.metrics_pusher = MetricsPusher()
      self.metrics_pusher.register_task(
          self._push_model_ids_info,
          PUSH_MULTIPLEXED_MODEL_IDS_INTERVAL_S,
      )
      self.metrics_pusher.start()
  
file_path: ray/python/ray/serve/multiplex.py
imports: import asyncio
  import inspect
  import logging
  import time
  from collections import OrderedDict
  from typing import Any, Callable, List, Set
  from ray.serve import metrics
  from ray.serve._private.common import DeploymentID, MultiplexedReplicaInfo
  from ray.serve._private.constants import DEFAULT_LATENCY_BUCKET_MS, PUSH_MULTIPLEXED_MODEL_IDS_INTERVAL_S, SERVE_LOGGER_NAME
  from ray.serve._private.usage import ServeUsageTag
  from ray.serve._private.utils import MetricsPusher
  from ray.serve.context import _get_global_client, _get_internal_replica_context
label_desc: Use image node_id 1 to create a new _ModelMultiplexWrapper object  with example: obj = _ModelMultiplexWrapper(model_load_func, _arg, max_num_models_per_replica)