class_docstr: Experimental class for accelerated execution.
  
  This class should not be called directly. Instead, create
  a ray.dag and call experimental_compile().
  
  See REP https://github.com/ray-project/enhancements/pull/48 for more
  information.
class_docstr_tok: ["Experimental","class","for","accelerated","execution",".","This","class","should","not","be","called","directly",".","Instead",",","create","a","ray.dag","and","call","experimental_compile","(",")",".","See","REP","https",":","\/\/github.com\/ray-project\/enhancements\/pull\/48","for","more","information","."]
docstr: Compile an execution path. This allocates channels for adjacent
  tasks to send/receive values. An infinite task is submitted to each
  actor in the DAG that repeatedly receives from input channel(s) and
  sends to output channel(s).
  
  This function is idempotent and will cache the previously allocated
  channels.
  
  Returns:
      A tuple of (input channel, output channel(s)). The input channel
      that should be used by the caller to submit a DAG execution. The
      output channel(s) should be read by the caller to get the DAG
      output.
docstr_tok: ["Compile","an","execution","path",".","This","allocates","channels","for","adjacent","tasks","to","send\/receive","values",".","An","infinite","task","is","submitted","to","each","actor","in","the","DAG","that","repeatedly","receives","from","input","channel","(","s",")","and","sends","to","output","channel","(","s",")",".","This","function","is","idempotent","and","will","cache","the","previously","allocated","channels",".","Returns",":","A","tuple","of","(","input","channel",",","output","channel","(","s",")",")",".","The","input","channel","that","should","be","used","by","the","caller","to","submit","a","DAG","execution",".","The","output","channel","(","s",")","should","be","read","by","the","caller","to","get","the","DAG","output","."]
code: def _get_or_compile(
      self,
  ) -> Tuple[Channel, Union[Channel, List[Channel]]]:
      """Compile an execution path. This allocates channels for adjacent
      tasks to send/receive values. An infinite task is submitted to each
      actor in the DAG that repeatedly receives from input channel(s) and
      sends to output channel(s).
  
      This function is idempotent and will cache the previously allocated
      channels.
  
      Returns:
          A tuple of (input channel, output channel(s)). The input channel
          that should be used by the caller to submit a DAG execution. The
          output channel(s) should be read by the caller to get the DAG
          output.
      """
      from ray.dag import (
          DAGNode,
          InputNode,
          MultiOutputNode,
          ClassMethodNode,
      )
  
      if self.input_task_idx is None:
          self._preprocess()
  
      if self.dag_input_channel is not None:
          assert self.dag_output_channels is not None
          return (
              self.dag_input_channel,
              self.dag_output_channels,
          )
  
      queue = [self.input_task_idx]
      visited = set()
      # Create output buffers
      while queue:
          cur_idx = queue.pop(0)
          if cur_idx in visited:
              continue
          visited.add(cur_idx)
  
          task = self.idx_to_task[cur_idx]
          # Create an output buffer on the actor.
          assert task.output_channel is None
          if isinstance(task.dag_node, ClassMethodNode):
              fn = task.dag_node._get_remote_method("__ray_call__")
              task.output_channel = ray.get(
                  fn.remote(
                      do_allocate_channel,
                      buffer_size_bytes=self._buffer_size_bytes,
                      num_readers=task.num_readers,
                  )
              )
              self.actor_refs.add(task.dag_node._get_actor_handle())
          elif isinstance(task.dag_node, InputNode):
              task.output_channel = Channel(
                  buffer_size_bytes=self._buffer_size_bytes,
                  num_readers=task.num_readers,
              )
          else:
              assert isinstance(task.dag_node, MultiOutputNode)
  
          for idx in task.downstream_node_idxs:
              queue.append(idx)
  
      for node_idx, task in self.idx_to_task.items():
          if node_idx == self.input_task_idx:
              # We don't need to assign an actual task for the input node.
              continue
  
          if node_idx == self.output_task_idx:
              # We don't need to assign an actual task for the input node.
              continue
  
          resolved_args = []
          has_at_least_one_channel_input = False
          for arg in task.args:
              if isinstance(arg, DAGNode):
                  arg_idx = self.dag_node_to_idx[arg]
                  arg_channel = self.idx_to_task[arg_idx].output_channel
                  assert arg_channel is not None
                  resolved_args.append(arg_channel)
                  has_at_least_one_channel_input = True
              else:
                  resolved_args.append(arg)
          # TODO: Support no-input DAGs (use an empty object to signal).
          if not has_at_least_one_channel_input:
              raise ValueError(
                  "Compiled DAGs require each task to take a "
                  "ray.dag.InputNode or at least one other DAGNode as an "
                  "input"
              )
  
          # Assign the task with the correct input and output buffers.
          worker_fn = task.dag_node._get_remote_method("__ray_call__")
          self.worker_task_refs.append(
              worker_fn.options(concurrency_group="_ray_system").remote(
                  do_exec_compiled_task,
                  resolved_args,
                  task.dag_node.get_method_name(),
              )
          )
  
      self.dag_input_channel = self.idx_to_task[
          self.input_task_idx
      ].output_channel
  
      self.dag_output_channels = []
      for output in self.idx_to_task[self.output_task_idx].args:
          assert isinstance(output, DAGNode)
          output_idx = self.dag_node_to_idx[output]
          self.dag_output_channels.append(
              self.idx_to_task[output_idx].output_channel
          )
  
      assert self.dag_input_channel
      assert self.dag_output_channels
      assert [
          output_channel is not None
          for output_channel in self.dag_output_channels
      ]
      # If no MultiOutputNode was specified during the DAG creation, there is only
      # one output. Return a single output channel instead of a list of
      # channels.
      if self.has_single_output:
          assert len(self.dag_output_channels) == 1
          self.dag_output_channels = self.dag_output_channels[0]
  
      # Driver should ray.put on input, ray.get/release on output
      self._monitor = self._monitor_failures()
      return (
          self.dag_input_channel,
          self.dag_output_channels,
          self._monitor,
      )
  
imports: from collections import defaultdict
  from typing import Any, Dict, List, Tuple, Union, Optional
  import logging
  import threading
  import traceback
  import ray
  from ray.exceptions import RayTaskError
  from ray.experimental.channel import Channel
  from ray.util.annotations import DeveloperAPI
label_desc: Use image node_id 4 for calling the CompiledDAG obj's underlying member method code with example usage: obj._get_or_compile() and returns: self,  self,  self, self,  self