[serve] Double Serialization Optimization (#4532)

python/ray/experimental/serve/router/routers.py

@@ -124,8 +124,14 @@ class DeadlineAwareRouter:
             ACTOR_NOT_REGISTERED_MSG(actor_name))
 
         result_object_id = get_new_oid()
+
+        # Here, 'data_object_id' is either an ObjectID or an actual object.
+        # When it is an ObjectID, this is an optimization to avoid creating
+        # an extra copy of 'data' in the object store.
+        data_object_id = ray.worker.global_worker._current_task.arguments()[1]
+
         self.query_queues[actor_name].push(
-            SingleQuery(data, result_object_id, deadline_s))
+            SingleQuery(data_object_id, result_object_id, deadline_s))
 
         return [result_object_id]