在 Python 池中创建队列延迟而不阻塞

Question

我有一个大型程序（特别是一个函数），我正在尝试使用JoinableQueue 和多处理map_async 方法进行并行化。我正在使用的函数对多维数组执行多项操作，因此我将每个数组分成多个部分，每个部分独立评估；但是我需要尽早将其中一个数组缝合在一起，但是“缝合”发生在“评估”之前，我需要在 JoinableQueue 中引入某种延迟。我已经到处寻找可行的解决方案，但我对多处理非常陌生，其中大部分都超出了我的想象。

这个措辞可能会令人困惑 - 道歉。这是我的代码大纲（我不能把它全部都放出来，因为它很长，但如果需要我可以提供更多细节）

import numpy as np
import multiprocessing as mp
from multiprocessing import Pool, Pipe, JoinableQueue

def main_function(section_number):

    #define section sizes
    array_this_section = array[:,start:end+1,:]
    histogram_this_section = np.zeros((3, dataset_size, dataset_size))
    #start and end are defined according to the size of the array
    #dataset_size is to show that the histogram is a different size than the array

    for m in range(1,num_iterations+1):
        #do several operations- each section of the array 
                 #corresponds to a section on the histogram

        hist_queue.put(histogram_this_section)

        #each process sends their own part of the histogram outside of the pool 
                 #to be combined with every other part- later operations 
                 #in this function must use the full histogram

        hist_queue.join()
        full_histogram = full_hist_queue.get()
        full_hist_queue.task_done()

        #do many more operations


hist_queue = JoinableQueue()
full_hist_queue = JoinableQueue()

if __name__ == '__main__':
    pool = mp.Pool(num_sections)
    args = np.arange(num_sections)
    pool.map_async(main_function, args, chunksize=1)    

    #I need the map_async because the program is designed to display an output at the 
        #end of each iteration, and each output must be a compilation of all processes

    #a few variable definitions go here

    for m in range(1,num_iterations+1):
        for i in range(num_sections):
            temp_hist = hist_queue.get()    #the code hangs here because the queue 
                                            #is attempting to get before anything 
                                            #has been put
            hist_full += temp_hist
        for i in range(num_sections):
            hist_queue.task_done()
        for i in range(num_sections):
            full_hist_queue.put(hist_full)    #the full histogram is sent back into 
                                              #the pool


            full_hist_queue.join()

        #etc etc

    pool.close()
    pool.join()

Answer 1

我很确定您的问题是您如何创建 Queues 并尝试与子进程共享它们。如果您只是将它们作为全局变量，它们可能会在子进程中重新创建而不是继承（具体细节取决于您使用的操作系统和/或context 用于multiprocessing）。

解决此问题的更好方法是避免使用multiprocessing.Pool 来生成您的流程，而是自己为您的工作人员明确创建Process 实例。这样，您可以毫无困难地将 Queue 实例传递给需要它们的进程（技术上可以将队列传递给 Pool 工作人员，但这很尴尬）。

我会尝试这样的：

def worker_function(section_number, hist_queue, full_hist_queue): # take queues as arguments
    # ... the rest of the function can work as before
    # note, I renamed this from "main_function" since it's not running in the main process

if __name__ == '__main__':
    hist_queue = JoinableQueue()   # create the queues only in the main process
    full_hist_queue = JoinableQueue()  # the workers don't need to access them as globals

    processes = [Process(target=worker_function, args=(i, hist_queue, full_hist_queue)
                 for i in range(num_sections)]
    for p in processes:
        p.start()

    # ...

如果你的工作函数的不同阶段或多或少相互独立（也就是说，“做更多操作”步骤不直接依赖于它上面的“做几个操作”步骤，只依赖于@ 987654329@)，您也许可以保留Pool，而是将不同的步骤拆分为单独的函数，主进程可以通过多次调用池上的map 来调用这些函数。在这种方法中，您不需要使用自己的Queues，只需使用池中内置的Queues。这可能是最好的，特别是如果您将工作分成的“部分”数量与您计算机上的处理器核心数量不一致。您可以让Pool 匹配核心数量，让每个核心依次处理数据的几个部分。

大概是这样的：

def worker_make_hist(section_number):
    # do several operations, get a partial histogram
    return histogram_this_section

def worker_do_more_ops(section_number, full_histogram):
    # whatever...
    return some_result

if __name__ == "__main__":
    pool = multiprocessing.Pool() # by default the size will be equal to the number of cores

    for temp_hist in pool.imap_unordered(worker_make_hist, range(number_of_sections)):
        hist_full += temp_hist

    some_results = pool.starmap(worker_do_more_ops, zip(range(number_of_sections),
                                                        itertools.repeat(hist_full)))