code: def test_threaded_actor_creation_and_kill(ray_start_cluster):
      """Test the scenario where the threaded actors are created and killed."""
      cluster = ray_start_cluster
      NUM_CPUS_PER_NODE = 3
      NUM_NODES = 2
      for _ in range(NUM_NODES):
          cluster.add_node(num_cpus=NUM_CPUS_PER_NODE)
      ray.init(address=cluster.address)
  
      @ray.remote(num_cpus=0)
      class ThreadedActor:
          def __init__(self):
              self.received = []
              self.lock = threading.Lock()
  
          def add(self, seqno):
              time.sleep(1)
              with self.lock:
                  self.received.append(seqno)
  
          def get_all(self):
              with self.lock:
                  return self.received
  
          def ready(self):
              pass
  
          def terminate(self):
              ray.actor.exit_actor()
  
      # - Create threaded actors
      # - Submit many tasks.
      # - Ungracefully kill them in the middle.
      for _ in range(10):
          actors = [
              ThreadedActor.options(max_concurrency=10).remote()
              for _ in range(NUM_NODES * NUM_CPUS_PER_NODE)
          ]
          ray.get([actor.ready.remote() for actor in actors])
  
          for _ in range(10):
              for actor in actors:
                  actor.add.remote(1)
          time.sleep(0.5)
          for actor in actors:
              ray.kill(actor)
      ensure_cpu_returned(NUM_NODES * NUM_CPUS_PER_NODE)
  
      # - Create threaded actors
      # - Submit many tasks.
      # - Gracefully kill them in the middle.
      for _ in range(10):
          actors = [
              ThreadedActor.options(max_concurrency=10).remote()
              for _ in range(NUM_NODES * NUM_CPUS_PER_NODE)
          ]
          ray.get([actor.ready.remote() for actor in actors])
          for _ in range(10):
              for actor in actors:
                  actor.add.remote(1)
  
          time.sleep(0.5)
          for actor in actors:
              actor.terminate.remote()
      ensure_cpu_returned(NUM_NODES * NUM_CPUS_PER_NODE)
  
file_path: ray/python/ray/tests/test_threaded_actor.py
imports: import sys
  import threading
  import time
  import numpy
  import pytest
  import ray
  import ray._private.test_utils
  from ray._private.state import available_resources
label_desc: Use image node_id 4 for calling a global function with example usage: test_threaded_actor_creation_and_kill(ray_start_cluster) and returns: self