def _create_session(self):
    """Factory for the _RecoverableSession.

    Returns:
      A session, initialized or recovered as needed.
    """
    if self._is_chief:
      tf_sess = self._session_manager.prepare_session(
          self._master,
          saver=self._scaffold.saver,
          checkpoint_dir=self._checkpoint_dir,
          config=self._config,
          init_op=self._scaffold.init_op,
          init_feed_dict=self._scaffold.init_feed_dict,
          init_fn=self._scaffold.init_fn)
    else:
      tf_sess = self._session_manager.wait_for_session(
          self._master, config=self._config)
    # Keep the tf_sess for quick runs of global step when needed.
    self._tf_sess = tf_sess
    # We don't want coordinator to suppress any exception.
    self._coord = coordinator.Coordinator(clean_stop_exception_types=[])
    queue_runner.start_queue_runners(sess=tf_sess, coord=self._coord)
    return _CoordinatedSession(_HookedSession(tf_sess, self._hooks),
                               self._coord)

  def _feed_remote_queues_forever(
      self, sess_callback, outer_coordinator, tolerator):
    if tolerator is None:
      tolerator = failure_tolerator.FailureTolerator(limit=5)

    # In a retry loop, keep the remote queue runners going:
    while True:
      if outer_coordinator.should_stop():
        return

      inner_coordinator = coordinator.Coordinator()

      # Make sure inner_coordinator stops when outer_coordinator does:
      _link_coordinators(inner_coordinator, outer_coordinator)

      # Create a fresh session to use for remote queues:
      inner_session = sess_callback()
      inner_session.run(self._fake_op)  # Work around b/32749157, as above

      queue_runner.start_queue_runners(sess=inner_session,
                                       coord=inner_coordinator,
                                       collection=Feeder.REMOTE_QUEUE_RUNNERS)

      self._feeding_event.set()  # Notify that feeding has begun

      try:
        with tolerator.forgive():
          # Wait for a stop to be requested.
          inner_coordinator.wait_for_stop()

          # TODO(shoutis): If outer_coordinator.should_stop(), it
          # would be nice to interrupt the remote queue runners (which
          # may be blocked if their remote queue is full) -- but
          # there's no way currently; see b/32774422.

          # Cause any exceptions from the remote queue runners to be
          # reraised immediately, without waiting for their associated
          # threads to terminate like join() would. This means a retry
          # can begin immediately after any remote device fails,
          # rather than having to wait for any pending enqueues to
          # other remote devices to finish first.
          inner_coordinator.raise_requested_exception()

          # If this line is reached, there was a graceful shutdown
          # requested.

          # Request the outer coordinator to stop. Since
          # outer_coordinator.request_stop() is the currently only way
          # for inner_coordinator() to finish without failure, this is
          # redundant, but it's harmless and defends against infinite
          # hangs should code changes make it possible for
          # inner_coordinator to finish in other ways.
          outer_coordinator.request_stop()

          return
      except Exception as e:
        # Pass non-forgiven errors along to outer_coordinator:
        outer_coordinator.request_stop(e)
        raise

 def create_session(self):
     """Creates a coordinated session."""
     # Keep the tf_sess for unit testing.
     self.tf_sess = self._session_creator.create_session()
     # We don't want coordinator to suppress any exception.
     self.coord = coordinator.Coordinator(clean_stop_exception_types=[])
     queue_runner.start_queue_runners(sess=self.tf_sess, coord=self.coord)
     return _CoordinatedSession(_HookedSession(self.tf_sess, self._hooks), self.coord)

  def run_feeding_forever(self,
                          sess_callback,
                          outer_coordinator=None,
                          tolerator=None,
                          start_queue_runners=True):
    """Runs feeding forever.

    This method exits only if `outer_coordinator` has a stop requested
    or if a remote feed encounters an un-tolerated error. The most
    likely cause of `outer_coordinator` stopping besides a manual call
    to `request_stop()` is a `QueueRunner` thread reaching the end of
    its queue or encountering an error.

    Returns only after joining `outer_coordinator`.

    Args:
      sess_callback: A function which, when called, returns a Session
        to use for feeding. Can be called multiple times due to retries.
      outer_coordinator: If present, a `Coordinator` which the feeding
        process will respect. Will be created if omitted.
      tolerator: If present, a `failure_tolerator.FailureTolerator` which is
        used to manage retries of feeding the remote devices.
      start_queue_runners: Whether to start queue runners before
        beginning to feed the remote devices. Defaults to True. If
        False and no other mechanism is used to start queue runners, this
        method will hang forever without doing work.

    """
    # We use /two/ coordinators: one which runs normal queue
    # runners (outer_coordinator), and one which runs the remote
    # enqueues (using an inner coordinator) with retries and failure
    # tolerance. By using two coordinators, errors
    # encountered while running the remote enqueue ops don't cause the
    # outer_coordinator to be shut down.
    if outer_coordinator is None:
      outer_coordinator = coordinator.Coordinator()

    # Start the outer queue runners:
    if start_queue_runners:
      session = sess_callback()
      # Work around b/32749157 by running an operation before proceeding --
      # this way the session used for queue runners will be fully established
      # before we create another session with the same target.
      session.run(self._fake_op)
      queue_runner.start_queue_runners(sess=session,
                                       coord=outer_coordinator)

    if self._num_remote_feeds == 0:
      self._feeding_event.set()
      outer_coordinator.join()
      return
    else:
      try:
        self._feed_remote_queues_forever(
            sess_callback, outer_coordinator, tolerator)
      finally:
        self._feeding_event.set()
        outer_coordinator.join()

 def create_session(self):
   """Creates a coordinated session."""
   # Keep the tf_sess for unit testing.
   self.tf_sess = self._session_creator.create_session()
   # We don't want coordinator to suppress any exception.
   self.coord = coordinator.Coordinator(clean_stop_exception_types=[])
   queue_runner.start_queue_runners(sess=self.tf_sess, coord=self.coord)
   # Inform the hooks that a new session has been created.
   for hook in self._hooks:
     hook.after_create_session(self.tf_sess, self.coord)
   return _CoordinatedSession(
       _HookedSession(self.tf_sess, self._hooks), self.coord,
       self._stop_grace_period_secs)

def run_feeds(output_dict, feed_dicts, restore_checkpoint_path=None):
  """Run `output_dict` tensors with each input in `feed_dicts`.

  If `checkpoint_path` is supplied, restore from checkpoint. Otherwise, init all
  variables.

  Args:
    output_dict: A `dict` mapping string names to `Tensor` objects to run.
      Tensors must all be from the same graph.
    feed_dicts: Iterable of `dict` objects of input values to feed.
    restore_checkpoint_path: A string containing the path to a checkpoint to
      restore.

  Returns:
    A list of dicts of values read from `output_dict` tensors, one item in the
    list for each item in `feed_dicts`. Keys are the same as `output_dict`,
    values are the results read from the corresponding `Tensor` in
    `output_dict`.

  Raises:
    ValueError: if `output_dict` or `feed_dicts` is None or empty.
  """
  if not output_dict:
    raise ValueError('output_dict is invalid: %s.' % output_dict)
  if not feed_dicts:
    raise ValueError('feed_dicts is invalid: %s.' % feed_dicts)

  graph = contrib_ops.get_graph_from_inputs(output_dict.values())

  with graph.as_default() as g:
    with tf_session.Session('') as session:
      if restore_checkpoint_path:
        _restore_from_checkpoint(session, g, restore_checkpoint_path)
      else:
        session.run(variables.initialize_all_variables())
      session.run(variables.initialize_local_variables())
      session.run(data_flow_ops.initialize_all_tables())
      coord = Coordinator()
      try:
        queue_runner.start_queue_runners(session, coord=coord)
        return [_run_dict(session, output_dict, f) for f in feed_dicts]
      finally:
        coord.request_stop()

  def run(self,
          num_batches=None,
          graph=None,
          session=None,
          start_queues=True,
          initialize_variables=True,
          **kwargs):
    """Builds and runs the columns of the `DataFrame` and yields batches.

    This is a generator that yields a dictionary mapping column names to
    evaluated columns.

    Args:
      num_batches: the maximum number of batches to produce. If none specified,
        the returned value will iterate through infinite batches.
      graph: the `Graph` in which the `DataFrame` should be built.
      session: the `Session` in which to run the columns of the `DataFrame`.
      start_queues: if true, queues will be started before running and halted
        after producting `n` batches.
      initialize_variables: if true, variables will be initialized.
      **kwargs: Additional keyword arguments e.g. `num_epochs`.

    Yields:
      A dictionary, mapping column names to the values resulting from running
      each column for a single batch.
    """
    if graph is None:
      graph = ops.get_default_graph()
    with graph.as_default():
      if session is None:
        session = sess.Session()
      self_built = self.build(**kwargs)
      keys = list(self_built.keys())
      cols = list(self_built.values())
      if initialize_variables:
        if variables.local_variables():
          session.run(variables.initialize_local_variables())
        if variables.all_variables():
          session.run(variables.initialize_all_variables())
      if start_queues:
        coord = coordinator.Coordinator()
        threads = qr.start_queue_runners(sess=session, coord=coord)
      i = 0
      while num_batches is None or i < num_batches:
        i += 1
        try:
          values = session.run(cols)
          yield collections.OrderedDict(zip(keys, values))
        except errors.OutOfRangeError:
          break
      if start_queues:
        coord.request_stop()
        coord.join(threads)

def run_feeds_iter(output_dict, feed_dicts, restore_checkpoint_path=None):
  """Run `output_dict` tensors with each input in `feed_dicts`.

  If `restore_checkpoint_path` is supplied, restore from checkpoint. Otherwise,
  init all variables.

  Args:
    output_dict: A `dict` mapping string names to `Tensor` objects to run.
      Tensors must all be from the same graph.
    feed_dicts: Iterable of `dict` objects of input values to feed.
    restore_checkpoint_path: A string containing the path to a checkpoint to
      restore.

  Yields:
    A sequence of dicts of values read from `output_dict` tensors, one item
    yielded for each item in `feed_dicts`. Keys are the same as `output_dict`,
    values are the results read from the corresponding `Tensor` in
    `output_dict`.

  Raises:
    ValueError: if `output_dict` or `feed_dicts` is None or empty.
  """
  if not output_dict:
    raise ValueError('output_dict is invalid: %s.' % output_dict)
  if not feed_dicts:
    raise ValueError('feed_dicts is invalid: %s.' % feed_dicts)

  graph = contrib_ops.get_graph_from_inputs(output_dict.values())
  with graph.as_default() as g:
    with tf_session.Session('') as session:
      session.run(
          resources.initialize_resources(resources.shared_resources() +
                                         resources.local_resources()))
      if restore_checkpoint_path:
        _restore_from_checkpoint(session, g, restore_checkpoint_path)
      else:
        session.run(variables.global_variables_initializer())
      session.run(variables.local_variables_initializer())
      session.run(data_flow_ops.initialize_all_tables())
      coord = coordinator.Coordinator()
      threads = None
      try:
        threads = queue_runner.start_queue_runners(session, coord=coord)
        for f in feed_dicts:
          yield session.run(output_dict, f)
      finally:
        coord.request_stop()
        if threads:
          coord.join(threads, stop_grace_period_secs=120)

def evaluate(graph,
             output_dir,
             checkpoint_path,
             eval_dict,
             update_op=None,
             global_step_tensor=None,
             supervisor_master='',
             log_every_steps=10,
             feed_fn=None,
             max_steps=None):
  """Evaluate a model loaded from a checkpoint.

  Given `graph`, a directory to write summaries to (`output_dir`), a checkpoint
  to restore variables from, and a `dict` of `Tensor`s to evaluate, run an eval
  loop for `max_steps` steps, or until an exception (generally, an
  end-of-input signal from a reader operation) is raised from running
  `eval_dict`.

  In each step of evaluation, all tensors in the `eval_dict` are evaluated, and
  every `log_every_steps` steps, they are logged. At the very end of evaluation,
  a summary is evaluated (finding the summary ops using `Supervisor`'s logic)
  and written to `output_dir`.

  Args:
    graph: A `Graph` to train. It is expected that this graph is not in use
      elsewhere.
    output_dir: A string containing the directory to write a summary to.
    checkpoint_path: A string containing the path to a checkpoint to restore.
      Can be `None` if the graph doesn't require loading any variables.
    eval_dict: A `dict` mapping string names to tensors to evaluate. It is
      evaluated in every logging step. The result of the final evaluation is
      returned. If `update_op` is None, then it's evaluated in every step. If
      `max_steps` is `None`, this should depend on a reader that will raise an
      end-of-input exception when the inputs are exhausted.
    update_op: A `Tensor` which is run in every step.
    global_step_tensor: A `Variable` containing the global step. If `None`,
      one is extracted from the graph using the same logic as in `Supervisor`.
      Used to place eval summaries on training curves.
    supervisor_master: The master string to use when preparing the session.
    log_every_steps: Integer. Output logs every `log_every_steps` evaluation
      steps. The logs contain the `eval_dict` and timing information.
    feed_fn: A function that is called every iteration to produce a `feed_dict`
      passed to `session.run` calls. Optional.
    max_steps: Integer. Evaluate `eval_dict` this many times.

  Returns:
    A tuple `(eval_results, global_step)`:
    eval_results: A `dict` mapping `string` to numeric values (`int`, `float`)
      that are the result of running eval_dict in the last step. `None` if no
      eval steps were run.
    global_step: The global step this evaluation corresponds to.

  Raises:
    ValueError: if `output_dir` is empty.
  """
  if not output_dir:
    raise ValueError('Output directory should be non-empty %s.' % output_dir)
  with graph.as_default():
    global_step_tensor = contrib_variables.assert_or_get_global_step(
        graph, global_step_tensor)

    # Create or get summary op, global_step and saver.
    saver = _get_saver()
    local_init_op = _get_local_init_op()
    ready_op = _get_ready_op()

    session_manager = session_manager_lib.SessionManager(
        local_init_op=local_init_op,
        ready_op=ready_op)
    session, initialized = session_manager.recover_session(
        master=supervisor_master,
        saver=saver,
        checkpoint_dir=checkpoint_path)

    # Start queue runners.
    coord = coordinator.Coordinator()
    threads = queue_runner.start_queue_runners(session, coord)

  with session:
    if not initialized:
      logging.warning('Failed to initialize from %s.', checkpoint_path)
      # TODO(ipolosukhin): This should be failing, but old code relies on that.
      session.run(variables.global_variables_initializer())
      if checkpoint_path:
        _restore_from_checkpoint(session, graph, checkpoint_path, saver)

    current_global_step = session.run(global_step_tensor)
    eval_results = None
    # TODO(amodei): Fix this to run through the eval set exactly once.
    step = 0
    eval_step = None
    feed_dict = None
    logging.info('Eval steps [%d,%s) for training step %d.', step,
                 'inf' if max_steps is None
                 else str(max_steps), current_global_step)
    try:
      try:
        while (max_steps is None) or (step < max_steps):
          step += 1
          start_time = time.time()
#.........这里部分代码省略.........