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()每个进程都会收到其列表 block 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 的 block ，本质上是在每个进程上重复工作。尝试将 unsorted_sub_lists 修复为长度 500k，或者仅删除 Pool.map 调用中的 chunksize 参数