code: def test_task_barrier(ray_start_4_cpus):
      event_manager = RayEventManager()
  
      seen = set()
  
      def on_completion(barrier: Barrier):
          seen.update(barrier.get_results())
  
      barrier = Barrier(max_results=4, on_completion=on_completion)
  
      event_manager.track_futures(
          [
              succeeding.remote("a"),
              succeeding.remote("b"),
              succeeding.remote("c"),
              succeeding.remote("d"),
              sleeping.remote(2, "e"),
          ],
          on_result=barrier.arrive,
      )
  
      event_manager.wait(num_results=4)
  
      assert "a" in seen
      assert "b" in seen
      assert "c" in seen
      assert "d" in seen
      assert "e" not in seen
  
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 11 for calling a global function with example usage: test_task_barrier(ray_start_4_cpus) without return types