多进程池Join时因[Errno 61]连接拒绝死锁，求跨平台最优进程数方案

测试代码 #

from multiprocessing import Pool, Manager
from os import cpu_count

PROCESSES = cpu_count() // 2

N = 250

def add_to_list(the_list, the_value):
    try:
        the_list.append(the_value)
    except Exception as e:
        print(e)


def main():
    with Manager() as manager:
        myList = manager.list()
        pool = Pool(PROCESSES)
        for i in range(N):
            pool.apply_async(add_to_list, (myList, i))
        print('Closing...')
        pool.close()
        print('Joining...')
        pool.join()
        print(myList)

if __name__ == '__main__':
    main()

问题现象 #

系统cpu_count()返回20，当进程数设为10时代码正常输出0-249的列表；但将进程池设为默认20，或进程数≥18时，add_to_list函数多次抛出[Errno 61] Connection refused异常，且pool.join()会无限阻塞，程序无法正常结束。强制终止后栈信息指向return self._semlock.__enter__()，推测是Python信号量实现的平台相关问题。

平台信息 #

macOS 13.5.1
3 GHz 10-Core Intel Xeon W
32 GB 2666 MHz DDR4
Python 3.11.5

解决方法 #

1. 动态设置平台适配的安全进程数 #

针对不同平台设置进程数上限，避免触发macOS的信号量或连接限制：

import sys
from os import cpu_count

def get_safe_process_count():
    core_count = cpu_count() or 4  # 当无法获取核心数时用默认值4
    if sys.platform == "darwin":  # 判断是否为macOS
        # 限制为核心数的80%，且最高不超过16，平衡性能与稳定性
        return min(int(core_count * 0.8), 16)
    else:
        # 其他平台可以用核心数的一半，或根据需求调整
        return core_count // 2 if core_count > 2 else core_count

PROCESSES = get_safe_process_count()

2. 替换跨进程共享数据结构 #

如果业务允许，尽量避免使用Manager类共享数据，改用进程返回结果后本地汇总的方式，减少跨进程通信的压力：

from multiprocessing import Pool
from os import cpu_count

PROCESSES = cpu_count() // 2

N = 250

def process_value(val):
    return val

def main():
    with Pool(PROCESSES) as pool:
        # 用map批量处理，结果直接返回后汇总
        results = pool.map(process_value, range(N))
    print(results)

if __name__ == '__main__':
    main()

3. 升级Python版本 #

部分旧版本Python在macOS的多进程实现存在bug，升级到3.12及以上版本，可能已经修复了该平台下的信号量或连接问题。

内容的提问来源于stack exchange，提问作者jackal