[Serve] Rewrite Router to be Embeddable #12019

simon-mo · 2020-11-14T03:16:29Z

Why are these changes needed?

This PR refactor the router into an embeddable object. It:

utilizes long polling to fetch config from the controller
has a simpler job (assign query to replica)

Now you can just call await router.assign_query(query) to retrieve an object ref. The method is async and it will wait until a replica is freed. The ServeHandle now utilizes this embeddable router (instead of calling to the HTTPProxy actor). For the handle to be accessible in sync context, we start a thread hosting the asyncio loop that runs the controller.

Related issue number

Checks

I've run scripts/format.sh to lint the changes in this PR.
I've included any doc changes needed for https://docs.ray.io/en/master/.
I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
Testing Strategy
- Unit tests
- Release tests
- This PR is not tested :(

edoakes

Looks good overall, this is significantly simpler, which is a great sign :)

edoakes · 2020-11-16T03:02:34Z

python/ray/serve/api.py

-            router_chosen,
-            endpoint_name,
-        )
+        if endpoint_name not in self._handle_cache:


When do we remove from this cache?

Can we do it when an endpoint is deleted? Should be some kind of hook for that.

Good catch. I also added a comment in the constructor.

edoakes · 2020-11-16T03:04:24Z

python/ray/serve/endpoint_policy.py


 import numpy as np

 from ray.serve.utils import logger


+@lru_cache(maxsize=128)


Good idea! It's probably worth benchmarking this.

stdlib is 10x faster than numpy.

and this generally task ~10us in cache miss

edoakes · 2020-11-16T03:04:41Z

python/ray/serve/endpoint_policy.py


 import numpy as np

 from ray.serve.utils import logger


+@lru_cache(maxsize=128)
+def deterministic_hash(key: bytes) -> float:
+    """Given an arbitrary string, return a value between 0 and 1"""


Suggested change

"""Given an arbitrary string, return a value between 0 and 1"""

"""Given an arbitrary string, return a deterministic value between 0 and 1."""

edoakes · 2020-11-16T03:05:15Z

python/ray/serve/endpoint_policy.py

+    def flush(self, query) -> List[str]:
+        """Assign a query to a list of backends.


So much nicer!

edoakes · 2020-11-16T03:05:33Z

python/ray/serve/endpoint_policy.py

-        from the endpoint queue and pushing them onto a backend queue. The
-        method must also return a set of all backend tags so that the caller
-        knows which backend_queues to flush.
+    def flush(self, query) -> List[str]:


Should we rename this method btw? Something like "assign_query" ?

yeah i changed it to assign. good catch!

edoakes · 2020-11-16T03:06:43Z

python/ray/serve/handle.py

+def create_or_get_async_loop_in_thread():
+    global global_async_loop
+    if global_async_loop is None:
+        # We have to asyncio's loop because uvloop will segfault when there


forgot a word

this seems pretty hacky/dangerous. seems like a red flag that we're doing something wrong to me?

edoakes · 2020-11-16T03:08:33Z

python/ray/serve/handle.py

+            asyncio.run_coroutine_threadsafe(
+                self.router.setup_in_async_loop(),
+                self.async_loop,


why do we use run_coroutine_threadsafe here but create_task down below?

create_task isn't thread safe :-)

edoakes · 2020-11-16T03:09:18Z

python/ray/serve/handle.py

+        future = asyncio.run_coroutine_threadsafe(coro, self.async_loop)
+        return future.result()
+
+    async def remote_async(self, request_data, **kwargs):


please mark this experimental for now so that we can change the API in the near future if we want to

edoakes · 2020-11-16T03:09:57Z

python/ray/serve/router.py

+        # NOTE(simon): we separated this method to run all the setup in asyncio
+        # loop.


this comment doesn't make sense to me out of context. Separated it from what?

edoakes · 2020-11-16T03:10:49Z

python/ray/serve/router.py

+            self.backend_replicas[backend_tag].set_max_concurrent_queries(
+                config.max_concurrent_queries)
+
+    async def enqueue_request(


there aren't really queues anymore, maybe we should rename this :) --> assign_request? send_request?

edoakes · 2020-11-16T03:11:46Z

python/ray/serve/router.py

-        kwargs = pickle.loads(value)
-        return Query(**kwargs)
+class ReplicaSet:
+    """Data structure represents a set of replica actor handles"""


Suggested change

"""Data structure represents a set of replica actor handles"""

"""Data structure representing a set of replica actor handles."""

edoakes · 2020-11-16T03:11:58Z

python/ray/serve/router.py

+    """Data structure represents a set of replica actor handles"""
+
+    def __init__(self):
+        # The initialization setup before we receive them from long polling


before we receive what from long polling?

removed. the NOTE below already explained it.

edoakes · 2020-11-16T03:12:19Z

python/ray/serve/router.py

+        self.max_concurrent_queries: int = 8
+        self.in_flight_queries: Dict[ActorHandle, set] = dict()
+
+        # The iterator used for load balancing among replicas.


please summarize the existing policy

edoakes · 2020-11-16T03:12:44Z

python/ray/serve/router.py

+        # the same node.
+        self.replica_iterator = itertools.cycle(self.in_flight_queries.keys())
+
+        # Used to unblock the `await all_in_flight_queries` wait. A newly added


don't know what that wait is out of context, please describe it in higher level terms rather than code

edoakes · 2020-11-16T03:13:52Z

python/ray/serve/router.py

+                self.in_flight_queries.keys())
+            self.config_updated_event.set()
+
+    def _try_assign_replica(self, query: Query) -> Optional[ray.ObjectRef]:


please comment when this returns a ref vs. None and what None signals

edoakes · 2020-11-16T03:14:34Z

python/ray/serve/router.py

+        return len(done)
+
+    async def assign_replica(self, query: Query) -> ray.ObjectRef:
+        """Given a query, submit it to a replica and returns the object ref.


Suggested change

"""Given a query, submit it to a replica and returns the object ref.

"""Given a query, submit it to a replica and return the object ref.

edoakes · 2020-11-16T03:14:54Z

python/ray/serve/router.py

+        if assigned_ref is not None:
+            return assigned_ref
+
+        # Maybe there exist a free replica, we just need to refresh our queries


Suggested change

# Maybe there exist a free replica, we just need to refresh our queries

# Maybe there exists a free replica, we just need to refresh our query

edoakes · 2020-11-16T03:15:09Z

python/ray/serve/router.py

+        # Maybe there exist a free replica, we just need to refresh our queries
+        # tracker.
+        num_finished = self._drain_completed_object_refs()
+        # All replicas are really busy, let's wait for one query to complete or


Suggested change

# All replicas are really busy, let's wait for one query to complete or

# All replicas are really busy, wait for a query to complete or

edoakes · 2020-11-16T03:15:16Z

python/ray/serve/router.py

+        # tracker.
+        num_finished = self._drain_completed_object_refs()
+        # All replicas are really busy, let's wait for one query to complete or
+        # the config is updated.


Suggested change

# the config is updated.

# the config to be updated.

edoakes · 2020-11-16T03:16:34Z

python/ray/serve/router.py

+                self.config_updated_event.clear()
+        # We are pretty sure a free replica is ready now, let's recurse and
+        # assign this query a replica.
+        return await self.assign_replica(query)


I'd prefer not to use recursion here to avoid any issues with Python stack limits in case of weird behavior. The logic is potentially easier to follow that way too.

simon-mo · 2020-11-16T17:31:49Z

python/ray/serve/api.py

-            router_chosen,
-            endpoint_name,
-        )
+        if endpoint_name not in self._handle_cache:


Good catch. I also added a comment in the constructor.

simon-mo · 2020-11-16T17:32:19Z

python/ray/serve/endpoint_policy.py


 import numpy as np

 from ray.serve.utils import logger


+@lru_cache(maxsize=128)


stdlib is 10x faster than numpy.

simon-mo · 2020-11-16T17:34:17Z

python/ray/serve/endpoint_policy.py


 import numpy as np

 from ray.serve.utils import logger


+@lru_cache(maxsize=128)


and this generally task ~10us in cache miss

simon-mo · 2020-11-16T17:35:43Z

python/ray/serve/endpoint_policy.py

-        from the endpoint queue and pushing them onto a backend queue. The
-        method must also return a set of all backend tags so that the caller
-        knows which backend_queues to flush.
+    def flush(self, query) -> List[str]:


yeah i changed it to assign. good catch!

simon-mo · 2020-11-16T17:45:56Z

python/ray/serve/handle.py

+            asyncio.run_coroutine_threadsafe(
+                self.router.setup_in_async_loop(),
+                self.async_loop,


create_task isn't thread safe :-)

simon-mo · 2020-11-16T17:50:09Z

python/ray/serve/router.py

+    """Data structure represents a set of replica actor handles"""
+
+    def __init__(self):
+        # The initialization setup before we receive them from long polling


removed. the NOTE below already explained it.

simon-mo · 2020-11-16T18:11:23Z

python/ray/serve/handle.py

+
+def create_or_get_async_loop_in_thread():
+    global global_async_loop
+    if global_async_loop is None:


fyi the resetting the default event loop policy is gone (probably some false positive during testing and debugging)

edoakes

LGTM

edoakes · 2020-11-16T18:25:04Z

python/ray/serve/handle.py

+        assert self.sync, "handle.remote() should be called from sync handle."
+        coro = self._remote(request_data, kwargs)
+        future = asyncio.run_coroutine_threadsafe(coro, self.async_loop)
+        return future.result()


Hmm, this doesn't seem like the interface we want -- this will throw an exception if the future's result isn't yet availabe

Actually, this future is a concurrent.futures.Future object, which will block until the result is ready when you call future.result().

Wow, that's quite confusing...

edoakes · 2020-11-16T18:52:05Z

@simon-mo please add a description to the PR before merging.

Pass test_api

e1d49ad

edoakes reviewed Nov 16, 2020

View reviewed changes

edoakes self-assigned this Nov 16, 2020

edoakes reviewed Nov 16, 2020

View reviewed changes

Address round of comments

5b8417e

simon-mo commented Nov 16, 2020

View reviewed changes

edoakes approved these changes Nov 16, 2020

View reviewed changes

edoakes reviewed Nov 16, 2020

View reviewed changes

simon-mo added 3 commits November 16, 2020 19:26

Fix tests

8202d8e

Lint and fix more tests

58c30f5

Deflake tests

a9a6014

simon-mo merged commit d7c95a4 into ray-project:master Nov 17, 2020

	"""Given an arbitrary string, return a value between 0 and 1"""
	"""Given an arbitrary string, return a deterministic value between 0 and 1."""

		def flush(self, query) -> List[str]:
		"""Assign a query to a list of backends.

		# NOTE(simon): we separated this method to run all the setup in asyncio
		# loop.

	"""Data structure represents a set of replica actor handles"""
	"""Data structure representing a set of replica actor handles."""

	"""Given a query, submit it to a replica and returns the object ref.
	"""Given a query, submit it to a replica and return the object ref.

	# Maybe there exist a free replica, we just need to refresh our queries
	# Maybe there exists a free replica, we just need to refresh our query

	# All replicas are really busy, let's wait for one query to complete or
	# All replicas are really busy, wait for a query to complete or

[Serve] Rewrite Router to be Embeddable #12019

[Serve] Rewrite Router to be Embeddable #12019

Conversation

simon-mo commented Nov 14, 2020 • edited Loading

Why are these changes needed?

Related issue number

Checks

edoakes left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

edoakes left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

simon-mo Nov 16, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

edoakes commented Nov 16, 2020

simon-mo commented Nov 14, 2020 •

edited

Loading

simon-mo Nov 16, 2020 •

edited

Loading