Method take

tensorflow/python/ops/work_queue.py:360–399

Take work from the work queue.

(self)

          lambda: create_with_prompt)

  def take(self):
    """Take work from the work queue."""
    def remote_take():
      """Take work from remote worker."""
      with ops.name_scope(self.name):
        with ops.device(self._remote_device):
          taken = gen_work_queue_ops.work_queue_take(
              self._handle,
              num_clients=self.num_clients)

          work_bak = control_flow_ops.no_op()
          if self._local_work_mgr:
            work_bak = gen_work_queue_ops.save_local_work(
                taken,
                job_name=self._local_work_mgr.job_name,
                task_index=self._local_work_mgr.task_index,
                restore_works_dir=self._local_work_mgr.restore_works_dir)
      with ops.control_dependencies([work_bak]):
        with ops.device(self._local_device):
          local_work = array_ops.identity(taken)
          return local_work

    def local_take():
      """Take work from local worker."""
      assert self._local_work_mgr, 'local_work_mgr should not be None.'
      return self._local_work_mgr.take()

    if self._local_work_mgr:
      with ops.name_scope(self.name):
        with ops.device(self._local_device):
          with ops.device('/cpu:0'):
            local_workqueue_empty = self._local_work_mgr.local_workqueue_empty()
            local_work = control_flow_ops.cond(
                local_workqueue_empty, remote_take, local_take)
    else:
      local_work = remote_take()

    if self._prefix is None:
      return local_work
    return string_ops.string_join([self._prefix, local_work])

  def input_producer(self):
    """Returns a FIFOQueue as input producer.