python - 使用多处理将来自多个数据加载器的对象填充到队列中

Question

我致力于机器学习输入管道。我编写了一个数据加载器，它从大型 .hdf 文件中读取数据并返回切片，每个切片大约需要 2 秒。因此，我想使用一个队列，它从多个数据加载器接收对象，并可以通过下一个函数(如生成器)从队列返回单个对象。此外，填充队列的进程应该以某种方式在后台运行，当队列未满时重新填充队列。我无法让它正常工作。它使用单个数据加载器，给我 4 倍相同的切片..

import multiprocessing as mp

class Queue_Generator():
    def __init__(self, data_loader_list):
        self.pool = mp.Pool(4)
        self.data_loader_list = data_loader_list
        self.queue = mp.Queue(maxsize=16)
        self.pool.map(self.fill_queue, self.data_loader_list)
    def fill_queue(self,gen):
        self.queue.put(next(gen))
    def __next__(self):
        yield self.queue.get()

我从中得到什么:NotImplementedError:池对象无法在进程之间传递或腌制提前致谢

Answer 1

您的具体错误意味着当您将类方法传递给池时，您不能将池作为类的一部分。我的建议可能如下:

import multiprocessing as mp
from queue import Empty


class QueueGenerator(object):
    def __init__(self, data_loader_list):
        self.data_loader_list = data_loader_list
        self.queue = mp.Queue(maxsize=16)

    def __iter__(self):
        processes = list()
        for _ in range(4):
            pr = mp.Process(target=fill_queue, args=(self.queue, self.data_loader_list))
            pr.start()
            processes.append(pr)
        return self

    def __next__(self):
        try:
            return self.queue.get(timeout=1) # this should have a value, otherwise your loop will never stop. make it something that ensures your processes have enough time to update the queue but not too long that your program freezes for an extended period of time after all information is processed
        except Empty:
            raise StopIteration

# have fill queue as a separate function
def fill_queue(queue, gen):
    while True:
        try:
            value = next(gen)
            queue.put(value)
        except StopIteration: # assumes the given data_loader_list is an iterator
            break
    print('stopping')


gen = iter(range(70))

qg = QueueGenerator(gen)


for val in qg:
    print(val)
# test if it works several times:
for val in qg:
    print(val)

我认为您要解决的下一个问题是让 data_loader_list 在每个单独的进程中提供新信息。但由于您没有提供任何相关信息，我无法帮助您。然而，上面确实为您提供了一种让进程填充队列的方法，然后将其作为迭代器传递出去。