多进程运行时出现Can't get attribute 'faker_function'错误的原因排查

为对比multiprocessing多进程与普通线性函数的运行速度，编写了如下代码：

from multiprocessing import Pool, Manager
import time
from faker import Faker

num_rows = 1000000

items = [
    ['Column_1', Faker(), "pyint", {}],
    ['Column_2', Faker(), "random_element", {"elements": ["Kayden", "Franklin", "Gabriel", "Vincent"]}],
    ['Column_3', Faker(), "random_element", {"elements": ["Miller", "Ward", "Edwards", "Parry"]}],
    ['Column_4', Faker(), "pyint", {}],
    ['Column_5', Faker(), "pyint", {}],
    ['Column_6', Faker(), "pyint", {}],
    ['Column_7', Faker(), "pyint", {}],
    ['Column_8', Faker(), "pyint", {}],
    ['Column_9', Faker(), "pyint", {}],
    ['Column_10', Faker(), "pyint", {}],
    ['Column_11', Faker(), "pyint", {}],
    ['Column_12', Faker(), "pyint", {}],
    ['Column_13', Faker(), "pyint", {}],
    ['Column_14', Faker(), "pyint", {}],
    ['Column_15', Faker(), "pyint", {}],
    ['Column_16', Faker(), "pyint", {}],
    ['Column_17', Faker(), "pyint", {}],
    ['Column_18', Faker(), "pyint", {}],
    ['Column_19', Faker(), "pyint", {}],
    ['Column_20', Faker(), "pyint", {}],
    ['Column_21', Faker(), "pyint", {}],
    ['Column_22', Faker(), "pyint", {}],
    ['Column_23', Faker(), "pyint", {}],
]

def concurrent():
    with Manager() as dict_manager:
        data_frame = dict_manager.dict()

        global faker_function
        def faker_function(params):
            items = []
            for _ in range(0, num_rows - 1):
                items.append(getattr(params[1], params[2])(**params[3]))
            data_frame[params[0]] = items

        curr_time = time.time()
        with Pool(10) as p:
            p.map(faker_function, items)
        elapsed = time.time() - curr_time

        print('Concurrent', elapsed)
        print('Dict size', len(data_frame))

def linear():
    data_frame = {}

    def faker_function(params):
        items = []
        for _ in range(0, num_rows - 1):
            items.append(getattr(params[1], params[2])(**params[3]))
        data_frame[params[0]] = items

    curr_time = time.time()
    for item in items:
        faker_function(item)
    elapsed = time.time() - curr_time

    print('Linear time', elapsed)
    print('Dict size', len(data_frame))

if __name__ == "__main__":
    concurrent()

    linear()

运行时进程抛出错误：

Can't get attribute 'faker_function' on <module '__mp_main__'

请问该问题的原因是什么？

• Python多进程的Pool启动子进程时，会通过**序列化（pickle）**传递任务函数和参数。你在concurrent函数内部定义的faker_function是嵌套函数，即便声明了global也无法被子进程正确识别：
  • 子进程会尝试在主模块（__mp_main__）的全局命名空间查找faker_function，但这个函数实际定义在concurrent内部，主模块全局不存在该函数。
  • global关键字仅让concurrent内部的faker_function成为全局变量，但这个赋值是在concurrent执行时才完成的，子进程启动时可能还未完成赋值；且子进程的全局命名空间与主进程完全隔离，无法共享这个函数定义。
• 嵌套函数会携带父函数的上下文（比如data_frame这个Manager字典），pickle无法序列化这种带上下文的嵌套函数，这也是导致函数无法被找到的深层原因。

将faker_function移到模块的全局作用域，通过Pool.map的返回值收集结果（无需依赖Manager.dict共享数据，减少进程间通信开销），示例代码如下：

from multiprocessing import Pool, Manager
import time
from faker import Faker

num_rows = 1000000

items = [
    ['Column_1', Faker(), "pyint", {}],
    ['Column_2', Faker(), "random_element", {"elements": ["Kayden", "Franklin", "Gabriel", "Vincent"]}],
    ['Column_3', Faker(), "random_element", {"elements": ["Miller", "Ward", "Edwards", "Parry"]}],
    ['Column_4', Faker(), "pyint", {}],
    ['Column_5', Faker(), "pyint", {}],
    ['Column_6', Faker(), "pyint", {}],
    ['Column_7', Faker(), "pyint", {}],
    ['Column_8', Faker(), "pyint", {}],
    ['Column_9', Faker(), "pyint", {}],
    ['Column_10', Faker(), "pyint", {}],
    ['Column_11', Faker(), "pyint", {}],
    ['Column_12', Faker(), "pyint", {}],
    ['Column_13', Faker(), "pyint", {}],
    ['Column_14', Faker(), "pyint", {}],
    ['Column_15', Faker(), "pyint", {}],
    ['Column_16', Faker(), "pyint", {}],
    ['Column_17', Faker(), "pyint", {}],
    ['Column_18', Faker(), "pyint", {}],
    ['Column_19', Faker(), "pyint", {}],
    ['Column_20', Faker(), "pyint", {}],
    ['Column_21', Faker(), "pyint", {}],
    ['Column_22', Faker(), "pyint", {}],
    ['Column_23', Faker(), "pyint", {}],
]

# 移到全局作用域的任务函数
def faker_function(params):
    col_name, faker_inst, method, kwargs = params
    items = []
    for _ in range(num_rows - 1):
        items.append(getattr(faker_inst, method)(**kwargs))
    return (col_name, items)

def concurrent():
    curr_time = time.time()
    with Pool(10) as p:
        results = p.map(faker_function, items)
    # 用返回结果直接构建字典
    data_frame = dict(results)
    elapsed = time.time() - curr_time

    print('Concurrent', elapsed)
    print('Dict size', len(data_frame))

def linear():
    data_frame = {}
    curr_time = time.time()
    for item in items:
        col_name, items_list = faker_function(item)
        data_frame[col_name] = items_list
    elapsed = time.time() - curr_time

    print('Linear time', elapsed)
    print('Dict size', len(data_frame))

if __name__ == "__main__":
    concurrent()
    linear()

内容的提问来源于Stack Exchange，提问作者lapots