多处理产生空闲进程并且不计算任何东西

Question

似乎有一连串关于多处理库溢出的问题和答案。我已经浏览了所有我能找到的相关内容，但没有找到直接与我的问题相关的内容。

我正在尝试将相同的功能并行应用于多个文件。但是，每当我开始处理时，计算机只会启动几个 python 实例，然后什么也不做。根本没有计算发生，进程只是闲置

我已经查看了所有关于溢出的类似问题，但似乎没有一个有我的空闲进程问题。

我做错了什么？

定义函数（例如缩写。检查以确保它有效）

import pandas as pd
import numpy as np
import glob
import os
#from timeit import default_timer as timer
import talib
from multiprocessing import Process


def example_function(file):

    df=pd.read_csv(file, header = 1)
    stock_name = os.path.basename(file)[:-4]
    macd, macdsignal, macdhist = talib.MACD(df.Close, fastperiod=12, slowperiod=26, signalperiod=9)

    df['macd'] = macdhist*1000
    print(f'stock{stock_name} processed')
    final_macd_report.append(df)

获取我要运行该功能的目录中所有文件的列表

import glob

path = r'C:\Users\josiahh\Desktop\big_test3/*'

files = [f for f in glob.glob(path, recursive=True)]

尝试多处理

import multiprocessing as mp
if __name__ == '__main__':

    p = mp.Pool(processes = 5)
    async_result = p.map_async(example_function, files)
    p.close()
    p.join()
    print("Complete")

任何帮助将不胜感激。

Answer 1

代码的结构没有任何问题，因此从您发布的内容中无法猜出某些问题。从非常简单的事情开始，然后逐步将其移至您实际尝试做的事情。您正在导入大量扩展（第 3 方）代码，问题可能出在任何地方。这是一个开始：

def example_function(arg):
    from time import sleep
    msg = "crunching " + str(arg)
    print(msg)
    sleep(arg)
    print("done " + msg)

if __name__ == '__main__':
    import multiprocessing as mp
    p = mp.Pool(processes = 5)
    async_result = p.map_async(example_function, reversed(range(15)))
    print("result", async_result.get())
    p.close()
    p.join()
    print("Complete")

对我来说，这在 64 位 Python 3.7.4 下的 Win10 上运行良好。适合你吗？

特别注意末尾的async_result.get()。这将显示一个包含 15 个 None 值的列表。你永远不会对你的async_result 做任何事情。因此，如果在工作进程中引发任何异常，它很可能会默默地消失。在这种情况下，.get()'ing 结果将（重新）引发主程序中的异常。

另外请确认您的files 列表实际上不是空的。我们也无法从这里猜测 ;-)

编辑

我将async_result.get() 移到了它自己的行中，就在map_async() 之后，以最大限度地提高在工作进程中揭示其他静默异常的机会。至少也要在你的代码中添加这么多。

Answer 2

虽然我本身没有发现任何错误，但我想提出一些更改建议。

一般来说，Pool 中的工作函数预计会返回一些内容。这个返回值被传回父进程。我喜欢用它作为状态报告。在工作进程中捕获异常也是一个好主意，以防万一。 例如：

def example_function(file):
    status = 'OK'
    try:
        df=pd.read_csv(file, header = 1)
        stock_name = os.path.basename(file)[:-4]
        macd, macdsignal, macdhist = talib.MACD(df.Close, fastperiod=12, slowperiod=26, signalperiod=9)
        df['macd'] = macdhist*1000
        final_macd_report.append(df)
   except:
       status = 'exception caught!'
   return {'filename': file, 'result': status}

（这只是一个简单的示例。您可能希望报告完整的异常回溯以帮助调试。）

如果工人长时间跑步，我希望尽快获得反馈。 所以我更喜欢使用imap_unordered，特别是如果某些任务可能比其他任务花费更长的时间。这将返回一个迭代器，该迭代器按照作业完成的顺序产生结果。

if __name__ == '__main__':

    with mp.Pool() as p:
        for res in p.imap_unordered(example_function, files):
            print(res)

通过这种方式，您可以明确地证明工人已完成工作，结果是什么以及是否发生了任何问题。

这比只从工作人员那里调用print 更可取。使用 stdout 缓冲和多个 worker 继承相同的输出流，当您真正看到某些东西时，就没有任何说法了。

编辑：如您所见，here、multiprocessing.Pool 不适用于交互式解释器，尤其是在 ms-windows 上。基本上，ms-windows 缺少让类 UNIX 系统复制进程的fork 系统调用。所以在 ms-windows 上，multiprocessing 必须尝试模仿fork，这意味着在子进程中导入原始程序文件。这不适用于 IPython 等交互式解释器。人们可能必须深入挖掘 Jupyter 和 multiprocessing 的内部结构才能找出问题的确切原因。

这个问题的解决方法似乎是在一个单独的模块中定义工作函数，然后在你的 IPython 代码中导入它。

它实际上在文档中提到multiprocessing.Pool 不适用于交互式解释器。见this section末尾的说明。