code: def test_callbacks(
      pool_4_processes, pool_4_processes_python_multiprocessing_lib
  ):
      Queue = (
          JoinableQueue
          if platform.system() == "Windows"
          else queue.Queue
      )
      callback_queue = Queue()
  
      def callback(result):
          callback_queue.put(result)
  
      def error_callback(error):
          callback_queue.put(error)
  
      # Will not error, check that callback is called.
      result = pool_4_processes.apply_async(
          callback_test_helper, ((0, [1]),), callback=callback
      )
      assert callback_queue.get() == 0
      result.get()
  
      # Will error, check that error_callback is called.
      result = pool_4_processes.apply_async(
          callback_test_helper,
          ((0, [0]),),
          error_callback=error_callback,
      )
      assert isinstance(callback_queue.get(), Exception)
      with pytest.raises(Exception, match="intentional failure"):
          result.get()
  
      # Ensure Ray's map_async behavior matches Multiprocessing's map_async
      process_pools = [
          pool_4_processes,
          pool_4_processes_python_multiprocessing_lib,
      ]
  
      for process_pool in process_pools:
          # Test error callbacks for map_async.
          test_callback_types = ["regular callback", "error callback"]
  
          for callback_type in test_callback_types:
              # Reinitialize queue to track number of callback calls made by
              # the current process_pool and callback_type in map_async
              callback_queue = Queue()
  
              indices, error_indices = list(range(100)), []
              if callback_type == "error callback":
                  error_indices = [2, 50, 98]
              result = process_pool.map_async(
                  callback_test_helper,
                  [(index, error_indices) for index in indices],
                  callback=callback,
                  error_callback=error_callback,
              )
              callback_results = None
              result.wait()
  
              callback_results = callback_queue.get()
              callback_queue.task_done()
  
              # Ensure that callback or error_callback was called only once
              assert callback_queue.qsize() == 0
  
              if callback_type == "regular callback":
                  assert result.successful()
              else:
                  assert not result.successful()
  
              if callback_type == "regular callback":
                  # Check that regular callback returned a list of all indices
                  for index in callback_results:
                      assert index in indices
                      indices.remove(index)
                  assert len(indices) == 0
              else:
                  # Check that error callback returned a single exception
                  assert isinstance(callback_results, Exception)
  
imports: import os
  import sys
  import platform
  import pytest
  import tempfile
  import time
  import random
  from collections import defaultdict
  import queue
  import math
  import ray
  from ray._private.test_utils import SignalActor
  from ray.util.multiprocessing import Pool, TimeoutError, JoinableQueue
  from ray.util.joblib import register_ray
  from joblib import parallel_backend, Parallel, delayed
label_desc: Use image node_id 17 for calling a global function with example usage: test_callbacks(pool_4_processes, pool_4_processes_python_multiprocessing_lib) without return types