Method _get_workers

tensorflow/python/ops/work_queue_test.py:297–342

(
      self, num_workers, workers,
      works, num_epochs=1, shuffle=True,
      restore_works_dir=None,
      prefix=None)

        thread.join()

  def _get_workers(
      self, num_workers, workers,
      works, num_epochs=1, shuffle=True,
      restore_works_dir=None,
      prefix=None):
    sessions = []
    graphs = []
    train_ops = []
    for worker_id in range(num_workers):
      graph = ops.Graph()
      is_chief = (worker_id == 0)
      with graph.as_default():
        worker_device = "/job:worker/task:%d/cpu:0" % (worker_id)
        with ops.device(device_setter.replica_device_setter(
            worker_device=worker_device,
            ps_device="/job:ps/task:0/cpu:0",
            ps_tasks=1)):

          local_work_mgr = None
          if restore_works_dir:
            local_work_mgr = LocalWorkMgr(
                'worker', worker_id, restore_works_dir)

          work_queue = WorkQueue(
              works, num_epochs=num_epochs, shuffle=shuffle,
              name='global_queue', prefix=prefix,
              local_work_mgr=local_work_mgr)
          dataset = work_queue.input_dataset()
          iterator = dataset.make_initializable_iterator()
          ops.add_to_collection(ops.GraphKeys.TABLE_INITIALIZERS,
                                iterator.initializer)
          data = iterator.get_next()
        # Creates MonitoredSession
        hooks = []
        if local_work_mgr:
          hooks.append(local_work_mgr.hook())
        sess = monitored_session.MonitoredTrainingSession(
            workers[worker_id].target,
            is_chief=is_chief,
            hooks=hooks)

      sessions.append(sess)
      graphs.append(graph)
      train_ops.append(data)

    return sessions, graphs, train_ops

  def test_multi_worker_dataset(self):
    def _run(train_op, sess, result):