wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-06 05:16:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

[serve] Only access backend_table in master actor (#7913)

Browse Source
This commit is contained in:
Edward Oakes
2020-04-07 10:12:39 -05:00
committed by GitHub
parent 48b48cc8c2
commit d3c310f408
3 changed files with 101 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/serve/api.py
+10 -85
View File
@@ -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
python/ray/serve/global_state.py
+83 -14
View File
@@ -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]
python/ray/serve/tests/test_api.py
+8 -8
View File
@@ -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())