code: def test_many_futures(ray_start_4_cpus, results_per_wait):
      """Schedule 500 succeeding and failing futures.
  
      Check that the callbacks get triggered correctly, independent of the number
      of results we await per call to RayEventManager.wait().
      """
      num_futures = 500
  
      event_manager = RayEventManager()
  
      seen_results = set()
      seen_errors = set()
  
      def on_result(result: Any):
          seen_results.add(result)
  
      def on_error(exception: RayTaskError):
          seen_errors.add(exception.cause.args[0])
  
      for i in range(num_futures):
          event_manager.track_futures(
              [
                  succeeding.remote("a" + str(i)),
                  failing.remote(RuntimeError, "b" + str(i)),
              ],
              on_result=on_result,
              on_error=on_error,
          )
  
      while event_manager.num_futures > 0:
          event_manager.wait(num_results=results_per_wait)
  
      for i in range(num_futures):
          assert "a" + str(i) in seen_results
          assert "b" + str(i) in seen_errors
  
file_path: ray/python/ray/air/tests/execution/test_event_manager.py
imports: import time
  from typing import Any, Type
  import pytest
  import ray
  from ray.air.execution._internal import Barrier
  from ray.air.execution._internal.event_manager import RayEventManager
  from ray.exceptions import RayTaskError
label_desc: Use image node_id 9 for calling a global function with example usage: test_many_futures(ray_start_4_cpus, results_per_wait) without return types