python/ray/serve/replica.py
def create_replica_wrapper(
    name: str, import_path: str = None, serialized_deployment_def: bytes = None
):
    """Creates a replica class wrapping the provided function or class.

    This approach is picked over inheritance to avoid conflict between user
    provided class and the RayServeReplica class.
    """

    if (import_path is None) and (serialized_deployment_def is None):
        raise ValueError(
            "Either the import_name or the serialized_deployment_def must "
            "be specified, but both were unspecified."
        )
    elif (import_path is not None) and (serialized_deployment_def is not None):
        raise ValueError(
            "Only one of either the import_name or the "
            "serialized_deployment_def must be specified, but both were "
            "specified."
        )

    # TODO(architkulkarni): Add type hints after upgrading cloudpickle
    class RayServeWrappedReplica(object):
        async def __init__(
            self,
            deployment_name,
            replica_tag,
            init_args,
            init_kwargs,
            deployment_config_proto_bytes: bytes,
            version: DeploymentVersion,
            controller_name: str,
            controller_namespace: str,
            detached: bool,
        ):
            configure_component_logger(
                component_type="deployment",
                component_name=deployment_name,
                component_id=replica_tag,
            )

            if import_path is not None:
                module_name, attr_name = parse_import_path(import_path)
                deployment_def = getattr(import_module(module_name), attr_name)
                # For ray or serve decorated class or function, strip to return
                # original body
                if isinstance(deployment_def, RemoteFunction):
                    deployment_def = deployment_def._function
                elif isinstance(deployment_def, ActorClass):
                    deployment_def = deployment_def.__ray_metadata__.modified_class
                elif isinstance(deployment_def, Deployment):
                    logger.warning(
                        f'The import path "{import_path}" contains a '
                        "decorated Serve deployment. The decorator's settings "
                        "are ignored when deploying via import path."
                    )
                    deployment_def = deployment_def.func_or_class

            else:
                deployment_def = cloudpickle.loads(serialized_deployment_def)

            deployment_config = DeploymentConfig.from_proto_bytes(
                deployment_config_proto_bytes
            )

            if inspect.isfunction(deployment_def):
                is_function = True
            elif inspect.isclass(deployment_def):
                is_function = False
            else:
                assert False, (
                    "deployment_def must be function, class, or "
                    "corresponding import path. Instead, it's type was "
                    f"{type(deployment_def)}."
                )

            # Set the controller name so that serve.connect() in the user's
            # code will connect to the instance that this deployment is running
            # in.
            ray.serve.context.set_internal_replica_context(
                deployment_name,
                replica_tag,
                controller_name,
                controller_namespace,
                servable_object=None,
            )

            assert controller_name, "Must provide a valid controller_name"

            controller_handle = ray.get_actor(
                controller_name, namespace=controller_namespace
            )

            # This closure initializes user code and finalizes replica
            # startup. By splitting the initialization step like this,
            # we can already access this actor before the user code
            # has finished initializing.
            # The supervising state manager can then wait
            # for allocation of this replica by using the `is_allocated`
            # method. After that, it calls `reconfigure` to trigger
            # user code initialization.
            async def initialize_replica():
                if is_function:
                    _callable = deployment_def
                else:
                    # This allows deployments to define an async __init__
                    # method (required for FastAPI).
                    _callable = deployment_def.__new__(deployment_def)
                    await sync_to_async(_callable.__init__)(*init_args, **init_kwargs)

                # Setting the context again to update the servable_object.
                ray.serve.context.set_internal_replica_context(
                    deployment_name,
                    replica_tag,
                    controller_name,
                    controller_namespace,
                    servable_object=_callable,
                )

                self.replica = RayServeReplica(
                    _callable,
                    deployment_name,
                    replica_tag,
                    deployment_config,
                    deployment_config.user_config,
                    version,
                    is_function,
                    controller_handle,
                )

            # Is it fine that replica is None here?
            # Should we add a check in all methods that use self.replica
            # or, alternatively, create an async get_replica() method?
            self.replica = None
            self._initialize_replica = initialize_replica

        @ray.method(num_returns=2)
        async def handle_request(
            self,
            pickled_request_metadata: bytes,
            *request_args,
            **request_kwargs,
        ):
            # The request metadata should be pickled for performance.
            request_metadata: RequestMetadata = pickle.loads(pickled_request_metadata)

            # Directly receive input because it might contain an ObjectRef.
            query = Query(request_args, request_kwargs, request_metadata)
            return await self.replica.handle_request(query)

        async def is_allocated(self) -> str:
            """poke the replica to check whether it's alive.

            When calling this method on an ActorHandle, it will complete as
            soon as the actor has started running. We use this mechanism to
            detect when a replica has been allocated a worker slot.
            At this time, the replica can transition from PENDING_ALLOCATION
            to PENDING_INITIALIZATION startup state.

            Return the NodeID of this replica
            """
            return ray.get_runtime_context().node_id

        async def reconfigure(
            self, user_config: Optional[Any] = None, _after: Optional[Any] = None
        ) -> Tuple[DeploymentConfig, DeploymentVersion]:
            # Unused `_after` argument is for scheduling: passing an ObjectRef
            # allows delaying reconfiguration until after this call has returned.
            if self.replica is None:
                await self._initialize_replica()
            if user_config is not None:
                await self.replica.reconfigure(user_config)

            return self.get_metadata()

        def get_metadata(self) -> Tuple[DeploymentConfig, DeploymentVersion]:
            return self.replica.deployment_config, self.replica.version

        async def prepare_for_shutdown(self):
            if self.replica is not None:
                return await self.replica.prepare_for_shutdown()

        async def check_health(self):
            await self.replica.check_health()

    RayServeWrappedReplica.__name__ = name
    return RayServeWrappedReplica
Creates a replica class wrapping the provided function or class.

    This approach is picked over inheritance to avoid conflict between user
    provided class and the RayServeReplica class.
    
Creates a replica class wrapping the provided function or class.
This approach is picked over inheritance to avoid conflict between user
provided class and the RayServeReplica class.
Creates a replica class wrapping the provided function or class . This approach is picked over inheritance to avoid conflict between user provided class and the RayServeReplica class .
def create_replica_wrapper(
    name: str, import_path: str = None, serialized_deployment_def: bytes = None
):
    if (import_path is None) and (serialized_deployment_def is None):
        raise ValueError(
            "Either the import_name or the serialized_deployment_def must "
            "be specified, but both were unspecified."
        )
    elif (import_path is not None) and (serialized_deployment_def is not None):
        raise ValueError(
            "Only one of either the import_name or the "
            "serialized_deployment_def must be specified, but both were "
            "specified."
        )
    class RayServeWrappedReplica(object):
        async def __init__(
            self,
            deployment_name,
            replica_tag,
            init_args,
            init_kwargs,
            deployment_config_proto_bytes: bytes,
            version: DeploymentVersion,
            controller_name: str,
            controller_namespace: str,
            detached: bool,
        ):
            configure_component_logger(
                component_type="deployment",
                component_name=deployment_name,
                component_id=replica_tag,
            )
            if import_path is not None:
                module_name, attr_name = parse_import_path(import_path)
                deployment_def = getattr(import_module(module_name), attr_name)
                if isinstance(deployment_def, RemoteFunction):
                    deployment_def = deployment_def._function
                elif isinstance(deployment_def, ActorClass):
                    deployment_def = deployment_def.__ray_metadata__.modified_class
                elif isinstance(deployment_def, Deployment):
                    logger.warning(
                        f'The import path "{import_path}" contains a '
                        "decorated Serve deployment. The decorator's settings "
                        "are ignored when deploying via import path."
                    )
                    deployment_def = deployment_def.func_or_class
            else:
                deployment_def = cloudpickle.loads(serialized_deployment_def)
            deployment_config = DeploymentConfig.from_proto_bytes(
                deployment_config_proto_bytes
            )
            if inspect.isfunction(deployment_def):
                is_function = True
            elif inspect.isclass(deployment_def):
                is_function = False
            else:
                assert False, (
                    "deployment_def must be function, class, or "
                    "corresponding import path. Instead, it's type was "
                    f"{type(deployment_def)}."
                )
            ray.serve.context.set_internal_replica_context(
                deployment_name,
                replica_tag,
                controller_name,
                controller_namespace,
                servable_object=None,
            )
            assert controller_name, "Must provide a valid controller_name"
            controller_handle = ray.get_actor(
                controller_name, namespace=controller_namespace
            )
            async def initialize_replica():
                if is_function:
                    _callable = deployment_def
                else:
                    _callable = deployment_def.__new__(deployment_def)
                    await sync_to_async(_callable.__init__)(*init_args, **init_kwargs)
                ray.serve.context.set_internal_replica_context(
                    deployment_name,
                    replica_tag,
                    controller_name,
                    controller_namespace,
                    servable_object=_callable,
                )
                self.replica = RayServeReplica(
                    _callable,
                    deployment_name,
                    replica_tag,
                    deployment_config,
                    deployment_config.user_config,
                    version,
                    is_function,
                    controller_handle,
                )
            self.replica = None
            self._initialize_replica = initialize_replica
        @ray.method(num_returns=2)
        async def handle_request(
            self,
            pickled_request_metadata: bytes,
            *request_args,
            **request_kwargs,
        ):
            request_metadata: RequestMetadata = pickle.loads(pickled_request_metadata)
            query = Query(request_args, request_kwargs, request_metadata)
            return await self.replica.handle_request(query)
        async def is_allocated(self) -> str:
            return ray.get_runtime_context().node_id
        async def reconfigure(
            self, user_config: Optional[Any] = None, _after: Optional[Any] = None
        ) -> Tuple[DeploymentConfig, DeploymentVersion]:
            if self.replica is None:
                await self._initialize_replica()
            if user_config is not None:
                await self.replica.reconfigure(user_config)
            return self.get_metadata()
        def get_metadata(self) -> Tuple[DeploymentConfig, DeploymentVersion]:
            return self.replica.deployment_config, self.replica.version
        async def prepare_for_shutdown(self):
            if self.replica is not None:
                return await self.replica.prepare_for_shutdown()
        async def check_health(self):
            await self.replica.check_health()
    RayServeWrappedReplica.__name__ = name
    return RayServeWrappedReplica
def create_replica_wrapper ( name : str , import_path : str = None , serialized_deployment_def : bytes = None ) : if ( import_path is None ) and ( serialized_deployment_def is None ) : raise ValueError ( "Either the import_name or the serialized_deployment_def must " "be specified, but both were unspecified." ) elif ( import_path is not None ) and ( serialized_deployment_def is not None ) : raise ValueError ( "Only one of either the import_name or the " "serialized_deployment_def must be specified, but both were " "specified." ) class RayServeWrappedReplica ( object ) : async def __init__ ( self , deployment_name , replica_tag , init_args , init_kwargs , deployment_config_proto_bytes : bytes , version : DeploymentVersion , controller_name : str , controller_namespace : str , detached : bool , ) : configure_component_logger ( component_type = "deployment" , component_name = deployment_name , component_id = replica_tag , ) if import_path is not None : module_name , attr_name = parse_import_path ( import_path ) deployment_def = getattr ( import_module ( module_name ) , attr_name ) if isinstance ( deployment_def , RemoteFunction ) : deployment_def = deployment_def . _function elif isinstance ( deployment_def , ActorClass ) : deployment_def = deployment_def . __ray_metadata__ . modified_class elif isinstance ( deployment_def , Deployment ) : logger . warning ( f'The import path "{import_path}" contains a ' "decorated Serve deployment. The decorator's settings " "are ignored when deploying via import path." ) deployment_def = deployment_def . func_or_class else : deployment_def = cloudpickle . loads ( serialized_deployment_def ) deployment_config = DeploymentConfig . from_proto_bytes ( deployment_config_proto_bytes ) if inspect . isfunction ( deployment_def ) : is_function = True elif inspect . isclass ( deployment_def ) : is_function = False else : assert False , ( "deployment_def must be function, class, or " "corresponding import path. Instead, it's type was " f"{type(deployment_def)}." ) ray . serve . context . set_internal_replica_context ( deployment_name , replica_tag , controller_name , controller_namespace , servable_object = None , ) assert controller_name , "Must provide a valid controller_name" controller_handle = ray . get_actor ( controller_name , namespace = controller_namespace ) async def initialize_replica ( ) : if is_function : _callable = deployment_def else : _callable = deployment_def . __new__ ( deployment_def ) await sync_to_async ( _callable . __init__ ) ( * init_args , ** init_kwargs ) ray . serve . context . set_internal_replica_context ( deployment_name , replica_tag , controller_name , controller_namespace , servable_object = _callable , ) self . replica = RayServeReplica ( _callable , deployment_name , replica_tag , deployment_config , deployment_config . user_config , version , is_function , controller_handle , ) self . replica = None self . _initialize_replica = initialize_replica @ ray . method ( num_returns = 2 ) async def handle_request ( self , pickled_request_metadata : bytes , * request_args , ** request_kwargs , ) : request_metadata : RequestMetadata = pickle . loads ( pickled_request_metadata ) query = Query ( request_args , request_kwargs , request_metadata ) return await self . replica . handle_request ( query ) async def is_allocated ( self ) -> str : """poke the replica to check whether it's alive.

            When calling this method on an ActorHandle, it will complete as
            soon as the actor has started running. We use this mechanism to
            detect when a replica has been allocated a worker slot.
            At this time, the replica can transition from PENDING_ALLOCATION
            to PENDING_INITIALIZATION startup state.

            Return the NodeID of this replica
            """ return ray . get_runtime_context ( ) . node_id async def reconfigure ( self , user_config : Optional [ Any ] = None , _after : Optional [ Any ] = None ) -> Tuple [ DeploymentConfig , DeploymentVersion ] : if self . replica is None : await self . _initialize_replica ( ) if user_config is not None : await self . replica . reconfigure ( user_config ) return self . get_metadata ( ) def get_metadata ( self ) -> Tuple [ DeploymentConfig , DeploymentVersion ] : return self . replica . deployment_config , self . replica . version async def prepare_for_shutdown ( self ) : if self . replica is not None : return await self . replica . prepare_for_shutdown ( ) async def check_health ( self ) : await self . replica . check_health ( ) RayServeWrappedReplica . __name__ = name return RayServeWrappedReplica
Creates a replica class wrapping the provided function or class.
Creates a replica class wrapping the provided function or class .
"""Creates a replica class wrapping the provided function or class.

    This approach is picked over inheritance to avoid conflict between user
    provided class and the RayServeReplica class.
    """ # TODO(architkulkarni): Add type hints after upgrading cloudpickle # For ray or serve decorated class or function, strip to return # original body # Set the controller name so that serve.connect() in the user's # code will connect to the instance that this deployment is running # in. # This closure initializes user code and finalizes replica # startup. By splitting the initialization step like this, # we can already access this actor before the user code # has finished initializing. # The supervising state manager can then wait # for allocation of this replica by using the `is_allocated` # method. After that, it calls `reconfigure` to trigger # user code initialization. # This allows deployments to define an async __init__ # method (required for FastAPI). # Setting the context again to update the servable_object. # Is it fine that replica is None here? # Should we add a check in all methods that use self.replica # or, alternatively, create an async get_replica() method? # The request metadata should be pickled for performance. # Directly receive input because it might contain an ObjectRef. """poke the replica to check whether it's alive.

            When calling this method on an ActorHandle, it will complete as
            soon as the actor has started running. We use this mechanism to
            detect when a replica has been allocated a worker slot.
            At this time, the replica can transition from PENDING_ALLOCATION
            to PENDING_INITIALIZATION startup state.

            Return the NodeID of this replica
            """ # Unused `_after` argument is for scheduling: passing an ObjectRef # allows delaying reconfiguration until after this call has returned.
name
str
import_path
str
serialized_deployment_def
bytes