[core][experimental] AccDag: Support more than one tasks per actor (#…

…44259)

Accelerated Dag currently does not support multiple methods within the same actor. This PR adds the support.

In this PR, the methods from the same actor are first grouped together, and then executed in the same execution loop based on their submission (binding) order.
Related issue number

Closes #44194

---------

Signed-off-by: Rui Qiao <[email protected]>

python/ray/actor.py

@@ -237,6 +237,7 @@ def _bind(
         _generator_backpressure_num_objects=None,
     ):
         from ray.dag.class_node import (
+            BIND_INDEX_KEY,
             PARENT_CLASS_NODE_KEY,
             PREV_CLASS_METHOD_CALL_KEY,
             ClassMethodNode,
@@ -259,7 +260,9 @@ def _bind(
         other_args_to_resolve = {
             PARENT_CLASS_NODE_KEY: actor,
             PREV_CLASS_METHOD_CALL_KEY: None,
+            BIND_INDEX_KEY: actor._ray_dag_bind_index,
         }
+        actor._ray_dag_bind_index += 1
 
         node = ClassMethodNode(
             self._method_name,
@@ -1306,6 +1309,11 @@ def __init__(
             actor_creation_function_descriptor
         )
         self._ray_function_descriptor = {}
+        # This is incremented each time `bind()` is called on an actor handle
+        # (in Ray DAGs), therefore capturing the bind order of the actor methods.
+        # TODO: this does not work properly if the caller has two copies of the
+        # same actor handle, and needs to be fixed.
+        self._ray_dag_bind_index = 0
 
         if not self._ray_is_cross_language:
             assert isinstance(

python/ray/dag/class_node.py

@@ -4,7 +4,11 @@
 from ray.dag.dag_node import DAGNode
 from ray.dag.input_node import InputNode
 from ray.dag.format_utils import get_dag_node_str
-from ray.dag.constants import PARENT_CLASS_NODE_KEY, PREV_CLASS_METHOD_CALL_KEY
+from ray.dag.constants import (
+    PARENT_CLASS_NODE_KEY,
+    PREV_CLASS_METHOD_CALL_KEY,
+    BIND_INDEX_KEY,
+)
 from ray.util.annotations import DeveloperAPI
 
 from typing import Any, Dict, List, Union, Tuple, Optional
@@ -156,6 +160,10 @@ def __init__(
         self._prev_class_method_call: Optional[
             ClassMethodNode
         ] = other_args_to_resolve.get(PREV_CLASS_METHOD_CALL_KEY, None)
+        # The index/order when bind() is called on this class method
+        self._bind_index: Optional[int] = other_args_to_resolve.get(
+            BIND_INDEX_KEY, None
+        )
 
         # The actor creation task dependency is encoded as the first argument,
         # and the ordering dependency as the second, which ensures they are
@@ -203,6 +211,9 @@ def __str__(self) -> str:
     def get_method_name(self) -> str:
         return self._method_name
 
+    def _get_bind_index(self) -> int:
+        return self._bind_index
+
     def _get_remote_method(self, method_name):
         method_body = getattr(self._parent_class_node, method_name)
         return method_body