code: def test_fault_tolerance_advanced_tree(shutdown_only, death_list):
      """
      Test fault tolerance for a more complicated execution graph.
      """
      # Root should always be finish
      execution_graph = {
          "root": [
              (NORMAL_TASK, "A"),
              (ACTOR_TASK, "B"),
              (NORMAL_TASK, "C"),
              (ACTOR_TASK, "D"),
          ],
          "A": [(ACTOR_TASK, "Aa"), (NORMAL_TASK, "Ab")],
          "C": [(ACTOR_TASK, "Ca"), (NORMAL_TASK, "Cb")],
          "D": [
              (NORMAL_TASK, "Da"),
              (NORMAL_TASK, "Db"),
              (ACTOR_TASK, "Dc"),
              (ACTOR_TASK, "Dd"),
          ],
          "Aa": [],
          "Ab": [
              (ACTOR_TASK, "Aba"),
              (NORMAL_TASK, "Abb"),
              (NORMAL_TASK, "Abc"),
          ],
          "Ca": [(ACTOR_TASK, "Caa"), (NORMAL_TASK, "Cab")],
          "Abb": [(NORMAL_TASK, "Abba")],
          "Abc": [],
          "Abba": [(NORMAL_TASK, "Abbaa"), (ACTOR_TASK, "Abbab")],
          "Abbaa": [(NORMAL_TASK, "Abbaaa"), (ACTOR_TASK, "Abbaab")],
      }
  
      ray.init(_system_config=_SYSTEM_CONFIG)
  
      @ray.remote
      class Killer:
          def __init__(self, death_list):
              self.idx_ = 0
              self.death_list_ = death_list
              self.kill_started = False
              self.name_to_pids = {}
  
          async def start_killing(self):
              self.kill_started = True
  
          async def next_to_kill(self):
              while not self.kill_started:
                  await asyncio.sleep(0.5)
  
              # if no more tasks to kill - simply sleep to keep all running tasks blocked.
              while self.idx_ >= len(self.death_list_):
                  await asyncio.sleep(999)
  
              to_kill = self.death_list_[self.idx_]
              return to_kill
  
          async def report_pid(self, name, pid):
              self.name_to_pids[name] = (pid, None)
  
          async def get_pids(self):
              return self.name_to_pids
  
          async def all_killed(self):
              while self.idx_ < len(self.death_list_):
                  await asyncio.sleep(0.5)
  
          async def advance_next(self):
              self.idx_ += 1
  
      def run_children(my_name, killer, execution_graph):
          children = execution_graph.get(my_name, [])
          for task_type, child_name in children:
              if task_type == NORMAL_TASK:
                  task.options(name=child_name).remote(
                      child_name, killer, execution_graph
                  )
              else:
                  a = Actor.remote()
                  a.actor_task.options(name=child_name).remote(
                      child_name, killer, execution_graph
                  )
  
          # Block until killed
          while True:
              to_fail, fail_kind = ray.get(killer.next_to_kill.remote())
              if to_fail == my_name:
                  ray.get(killer.advance_next.remote())
                  if fail_kind == "exit_kill":
                      os._exit(1)
                  elif fail_kind == "exit_exception":
                      raise ValueError("Killed by test")
                  else:
                      assert (
                          False
                      ), f"Test invalid kill options: {fail_kind}"
              else:
                  # Sleep a bit to wait for death.
                  time.sleep(0.1)
  
      @ray.remote(max_task_retries=0, max_restarts=0)
      class Actor:
          def actor_task(self, my_name, killer, execution_graph):
              ray.get(killer.report_pid.remote(my_name, os.getpid()))
              run_children(my_name, killer, execution_graph)
  
      @ray.remote(max_retries=0)
      def task(my_name, killer, execution_graph):
          ray.get(killer.report_pid.remote(my_name, os.getpid()))
          run_children(my_name, killer, execution_graph)
  
      killer = Killer.remote(death_list)
      # Kick off the workload
      task.options(name="root").remote("root", killer, execution_graph)
  
      # Calculate all expected tasks
      tasks = []
  
      def add_task_recur(task):
          # Not all tasks are keyed in the execution graph.
          tasks.append(task)
          children = execution_graph.get(task, [])
          for _, child in children:
              add_task_recur(child)
  
      add_task_recur("root")
  
      def tasks_in_execution_graph_all_running():
          running_tasks = [
              task["name"]
              for task in list_tasks(
                  filters=[("state", "=", "RUNNING")], limit=10000
              )
          ]
          for task in tasks:
              assert task in running_tasks, f"Task {task} not running"
          print("All tasks in execution graph are running")
          return True
  
      wait_for_condition(
          tasks_in_execution_graph_all_running,
          timeout=20,
          retry_interval_ms=2000,
      )
  
      # Starts killing :O
      print("start killing")
      ray.get(killer.start_killing.remote())
      print("waiting for all killed")
      ray.get(killer.all_killed.remote())
      print("all killed")
  
      wait_for_condition(
          verify_tasks_running_or_terminated,
          task_pids=ray.get(killer.get_pids.remote()),
          expect_num_tasks=len(tasks),
          timeout=30,
          retry_interval_ms=500,
      )
  
imports: import asyncio
  from collections import defaultdict
  import os
  from typing import Dict
  import pytest
  import sys
  import time
  from ray._private import ray_constants
  from functools import reduce
  import ray
  from ray._private.state_api_test_utils import PidActor, get_state_api_manager, verify_tasks_running_or_terminated, verify_failed_task
  from ray.util.state.common import ListApiOptions, StateResource
  from ray._private.test_utils import async_wait_for_condition_async_predicate, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
  from ray.util.state import StateApiClient, list_actors, list_jobs, list_tasks
label_desc: Use image node_id 14 for calling a global function with example usage: test_fault_tolerance_advanced_tree(shutdown_only, death_list) and returns: True, to_kill, self