如何解决concurrent.futures与multiprocessing.Process混用的弃用警告

问题背景 #

测试场景包含：

• 测试对象Obj通过ProcessPoolExecutor处理CPU密集型任务
• 用multiprocessing.Process启动模拟外部服务器
• 运行测试时触发DeprecationWarning，提示多线程环境下使用fork可能导致死锁，且尝试全局设置multiprocessing.set_start_method("spawn")无效

复现代码 #

pytest环境内代码 #

import asyncio
import concurrent.futures
import http.server
import multiprocessing
import os

import pytest
import pytest_asyncio

def hello_world():
    print("Hello, World!")

class Obj():
    def __init__(self):
        self.pool = concurrent.futures.ProcessPoolExecutor(max_workers=1)
        asyncio.get_running_loop().run_in_executor(self.pool, hello_world)

@pytest_asyncio.fixture
async def obj_to_test():
    return Obj()

@pytest.fixture
def mock_server(tmpdir):
    def server(path, host, port):
        os.chdir(path)
        httpd = http.server.HTTPServer(
            (host, port),
            http.server.SimpleHTTPRequestHandler
        )
        httpd.serve_forever()

    process = multiprocessing.Process(target=server, args=(".", "127.0.0.1", 9999))
    process.start()
    yield process
    process.terminate()
    process.join()

def test_something(obj_to_test, mock_server):
    # obj_to_test与mock_server交互逻辑
    pass

pytest环境外复现代码 #

import concurrent.futures
import multiprocessing
import time

pool = concurrent.futures.ProcessPoolExecutor(max_workers=1)
job = pool.submit(lambda: time.sleep(5))

process = multiprocessing.Process(target=lambda: None)
process.start()
process.terminate()

运行命令及警告：

PYTHONWARNINGS=default python3 t5.py
/usr/local/lib/python3.12/multiprocessing/popen_fork.py:66: DeprecationWarning: This process (pid=1024) is multi-threaded, use of fork() may lead to deadlocks in the child.
  self.pid = os.fork()

解决方案 #

核心思路 #

Python 3.12中，主进程存在多线程（如asyncio事件循环线程）时，调用fork()会触发警告——因为fork会复制主进程的线程状态，但子进程不会启动这些线程，极易引发死锁。改用spawn启动方式可避免该问题，它会启动全新的Python进程，不继承主进程的线程。

具体修改 #

1. 给ProcessPoolExecutor指定spawn上下文
   修改Obj类的初始化代码，显式指定进程池使用spawn启动方式：

class Obj():
    def __init__(self):
        # 创建spawn类型的多进程上下文
        ctx = multiprocessing.get_context('spawn')
        # 给ProcessPoolExecutor传入该上下文
        self.pool = concurrent.futures.ProcessPoolExecutor(max_workers=1, mp_context=ctx)
        asyncio.get_running_loop().run_in_executor(self.pool, hello_world)

2. 给模拟服务器的Process指定spawn上下文
   修改mock_server fixture，使用spawn上下文创建进程：

@pytest.fixture
def mock_server(tmpdir):
    def server(path, host, port):
        os.chdir(path)
        httpd = http.server.HTTPServer(
            (host, port),
            http.server.SimpleHTTPRequestHandler
        )
        httpd.serve_forever()

    # 创建spawn类型的上下文
    ctx = multiprocessing.get_context('spawn')
    # 通过上下文创建Process
    process = ctx.Process(target=server, args=(str(tmpdir), "127.0.0.1", 9999))
    process.start()
    yield process
    process.terminate()
    process.join()

为什么全局设置set_start_method("spawn")无效？ #

• pytest-asyncio在运行测试前已启动asyncio事件循环，此时全局调用set_start_method会抛出RuntimeError（不能在进程启动后修改启动方式）。
• spawn启动方式要求目标函数和参数可序列化（pickle），全局设置可能会导致某些测试对象无法序列化，而单独给每个进程/进程池指定上下文更灵活，可针对性处理序列化问题。

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