多处理脚本卡住

Question

我有以下Python代码：

def workPackage(args):
   try:      
    outputdata                  = dict()
    iterator                        = 1
    for name in outputnames:
        outputdata[name]            = []
    for filename in filelist:
        read_data                   = np.genfromtxt(filename, comments="#", unpack=True, names=datacolnames, delimiter=";")
        mean_va1                    = np.mean(read_data["val1"])
        mean_va2                    = np.mean(read_data["val2"])
        outputdata[outputnames[0]].append(read_data["setpoint"][0])
        outputdata[outputnames[1]].append(mean_val1)
        outputdata[outputnames[2]].append(mean_val2)        
        outputdata[outputnames[3]].append(mean_val1-mean_val2)
        outputdata[outputnames[4]].append((mean_val1-mean_val2)/read_data["setpoint"][0]*100)
        outputdata[outputnames[5]].append(2*np.std(read_data["val1"]))
        outputdata[outputnames[6]].append(2*np.std(read_data["val2"]))      

        print("Process "+str(identifier+1)+": "+str(round(100*(iterator/len(filelist)),1))+"% complete")
        iterator    = iterator+1
    queue.put (outputdata)
 except:
 some message
if __name__ == '__main__':
"Main script"

此代码用于评估大量测量数据。我总共在多个目录中获得了大约 900 个文件(总共约 13GB(。主脚本确定所有文件路径并将它们存储在 4 个块中。每个块(文件路径列表(都提供给一个进程。

    try:
      print("Distributing the workload on "+str(numberOfProcesses)+" processes...")                     
      for i in range(0,numberOfProcesses):
        q[i]                = multiprocessing.Queue()
        Processes[i]        = multiprocessing.Process(target=workPackage, args=(filelistChunks[i], colnames, outputdatanames, i, q[i]))
        Processes[i].start()
      for i in range(0,numberOfProcesses):
        Processes[i].join()
    except:
       print("Exception while processing stuff...")

之后，从队列中读取 restuls 并将其存储到输出文件中。现在这是我的问题：该脚本启动 4 个进程，每个进程运行到 100%(请参阅 workPackage 函数中的打印(。它们不会同时完成，而是在大约 2 分钟内完成。但随后脚本就停止了。如果我通过简单地剪切文件列表来限制要处理的数据量，它有时会运行到最后，但有时不会。我不明白，为什么脚本在所有进程达到 100% 后就会卡住。

我真的不知道那里发生了什么。

Answer 1

您使用 queue.put(( 将项目添加到队列中，然后调用 queue.join((，但我看不到您在哪里调用 queue.get(( 或 queue.task_done((。 Join 不会释放线程，直到队列为空并且对每个项目调用了 task_done((。