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 _monitor_failures(self):
      outer = self
  
      class Monitor(threading.Thread):
          def __init__(self):
              super().__init__(daemon=True)
              self.in_teardown = False
  
          def teardown(self):
              if self.in_teardown:
                  return
              logger.info("Tearing down compiled DAG")
              self.in_teardown = True
              for actor in outer.actor_refs:
                  logger.info(
                      f"Cancelling compiled worker on actor: {actor}"
                  )
                  try:
                      ray.get(
                          actor.__ray_call__.remote(
                              do_cancel_compiled_task
                          )
                      )
                  except Exception:
                      logger.exception("Error cancelling worker task")
                      pass
              logger.info("Waiting for worker tasks to exit")
              for ref in outer.worker_task_refs:
                  try:
                      ray.get(ref)
                  except Exception:
                      pass
              logger.info("Teardown complete")
  
          def run(self):
              try:
                  ray.get(outer.worker_task_refs)
              except Exception as e:
                  logger.debug(
                      f"Handling exception from worker tasks: {e}"
                  )
                  if self.in_teardown:
                      return
                  if isinstance(outer.dag_output_channels, list):
                      for output_channel in outer.dag_output_channels:
                          output_channel.close()
                  else:
                      outer.dag_output_channels.close()
                  self.teardown()
  
      monitor = Monitor()
      monitor.start()
      return 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 5 for calling the CompiledDAG obj's underlying member method code with example usage: obj._monitor_failures() and returns: monitor