wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-09-11 12:43:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

service -> endpoint in router (#8269)

Browse Source
This commit is contained in:
Edward Oakes
2020-05-01 11:55:34 -05:00
committed by GitHub
parent 421b3c9d8b
commit 63bc7dc522
4 changed files with 76 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/serve/router.py
+30 -30
View File
@@ -104,18 +104,18 @@ class Router:
async def __init__(self):
# Note: Several queues are used in the router
# - When a request come in, it's placed inside its corresponding
# service_queue.
# - The service_queue is dequed during flush operation, which moves
# endpoint_queue.
# - The endpoint_queue is dequeued during flush operation, which moves
# the queries to backend buffer_queue. Here we match a request
# for a service to a backend given some policy.
# for an endpoint to a backend given some policy.
# - The worker_queue is used to collect idle actor handle. These
# handles are dequed during the second stage of flush operation,
# which assign queries in buffer_queue to actor handle.
# -- Queues -- #
# service_name -> request queue
self.service_queues: DefaultDict[asyncio.Queue[Query]] = defaultdict(
# endpoint_name -> request queue
self.endpoint_queues: DefaultDict[asyncio.Queue[Query]] = defaultdict(
asyncio.Queue)
# backend_name -> worker request queue
self.worker_queues: DefaultDict[asyncio.Queue[
@@ -125,7 +125,7 @@ class Router:
# -- Metadata -- #
# service_name -> traffic_policy
# endpoint_name -> traffic_policy
self.traffic = defaultdict(dict)
# backend_name -> backend_config
self.backend_info = dict()
@@ -180,8 +180,8 @@ class Router:
async def enqueue_request(self, request_meta, *request_args,
**request_kwargs):
service = request_meta.service
logger.debug("Received a request for service {}".format(service))
endpoint = request_meta.endpoint
logger.debug("Received a request for endpoint {}".format(endpoint))
# check if the slo specified is directly the
# wall clock time
@@ -197,7 +197,7 @@ class Router:
request_slo_ms,
call_method=request_meta.call_method,
async_future=asyncio.get_event_loop().create_future())
await self.service_queues[service].put(query)
await self.endpoint_queues[endpoint].put(query)
await self.flush()
# Note: a future change can be to directly return the ObjectID from
@@ -244,21 +244,21 @@ class Router:
# on it.
worker_handle.__ray_terminate__.remote()
async def set_traffic(self, service, traffic_dict):
logger.debug("Setting traffic for service %s to %s", service,
async def set_traffic(self, endpoint, traffic_dict):
logger.debug("Setting traffic for endpoint %s to %s", endpoint,
traffic_dict)
self.traffic[service] = traffic_dict
self.traffic[endpoint] = traffic_dict
await self.flush()
async def remove_service(self, service):
logger.debug("Removing service {}".format(service))
async def remove_endpoint(self, endpoint):
logger.debug("Removing endpoint {}".format(endpoint))
async with self.flush_lock:
await self._flush_service_queues()
await self._flush_endpoint_queues()
await self._flush_buffer_queues()
if service in self.service_queues:
del self.service_queues[service]
if service in self.traffic:
del self.traffic[service]
if endpoint in self.endpoint_queues:
del self.endpoint_queues[endpoint]
if endpoint in self.traffic:
del self.traffic[endpoint]
async def set_backend_config(self, backend, config):
logger.debug("Setting backend config for "
@@ -268,7 +268,7 @@ class Router:
async def remove_backend(self, backend):
logger.debug("Removing backend {}".format(backend))
async with self.flush_lock:
await self._flush_service_queues()
await self._flush_endpoint_queues()
await self._flush_buffer_queues()
if backend in self.backend_info:
del self.backend_info[backend]
@@ -284,11 +284,11 @@ class Router:
method invocation.
"""
async with self.flush_lock:
await self._flush_service_queues()
await self._flush_endpoint_queues()
await self._flush_buffer_queues()
def _get_available_backends(self, service):
backends_in_policy = set(self.traffic[service].keys())
def _get_available_backends(self, endpoint):
backends_in_policy = set(self.traffic[endpoint].keys())
available_workers = {
backend
for backend, queues in self.worker_queues.items()
@@ -296,25 +296,25 @@ class Router:
}
return list(backends_in_policy.intersection(available_workers))
async def _flush_service_queues(self):
"""Selects the backend and puts the service queue query to the buffer
async def _flush_endpoint_queues(self):
"""Selects the backend and puts the endpoint queue query to the buffer
Expected Implementation:
The implementer is expected to access and manipulate
self.service_queues : dict[str,Deque]
self.endpoint_queues : dict[str,Deque]
self.buffer_queues : dict[str,sortedlist]
For registering the implemented policies register at policy.py
Expected Behavior:
the Deque of all services in self.service_queues linked with
the Deque of all endpoints in self.endpoint_queues linked with
atleast one backend must be empty irrespective of whatever
backend policy is implemented.
"""
raise NotImplementedError(
"This method should be implemented by child class.")
# flushes the buffer queue and assigns work to workers
# Flushes the buffer queue and assigns work to workers.
async def _flush_buffer_queues(self):
for service in self.traffic.keys():
ready_backends = self._get_available_backends(service)
for endpoint in self.traffic:
ready_backends = self._get_available_backends(endpoint)
for backend in ready_backends:
# no work available
if len(self.buffer_queues[backend]) == 0: