def testName(self):
   with ops.name_scope("scope"):
     queue = data_flow_ops.FIFOQueue(10, dtypes.float32, name="queue")
   qr = queue_runner_impl.QueueRunner(queue, [control_flow_ops.no_op()])
   self.assertEqual("scope/queue", qr.name)
   queue_runner_impl.add_queue_runner(qr)
   self.assertEqual(
       1, len(ops.get_collection(ops.GraphKeys.QUEUE_RUNNERS, "scope")))

def boston_input_fn_with_queue(num_epochs=None):
  features, labels = boston_input_fn(num_epochs=num_epochs)

  # Create a minimal queue runner.
  fake_queue = data_flow_ops.FIFOQueue(30, dtypes.int32)
  queue_runner = queue_runner_impl.QueueRunner(fake_queue,
                                               [constant_op.constant(0)])
  queue_runner_impl.add_queue_runner(queue_runner)

  return features, labels

 def testStartQueueRunnersRaisesIfNotASession(self):
   zero64 = constant_op.constant(0, dtype=dtypes.int64)
   var = variables.VariableV1(zero64)
   count_up_to = var.count_up_to(3)
   queue = data_flow_ops.FIFOQueue(10, dtypes.float32)
   init_op = variables.global_variables_initializer()
   qr = queue_runner_impl.QueueRunner(queue, [count_up_to])
   queue_runner_impl.add_queue_runner(qr)
   with self.cached_session():
     init_op.run()
     with self.assertRaisesRegexp(TypeError, "tf.Session"):
       queue_runner_impl.start_queue_runners("NotASession")

 def testStartQueueRunnersIgnoresMonitoredSession(self):
   zero64 = constant_op.constant(0, dtype=dtypes.int64)
   var = variables.VariableV1(zero64)
   count_up_to = var.count_up_to(3)
   queue = data_flow_ops.FIFOQueue(10, dtypes.float32)
   init_op = variables.global_variables_initializer()
   qr = queue_runner_impl.QueueRunner(queue, [count_up_to])
   queue_runner_impl.add_queue_runner(qr)
   with self.cached_session():
     init_op.run()
     threads = queue_runner_impl.start_queue_runners(
         monitored_session.MonitoredSession())
     self.assertFalse(threads)

  def _testScopedExportWithQueue(self, test_dir, exported_filename):
    graph = ops.Graph()
    with graph.as_default():
      with ops.name_scope("queue1"):
        input_queue = data_flow_ops.FIFOQueue(10, dtypes.float32)
        enqueue = input_queue.enqueue((9876), name="enqueue")
        close = input_queue.close(name="close")
        qr = queue_runner_impl.QueueRunner(input_queue, [enqueue], close)
        queue_runner_impl.add_queue_runner(qr)
        input_queue.dequeue(name="dequeue")

      orig_meta_graph, _ = meta_graph.export_scoped_meta_graph(
          filename=os.path.join(test_dir, exported_filename),
          graph=ops.get_default_graph(),
          export_scope="queue1")

    return orig_meta_graph

 def testStartQueueRunners(self):
   # CountUpTo will raise OUT_OF_RANGE when it reaches the count.
   zero64 = constant_op.constant(0, dtype=dtypes.int64)
   var = variables.VariableV1(zero64)
   count_up_to = var.count_up_to(3)
   queue = data_flow_ops.FIFOQueue(10, dtypes.float32)
   init_op = variables.global_variables_initializer()
   qr = queue_runner_impl.QueueRunner(queue, [count_up_to])
   queue_runner_impl.add_queue_runner(qr)
   with self.cached_session() as sess:
     init_op.run()
     threads = queue_runner_impl.start_queue_runners(sess)
     for t in threads:
       t.join()
     self.assertEqual(0, len(qr.exceptions_raised))
     # The variable should be 3.
     self.assertEqual(3, var.eval())

  def testMultiThreadedEstimateDataDistribution(self):
    num_classes = 10

    # Set up graph.
    random_seed.set_random_seed(1234)
    label = math_ops.cast(
        math_ops.round(random_ops.random_uniform([1]) * num_classes),
        dtypes_lib.int32)

    prob_estimate = sampling_ops._estimate_data_distribution(  # pylint: disable=protected-access
        label, num_classes)
    # Check that prob_estimate is well-behaved in a multithreaded context.
    _, _, [prob_estimate] = sampling_ops._verify_input(  # pylint: disable=protected-access
        [], label, [prob_estimate])

    # Use queues to run multiple threads over the graph, each of which
    # fetches `prob_estimate`.
    queue = data_flow_ops.FIFOQueue(
        capacity=25,
        dtypes=[prob_estimate.dtype],
        shapes=[prob_estimate.get_shape()])
    enqueue_op = queue.enqueue([prob_estimate])
    queue_runner_impl.add_queue_runner(
        queue_runner_impl.QueueRunner(queue, [enqueue_op] * 25))
    out_tensor = queue.dequeue()

    # Run the multi-threaded session.
    with self.cached_session() as sess:
      # Need to initialize variables that keep running total of classes seen.
      variables.global_variables_initializer().run()

      coord = coordinator.Coordinator()
      threads = queue_runner_impl.start_queue_runners(coord=coord)

      for _ in range(25):
        sess.run([out_tensor])

      coord.request_stop()
      coord.join(threads)