Mac 上的 Python Multiprocessing Pool.map() 以串行方式而不是并行方式运行进程

Question

我一直在编写一个脚本来对大型数据集进行排序和过滤，并将工作拆分到多个 CPU 内核上（通过使用多个进程），但是，Python 似乎一次启动每个进程，以串行方式而不是并行方式运行它们.

我已经剥离了代码，因此它基本上没有做任何有用的事情（它生成一个随机数列表并简单地将它们全部删除），问题仍然存在。这是 Mac 上 Python 的问题吗？

我在 OS X 10.13.6 上运行 Python 3.7.1。

这是完整的代码：

import math
import multiprocessing
import os
import random
import sys
import timeit


def delete_all(passed_nums):

    print("Started process: {}, {}".format(multiprocessing.current_process(), os.getpid()))
    while (len(passed_nums) > 0):
        passed_nums.remove(passed_nums[0])
    print("Finished process: {}, {}".format(multiprocessing.current_process(), os.getpid()))

    return passed_nums


def chunksl(l, n):

    i = [l[i:i + n] for i in range(0, len(l), n)]
    return i


def main():

    rnd_nums = random.sample(range(1, 1000000), 500000)
    num_processes = 1
    Pool = multiprocessing.Pool(num_processes)

    list_chunk_size_per_core = int(math.ceil(len(rnd_nums)/float(num_processes)))

    unsorted_sub_lists = list(chunksl(rnd_nums, list_chunk_size_per_core))

    print("Number of CPUs:  {}".format(num_processes))
    print("Chunk size per CPU: {}".format(list_chunk_size_per_core))
    print("Number of chunks: {}".format(len(unsorted_sub_lists)))

    start_time = timeit.default_timer()
    sorted_sub_lists = Pool.map(delete_all, unsorted_sub_lists, list_chunk_size_per_core)
    end_time = timeit.default_timer()
    print('Duration: {}'.format(end_time - start_time))

    return True


if __name__ == '__main__':
    sys.exit(main())

这是num_processes = 1的输出：

Number of CPUs:  1
Chunk size per CPU: 500000
Number of chunks: 1
Started process: <ForkProcess(ForkPoolWorker-1, started daemon)>, 1617
Finished process: <ForkProcess(ForkPoolWorker-1, started daemon)>, 1617
Duration: 23.922029328999997

这是num_processes = 2的输出：

Number of CPUs:  2
Chunk size per CPU: 250000
Number of chunks: 2
Started process: <ForkProcess(ForkPoolWorker-1, started daemon)>, 1630
Finished process: <ForkProcess(ForkPoolWorker-1, started daemon)>, 1630
Started process: <ForkProcess(ForkPoolWorker-1, started daemon)>, 1630
Finished process: <ForkProcess(ForkPoolWorker-1, started daemon)>, 1630
Duration: 11.938197925

最后，这是num_processes = 1 的输出，但列表的大小减少到 250,000 个条目而不是 500,000 个：

Number of CPUs:  1
Chunk size per CPU: 250000
Number of chunks: 1
Started process: <ForkProcess(ForkPoolWorker-1, started daemon)>, 1639
Finished process: <ForkProcess(ForkPoolWorker-1, started daemon)>, 1639
Duration: 5.904828338

可以看出，num_processes = 2 时脚本运行速度更快，但不是因为它是并行运行进程，而是因为删除两个 250k 项目列表中的所有条目比删除一个 500k 项目列表中的所有条目更快（ num_processes = 2 时的输出是上次运行时长的两倍 num_processes = 1 但列表大小减少到 250k 条目，这也大约是第一次运行时间的四分之一）。

我的理解是，当启动一个新进程时，使用Pool.map() 每个进程都会收到其列表块unsorted_sub_lists 的完整副本，这意味着多个进程不会阻止尝试访问原始unsorted_sub_lists同时列出。 Python 没有通过引用传递给新进程。我可以在脚本末尾打印列表unsorted_sub_lists 并且原始内容仍然存在所以我假设我的理解是正确的？

Answer 1

对于n 进程，变量unsorted_sub_lists 具有n 元素。因此，当您传递 chunksize=list_chunk_size_per_core 时，其中 list_chunk_size_per_core 为 250k，您将长度为 2 的列表分块为最大长度为 250k 的块，实质上是重复每个进程的工作。尝试将您的 unsorted_sub_lists 修复为长度 500k，或者仅删除 Pool.map 调用中的 chunksize 参数