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: An infinite loop waiting for commands from the remote object.
  
  Each job will receive two kinds of message from the remote object:
  
  1. When the remote object calls a function, job will run the
     function on the local instance and return the results to the
     remote object.
  2. When the remote object is deleted, the job will quit and release
     related computation resources.
  
  Args:
      reply_socket (sockert): main socket to accept commands of remote object.
      job_address (String): address of reply_socket.
docstr_tok: ["An","infinite","loop","waiting","for","commands","from","the","remote","object",".","Each","job","will","receive","two","kinds","of","message","from","the","remote","object",":","1",".","When","the","remote","object","calls","a","function",",","job","will","run","the","function","on","the","local","instance","and","return","the","results","to","the","remote","object",".","2",".","When","the","remote","object","is","deleted",",","the","job","will","quit","and","release","related","computation","resources",".","Args",":","reply_socket","(","sockert",")",":","main","socket","to","accept","commands","of","remote","object",".","job_address","(","String",")",":","address","of","reply_socket","."]
code: def single_task(self, obj, reply_socket, job_address):
      """An infinite loop waiting for commands from the remote object.
  
      Each job will receive two kinds of message from the remote object:
  
      1. When the remote object calls a function, job will run the
         function on the local instance and return the results to the
         remote object.
      2. When the remote object is deleted, the job will quit and release
         related computation resources.
  
      Args:
          reply_socket (sockert): main socket to accept commands of remote object.
          job_address (String): address of reply_socket.
      """
  
      while True:
          message = reply_socket.recv_multipart()
          tag = message[0]
          if tag in [
              remote_constants.CALL_TAG,
              remote_constants.GET_ATTRIBUTE_TAG,
              remote_constants.SET_ATTRIBUTE_TAG,
          ]:
              try:
                  if tag == remote_constants.CALL_TAG:
                      function_name = to_str(message[1])
                      data = message[2]
                      args, kwargs = loads_argument(data)
  
                      # Redirect stdout to stdout.log temporarily
                      with redirect_output_to_file(
                          self.logfile_path, os.devnull
                      ):
                          ret = getattr(obj, function_name)(
                              *args, **kwargs
                          )
  
                      ret = dumps_return(ret)
                      reply_socket.send_multipart(
                          [
                              remote_constants.NORMAL_TAG,
                              ret,
                              dumps_return(set(obj.__dict__.keys())),
                          ]
                      )
  
                  elif tag == remote_constants.GET_ATTRIBUTE_TAG:
                      attribute_name = to_str(message[1])
                      with redirect_output_to_file(
                          self.logfile_path, os.devnull
                      ):
                          ret = getattr(obj, attribute_name)
                      ret = dumps_return(ret)
                      reply_socket.send_multipart(
                          [remote_constants.NORMAL_TAG, ret]
                      )
                  elif tag == remote_constants.SET_ATTRIBUTE_TAG:
                      attribute_name = to_str(message[1])
                      attribute_value = loads_return(message[2])
                      with redirect_output_to_file(
                          self.logfile_path, os.devnull
                      ):
                          setattr(obj, attribute_name, attribute_value)
                      reply_socket.send_multipart(
                          [
                              remote_constants.NORMAL_TAG,
                              dumps_return(set(obj.__dict__.keys())),
                          ]
                      )
                  else:
                      pass
  
              except Exception as e:
                  # reset the job
  
                  error_str = str(e)
                  logger.error(error_str)
  
                  if type(e) == AttributeError:
                      reply_socket.send_multipart(
                          [
                              remote_constants.ATTRIBUTE_EXCEPTION_TAG,
                              to_byte(error_str),
                          ]
                      )
                      raise AttributeError
  
                  elif type(e) == SerializeError:
                      reply_socket.send_multipart(
                          [
                              remote_constants.SERIALIZE_EXCEPTION_TAG,
                              to_byte(error_str),
                          ]
                      )
                      raise SerializeError
  
                  elif type(e) == DeserializeError:
                      reply_socket.send_multipart(
                          [
                              remote_constants.DESERIALIZE_EXCEPTION_TAG,
                              to_byte(error_str),
                          ]
                      )
                      raise DeserializeError
  
                  else:
                      traceback_str = str(traceback.format_exc())
                      logger.error(
                          "traceback:\n{}".format(traceback_str)
                      )
                      reply_socket.send_multipart(
                          [
                              remote_constants.EXCEPTION_TAG,
                              to_byte(
                                  error_str
                                  + "\ntraceback:\n"
                                  + traceback_str
                              ),
                          ]
                      )
                      break
  
          # receive DELETE_TAG from actor, and stop replying worker heartbeat
          elif tag == remote_constants.KILLJOB_TAG:
              reply_socket.send_multipart([remote_constants.NORMAL_TAG])
              logger.warning(
                  "An actor exits and this job {} will exit.".format(
                      job_address
                  )
              )
              break
          else:
              logger.error(
                  "The job receives an unknown message: {}".format(
                      message
                  )
              )
              raise NotImplementedError
  
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 8 for calling the Job obj's underlying member method code with example usage: obj.single_task(obj, reply_socket, job_address) without return types