class_docstr: Base class for the job.
  
  After establishing connection with the remote object, the job will
  create a remote class instance locally and enter an infinite loop
  in a separate process, waiting for commands from the remote object.
class_docstr_tok: ["Base","class","for","the","job",".","After","establishing","connection","with","the","remote","object",",","the","job","will","create","a","remote","class","instance","locally","and","enter","an","infinite","loop","in","a","separate","process",",","waiting","for","commands","from","the","remote","object","."]
docstr: Create three sockets for each job in the main process.
  
  (1) job_socket(functional socket): sends job_address and heartbeat_address to worker.
  (2) reply_client_ping_socket: replies ping message of client.
  (3) remove_job_socket: sends a command to the corresponding worker to remove the job.
                         Used to ask the worker removing the dead job immediately 
                         instead of waiting for the heartbeat failure.
  
  Create two heartbeat server threads for each job:
  (1) worker_heartbeat_server_thread: reply heartbeat signal from the worker.
  (2) client_heartbeat_client_thread: send heartbeat signal to the client.
docstr_tok: ["Create","three","sockets","for","each","job","in","the","main","process",".","(","1",")","job_socket","(","functional","socket",")",":","sends","job_address","and","heartbeat_address","to","worker",".","(","2",")","reply_client_ping_socket",":","replies","ping","message","of","client",".","(","3",")","remove_job_socket",":","sends","a","command","to","the","corresponding","worker","to","remove","the","job",".","Used","to","ask","the","worker","removing","the","dead","job","immediately","instead","of","waiting","for","the","heartbeat","failure",".","Create","two","heartbeat","server","threads","for","each","job",":","(","1",")","worker_heartbeat_server_thread",":","reply","heartbeat","signal","from","the","worker",".","(","2",")","client_heartbeat_client_thread",":","send","heartbeat","signal","to","the","client","."]
code: def _create_sockets(self):
      """Create three sockets for each job in the main process.
  
      (1) job_socket(functional socket): sends job_address and heartbeat_address to worker.
      (2) reply_client_ping_socket: replies ping message of client.
      (3) remove_job_socket: sends a command to the corresponding worker to remove the job.
                             Used to ask the worker removing the dead job immediately
                             instead of waiting for the heartbeat failure.
  
      Create two heartbeat server threads for each job:
      (1) worker_heartbeat_server_thread: reply heartbeat signal from the worker.
      (2) client_heartbeat_client_thread: send heartbeat signal to the client.
      """
      # wait for another process to create reply socket
      self.job_address = self.job_address_receiver.recv()
      self.job_id = self.job_id_receiver.recv()
  
      self.ctx = zmq.Context()
      # create the job_socket
      self.job_socket = create_client_socket(
          self.ctx, self.worker_address, heartbeat_timeout=True
      )
  
      # a thread that reply ping signals from the client
      reply_client_ping_socket, port = create_server_socket(self.ctx)
      reply_client_ping_address = "{}:{}".format(self.job_ip, port)
      ping_thread = threading.Thread(
          target=self._reply_ping, args=(reply_client_ping_socket,)
      )
      ping_thread.setDaemon(True)
      ping_thread.start()
  
      # a thread that reply heartbeat signals from the worker
      def worker_heartbeat_exit_callback_func():
          logger.warning(
              "[Job]lost connection with the worker, will exit"
          )
          os._exit(1)
  
      worker_heartbeat_server_thread = HeartbeatServerThread(
          heartbeat_exit_callback_func=worker_heartbeat_exit_callback_func
      )
      worker_heartbeat_server_thread.setDaemon(True)
      worker_heartbeat_server_thread.start()
  
      # sends job information to the worker
      initialized_job = InitializedJob(
          self.job_address,
          worker_heartbeat_server_thread.get_address(),
          reply_client_ping_address,
          None,
          self.pid,
          self.job_id,
          self.log_server_address,
      )
  
      try:
          self.job_socket.send_multipart(
              [
                  remote_constants.NORMAL_TAG,
                  cloudpickle.dumps(initialized_job),
              ]
          )
          message = self.job_socket.recv_multipart()
      except zmq.error.Again as e:
          logger.warning(
              "[Job] Cannot connect to the worker {}. ".format(
                  self.worker_address
              )
              + "Job will quit."
          )
          self.job_socket.close(0)
          os._exit(0)
  
      tag = message[0]
      assert tag == remote_constants.NORMAL_TAG
      # create the remove_job_socket
      remove_job_address = to_str(message[1])
      self.worker_pid = int(to_str(message[2]))
      worker_heartbeat_server_thread.set_host_pid(self.worker_pid)
      self.remove_job_socket = self.ctx.socket(zmq.REQ)
      self.remove_job_socket.setsockopt(
          zmq.RCVTIMEO, remote_constants.HEARTBEAT_TIMEOUT_S * 1000
      )
      self.remove_job_socket.connect(
          "tcp://{}".format(remove_job_address)
      )
  
imports: import os
  import compatible_trick
  import argparse
  import cloudpickle
  import pickle
  import psutil
  import re
  import sys
  import tempfile
  import shutil
  import threading
  import time
  import traceback
  import zmq
  import importlib
  import parl
  from multiprocessing import Process, Pipe
  from parl.utils import to_str, to_byte, get_ip_address, logger
  from parl.remote.communication import loads_argument, loads_return, dumps_argument, dumps_return
  from parl.remote import remote_constants
  from parl.utils.exceptions import SerializeError, DeserializeError
  from parl.remote.message import InitializedJob
  from parl.remote.utils import redirect_output_to_file
  from parl.remote.remote_class_serialization import load_remote_class
  from parl.remote.zmq_utils import create_server_socket, create_client_socket
  from parl.remote.grpc_heartbeat import HeartbeatServerThread, HeartbeatClientThread
label_desc: Use image node_id 2 for calling the Job obj's underlying member method code with example usage: obj._create_sockets() without return types