在tensorflow的输入管道中进行线程处理

Question

背景

张量流中的典型输入管道如下所示:

                  tf.train.string_input_producer(list_of_filenames)
                         (creates queue of filenames)
                                   |
                                  \|/
           fixed length reader reads records from the files
                                   |
                                  \|/
    Read records are decoded and processed(eg if dealing with images then cropping,flipping etc)
                                   |
                                  \|/
            tf.train.shuffle_batch(tensors,num_threads)
        (creates a shuffling queue and returns batches of tensors) 

问题

Q1)函数tf.train.string_input_producer()中没有num_threads的参数.这是否意味着只有一个线程专用于从文件名队列中读取文件名？

Q2)函数tf.train.shuffle_batch()的num_threads参数的范围是什么,即这里提到的线程数用于读取,解码和处理文件,还是仅用于创建批量的张量？

Q3)有没有办法打印哪个线程从特定文件读取文件名或记录,即每个线程完成的工作记录？

Answer 1

所有数据加载操作都在张量流图中执行，您需要做的是启动一个或多个线程来迭代读取器/入队操作。Tensorflow 提供了一个 QueueRunner 类来完成此任务。Coordinator 类允许您非常轻松地管理线程。

https://www.tensorflow.org/programmers_guide/threading_and_queues

这是上面链接中的示例代码：

# Create a queue runner that will run 4 threads in parallel to enqueue
# examples.
qr = tf.train.QueueRunner(queue, [enqueue_op] * 4)

# Launch the graph.
sess = tf.Session()
# Create a coordinator, launch the queue runner threads.
coord = tf.train.Coordinator()
enqueue_threads = qr.create_threads(sess, coord=coord, start=True)
# Run the training loop, controlling termination with the coordinator.
for step in xrange(1000000):
    if coord.should_stop():
        break
    sess.run(train_op)
# When done, ask the threads to stop.
coord.request_stop()
# And wait for them to actually do it.
coord.join(enqueue_threads)

如果您在图表之外加载/预处理样本（在您自己的代码中，不使用 TF 操作），那么您不会使用 QueueRunner，而是使用您自己的类sess.run(enqueue_op, feed_dict={...})在循环中使用命令对数据进行排队。

Q1：线程数的处理方式：qr.create_threads(sess, coord=coord, start=True)

Q2：TF 会话是线程安全的，每次调用都会tf.run(...)看到截至其开始时当前变量的一致快照。您的 QueueRunner 排队操作可以运行任意数量的线程。他们将以线程安全的方式排队。

Q3：我自己没有使用过tf.train.string_input_producer，但我认为您必须稍后在图表中请求dequeued数据的张量，只需将该张量添加到您的请求列表中sess.run([train_op, dequeue_op])