diff --git a/python/ray/serve/api.py b/python/ray/serve/api.py index 1c76c8779..aaff4737e 100644 --- a/python/ray/serve/api.py +++ b/python/ray/serve/api.py @@ -167,10 +167,9 @@ def create_endpoint(endpoint_name, route=None, methods=["GET"]): blocking (bool): If true, the function will wait for service to be registered before returning """ - methods = [m.upper() for m in methods] ray.get( global_state.master_actor.create_endpoint.remote( - route, endpoint_name, methods)) + route, endpoint_name, [m.upper() for m in methods])) @_ensure_connected @@ -181,52 +180,20 @@ def set_backend_config(backend_tag, backend_config): backend_tag(str): A registered backend. backend_config(BackendConfig) : Desired backend configuration. """ - assert (backend_tag in global_state.backend_table.list_backends() - ), "Backend {} is not registered.".format(backend_tag) - assert isinstance(backend_config, - BackendConfig), ("backend_config must be" - " of instance BackendConfig") - backend_config_dict = dict(backend_config) - old_backend_config_dict = global_state.backend_table.get_info(backend_tag) - - if (not old_backend_config_dict["has_accept_batch_annotation"] - and backend_config.max_batch_size is not None): - raise batch_annotation_not_found - - global_state.backend_table.register_info(backend_tag, backend_config_dict) - - # inform the router about change in configuration - # particularly for setting max_batch_size - router = global_state.get_router() - ray.get(router.set_backend_config.remote(backend_tag, backend_config_dict)) - - # checking if replicas need to be restarted - # Replicas are restarted if there is any change in the backend config - # related to restart_configs - # TODO(alind) : have replica restarting policies selected by the user - - need_to_restart_replicas = any( - old_backend_config_dict[k] != backend_config_dict[k] - for k in BackendConfig.restart_on_change_fields) - if need_to_restart_replicas: - # kill all the replicas for restarting with new configurations - _scale(backend_tag, 0) - - # scale the replicas with new configuration - _scale(backend_tag, backend_config_dict["num_replicas"]) + ray.get( + global_state.master_actor.set_backend_config.remote( + backend_tag, backend_config)) @_ensure_connected def get_backend_config(backend_tag): - """get the backend configuration for a backend tag + """Get the backend configuration for a backend tag. Args: backend_tag(str): A registered backend. """ - assert (backend_tag in global_state.backend_table.list_backends() - ), "Backend {} is not registered.".format(backend_tag) - backend_config_dict = global_state.backend_table.get_info(backend_tag) - return BackendConfig(**backend_config_dict) + return ray.get( + global_state.master_actor.get_backend_config.remote(backend_tag)) def _backend_accept_batch(func_or_class): @@ -290,51 +257,9 @@ def create_backend(func_or_class, "Backend must be a function or class, it is {}.".format( type(func_or_class))) - backend_config_dict = dict(backend_config) - - # save creator which starts replicas - global_state.backend_table.register_backend(backend_tag, creator) - - # save information about configurations needed to start the replicas - global_state.backend_table.register_info(backend_tag, backend_config_dict) - - # save the initial arguments needed by replicas - global_state.backend_table.save_init_args(backend_tag, arg_list) - - # set the backend config inside the router - # particularly for max-batch-size - router = global_state.get_router() - ray.get(router.set_backend_config.remote(backend_tag, backend_config_dict)) - _scale(backend_tag, backend_config_dict["num_replicas"]) - - -@_ensure_connected -def _scale(backend_tag, num_replicas): - """Set the number of replicas for backend_tag. - - Args: - backend_tag (str): A registered backend. - num_replicas (int): Desired number of replicas - """ - assert (backend_tag in global_state.backend_table.list_backends() - ), "Backend {} is not registered.".format(backend_tag) - assert num_replicas >= 0, ("Number of replicas must be" - " greater than or equal to 0.") - - replicas = global_state.backend_table.list_replicas(backend_tag) - current_num_replicas = len(replicas) - delta_num_replicas = num_replicas - current_num_replicas - - if delta_num_replicas > 0: - for _ in range(delta_num_replicas): - ray.get( - global_state.master_actor.start_backend_replica.remote( - backend_tag)) - elif delta_num_replicas < 0: - for _ in range(-delta_num_replicas): - ray.get( - global_state.master_actor.remove_backend_replica.remote( - backend_tag)) + ray.get( + global_state.master_actor.create_backend.remote( + backend_tag, creator, backend_config, arg_list)) @_ensure_connected diff --git a/python/ray/serve/global_state.py b/python/ray/serve/global_state.py index af39d2d8e..aefa6e451 100644 --- a/python/ray/serve/global_state.py +++ b/python/ray/serve/global_state.py @@ -1,6 +1,7 @@ import ray from ray.serve.backend_config import BackendConfig from ray.serve.constants import (SERVE_MASTER_NAME, ASYNC_CONCURRENCY) +from ray.serve.exceptions import batch_annotation_not_found from ray.serve.http_proxy import HTTPProxyActor from ray.serve.kv_store_service import (BackendTable, RoutingTable, TrafficPolicyTable) @@ -31,9 +32,6 @@ class ServeMaster: self.http_proxy = None self.metric_monitor = None - def get_kv_store_connector(self): - return self.kv_store_connector - def get_traffic_policy(self, endpoint_name): return self.policy_table.list_traffic_policy()[endpoint_name] @@ -71,7 +69,26 @@ class ServeMaster: "Metric monitor not started yet.") return [self.metric_monitor] - def start_backend_replica(self, backend_tag): + def _list_replicas(self, backend_tag): + return self.backend_table.list_replicas(backend_tag) + + def scale_replicas(self, backend_tag, num_replicas): + assert (backend_tag in self.backend_table.list_backends() + ), "Backend {} is not registered.".format(backend_tag) + assert num_replicas >= 0, ("Number of replicas must be" + " greater than or equal to 0.") + + current_num_replicas = len(self._list_replicas(backend_tag)) + delta_num_replicas = num_replicas - current_num_replicas + + if delta_num_replicas > 0: + for _ in range(delta_num_replicas): + self._start_backend_replica(backend_tag) + elif delta_num_replicas < 0: + for _ in range(-delta_num_replicas): + self._remove_backend_replica(backend_tag) + + def _start_backend_replica(self, backend_tag): assert (backend_tag in self.backend_table.list_backends() ), "Backend {} is not registered.".format(backend_tag) @@ -98,13 +115,12 @@ class ServeMaster: self.backend_table.add_replica(backend_tag, replica_tag) self.get_metric_monitor()[0].add_target.remote(runner_handle) - def remove_backend_replica(self, backend_tag): + def _remove_backend_replica(self, backend_tag): assert (backend_tag in self.backend_table.list_backends() ), "Backend {} is not registered.".format(backend_tag) - assert ( - len(self.backend_table.list_replicas(backend_tag)) > 0 - ), "Backend {} does not have enough replicas to be removed.".format( - backend_tag) + assert (len(self._list_replicas(backend_tag)) > + 0), "Tried to remove replica from empty backend ({}).".format( + backend_tag) replica_tag = self.backend_table.remove_replica(backend_tag) assert replica_tag in self.tag_to_actor_handles @@ -159,6 +175,64 @@ class ServeMaster: self.route_table.list_service( include_methods=True, include_headless=False))) + def create_backend(self, backend_tag, creator, backend_config, arg_list): + backend_config_dict = dict(backend_config) + + # Save creator which starts replicas. + self.backend_table.register_backend(backend_tag, creator) + + # Save information about configurations needed to start the replicas. + self.backend_table.register_info(backend_tag, backend_config_dict) + + # Save the initial arguments needed by replicas. + self.backend_table.save_init_args(backend_tag, arg_list) + + # Set the backend config inside the router + # (particularly for max-batch-size). + [router] = self.get_router() + ray.get( + router.set_backend_config.remote(backend_tag, backend_config_dict)) + self.scale_replicas(backend_tag, backend_config_dict["num_replicas"]) + + def set_backend_config(self, backend_tag, backend_config): + assert (backend_tag in self.backend_table.list_backends() + ), "Backend {} is not registered.".format(backend_tag) + assert isinstance(backend_config, + BackendConfig), ("backend_config must be" + " of instance BackendConfig") + backend_config_dict = dict(backend_config) + old_backend_config_dict = self.backend_table.get_info(backend_tag) + + if (not old_backend_config_dict["has_accept_batch_annotation"] + and backend_config.max_batch_size is not None): + raise batch_annotation_not_found + + self.backend_table.register_info(backend_tag, backend_config_dict) + + # Inform the router about change in configuration + # (particularly for setting max_batch_size). + [router] = self.get_router() + ray.get( + router.set_backend_config.remote(backend_tag, backend_config_dict)) + + # Restart replicas if there is a change in the backend config related + # to restart_configs. + need_to_restart_replicas = any( + old_backend_config_dict[k] != backend_config_dict[k] + for k in BackendConfig.restart_on_change_fields) + if need_to_restart_replicas: + # Kill all the replicas for restarting with new configurations. + self.scale_replicas(backend_tag, 0) + + # Scale the replicas with the new configuration. + self.scale_replicas(backend_tag, backend_config_dict["num_replicas"]) + + def get_backend_config(self, backend_tag): + assert (backend_tag in self.backend_table.list_backends() + ), "Backend {} is not registered.".format(backend_tag) + backend_config_dict = self.backend_table.get_info(backend_tag) + return BackendConfig(**backend_config_dict) + class GlobalState: """Encapsulate all global state in the serving system. @@ -174,11 +248,6 @@ class GlobalState: master_actor = ray.util.get_actor(SERVE_MASTER_NAME) self.master_actor = master_actor - # Connect to all the tables. - kv_store_connector = ray.get( - self.master_actor.get_kv_store_connector.remote()) - self.backend_table = BackendTable(kv_store_connector) - def get_router(self): return ray.get(self.master_actor.get_router.remote())[0] diff --git a/python/ray/serve/tests/test_api.py b/python/ray/serve/tests/test_api.py index f0a4bd772..a93219396 100644 --- a/python/ray/serve/tests/test_api.py +++ b/python/ray/serve/tests/test_api.py @@ -182,16 +182,16 @@ def test_killing_replicas(serve_instance): b_config = BackendConfig(num_replicas=3, num_cpus=2) serve.create_backend(Simple, "simple:v1", backend_config=b_config) global_state = serve.api._get_global_state() - old_replica_tag_list = global_state.backend_table.list_replicas( - "simple:v1") + old_replica_tag_list = ray.get( + global_state.master_actor._list_replicas.remote("simple:v1")) bnew_config = serve.get_backend_config("simple:v1") # change the config bnew_config.num_cpus = 1 # set the config serve.set_backend_config("simple:v1", bnew_config) - new_replica_tag_list = global_state.backend_table.list_replicas( - "simple:v1") + new_replica_tag_list = ray.get( + global_state.master_actor._list_replicas.remote("simple:v1")) new_all_tag_list = list( ray.get(global_state.master_actor.get_all_handles.remote()).keys()) @@ -216,16 +216,16 @@ def test_not_killing_replicas(serve_instance): b_config = BackendConfig(num_replicas=3, max_batch_size=2) serve.create_backend(BatchSimple, "bsimple:v1", backend_config=b_config) global_state = serve.api._get_global_state() - old_replica_tag_list = global_state.backend_table.list_replicas( - "bsimple:v1") + old_replica_tag_list = ray.get( + global_state.master_actor._list_replicas.remote("bsimple:v1")) bnew_config = serve.get_backend_config("bsimple:v1") # change the config bnew_config.max_batch_size = 5 # set the config serve.set_backend_config("bsimple:v1", bnew_config) - new_replica_tag_list = global_state.backend_table.list_replicas( - "bsimple:v1") + new_replica_tag_list = ray.get( + global_state.master_actor._list_replicas.remote("bsimple:v1")) new_all_tag_list = list( ray.get(global_state.master_actor.get_all_handles.remote()).keys())