code: def test_dag_fault_tolerance(ray_start_regular_shared):
      actors = [
          Actor.remote(0, fail_after=100, sys_exit=False)
          for _ in range(4)
      ]
      with InputNode() as i:
          out = [a.inc.bind(i) for a in actors]
          dag = MultiOutputNode(out)
  
      compiled_dag = dag.experimental_compile()
  
      for i in range(99):
          output_channels = compiled_dag.execute(1)
          # TODO(swang): Replace with fake ObjectRef.
          results = [chan.begin_read() for chan in output_channels]
          assert results == [i + 1] * 4
          for chan in output_channels:
              chan.end_read()
  
      with pytest.raises(ValueError):
          for i in range(99):
              output_channels = compiled_dag.execute(1)
              for chan in output_channels:
                  chan.begin_read()
              for chan in output_channels:
                  chan.end_read()
  
      compiled_dag.teardown()
  
imports: import logging
  import os
  import random
  import sys
  import time
  import pytest
  import ray
  import ray.cluster_utils
  from ray.exceptions import RaySystemError
  from ray.dag import InputNode, MultiOutputNode
  from ray.tests.conftest import *
label_desc: Use image node_id 7 for calling a global function with example usage: test_dag_fault_tolerance(ray_start_regular_shared) without return types