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: Before compiling, preprocess the DAG to build an index from task to
  upstream and downstream tasks, and to set the input and output node(s)
  of the DAG.
  
  This function is idempotent.
docstr_tok: ["Before","compiling",",","preprocess","the","DAG","to","build","an","index","from","task","to","upstream","and","downstream","tasks",",","and","to","set","the","input","and","output","node","(","s",")","of","the","DAG",".","This","function","is","idempotent","."]
code: def _preprocess(self) -> None:
      """Before compiling, preprocess the DAG to build an index from task to
      upstream and downstream tasks, and to set the input and output node(s)
      of the DAG.
  
      This function is idempotent.
      """
      from ray.dag import (
          DAGNode,
          ClassMethodNode,
          FunctionNode,
          InputAttributeNode,
          InputNode,
          MultiOutputNode,
      )
  
      self.input_task_idx, self.output_task_idx = None, None
      self.actor_task_count.clear()
  
      # For each task node, set its upstream and downstream task nodes.
      for idx, task in self.idx_to_task.items():
          dag_node = task.dag_node
          if not (
              isinstance(dag_node, InputNode)
              or isinstance(dag_node, MultiOutputNode)
              or isinstance(dag_node, ClassMethodNode)
          ):
              if isinstance(dag_node, InputAttributeNode):
                  # TODO(swang): Support multi args.
                  raise NotImplementedError(
                      "Compiled DAGs currently do not support kwargs or "
                      "multiple args for InputNode"
                  )
              elif isinstance(dag_node, FunctionNode):
                  # TODO(swang): Support non-actor tasks.
                  raise NotImplementedError(
                      "Compiled DAGs currently only support actor method nodes"
                  )
              else:
                  raise ValueError(
                      f"Found unsupported node of type {type(task.dag_node)}"
                  )
  
          if isinstance(dag_node, ClassMethodNode):
              actor_handle = dag_node._get_actor_handle()
              if actor_handle is None:
                  raise ValueError(
                      "Compiled DAGs can only bind methods to an actor "
                      "that is already created with Actor.remote()"
                  )
              self.actor_task_count[actor_handle._actor_id] += 1
  
          for arg in task.args:
              if isinstance(arg, DAGNode):
                  arg_idx = self.dag_node_to_idx[arg]
                  self.idx_to_task[arg_idx].downstream_node_idxs.add(
                      idx
                  )
  
      for actor_id, task_count in self.actor_task_count.items():
          if task_count > 1:
              raise NotImplementedError(
                  "Compiled DAGs can contain at most one task per actor handle. "
                  f"Actor with ID {actor_id} appears {task_count}x."
              )
  
      # Find the input node to the DAG.
      for idx, task in self.idx_to_task.items():
          if isinstance(task.dag_node, InputNode):
              assert (
                  self.input_task_idx is None
              ), "more than one InputNode found"
              self.input_task_idx = idx
      # TODO: Support no-input DAGs (use an empty object to signal).
      if self.input_task_idx is None:
          raise NotImplementedError(
              "Compiled DAGs currently require exactly one InputNode"
          )
  
      # Find the (multi-)output node to the DAG.
      for idx, task in self.idx_to_task.items():
          if len(task.downstream_node_idxs) == 0:
              assert (
                  self.output_task_idx is None
              ), "More than one output node found"
              self.output_task_idx = idx
  
      assert self.output_task_idx is not None
      output_node = self.idx_to_task[self.output_task_idx].dag_node
      # Add an MultiOutputNode to the end of the DAG if it's not already there.
      if not isinstance(output_node, MultiOutputNode):
          self.has_single_output = True
          output_node = MultiOutputNode([output_node])
          self._add_node(output_node)
          self.output_task_idx = self.dag_node_to_idx[output_node]
          # Preprocess one more time so that we have the right output node
          # now.
          self._preprocess()
  
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 3 for calling the CompiledDAG obj's underlying member method code with example usage: obj._preprocess() without return types