[Serve] Use pickle instead of clouldpickle (#9479)

python/ray/serve/BUILD

@@ -90,6 +90,14 @@ py_test(
     deps = [":serve_lib"],
 )
 
+py_test(
+    name = "test_regression",
+    size = "small",
+    srcs = serve_tests_srcs,
+    tags = ["exclusive"],
+    deps = [":serve_lib"],
+)
+
 
 py_test(
     name = "test_util",

python/ray/serve/request_params.py

@@ -42,7 +42,7 @@ class RequestMetadata:
         return current_time_ms + slo_ms
 
     def ray_serialize(self):
-        return pickle.dumps(self.__dict__, protocol=5)
+        return pickle.dumps(self.__dict__)
 
     @staticmethod
     def ray_deserialize(value):

python/ray/serve/router.py

@@ -3,10 +3,10 @@ import copy
 from collections import defaultdict, deque
 import time
 from typing import DefaultDict, List
+import pickle
 
 import blist
 
-import ray.cloudpickle as pickle
 from ray.exceptions import RayTaskError
 
 import ray
@@ -50,7 +50,7 @@ class Query:
         # worker without removing async_future.
         clone = copy.copy(self).__dict__
         clone.pop("async_future")
-        return pickle.dumps(clone, protocol=5)
+        return pickle.dumps(clone)
 
     @staticmethod
     def ray_deserialize(value):

python/ray/serve/tests/test_regression.py

@@ -0,0 +1,38 @@
+import numpy as np
+import requests
+
+from ray import serve
+
+
+def test_np_in_composed_model(serve_instance):
+    # https://github.com/ray-project/ray/issues/9441
+    # AttributeError: 'bytes' object has no attribute 'readonly'
+    # in cloudpickle _from_numpy_buffer
+
+    def sum_model(_request, data=None):
+        return np.sum(data)
+
+    class ComposedModel:
+        def __init__(self):
+            self.model = serve.get_handle("sum_model")
+
+        async def __call__(self, _request):
+            data = np.ones((10, 10))
+            result = await self.model.remote(data=data)
+            return result
+
+    serve.create_backend("sum_model", sum_model)
+    serve.create_endpoint("sum_model", backend="sum_model")
+    serve.create_backend("model", ComposedModel)
+    serve.create_endpoint(
+        "model", backend="model", route="/model", methods=['GET'])
+
+    result = requests.get("http://127.0.0.1:8000/model")
+    assert result.status_code == 200
+    assert result.json() == 100.0
+
+
+if __name__ == "__main__":
+    import sys
+    import pytest
+    sys.exit(pytest.main(["-v", "-s", __file__]))