up

Signed-off-by: Ubuntu <[email protected]>
ray-project · Sep 18, 2024 · 6b66049 · 6b66049
1 parent d060ec2
commit 6b66049
Showing 1 changed file with 4 additions and 4 deletions.
diff --git a/python/ray/dag/compiled_dag_node.py b/python/ray/dag/compiled_dag_node.py
@@ -206,7 +206,7 @@ def do_stream_tasks(
             if done:
                 break
             for operation in schedule:
-                done = tasks[operation.exec_task_idx].exec_operation(
+                done = tasks[operation.exec_task_idx].stream_operation(
                     self, operation, exec_stream
                 )
                 if done:
@@ -264,7 +264,7 @@ def do_profile_stream_tasks(
                 if done:
                     break
                 for operation in schedule:
-                    done = tasks[operation.exec_task_idx].exec_operation(
+                    done = tasks[operation.exec_task_idx].stream_operation(
                         self, operation, exec_stream
                     )
                     if done:
@@ -613,7 +613,7 @@ def set_stream_buffer(self, op: _DAGNodeOperation, data: Any):
             op: The operation that generates the intermediate result.
             data: The intermediate result of a READ or COMPUTE operation.
         """
-        self._stream_buffer[op.next_operation] = data
+        self._stream_buffer[op.next_operation()] = data
 
     def reset_stream_buffer(self, op: _DAGNodeOperation) -> Any:
         """
@@ -708,7 +708,7 @@ def _stream_write(self, op: _DAGNodeOperation) -> bool:
         Returns:
             True if system error occurs and exit the loop; otherwise, False.
         """
-        output_val, exec_event = self.reset_intermediate_buffer(op)
+        output_val, exec_event = self.reset_stream_buffer(op)
         exit = False
         exec_event.synchronize()
         try: