code: def test_resolve_many(
      ray_start_4_cpus, resource_manager_cls, num_tasks
  ):
      """Schedule ``num_tasks`` tasks and wait until ``wait_for_events`` of them resolve.
  
      Every resolved task will increase a counter by its return value (1).
      """
      actor_manager = RayActorManager(
          resource_manager=resource_manager_cls()
      )
  
      seen = {"data": 0}
  
      def result_callback(tracked_actor, result):
          seen["data"] += result
  
      tracked_actor = actor_manager.add_actor(
          cls=Actor,
          kwargs={},
          resource_request=ResourceRequest([{"CPU": 4}]),
      )
      actor_manager.next()
  
      for i in range(num_tasks):
          actor_manager.schedule_actor_task(
              tracked_actor,
              "foo",
              (1, False),
              on_result=result_callback,
          )
  
      for i in range(num_tasks):
          actor_manager.next()
          assert seen["data"] == i + 1
  
file_path: ray/python/ray/air/tests/execution/test_tracked_actor_task.py
imports: from collections import Counter
  import pytest
  import ray
  from ray.air import ResourceRequest
  from ray.air.execution import FixedResourceManager, PlacementGroupResourceManager
  from ray.air.execution._internal.actor_manager import RayActorManager
label_desc: Use image node_id 3 for calling a global function with example usage: test_resolve_many(ray_start_4_cpus, resource_manager_cls, num_tasks) without return types