Revert "[Serve] Fix ServeHandle serialization (ray-project#13695)" (ray-project#13753)

simon-mo · fishbone · commit 405c8d779452 · 2021-02-15T20:23:43.000-08:00
This reverts commit 202fbdf.
diff --git a/python/ray/serve/api.py b/python/ray/serve/api.py
@@ -66,8 +66,6 @@ def check(self, *args, **kwargs):
 
 class ThreadProxiedRouter:
     def __init__(self, controller_handle, sync: bool):
-        self.controller_handle = controller_handle
-        self.sync = sync
         self.router = Router(controller_handle)
 
         if sync:
@@ -94,11 +92,6 @@ def _remote(self, endpoint_name, handle_options, request_data,
                                           **kwargs)
         return coro
 
-    def __reduce__(self):
-        deserializer = ThreadProxiedRouter
-        serialized_data = (self.controller_handle, self.sync)
-        return deserializer, serialized_data
-
 
 class Client:
     def __init__(self,
diff --git a/python/ray/serve/handle.py b/python/ray/serve/handle.py
@@ -4,6 +4,8 @@
 from typing import Any, Dict, Optional, Union
 from enum import Enum
 
+from ray.serve.router import Router
+
 
 @dataclass(frozen=True)
 class HandleOptions:
@@ -38,11 +40,10 @@ class RayServeHandle:
        # raises RayTaskError Exception
     """
 
-    def __init__(
-            self,
-            router,  # ThreadProxiedRouter
-            endpoint_name,
-            handle_options: Optional[HandleOptions] = None):
+    def __init__(self,
+                 router: Router,
+                 endpoint_name,
+                 handle_options: Optional[HandleOptions] = None):
         self.router = router
         self.endpoint_name = endpoint_name
         self.handle_options = handle_options or HandleOptions()
@@ -77,7 +78,7 @@ def options(self,
     async def remote(self,
                      request_data: Optional[Union[Dict, Any]] = None,
                      **kwargs):
-        """Issue an asynchronous request to the endpoint.
+        """Issue an asynchrounous request to the endpoint.
 
         Returns a Ray ObjectRef whose results can be waited for or retrieved
         using ray.wait or ray.get (or ``await object_ref``), respectively.
@@ -97,12 +98,6 @@ async def remote(self,
     def __repr__(self):
         return f"{self.__class__.__name__}(endpoint='{self.endpoint_name}')"
 
-    def __reduce__(self):
-        deserializer = RayServeHandle
-        serialized_data = (self.router, self.endpoint_name,
-                           self.handle_options)
-        return deserializer, serialized_data
-
 
 class RayServeSyncHandle(RayServeHandle):
     def remote(self, request_data: Optional[Union[Dict, Any]] = None,
@@ -128,9 +123,3 @@ def remote(self, request_data: Optional[Union[Dict, Any]] = None,
         future: concurrent.futures.Future = asyncio.run_coroutine_threadsafe(
             coro, self.router.async_loop)
         return future.result()
-
-    def __reduce__(self):
-        deserializer = RayServeSyncHandle
-        serialized_data = (self.router, self.endpoint_name,
-                           self.handle_options)
-        return deserializer, serialized_data
diff --git a/python/ray/serve/tests/test_handle.py b/python/ray/serve/tests/test_handle.py
@@ -1,51 +1,9 @@
 import requests
-import pytest
+
 import ray
 from ray import serve
 
 
-@pytest.mark.asyncio
-async def test_async_handle_serializable(serve_instance):
-    client = serve_instance
-
-    def f(_):
-        return "hello"
-
-    client.create_backend("f", f)
-    client.create_endpoint("f", backend="f")
-
-    @ray.remote
-    class TaskActor:
-        async def task(self, handle):
-            ref = await handle.remote()
-            output = await ref
-            return output
-
-    handle = client.get_handle("f", sync=False)
-
-    task_actor = TaskActor.remote()
-    result = await task_actor.task.remote(handle)
-    assert result == "hello"
-
-
-def test_sync_handle_serializable(serve_instance):
-    client = serve_instance
-
-    def f(_):
-        return "hello"
-
-    client.create_backend("f", f)
-    client.create_endpoint("f", backend="f")
-
-    @ray.remote
-    def task(handle):
-        return ray.get(handle.remote())
-
-    handle = client.get_handle("f", sync=True)
-    result_ref = task.remote(handle)
-    assert ray.get(result_ref) == "hello"
-
-
 def test_handle_in_endpoint(serve_instance):
     client = serve_instance
 
