class_docstr: Base class for the remote client.
  
  For each training task, there is a global client in the cluster which
  submits jobs to the master node. Different `@parl.remote_class` objects
  connect to the same global client in a training task.
  
  Attributes:
      submit_job_socket (zmq.Context.socket): A socket which submits job to
                                              the master node.
      pyfiles (bytes): A serialized dictionary containing the code of python
                       files in local working directory.
      executable_path (str): File path of the executable python script.
      start_time (time): A timestamp to record the start time of the program.
class_docstr_tok: ["Base","class","for","the","remote","client",".","For","each","training","task",",","there","is","a","global","client","in","the","cluster","which","submits","jobs","to","the","master","node",".","Different","`","@","parl.remote_class","`","objects","connect","to","the","same","global","client","in","a","training","task",".","Attributes",":","submit_job_socket","(","zmq.Context.socket",")",":","A","socket","which","submits","job","to","the","master","node",".","pyfiles","(","bytes",")",":","A","serialized","dictionary","containing","the","code","of","python","files","in","local","working","directory",".","executable_path","(","str",")",":","File","path","of","the","executable","python","script",".","start_time","(","time",")",":","A","timestamp","to","record","the","start","time","of","the","program","."]
docstr: Send a job to the Master node.
  
  When a `@parl.remote_class` object is created, the global client
  sends a job to the master node. Then the master node will allocate
  a vacant job from its job pool to the remote object.
  
  Args:
      max_memory (float): Maximum memory (MB) can be used by each remote
                          instance, the unit is in MB and default value is
                          none(unlimited).
      n_gpu (int): Number of GPUs can used in this remote instance.
  Returns:
      An ``InitializedJob`` that has information about available job address.
docstr_tok: ["Send","a","job","to","the","Master","node",".","When","a","`","@","parl.remote_class","`","object","is","created",",","the","global","client","sends","a","job","to","the","master","node",".","Then","the","master","node","will","allocate","a","vacant","job","from","its","job","pool","to","the","remote","object",".","Args",":","max_memory","(","float",")",":","Maximum","memory","(","MB",")","can","be","used","by","each","remote","instance",",","the","unit","is","in","MB","and","default","value","is","none","(","unlimited",")",".","n_gpu","(","int",")",":","Number","of","GPUs","can","used","in","this","remote","instance",".","Returns",":","An","``","InitializedJob","``","that","has","information","about","available","job","address","."]
code: def submit_job(self, max_memory, n_gpu, job_is_alive):
      """Send a job to the Master node.
  
      When a `@parl.remote_class` object is created, the global client
      sends a job to the master node. Then the master node will allocate
      a vacant job from its job pool to the remote object.
  
      Args:
          max_memory (float): Maximum memory (MB) can be used by each remote
                              instance, the unit is in MB and default value is
                              none(unlimited).
          n_gpu (int): Number of GPUs can used in this remote instance.
      Returns:
          An ``InitializedJob`` that has information about available job address.
      """
      if self.connected_to_master:
          while True:
              self.lock.acquire()
              n_cpu = 0 if n_gpu > 0 else 1
              self.submit_job_socket.send_multipart(
                  [
                      remote_constants.CLIENT_SUBMIT_TAG,
                      to_byte(self.reply_master_heartbeat_address),
                      to_byte(self.client_id),
                      to_byte(str(n_cpu)),
                      to_byte(str(n_gpu)),
                  ]
              )
              message = self.submit_job_socket.recv_multipart()
              self.lock.release()
              tag = message[0]
              if tag == remote_constants.NORMAL_TAG:
                  job_info = cloudpickle.loads(message[1])
                  job_ping_address = job_info.ping_heartbeat_address
  
                  self.lock.acquire()
                  instance_id = self._check_job(
                      job_ping_address,
                      max_memory,
                      job_info.allocated_gpu.gpu,
                  )
                  self.lock.release()
                  if instance_id != -1:
                      self.instance_id_to_job[
                          instance_id
                      ] = job_is_alive
                      return job_info
              # no vacant CPU resources, cannot submit a new job
              elif tag == remote_constants.CPU_TAG:
                  # wait 1 second to avoid requesting in a high frequency.
                  time.sleep(1)
                  return None
              # no vacant GPU resources, cannot submit a new job
              elif tag == remote_constants.GPU_TAG:
                  # wait 5 second to avoid requesting in a high frequency.
                  time.sleep(1)
                  return None
              elif tag == remote_constants.REJECT_GPU_JOB_TAG:
                  error_message = "[Client] Request fails. It is not allowed to request GPU resource from a CPU cluster."
                  logger.error(error_message)
                  raise Exception(error_message)
              elif tag == remote_constants.REJECT_CPU_JOB_TAG:
                  error_message = "[Client] Request fails. It is not allowed to request CPU resource from a GPU cluster."
                  logger.error(error_message)
                  raise Exception(error_message)
              elif tag == remote_constants.REJECT_INVALID_GPU_JOB_TAG:
                  error_message = (
                      "[Client] request {} GPUs, but rejected.".format(
                          n_gpu
                      )
                  )
                  logger.error(error_message)
                  raise Exception(error_message)
              else:
                  raise NotImplementedError
      else:
          raise Exception(
              "Client can not submit job to the master, please check if master is connected."
          )
      return None
  
imports: import cloudpickle
  import datetime
  import os
  import socket
  import sys
  import threading
  import zmq
  import parl
  import time
  import glob
  import multiprocessing
  from parl.utils import to_str, to_byte, get_ip_address, logger, isnotebook
  from parl.remote.utils import get_subfiles_recursively
  from parl.remote import remote_constants
  from parl.remote.grpc_heartbeat import HeartbeatServerThread, HeartbeatServerProcess
  from parl.remote.utils import get_version
label_desc: Use image node_id 12 for calling the Client obj's underlying member method code with example usage: obj.submit_job(max_memory, n_gpu, job_is_alive) and returns: None, job_info, None, None