多进程中使用ctypes结构遇PicklingError问题求助

问题原因 #

你遇到的PicklingError是因为动态生成的ctypes数组类型（比如c_char_Array_4这类）无法被Python的pickle机制序列化。单个ctypes基本类型（如c_int、c_char）自带pickle支持，但通过(c_char * N)动态创建的数组类是临时生成的，pickle在反序列化时找不到对应的类定义，因此报错。

解决方案 #

方案1：转成Python原生类型传递（最简单） #

把ctypes数组转换成Python原生的bytes或列表，传递到子进程后再转回ctypes数组，避开直接序列化ctypes数组类型。

修改后的代码示例：

import concurrent.futures
import time
from ctypes import *


def test_c_val(c_val):
    # 若传入的是bytes，转回对应的ctypes数组
    if isinstance(c_val, bytes):
        arr_type = c_char * len(c_val)
        c_val = arr_type.from_buffer_copy(c_val)
    
    # 兼容单个值和数组的打印逻辑
    print(c_val.value if hasattr(c_val, 'value') else bytes(c_val))
    return c_val.value if hasattr(c_val, 'value') else bytes(c_val)

test_int = c_int(55)
test_char = c_char(str(6).encode())
arr = [str(i).encode() for i in range(4)]
test_c_array = (c_char * len(arr))(*arr)

futures = []
with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
    futures.append(executor.submit(test_c_val, test_int))
    futures.append(executor.submit(test_c_val, test_char))
    # 将ctypes数组转为bytes传递
    futures.append(executor.submit(test_c_val, bytes(test_c_array)))
    time.sleep(2)
    
for future in futures:
    try:
        print(f"结果：{future.result()}")
    except Exception as e:
        print(f"错误：{e}")

方案2：使用共享内存（适合大数据量） #

如果数组体积较大，转原生类型的拷贝开销高，可以用multiprocessing的共享内存，让子进程直接访问内存区域，无需序列化。

示例代码：

import concurrent.futures
import time
from ctypes import *
import multiprocessing


def test_c_val(c_val):
    print(c_val.value)
    return c_val.value

def test_c_shared_arr(arr):
    # 将共享内存对象转为ctypes数组
    c_arr = cast(arr._obj, POINTER(c_char * len(arr))).contents
    print(bytes(c_arr))
    return bytes(c_arr)

if __name__ == "__main__":
    arr = [str(i).encode() for i in range(4)]
    # 创建共享内存的ctypes数组
    shared_arr = multiprocessing.Array(c_char, len(arr))
    # 写入数据
    for i, b in enumerate(arr):
        shared_arr[i] = b[0]
    
    test_int = c_int(55)
    test_char = c_char(str(6).encode())

    futures = []
    with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
        futures.append(executor.submit(test_c_val, test_int))
        futures.append(executor.submit(test_c_val, test_char))
        futures.append(executor.submit(test_c_shared_arr, shared_arr))
        time.sleep(2)
    
    for future in futures:
        try:
            print(f"结果：{future.result()}")
        except Exception as e:
            print(f"错误：{e}")

方案3：自定义pickle序列化规则（最灵活） #

给动态生成的ctypes数组类型注册pickle的序列化和反序列化函数，让pickle能正确识别并处理这类类型。

示例代码：

import concurrent.futures
import time
from ctypes import *
import pickle


# 定义ctypes数组的pickle序列化逻辑
def pickle_ctypes_array(arr):
    # 保存数组的元素类型、长度和数据
    elem_type = arr._type_
    length = len(arr)
    data = bytes(arr)
    return unpickle_ctypes_array, (elem_type, length, data)

# 定义反序列化逻辑
def unpickle_ctypes_array(elem_type, length, data):
    arr_type = elem_type * length
    return arr_type.from_buffer_copy(data)

# 给所有ctypes数组类型注册pickle处理函数
import ctypes
for name in dir(ctypes):
    cls = getattr(ctypes, name)
    if isinstance(cls, type) and issubclass(cls, ctypes.Array):
        pickle.register(cls, pickle_ctypes_array)


def test_c_val(c_val):
    print(c_val.value if hasattr(c_val, 'value') else bytes(c_val))
    return c_val.value if hasattr(c_val, 'value') else bytes(c_val)

test_int = c_int(55)
test_char = c_char(str(6).encode())
arr = [str(i).encode() for i in range(4)]
test_c_array = (c_char * len(arr))(*arr)

futures = []
with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
    futures.append(executor.submit(test_c_val, test_int))
    futures.append(executor.submit(test_c_val, test_char))
    futures.append(executor.submit(test_c_val, test_c_array))
    time.sleep(2)
    
for future in futures:
    try:
        print(f"结果：{future.result()}")
    except Exception as e:
        print(f"错误：{e}")

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