解决Python类中使用Multiprocessing时的weakref pickle错误

问题背景 #

开发Python FastAPI应用时，使用multiprocessing库启动并行的忙碌循环进程。最初在main.py的启动阶段创建进程完全正常，后来为了让特定FastAPI端点也能控制这些进程，将进程创建逻辑移至单例类thread_utils中，供启动流程和端点调用。但移入类后，启动多个进程时触发如下错误：

报错信息 #

File "/home/user/.local/lib/python3.8/site-packages/starlette/routing.py", line 677, in lifespan
    async with self.lifespan_context(app) as maybe_state:
  File "/home/user/.local/lib/python3.8/site-packages/starlette/routing.py", line 566, in __aenter__
    await self._router.startup()
  File "/home/user/.local/lib/python3.8/site-packages/starlette/routing.py", line 654, in startup
    await handler()
  File "/home/user/Projects/test/test-api/api/main.py", line 127, in startup_event
    thread_utils.launch_all_threads()
  File "/home/user/Projects/test/test-api/api/core/thread_utils.py", line 66, in launch_all_threads
    p.start()
  File "/usr/lib/python3.8/multiprocessing/process.py", line 121, in start
    self._popen = self._Popen(self)
  File "/usr/lib/python3.8/multiprocessing/context.py", line 224, in _Popen
    return _default_context.get_context().Process._Popen(process_obj)
  File "/usr/lib/python3.8/multiprocessing/context.py", line 284, in _Popen
    return Popen(process_obj)
  File "/usr/lib/python3.8/multiprocessing/popen_spawn_posix.py", line 32, in __init__
    super().__init__(process_obj)
  File "/usr/lib/python3.8/multiprocessing/popen_fork.py", line 19, in __init__
    self._launch(process_obj)
  File "/usr/lib/python3.8/multiprocessing/popen_spawn_posix.py", line 47, in _launch
    reduction.dump(process_obj, fp)
  File "/usr/lib/python3.8/multiprocessing/reduction.py", line 60, in dump
    ForkingPickler(file, protocol).dump(obj)
TypeError: cannot pickle 'weakref' object

原本在main.py中正常运行的代码，移入类后仅能启动一个进程，启动多个就会触发该TypeError。曾推测是Python进程fork机制问题，参考过CPython相关issue和multiprocessing文档的安全导入说明，但并未解决。

出错的代码示例 #

main.py #

from .core.thread_utils import thread_utils
thread_utils = thread_utils()
...
@app.on_event("startup") # FastAPI启动事件
async def startup_event():
    # 之前在这里创建进程完全正常
    thread_utils.launch_all_threads()

core/thread_utils.py #

from multiprocessing import Process
from ..core import busy_loop

class thread_utils:
    _self = None

    def __new__(cls):
        if cls._self is None:
            cls._self = super().__new__(cls)
            return cls._self

    def launch_runner_thread(self, t:int):
        busy_loop(t)

    def launch_all_threads(self):
        t_list = [1,2]
        for t in t_list:
            p = Process(target=self.launch_runner_thread, args=(t,))
            p.start()

解决方法 #

经@pts建议，将单例类改为无类的函数形式后问题解决，工作代码如下：

main.py #

from .core.thread_utils import launch_all_threads
...
@app.on_event("startup")
async def startup_event():
    launch_all_threads()

core/thread_utils.py #

from multiprocessing import Process
from ..core import busy_loop

def launch_runner_thread(t:int):
    busy_loop(t)

def launch_all_threads():
    t_list = [1,2]
    for t in t_list:
        p = Process(target=launch_runner_thread, args=(t,))
        p.start()

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