### FastAPI 中使用 PyTorch 多进程时解决 `'not using fork'` 错误的解决方案 当在 FastAPI 应用中集成 PyTorch 并尝试使用多进程功能时，可能会遇到与启动方法相关的错误。这是因为某些操作系统（特别是 macOS 和 Windows）默认不支持 `fork` 方法来创建子进程，而是强制使用更安全但也更复杂的 `spawn` 或 `forkserver` 方法。 以下是针对该问题的具体分析和解决方案： #### 1. **错误原因** PyTorch 的多进程模块 (`torch.multiprocessing`) 默认会根据平台选择合适的启动方法。然而，在 Unix-like 系统上，默认可能仍然是 `fork` 方法。这种方法虽然高效，但它会在子进程中复制父进程的所有状态，这可能导致 CUDA 初始化失败或其它不可预见的行为[^1]。因此，推荐显式设置为 `spawn` 模式以避免此类问题。 此外，如果直接将自定义类的对象（如模型实例或处理器对象）作为参数传递给子进程，则可能出现序列化错误（例如 `Can't pickle local object`），这是由于这些对象无法被正确序列化所致[^2]。 --- #### 2. **解决方案** ##### （1）切换至 `spawn` 启动方法 可以通过以下代码确保所有子进程都采用 `spawn` 方式启动： ```python import torch.multiprocessing as mp mp.set_start_method('spawn', force=True) ``` 此命令应在程序入口处尽早执行，以便覆盖默认行为。注意，`force=True` 参数表示即使已设置了其他启动方法也会强行改为 `spawn`[^3]。 ##### （2）重构参数传递逻辑 对于需要传入子进程的目标函数而言，应当避免直接传递大型复杂对象（比如神经网络模型）。取而代之的是让每个子进程自行加载所需的数据结构。例如修改原代码如下： ```python from torch.multiprocessing import Process, Queue def worker_process(rank, queue): # 在每个子进程中重新加载模型和其他必要组件 model = load_model_from_checkpoint() image_processor = ImageProcessor() while True: data = queue.get() if data is None: # 结束信号 break processed_data = image_processor(data) output = model(processed_data) print(f"Process {rank} finished processing.") if __name__ == "__main__": manager = Queue(maxsize=10) num_processes = 4 processes = [] for i in range(num_processes): p = Process(target=worker_process, args=(i, manager)) p.start() processes.append(p) # 发送任务到队列... for _ in range(20): # 假设共有20条数据待处理 manager.put(some_input_data()) # 添加结束标志 for _ in range(num_processes): manager.put(None) # 等待所有子进程完成 for p in processes: p.join() ``` 在此版本中，我们不再向子进程传递预构建好的模型或图像处理器实例；相反，它们各自负责初始化自己的资源。这种做法不仅规避了序列化的难题，还提高了系统的鲁棒性[^2]。 ##### （3）结合 FastAPI 使用 最后一步就是把上述逻辑嵌套进 FastAPI APIRoute 定义里去。下面给出一个简单的例子展示如何做到这一点： ```python from fastapi import FastAPI, BackgroundTasks app = FastAPI() @app.post("/process/") async def process_images(background_tasks: BackgroundTasks, images: List[str]): input_queue = Queue(len(images)) def enqueue_inputs(): nonlocal input_queue for img_path in images: input_queue.put(img_path) background_tasks.add_task(enqueue_inputs) results = [] workers_count = min(mp.cpu_count(), len(images)) pool = Pool(workers_count, initializer=init_worker, initargs=(input_queue,)) try: results.extend(pool.imap_unordered(run_inference_on_image, range(workers_count))) finally: pool.terminate() pool.close() return {"results": results} ``` 这里利用了背景任务机制来异步填充输入队列，并通过进程池分配计算密集型作业给各个工作者节点处理[^4]。 --- ###