class_docstr: Actor definition for replicas of Ray Serve deployments.
  
  This class defines the interface that the controller and deployment handles
  (i.e., from proxies and other replicas) use to interact with a replica.
  
  All interaction with the user-provided callable is done via the
  `UserCallableWrapper` class.
class_docstr_tok: ["Actor","definition","for","replicas","of","Ray","Serve","deployments",".","This","class","defines","the","interface","that","the","controller","and","deployment","handles","(","i.e.",",","from","proxies","and","other","replicas",")","use","to","interact","with","a","replica",".","All","interaction","with","the","user-provided","callable","is","done","via","the","`","UserCallableWrapper","`","class","."]
docstr: Context manager that wraps user method calls.
  
  1) Sets the request context var with appropriate metadata.
  2) Records the access log message (if not disabled).
  3) Records per-request metrics via the metrics manager.
docstr_tok: ["Context","manager","that","wraps","user","method","calls",".","1",")","Sets","the","request","context","var","with","appropriate","metadata",".","2",")","Records","the","access","log","message","(","if","not","disabled",")",".","3",")","Records","per-request","metrics","via","the","metrics","manager","."]
code: def _wrap_user_method_call(self, request_metadata: RequestMetadata):
      """Context manager that wraps user method calls.
  
      1) Sets the request context var with appropriate metadata.
      2) Records the access log message (if not disabled).
      3) Records per-request metrics via the metrics manager.
      """
      ray.serve.context._serve_request_context.set(
          ray.serve.context._RequestContext(
              request_metadata.route,
              request_metadata.request_id,
              self._deployment_id.app,
              request_metadata.multiplexed_model_id,
              request_metadata.grpc_context,
          )
      )
  
      start_time = time.time()
      user_exception = None
      try:
          yield
      except Exception as e:
          user_exception = e
          logger.error(f"Request failed:\n{e}")
          if ray.util.pdb._is_ray_debugger_enabled():
              ray.util.pdb._post_mortem()
  
      latency_ms = (time.time() - start_time) * 1000
      if user_exception is None:
          status_str = "OK"
      elif isinstance(user_exception, asyncio.CancelledError):
          status_str = "CANCELLED"
      else:
          status_str = "ERROR"
  
      logger.info(
          access_log_msg(
              method=request_metadata.call_method,
              status=status_str,
              latency_ms=latency_ms,
          ),
          extra={"serve_access_log": True},
      )
      self._metrics_manager.record_request_metrics(
          route=request_metadata.route,
          status_str=status_str,
          latency_ms=latency_ms,
          was_error=user_exception is not None,
      )
  
      if user_exception is not None:
          raise user_exception from None
  
imports: import asyncio
  import inspect
  import logging
  import os
  import pickle
  import time
  import traceback
  from contextlib import contextmanager
  from importlib import import_module
  from typing import Any, AsyncGenerator, Callable, Dict, Optional, Tuple, Union
  import starlette.responses
  import ray
  from ray import cloudpickle
  from ray._private.utils import get_or_create_event_loop
  from ray.actor import ActorClass
  from ray.remote_function import RemoteFunction
  from ray.serve import metrics
  from ray.serve._private.autoscaling_metrics import InMemoryMetricsStore
  from ray.serve._private.common import DeploymentID, ReplicaName, ReplicaTag, ServeComponentType, StreamingHTTPRequest, gRPCRequest
  from ray.serve._private.config import DeploymentConfig
  from ray.serve._private.constants import DEFAULT_LATENCY_BUCKET_MS, GRPC_CONTEXT_ARG_NAME, HEALTH_CHECK_METHOD, RAY_SERVE_GAUGE_METRIC_SET_PERIOD_S, RAY_SERVE_REPLICA_AUTOSCALING_METRIC_RECORD_PERIOD_S, RECONFIGURE_METHOD, REPLICA_CONTROL_PLANE_CONCURRENCY_GROUP, SERVE_CONTROLLER_NAME, SERVE_LOGGER_NAME, SERVE_NAMESPACE
  from ray.serve._private.http_util import ASGIAppReplicaWrapper, ASGIArgs, ASGIReceiveProxy, MessageQueue, Response
  from ray.serve._private.logging_utils import access_log_msg, configure_component_cpu_profiler, configure_component_logger, configure_component_memory_profiler, get_component_logger_file_path
  from ray.serve._private.router import RequestMetadata
  from ray.serve._private.utils import MetricsPusher, merge_dict, parse_import_path, wrap_to_ray_error
  from ray.serve._private.version import DeploymentVersion
  from ray.serve.config import AutoscalingConfig
  from ray.serve.deployment import Deployment
  from ray.serve.exceptions import RayServeException
  from ray.serve.schema import LoggingConfig
label_desc: Use image node_id 4 for calling the ReplicaActor obj's underlying member method code with example usage: obj._wrap_user_method_call(request_metadata) without return types