code: def test_named_actor_max_task_retries(ray_init_with_task_retry_delay):
      @ray.remote(num_cpus=0)
      class Counter:
          def __init__(self):
              self.count = 0
              self.event = asyncio.Event()
  
          def increment(self):
              self.count += 1
              self.event.set()
  
          async def wait_for_count(self, count):
              while True:
                  if self.count >= count:
                      return
                  await self.event.wait()
                  self.event.clear()
  
      @ray.remote
      class ActorToKill:
          def __init__(self, counter):
              counter.increment.remote()
  
          def run(self, counter, signal):
              counter.increment.remote()
              ray.get(signal.wait.remote())
  
      @ray.remote
      class CallingActor:
          def __init__(self):
              self.actor = ray.get_actor("a")
  
          def call_other(self, counter, signal):
              return ray.get(self.actor.run.remote(counter, signal))
  
      init_counter = Counter.remote()
      run_counter = Counter.remote()
      signal = SignalActor.remote()
  
      # Start the two actors, wait for ActorToKill's constructor to run.
      a = ActorToKill.options(
          name="a", max_restarts=-1, max_task_retries=-1
      ).remote(init_counter)
      c = CallingActor.remote()
      ray.get(init_counter.wait_for_count.remote(1), timeout=30)
  
      # Signal the CallingActor to call ActorToKill, wait for it to be running,
      # then kill ActorToKill.
      # Verify that this causes ActorToKill's constructor to run a second time
      # and the run method to begin a second time.
      ref = c.call_other.remote(run_counter, signal)
      ray.get(run_counter.wait_for_count.remote(1), timeout=30)
      ray.kill(a, no_restart=False)
      ray.get(init_counter.wait_for_count.remote(2), timeout=30)
      ray.get(run_counter.wait_for_count.remote(2), timeout=30)
  
      # Signal the run method to finish, verify that the CallingActor returns.
      signal.send.remote()
      ray.get(ref, timeout=30)
  
imports: import atexit
  import asyncio
  import collections
  import numpy
  import os
  import pytest
  import signal
  import sys
  import time
  import ray
  from ray.actor import exit_actor
  import ray.cluster_utils
  from ray._private.test_utils import wait_for_condition, wait_for_pid_to_exit, generate_system_config_map, SignalActor
label_desc: Use image node_id 5 for calling a global function with example usage: test_named_actor_max_task_retries(ray_init_with_task_retry_delay) and returns: ray