class_docstr: A thread-safe data structure used to respond to resource request,
     maintaining the available computation resources in each worker.
  
  Attributes:
      worker_hostname (dict): A dict to record worker hostname.
      worker_vacant_jobs (dict): Record which vacant jobs does each
                                 worker has.
      worker_used_jobs (dict): Record which used jobs does each
                                 worker has.
      master_ip (str): IP address of the master node.
class_docstr_tok: ["A","thread-safe","data","structure","used","to","respond","to","resource","request",",","maintaining","the","available","computation","resources","in","each","worker",".","Attributes",":","worker_hostname","(","dict",")",":","A","dict","to","record","worker","hostname",".","worker_vacant_jobs","(","dict",")",":","Record","which","vacant","jobs","does","each","worker","has",".","worker_used_jobs","(","dict",")",":","Record","which","used","jobs","does","each","worker","has",".","master_ip","(","str",")",":","IP","address","of","the","master","node","."]
docstr: Return a job when the client submits a job.
  
  If there is no vacant CPU and GPU in the cluster, this will return None.
  
  Args:
      n_cpu (int): request a Job with n_cpu CPUs
      n_gpu (int): request a Job with n_gpu GPUs
  
  Return:
      An ``InitializedJob`` that has information about available job address.
docstr_tok: ["Return","a","job","when","the","client","submits","a","job",".","If","there","is","no","vacant","CPU","and","GPU","in","the","cluster",",","this","will","return","None",".","Args",":","n_cpu","(","int",")",":","request","a","Job","with","n_cpu","CPUs","n_gpu","(","int",")",":","request","a","Job","with","n_gpu","GPUs","Return",":","An","``","InitializedJob","``","that","has","information","about","available","job","address","."]
code: def request_job(self, n_cpu=1, n_gpu=0):
      """Return a job when the client submits a job.
  
      If there is no vacant CPU and GPU in the cluster, this will return None.
  
      Args:
          n_cpu (int): request a Job with n_cpu CPUs
          n_gpu (int): request a Job with n_gpu GPUs
  
      Return:
          An ``InitializedJob`` that has information about available job address.
      """
      with self.lock:
          candidates = self.filter(n_cpu, n_gpu)
          if candidates:
              random.shuffle(candidates)
              worker_address = candidates[0]
              job_address, job = self.worker_vacant_jobs[
                  worker_address
              ].popitem()
              job.allocated_cpu, job.allocated_gpu = self.allocate(
                  worker_address, n_cpu, n_gpu
              )
              self.worker_used_jobs[worker_address][job_address] = job
              return job
          logger.warning("No enough cpu/gpu resources at the moment")
          return None
  
imports: import threading
  import random
  from collections import defaultdict
  from collections import namedtuple
  from parl.utils import logger
  from parl.remote import remote_constants
  from parl.remote.cpu_resource import CpuResource
  from parl.remote.gpu_resource import GpuResource
  from parl.remote.message import AllocatedCpu
  from parl.remote.message import AllocatedGpu
label_desc: Use image node_id 8 for calling the WorkerManager obj's underlying member method code with example usage: obj.request_job(n_cpu, n_gpu) and returns: None, job