Tensorflow:批处理输入队列,然后更改队列源

Question

我有一个模型在一组图像上运行并使用它们计算一些统计数据 - 为简单起见,它只是输出集合中的平均图像(它比实际中的图像更多).我有许多包含图像的目录,我想从每个目录中获取输出.每个目录中都有可变数量的图像.

我为我的脚本构建了一次图形,输出变量和损失函数.输入使用稍微适应的代码进行批处理.我改编它采取一系列路径,我使用可变大小的占位符.我从这里得到了灵感.

然后我遍历目录并运行以下内容:

1. 初始化变量(这会从先前目录中计算的结果重置先前的输出变量)
2. 将image paths变量设置为新目录中的当前文件数组: sess.run(image_paths.initializer, feed_dict={image_paths_initializer: image_paths})
3. 启动队列运行: queue_threads = tf.train.start_queue_runners(sess=sess, coord=coord)
4. 运行一些时代以获得结果
5. 关闭线程 coord.request_stop(); coord.join(queue_threads); coord.clear_stop()
6. 返回结果,保存结果,移动到下一个目录...

问题是,当涉及到第二个目录时,队列运行程序线程拒绝启动(我可以通过调试queue_threads变量来看到这一点).这给出了如下错误:

Compute status: Aborted: FIFOQueue '_1_input_producer' is closed.
Compute status: Aborted: RandomShuffleQueue '_0_shuffle_batch/random_shuffle_queue' is closed.

如果我不关闭线程(并且不再次启动它们),那么它们不会从新目录生成文件 - 它们忽略了(2)中的变量赋值操作.是不是可以像这样重启队列？

我已经尝试在他们自己的单独会话中设置队列并从中提取批次,但这给了我各种CUDA /内存错误.如果我这样做并添加调试停止,我可以让它在它到达之前运行很远 - 但我不知道是否有可能在不相交的会话/图形之间添加控制依赖项？

对于每个新目录,都可以从头开始,但这会给我试图避免的进程增加很多开销.我已经做了类似的事情(即,重置变量并使用不同的输入重新运行)没有队列,它节省了大量的时间,所以我知道这有点工作.

你们中的任何一个人都可以想出一个出路吗？

Answer 1

string_input_producer是FIFOQueue+ QueueRunner.如果您使用a FIFOQueue并手动排队,则可以获得更多控制权.像这样的东西

filename_queue = tf.FIFOQueue(100, tf.string)
enqueue_placeholder = tf.placeholder(dtype=tf.string)
enqueue_op = filename_queue.enqueue(enqueue_placeholder)

config = tf.ConfigProto()
config.operation_timeout_in_ms=2000  # for debugging queue hangs
sess = tf.InteractiveSession(config=config)
coord = tf.train.Coordinator()
threads = tf.train.start_queue_runners(coord=coord)

sess.run([enqueue_op], feed_dict={enqueue_placeholder:"/temp/dir1/0"})
sess.run([enqueue_op], feed_dict={enqueue_placeholder:"/temp/dir1/1"})

# do stats for /temp/dir1

sess.run([enqueue_op], feed_dict={enqueue_placeholder:"/temp/dir2/0"})
sess.run([enqueue_op], feed_dict={enqueue_placeholder:"/temp/dir2/1"})

# do stats for /temp/dir2

coord.request_stop()
coord.join(threads)