benchmark_data_reading.py 文件源码

def test_input_producer(fname):
    import pdb
    pdb.set_trace()

    with tf.Session() as sess:
        strings = [b"to", b"be", b"or", b"not", b"to", b"be"]
        num_epochs = 3
        queue = tf.train.string_input_producer(
            strings, num_epochs=num_epochs, shuffle=False)
        dequeue_many = queue.dequeue_many(len(strings) * num_epochs)
        dequeue = queue.dequeue()
        tf.initialize_all_variables().run()
        tf.initialize_local_variables().run()
        threads = tf.train.start_queue_runners()

        # No randomness, so just see repeated copies of the input.
        output = dequeue_many.eval()
        self.assertAllEqual(strings * num_epochs, output)

        # Reached the limit.
        with self.assertRaises(tf.errors.OutOfRangeError):
            dequeue.eval()
        for thread in threads:
            thread.join()