code: def _start_ray_worker_nodes(
      *,
      spark,
      spark_job_group_id,
      spark_job_group_desc,
      num_worker_nodes,
      using_stage_scheduling,
      ray_head_ip,
      ray_head_port,
      ray_temp_dir,
      num_cpus_per_node,
      num_gpus_per_node,
      heap_memory_per_node,
      object_store_memory_per_node,
      worker_node_options,
      collect_log_to_path,
      autoscale_mode,
      spark_job_server_port,
  ):
      # NB:
      # In order to start ray worker nodes on spark cluster worker machines,
      # We launch a background spark job:
      #  1. Each spark task launches one ray worker node. This design ensures all ray
      #     worker nodes have the same shape (same cpus / gpus / memory configuration).
      #     If ray worker nodes have a non-uniform shape, the Ray cluster setup will
      #     be non-deterministic and could create issues with node sizing.
      #  2. A ray worker node is started via the `ray start` CLI. In each spark task,
      #     a child process is started and will execute a `ray start ...` command in
      #     blocking mode.
      #  3. Each task will acquire a file lock for 10s to ensure that the ray worker
      #     init will acquire a port connection to the ray head node that does not
      #     contend with other worker processes on the same Spark worker node.
      #  4. When the ray cluster is shutdown, killing ray worker nodes is implemented by
      #     `sparkContext.cancelJobGroup` to cancel the background spark job, sending a
      #     SIGKILL signal to all spark tasks. Once the spark tasks are killed,
      #     `ray_start_node` process detects parent died event then it kills ray
      #     worker node.
  
      def ray_cluster_job_mapper(_):
          from pyspark.taskcontext import TaskContext
  
          _worker_logger = logging.getLogger("ray.util.spark.worker")
  
          context = TaskContext.get()
  
          (
              worker_port_range_begin,
              worker_port_range_end,
          ) = _preallocate_ray_worker_port_range()
  
          # 10001 is used as ray client server port of global mode ray cluster.
          ray_worker_node_dashboard_agent_port = get_random_unused_port(
              ray_head_ip, min_port=10002, max_port=20000
          )
          ray_worker_node_cmd = [
              sys.executable,
              "-m",
              "ray.util.spark.start_ray_node",
              f"--num-cpus={num_cpus_per_node}",
              "--block",
              f"--address={ray_head_ip}:{ray_head_port}",
              f"--memory={heap_memory_per_node}",
              f"--object-store-memory={object_store_memory_per_node}",
              f"--min-worker-port={worker_port_range_begin}",
              f"--max-worker-port={worker_port_range_end - 1}",
              f"--dashboard-agent-listen-port={ray_worker_node_dashboard_agent_port}",
              *_convert_ray_node_options(worker_node_options),
          ]
          if ray_temp_dir is not None:
              ray_worker_node_cmd.append(f"--temp-dir={ray_temp_dir}")
  
          hook_entry = _create_hook_entry(
              is_global=(ray_temp_dir is None)
          )
  
          ray_worker_node_extra_envs = {
              RAY_ON_SPARK_COLLECT_LOG_TO_PATH: collect_log_to_path
              or "",
              RAY_ON_SPARK_START_RAY_PARENT_PID: str(os.getpid()),
              "RAY_ENABLE_WINDOWS_OR_OSX_CLUSTER": "1",
              **hook_entry.custom_environment_variables(),
          }
  
          if num_gpus_per_node > 0:
              task_resources = context.resources()
  
              if "gpu" not in task_resources:
                  raise RuntimeError(
                      "Couldn't get the gpu id, Please check the GPU resource "
                      "configuration"
                  )
              gpu_addr_list = [
                  int(addr.strip())
                  for addr in task_resources["gpu"].addresses
              ]
  
              available_physical_gpus = (
                  get_spark_task_assigned_physical_gpus(gpu_addr_list)
              )
              ray_worker_node_cmd.append(
                  f"--num-gpus={len(available_physical_gpus)}",
              )
              ray_worker_node_extra_envs[
                  "CUDA_VISIBLE_DEVICES"
              ] = ",".join(
                  [str(gpu_id) for gpu_id in available_physical_gpus]
              )
  
          _worker_logger.info(
              f"Start Ray worker, command: {' '.join(ray_worker_node_cmd)}"
          )
  
          try:
              if autoscale_mode:
                  # Notify job server the task has been launched.
                  requests.post(
                      url=(
                          f"http://{ray_head_ip}:{spark_job_server_port}"
                          "/notify_task_launched"
                      ),
                      json={
                          "spark_job_group_id": spark_job_group_id,
                      },
                  )
  
              # Note:
              # When a pyspark job cancelled, the UDF python worker process are killed by
              # signal "SIGKILL", then `start_ray_node` process will detect the parent
              # died event (see `ray.util.spark.start_ray_node.check_parent_alive`) and
              # then kill ray worker node process and execute cleanup routine.
              exec_cmd(
                  ray_worker_node_cmd,
                  synchronous=True,
                  extra_env=ray_worker_node_extra_envs,
              )
          except Exception as e:
              if autoscale_mode:
                  # In autoscaling mode, when Ray worker node is down, autoscaler will
                  # try to start new Ray worker node if necessary,
                  # but we use spark job to launch Ray worker node process,
                  # to avoid trigger spark task retries, we swallow exception here
                  # to make spark task exit normally.
                  _logger.warning(
                      f"Ray worker node process exit, reason: {repr(e)}."
                  )
              else:
                  raise
  
          # NB: Not reachable.
          yield 0
  
      spark.sparkContext.setJobGroup(
          spark_job_group_id,
          spark_job_group_desc,
      )
  
      # Starting a normal spark job (not barrier spark job) to run ray worker
      # nodes, the design purpose is:
      # 1. Using normal spark job, spark tasks can automatically retry
      # individually, we don't need to write additional retry logic, But, in
      # barrier mode, if one spark task fails, it will cause all other spark
      # tasks killed.
      # 2. Using normal spark job, we can support failover when a spark worker
      # physical machine crashes. (spark will try to re-schedule the spark task
      # to other spark worker nodes)
      # 3. Using barrier mode job, if the cluster resources does not satisfy
      # "idle spark task slots >= argument num_spark_task", then the barrier
      # job gets stuck and waits until enough idle task slots available, this
      # behavior is not user-friendly, on a shared spark cluster, user is hard
      # to estimate how many idle tasks available at a time, But, if using normal
      # spark job, it can launch job with less spark tasks (i.e. user will see a
      # ray cluster setup with less worker number initially), and when more task
      # slots become available, it continues to launch tasks on new available
      # slots, and user can see the ray cluster worker number increases when more
      # slots available.
      job_rdd = spark.sparkContext.parallelize(
          list(range(num_worker_nodes)), num_worker_nodes
      )
  
      if using_stage_scheduling:
          resource_profile = _create_resource_profile(
              num_cpus_per_node,
              num_gpus_per_node,
          )
          job_rdd = job_rdd.withResources(resource_profile)
  
      hook_entry = _create_hook_entry(is_global=(ray_temp_dir is None))
      hook_entry.on_spark_job_created(spark_job_group_id)
  
      job_rdd.mapPartitions(ray_cluster_job_mapper).collect()
  
imports: import copy
  import yaml
  import json
  import os
  import socket
  import sys
  import time
  import threading
  import logging
  import uuid
  import warnings
  import requests
  from packaging.version import Version
  from typing import Optional, Dict, Tuple, Type
  import ray
  import ray._private.services
  from ray.autoscaler._private.spark.node_provider import HEAD_NODE_ID
  from ray.util.annotations import DeveloperAPI, PublicAPI
  from ray._private.storage import _load_class
  from .utils import exec_cmd, is_port_in_use, get_random_unused_port, get_spark_session, get_spark_application_driver_host, is_in_databricks_runtime, get_spark_task_assigned_physical_gpus, get_avail_mem_per_ray_worker_node, get_max_num_concurrent_tasks, gen_cmd_exec_failure_msg, calc_mem_ray_head_node, _wait_service_up
  from .start_hook_base import RayOnSparkStartHook
  from .databricks_hook import DefaultDatabricksRayOnSparkStartHook
  from threading import Event
label_desc: Use image node_id 15 for calling a global function with example usage: _start_ray_worker_nodes() without return types