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","."]
code: def __init__(self, buffer_size_bytes: Optional[int]):
      self._buffer_size_bytes: Optional[int] = buffer_size_bytes
      if self._buffer_size_bytes is None:
          self._buffer_size_bytes = MAX_BUFFER_SIZE
      if (
          not isinstance(self._buffer_size_bytes, int)
          or self._buffer_size_bytes <= 0
      ):
          raise ValueError(
              "`buffer_size_bytes` must be a positive integer, found "
              f"{self._buffer_size_bytes}"
          )
  
      # idx -> CompiledTask.
      self.idx_to_task: Dict[int, "CompiledTask"] = {}
      # DAGNode -> idx.
      self.dag_node_to_idx: Dict["ray.dag.DAGNode", int] = {}
      # idx counter.
      self.counter: int = 0
  
      # Attributes that are set during preprocessing.
      # Preprocessing identifies the input node and output node.
      self.input_task_idx: Optional[int] = None
      self.output_task_idx: Optional[int] = None
      self.has_single_output: bool = False
      self.actor_task_count: Dict[
          "ray._raylet.ActorID", int
      ] = defaultdict(int)
  
      # Cached attributes that are set during compilation.
      self.dag_input_channel: Optional[Channel] = None
      self.dag_output_channels: Optional[Channel] = None
      # ObjectRef for each worker's task. The task is an infinite loop that
      # repeatedly executes the method specified in the DAG.
      self.worker_task_refs: List["ray.ObjectRef"] = []
      # Set of actors present in the DAG.
      self.actor_refs = set()
  
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 1 to create a new CompiledDAG object  with example: obj = CompiledDAG(buffer_size_bytes)