使用多处理从多个进程中对 tf.RandomShuffleQueue 进行排队

Question

我想使用多个进程 (not threads) 进行一些预处理并将结果排入一个 tf.RandomShuffleQueue 队列，我的主图可以使用该队列进行训练。

有没有办法做到这一点？

我的实际问题

我已将我的数据集转换为跨 256 个分片的 TFRecords。我想使用multiprocessing 启动 20 个进程，并让每个进程处理一系列分片。每个进程都应该读取图像，然后对其进行扩充并将它们推送到tf.RandomShuffleQueue，从中可以将输入提供给图形进行训练。

有些人建议我查看tensorflow 中的inception 示例。然而，这是一个非常不同的情况，因为只有数据分片的读取是由多个线程完成的 (not processes)，而预处理（例如 - 扩充）发生在主线程中。

Answer 1

(这旨在解决你的实际问题)

在另一个主题中，有人告诉您 Python 具有全局解释器锁 (GIL)，因此除非您使用多个进程，否则多核不会带来速度优势。

这可能是促使您使用multiprocessing 的原因。

但是，对于 TF，Python 通常仅用于构建“图”。实际执行发生在本机代码（或 GPU）中，其中 GIL 不起任何作用。

鉴于此，我建议让 TF 使用多线程。这可以使用intra_op_parallelism_threads 参数来控制，例如：

with tf.Session(graph=graph, 
    config=tf.ConfigProto(allow_soft_placement=True, 
    intra_op_parallelism_threads=20)) as sess:
    # ...

（旁注：如果你有一个 2-CPU、32 核系统，最好的参数可能是intra_op_parallelism_threads=16，这取决于很多因素）

Answer 2

评论：TFRecords 的酸洗没那么重要。 我可以传递包含分片 TFRecord 文件范围名称的列表列表。

我必须重新启动决策过程！

评论：我可以将它作为参数传递给 Pool.map()。

验证，如果 multiprocesing.Queue() 可以处理此问题。
张量函数的结果是Tensor object。
请尝试以下操作：

tensor_object = func(TFRecord)
q = multiprocessing.Manager().Queue()
q.put(tensor_object)
data = q.get()
print(data)

评论：如何确保所有进程都排入同一个队列？

这很简单enqueue 来自Pool.map(... 的结果 毕竟process 完成了。
或者，我们可以enqueue 并行，queueing 来自all processes 的数据。

但这样做取决于如上所述的 pickleabel 数据。

---

例如：

import multiprocessing as mp
def func(filename):
    TFRecord = read(filename)
    tensor_obj = tf.func(TFRecord)
    return tensor_obj

def main_Tensor(tensor_objs):
    tf = # ... instantiat Tensor Session
    rsq = tf.RandomShuffleQueue(...)
    for t in tensor_objs:
        rsq.enqueue(t)

if __name__ == '__main__':
    sharded_TFRecords = ['file1', 'file2']
    with mp.Pool(20) as pool:
        tensor_objs = pool.map(func, sharded_TFRecords)
        pool.join()

    main_Tensor(tensor_objs)

Answer 3

似乎推荐的使用multiprocessing 运行TF 的方法是为每个孩子创建一个单独的tf.Session，因为跨进程共享它是不可行的。

你可以看看this example，希望对你有帮助。

[编辑：旧答案]

您可以使用multiprocessing.Pool 并依靠其回调机制在结果准备好后立即将结果放入tf.RandomShuffleQueue。

这里有一个非常简单的例子来说明如何做到这一点。

from multiprocessing import Pool


class Processor(object):
    def __init__(self, random_shuffle_queue):
        self.queue = random_shuffle_queue
        self.pool = Pool()

    def schedule_task(self, task):
        self.pool.apply_async(processing_function, args=[task], callback=self.task_done)

    def task_done(self, results):
        self.queue.enqueue(results)

这里假设 Python 2，对于 Python 3，我建议使用 concurrent.futures.ProcessPoolExecutor。