code: def get_avail_mem_per_ray_worker_node(
      spark,
      object_store_memory_per_node,
      num_cpus_per_node,
      num_gpus_per_node,
  ):
      """
      Return the available heap memory and object store memory for each ray worker,
      and error / warning message if it has.
      Return value is a tuple of
      (ray_worker_node_heap_mem_bytes, ray_worker_node_object_store_bytes,
       error_message, warning_message)
      NB: We have one ray node per spark task.
      """
  
      def mapper(_):
          try:
              return _get_avail_mem_per_ray_worker_node(
                  num_cpus_per_node,
                  num_gpus_per_node,
                  object_store_memory_per_node,
              )
          except Exception as e:
              import traceback
  
              trace_msg = "\n".join(
                  traceback.format_tb(e.__traceback__)
              )
              return -1, -1, repr(e) + trace_msg, None
  
      # Running memory inference routine on spark executor side since the spark worker
      # nodes may have a different machine configuration compared to the spark driver
      # node.
      (
          inferred_ray_worker_node_heap_mem_bytes,
          inferred_ray_worker_node_object_store_bytes,
          err,
          warning_msg,
      ) = (
          spark.sparkContext.parallelize([1], 1)
          .map(mapper)
          .collect()[0]
      )
  
      if err is not None:
          raise RuntimeError(
              f"Inferring ray worker node available memory failed, error: {err}. "
              "You can bypass this error by setting following spark configs: "
              "spark.executorEnv.RAY_ON_SPARK_WORKER_CPU_CORES, "
              "spark.executorEnv.RAY_ON_SPARK_WORKER_GPU_NUM, "
              "spark.executorEnv.RAY_ON_SPARK_WORKER_PHYSICAL_MEMORY_BYTES, "
              "spark.executorEnv.RAY_ON_SPARK_WORKER_SHARED_MEMORY_BYTES."
          )
      if warning_msg is not None:
          _logger.warning(warning_msg)
      return (
          inferred_ray_worker_node_heap_mem_bytes,
          inferred_ray_worker_node_object_store_bytes,
      )
  
imports: import subprocess
  import os
  import sys
  import random
  import threading
  import collections
  import logging
  import time
label_desc: Use image node_id 18 for calling a global function with example usage: get_avail_mem_per_ray_worker_node(spark, object_store_memory_per_node, num_cpus_per_node, num_gpus_per_node) and returns: inferred_ray_worker_node_heap_mem_bytes,  inferred_ray_worker_node_object_store_bytes, _get_avail_mem_per_ray_worker_node, unknown,  unknown,  unknown,  None