code: def do_exec_compiled_task(
      self,
      inputs: List[Union[Any, Channel]],
      actor_method_name: str,
  ) -> None:
      """Generic actor method to begin executing a compiled DAG. This runs an
      infinite loop to repeatedly read input channel(s), execute the given
      method, and write output channel(s). It only exits if the actor dies or an
      exception is thrown.
  
      Args:
          inputs: The arguments to the task. Arguments that are not Channels will
              get passed through to the actor method. If the argument is a channel,
              it will be replaced by the value read from the channel before the
              method execute.
          actor_method_name: The name of the actual actor method to execute in
              the loop.
      """
      self._dag_cancelled = False
  
      try:
          self._input_channels = [
              i for i in inputs if isinstance(i, Channel)
          ]
          method = getattr(self, actor_method_name)
  
          resolved_inputs = []
          input_channel_idxs = []
          # Add placeholders for input channels.
          for idx, inp in enumerate(inputs):
              if isinstance(inp, Channel):
                  input_channel_idxs.append((idx, inp))
                  resolved_inputs.append(None)
              else:
                  resolved_inputs.append(inp)
  
          while True:
              for idx, channel in input_channel_idxs:
                  resolved_inputs[idx] = channel.begin_read()
  
              try:
                  output_val = method(*resolved_inputs)
              except Exception as exc:
                  backtrace = ray._private.utils.format_error_message(
                      "".join(
                          traceback.format_exception(
                              type(exc), exc, exc.__traceback__
                          )
                      ),
                      task_exception=True,
                  )
                  wrapped = RayTaskError(
                      function_name="do_exec_compiled_task",
                      traceback_str=backtrace,
                      cause=exc,
                  )
                  self._output_channel.write(wrapped)
              else:
                  if self._dag_cancelled:
                      raise RuntimeError("DAG execution cancelled")
                  self._output_channel.write(output_val)
  
              for _, channel in input_channel_idxs:
                  channel.end_read()
  
      except Exception:
          logging.exception("Compiled DAG task exited with exception")
          raise
  
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 2 for calling a global function with example usage: do_exec_compiled_task(inputs, actor_method_name) without return types