以并行模式运行命令行应用程序答案

【问题标题】：Run command line application in parallel mode以并行模式运行命令行应用程序
【发布时间】：2018-04-10 11:07:29
【问题描述】：

我正在尝试通过运行第三方可执行应用程序来实现简单的并行数据处理，该应用程序从我的 Python 应用程序启动并面临一些与并行执行相关的有趣问题。

用例非常简单。我有一个应通过第三方 .exe 文件处理的数据对象列表。我们称之为consoleApp.exe。每个数据对象都应该通过调用这个 consoleApp 处理几次。出于测试目的，此控制台应用程序只需将一些文本写入控制台，稍等片刻然后退出。

这是我的 Python 代码，它将执行此类处理

def ProcessFile(idx, row):
    config = UtilitySettings.ConfigFile + " some_arguments"
    config2 = UtilitySettings.config2File + " some_arguments_2"
    config3 = UtilitySettings.config3File + " some_arguments_3"
    config4 = UtilitySettings.config4File + " some_arguments_4"

    fullFileName = AppSettings.BinaryDataDirectory + row.FileName

    cmd1 = "ConsoleApp.exe" + ' ' + fullFileName + ' ' + config
    cmd2 = "ConsoleApp.exe" + ' ' + fullFileName + ' ' +  config2
    cmd3 = "ConsoleApp.exe" + ' ' + fullFileName + ' ' +  config3
    cmd4 = "ConsoleApp.exe" + ' ' + fullFileName + ' ' +  config4

    devnull = open(os.devnull, 'w')
    call(cmd1, stdout=devnull, stderr=devnull)
    call(cmd2, stdout=devnull, stderr=devnull)
    call(cmd3, stdout=devnull, stderr=devnull)

下一段代码将开始并行处理：

class ConvertDataCommand(ICommand):
    def Execute(self):
        startExecutionTime = time.time()
        result = CommandExecutionResult()

        try:
            geoDataFrame = geopandas.GeoDataFrame(objectInfo, crs=crs, geometry = objectInfo.geometry)

            # let's use cpu count - 1 . The last one core will be used by the currently executed parent thread
            coreCount = multiprocessing.cpu_count() - 1
            dataFrameLength = len(geoDataFrame)

            # splitting dataFrame to the chuncks to perform parallel processing
            chunksCount = dataFrameLength / coreCount if dataFrameLength % coreCount == 0 else dataFrameLength / coreCount + 1
            chunkedArr = np.array_split(geoDataFrame, chunksCount)

            for slice in chunkedArr:
                pool = multiprocessing.Pool(processes=coreCount)
                results = [pool.apply(ProcessFile, args=(idx, row)) for idx, row in slice.iterrows()]
                pool.close()
                pool.join()

            result.Success = True

        except BaseException as e:
            result.Success = False
            stacktrace = traceback.format_exc()
            Logger.Log(stacktrace)

        finally:
            result.ExecutionTime = time.time() - startExecutionTime
            return result

主要有趣的是，在我的例子中，并行处理比我们按顺序处理这些数据（142 秒）花费更多的时间（178 秒）（但它不应该）。看起来这个控制台应用程序被所有内核使用（但我希望每个进程都会调用 consoleApp.exe 的一个新实例并在每个进程中执行它）我在更改consoleApp.exe 的实现时发现了它。我只是写了一个无限循环并再次启动一个 Python 程序。然后打开进程资源管理器，它只显示了一个 ConsoleApp.exe 实例和我的 Python 应用程序创建的 3 个进程。

有人知道我做错了什么吗？

【问题讨论】：

标签： python windows python-3.x console-application

【解决方案1】：

哦，太好了。通过调用 pool.map() 函数而不是 pool.apply() 解决了这个问题（因为它会阻塞线程，直到结果准备好。有关更多详细信息，请参阅python-docs）

并更改调用方法的签名：

def ProcessFile(idx, row):

到：

def ProcessFile(args):
    idx, row = args

for slice in chunkedArr:
    pool = multiprocessing.Pool(processes=coresNum)

    data = [(idx, row) for idx, row in slice.iterrows()]
    results = pool.map(ProcessFile, data)

    pool.close()
    pool.join()

【讨论】：