class_docstr: Ray actor created by JobManager for each submitted job, responsible to
  setup runtime_env, execute given shell command in subprocess, update job
  status, persist job logs and manage subprocess group cleaning.
  
  One job supervisor actor maps to one subprocess, for one job_id.
  Job supervisor actor should fate share with subprocess it created.
docstr: Runs the entrypoint command as a child process, streaming stderr &
  stdout to given log files.
  
  Unix systems:
  Meanwhile we start a demon process and group driver
  subprocess in same pgid, such that if job actor dies, entire process
  group also fate share with it.
  
  Windows systems:
  A jobObject is created to enable fate sharing for the entire process group.
  
  Args:
      logs_path: File path on head node's local disk to store driver
          command's stdout & stderr.
  Returns:
      child_process: Child process that runs the driver command. Can be
          terminated or killed upon user calling stop().
docstr_tok: ["Runs","the","entrypoint","command","as","a","child","process",",","streaming","stderr","&","stdout","to","given","log","files",".","Unix","systems",":","Meanwhile","we","start","a","demon","process","and","group","driver","subprocess","in","same","pgid",",","such","that","if","job","actor","dies",",","entire","process","group","also","fate","share","with","it",".","Windows","systems",":","A","jobObject","is","created","to","enable","fate","sharing","for","the","entire","process","group",".","Args",":","logs_path",":","File","path","on","head","node","'s","local","disk","to","store","driver","command","'s","stdout","&","stderr",".","Returns",":","child_process",":","Child","process","that","runs","the","driver","command",".","Can","be","terminated","or","killed","upon","user","calling","stop","(",")","."]
code: def _exec_entrypoint(self, logs_path: str) -> subprocess.Popen:
      """
      Runs the entrypoint command as a child process, streaming stderr &
      stdout to given log files.
  
      Unix systems:
      Meanwhile we start a demon process and group driver
      subprocess in same pgid, such that if job actor dies, entire process
      group also fate share with it.
  
      Windows systems:
      A jobObject is created to enable fate sharing for the entire process group.
  
      Args:
          logs_path: File path on head node's local disk to store driver
              command's stdout & stderr.
      Returns:
          child_process: Child process that runs the driver command. Can be
              terminated or killed upon user calling stop().
      """
      with open(logs_path, "w") as logs_file:
          child_process = subprocess.Popen(
              self._entrypoint,
              shell=True,
              start_new_session=True,
              stdout=logs_file,
              stderr=subprocess.STDOUT,
              # Ray intentionally blocks SIGINT in all processes, so if the user wants
              # to stop job through SIGINT, we need to unblock it in the child process
              preexec_fn=lambda: signal.pthread_sigmask(
                  signal.SIG_UNBLOCK, {signal.SIGINT}
              )
              if sys.platform != "win32"
              and os.environ.get("RAY_JOB_STOP_SIGNAL") == "SIGINT"
              else None,
          )
          parent_pid = os.getpid()
          child_pid = child_process.pid
          # Create new pgid with new subprocess to execute driver command
  
          if sys.platform != "win32":
              try:
                  child_pgid = os.getpgid(child_pid)
              except ProcessLookupError:
                  # Process died before we could get its pgid.
                  return child_process
  
              # Open a new subprocess to kill the child process when the parent
              # process dies kill -s 0 parent_pid will succeed if the parent is
              # alive. If it fails, SIGKILL the child process group and exit
              subprocess.Popen(
                  f"while kill -s 0 {parent_pid}; do sleep 1; done; kill -9 -{child_pgid}",  # noqa: E501
                  shell=True,
                  # Suppress output
                  stdout=subprocess.DEVNULL,
                  stderr=subprocess.DEVNULL,
              )
  
          elif sys.platform == "win32" and win32api:
              # Create a JobObject to which the child process (and its children)
              # will be connected. This job object can be used to kill the child
              # processes explicitly or when the jobObject gets deleted during
              # garbage collection.
              self._win32_job_object = win32job.CreateJobObject(
                  None, ""
              )
              win32_job_info = win32job.QueryInformationJobObject(
                  self._win32_job_object,
                  win32job.JobObjectExtendedLimitInformation,
              )
              win32_job_info["BasicLimitInformation"][
                  "LimitFlags"
              ] = win32job.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
              win32job.SetInformationJobObject(
                  self._win32_job_object,
                  win32job.JobObjectExtendedLimitInformation,
                  win32_job_info,
              )
              child_handle = win32api.OpenProcess(
                  win32con.PROCESS_TERMINATE
                  | win32con.PROCESS_SET_QUOTA,
                  False,
                  child_pid,
              )
              win32job.AssignProcessToJobObject(
                  self._win32_job_object, child_handle
              )
  
          return child_process
  
imports: import asyncio
  import copy
  import json
  import logging
  import os
  import psutil
  import random
  import signal
  import string
  import subprocess
  import sys
  import time
  import traceback
  from asyncio.tasks import FIRST_COMPLETED
  from collections import deque
  from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy, SchedulingStrategyT
  import ray
  from ray._private.gcs_utils import GcsAioClient
  from ray._private.utils import run_background_task
  import ray._private.ray_constants
  from ray._private.runtime_env.constants import RAY_JOB_CONFIG_JSON_ENV_VAR
  from ray.actor import ActorHandle
  from ray.dashboard.consts import RAY_JOB_ALLOW_DRIVER_ON_WORKER_NODES_ENV_VAR, DEFAULT_JOB_START_TIMEOUT_SECONDS, RAY_JOB_START_TIMEOUT_SECONDS_ENV_VAR
  from ray.dashboard.modules.job.common import JOB_ID_METADATA_KEY, JOB_NAME_METADATA_KEY, JOB_ACTOR_NAME_TEMPLATE, JOB_LOGS_PATH_TEMPLATE, SUPERVISOR_ACTOR_RAY_NAMESPACE, JobInfo, JobInfoStorageClient
  from ray.dashboard.modules.job.utils import file_tail_iterator
  from ray.exceptions import ActorUnschedulableError, RuntimeEnvSetupError
  from ray.job_submission import JobStatus
  from ray._private.event.event_logger import get_event_logger
  from ray.core.generated.event_pb2 import Event
label_desc: Use image node_id 4 for calling the JobSupervisor obj's underlying member method code with example usage: obj._exec_entrypoint(logs_path) and returns: child_process, child_process