code: def test_start_many_actors(ray_start_4_cpus, resource_manager_cls):
      """Test that starting more actors than fit onto the cluster works.
  
      - Request 10 actors
      - 4 can be started. Assert they are started
      - Stop 2
      - Assert 2 are stopped and 2 new ones are started
      """
      actor_manager = RayActorManager(
          resource_manager=resource_manager_cls()
      )
  
      running_actors = []
      # stats keeps track of started/stopped actors
      stats = Counter()
  
      def start_callback(tracked_actor):
          running_actors.append(tracked_actor)
          stats["started"] += 1
  
      def stop_callback(tracked_actor):
          running_actors.remove(tracked_actor)
          stats["stopped"] += 1
  
      # start 10 actors
      expected_actors = []
      for i in range(10):
          tracked_actor = actor_manager.add_actor(
              cls=Actor,
              kwargs={"key": "val"},
              resource_request=ResourceRequest([{"CPU": 1}]),
              on_start=start_callback,
              on_stop=stop_callback,
              on_error=_raise(Failed),
          )
          expected_actors.append(tracked_actor)
  
      # wait for some actor starts
      for i in range(4):
          actor_manager.next()
  
      # we should now have 4 started actors
      assert stats["started"] == 4
      assert stats["stopped"] == 0
      assert len(running_actors) == 4
      assert set(running_actors) == set(expected_actors[:4])
  
      # stop 2 actors
      actor_manager.remove_actor(running_actors[0])
      actor_manager.remove_actor(running_actors[1])
  
      # Wait four times, twice for termination, twice for start
      for i in range(4):
          actor_manager.next()
  
      # we should have 4 running actors, 6 started and 2 stopped
      assert stats["started"] == 6
      assert stats["stopped"] == 2
      assert len(running_actors) == 4
  
imports: import threading
  import time
  from collections import Counter
  import gc
  from typing import Any, Optional, Type
  import pytest
  import ray
  from ray.air import ResourceRequest
  from ray.air.execution import FixedResourceManager, PlacementGroupResourceManager
  from ray.air.execution._internal import Barrier
  from ray.air.execution._internal.actor_manager import RayActorManager
label_desc: Use image node_id 6 for calling a global function with example usage: test_start_many_actors(ray_start_4_cpus, resource_manager_cls) without return types