Python multiprocessing用c_char_p/c_wchar_p共享值时无异常退出是否正常？

问题描述 #

尝试用multiprocessing.Value(ctypes.c_wchar_p, "")在进程间共享字符串时，程序无提示意外退出，退出码为0xC0000005。推测问题是主进程访问了子进程的内存，且该问题仅在使用ctypes.c_char_p/c_wchar_p时出现，线程场景下运行正常。这是预期行为吗？

复现代码 #

import multiprocessing
import threading
import ctypes
import time


class ProcessExample:

    def __init__(self, proc: bool):
        self.shared_bool = multiprocessing.Value(ctypes.c_bool, False)
        self.shared_string = multiprocessing.Value(ctypes.c_wchar_p, "one")
        self.shared_byte = multiprocessing.Value(ctypes.c_char_p, b"one")
        if proc:
            self.th_proc = multiprocessing.Process(target=self.target, args=(self.shared_bool, self.shared_string, self.shared_byte))
        else:
            self.th_proc = threading.Thread(target=self.target, args=(self.shared_bool, self.shared_string, self.shared_byte))
        self.th_proc.start()

    @staticmethod
    def target(shared_bool, shared_string, shared_byte):
        shared_bool.value = True
        shared_string.value = "two"
        shared_byte.value = b"two"

    def print_values(self):
        self.th_proc.join()
        print(self.shared_bool.value)
        print(self.shared_string.value)  # 此处程序退出
        print(self.shared_byte.value)  # 此处程序退出


if __name__ == "__main__":
    # example = ProcessExample(False)  # 线程模式正常运行
    example = ProcessExample(True)
    time.sleep(1)
    example.print_values()

程序输出 #

True

Process finished with exit code -1073741819 (0xC0000005) 

解答 #

这是预期行为，核心原因在于指针类型的内存特性：

• ctypes.c_char_p和ctypes.c_wchar_p是指针类型，multiprocessing.Value仅会在共享内存中存储指针本身，而指针指向的字符串数据属于进程私有内存，不在共享区域内。
• 子进程执行shared_string.value = "two"时，实际是将指针指向了子进程自身内存中的字符串地址，这个地址对主进程来说是无效的私有内存空间。
• 主进程尝试读取该指针指向的地址时，会触发Windows系统的内存访问违规错误（错误码0xC0000005），导致程序直接崩溃退出。
• 线程模式下所有线程共享同一进程的内存空间，指针指向的地址在所有线程中都有效，因此不会出现问题。

正确的进程间字符串共享方式 #

1. 使用multiprocessing.Array存储字符数组：需要预先分配足够的空间来容纳字符串

   # 示例：共享宽字符数组，长度设为10
   shared_string = multiprocessing.Array(ctypes.c_wchar, "one", 10)
   # 修改时需要先转为字符串再赋值
   shared_string[:3] = "two"
   

2. 使用multiprocessing.Manager代理共享：Manager通过进程间通信封装了数据共享，无需手动处理内存

   manager = multiprocessing.Manager()
   shared_string = manager.Value(str, "one")
   

3. 自定义序列化共享：将字符串序列化为字节流后存入共享内存，读取时再反序列化。

修改后的可运行示例 #

import multiprocessing
import threading
import ctypes
import time


class ProcessExample:

    def __init__(self, proc: bool):
        self.shared_bool = multiprocessing.Value(ctypes.c_bool, False)
        # 使用Manager创建可安全共享的字符串对象
        manager = multiprocessing.Manager()
        self.shared_string = manager.Value(str, "one")
        self.shared_byte = manager.Value(bytes, b"one")
        if proc:
            self.th_proc = multiprocessing.Process(target=self.target, args=(self.shared_bool, self.shared_string, self.shared_byte))
        else:
            self.th_proc = threading.Thread(target=self.target, args=(self.shared_bool, self.shared_string, self.shared_byte))
        self.th_proc.start()

    @staticmethod
    def target(shared_bool, shared_string, shared_byte):
        shared_bool.value = True
        shared_string.value = "two"
        shared_byte.value = b"two"

    def print_values(self):
        self.th_proc.join()
        print(self.shared_bool.value)
        print(self.shared_string.value)
        print(self.shared_byte.value)


if __name__ == "__main__":
    example = ProcessExample(True)
    time.sleep(1)
    example.print_values()

修改后输出 #

True
two
two

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