Python多进程Camera进程无法Join只能终止问题求助

背景 #

• 开发跨Windows 11和Debian（OrangePi5）的图像处理应用，基于Python multiprocessing 实现多进程架构，包含两个子进程：处理按钮IO操作的进程、分离相机功能的Camera进程
• 两个进程的实现逻辑统一：
  1. 配置3个队列：Data队列存储主输出元组；Config队列用于向工作进程发送配置方法与参数；Status队列接收配置确认信息
  2. 统一使用spawn方式启动进程
  3. 均包含start_process和stop_process方法：start初始化多进程字段并启动工作进程；stop设置停止事件，尝试join进程，失败则终止并递归记录详情

核心问题 #

• 按钮进程可正常join（超时≥0.1s即可），但Camera进程无论设置多久超时都无法join，只能通过terminate()强制终止
• 该问题在双系统、单独/同时运行两类进程时均复现；即使检测到进程已停止，join()仍会超时失败
• 推测存在资源阻塞，但不知具体排查方向；按钮进程传输不可变值，Camera进程传输1-2个np.array图像（通过image.copy()获取）、None或Exception

相关代码 #

按钮进程worker（可正常join） #

def _worker_process(self) -> None:
    try:
        while not self._stop_event.is_set():
            if not self._is_parent_alive():
                break

            if not self._config_queue.empty():
                config_message: tuple = self._config_queue.get()
                self._apply_config(config_message)

            is_pressed: bool = self._button.check_if_pressed()
            button_state_from_info, time_until_long_press = self._button.get_button_info(is_pressed)

            # Determine if the state should be updated
            state_idx: int = self._button_state_map.get(button_state_from_info, -1)
            update_state: bool = False

            # Button state pattern matching and state change record update here
            ...

            self._clear_queue(self._data_queue)
            self._data_queue_send(self._worker_but_st, time_until_long_press, st_ch_rec_tuple)

            time.sleep(0.05)

    except Exception:
        full_traceback = traceback.format_exc()
        self._clear_queue(self._data_queue)
        self._data_queue_send(exception=full_traceback)

Camera进程worker（无法join） #

def _worker_process(self) -> None:
    try:
        while not self._stop_event.is_set():

            if not self._is_parent_alive():
                break

            if not self._config_queue.empty():
                config_message: tuple = self._config_queue.get()
                self._apply_config(config_message)

            # Capture image(s)
            self.camera.capture()

            # Get the latest captured images based on exposure mode
            if self.camera.exp_mode == ExposureMode.SINGLE_EXP:
                latest_img_loexp = self.camera.get_last_img().copy()
                latest_img_hiexp = None
            elif self.camera.exp_mode == ExposureMode.DOUBLE_EXP:
                latest_img_loexp = self.camera.get_last_img_loexp().copy()
                latest_img_hiexp = self.camera.get_last_img_hiexp().copy()
            else:
                raise ValueError(f"Exposure mode ({self.camera.exp_mode}) either invalid or not recognised.")

            self._clear_queue(self._data_queue)
            self._data_queue_send(latest_img_loexp, latest_img_hiexp, None)

            time.sleep(0.05)

    except Exception:
        full_traceback = traceback.format_exc()
        self._clear_queue(self._data_queue)
        self._data_queue_send(exception=full_traceback)

共用的stop_process方法 #

def stop_process(self, is_recursive_call=False) -> None:
    
    self._stop_event.set()

    if self._process is not None and self._process.is_alive():
        try:
            self._process.join(timeout=0.2)

            # If the process is still alive and this is not a recursive call, terminate and retry
            if self._process.is_alive() and not is_recursive_call:
                print(f"{self.CLASS_NAME}: Failed to join worker process, terminating.")
                self._process.terminate()
                self.stop_process(is_recursive_call=True)
            else:
                self._report_on_stop_attempt()
        
        except KeyboardInterrupt as e:
            print(f"{self.CLASS_NAME}: KeyboardInterrupt in process joining, joining failed.")
            raise e

    elif self._process and not self._process.is_alive():
        print(f"{self.CLASS_NAME}: Unexpected stop behavior - process was not alive "
            f"on 'stop_process' ({self._process=}).")
    else:
        print(f"{self.CLASS_NAME}: Unexpected stop behavior - process was None.")

排查方向与解决方案建议 #

排查方向 #

• Camera硬件资源未释放：self.camera.capture()或图像获取方法可能阻塞在硬件调用上，比如相机驱动未响应、缓冲区未清空，导致进程无法退出循环
• 队列操作阻塞：_clear_queue或_data_queue_send处理大np.array时，若主进程未及时消费队列数据，会导致子进程卡在发送环节，无法响应停止事件
• 异常处理遗漏：相机操作可能抛出未被外层try捕获的底层异常，导致进程卡在错误点无法退出
• spawn模式资源继承问题：spawn模式会重新初始化进程，Camera实例可能在子进程中持有未正确初始化的硬件句柄，退出时无法释放
• 进程状态检测误差：is_alive()可能存在延迟，或进程已退出但资源未回收，导致join()超时

解决方案建议 #

1. 强制释放Camera资源：在Camera进程的_worker_process中添加finally块，确保退出时释放相机资源：

   def _worker_process(self) -> None:
       try:
           # 原循环逻辑...
       except Exception:
           # 原异常处理...
       finally:
           # 释放相机资源，假设Camera类有close方法
           if hasattr(self, 'camera'):
               self.camera.close()
               self.camera = None
   

2. 清空队列避免阻塞：在stop_process中先清空Data和Config队列，防止子进程卡在队列读写：

   def stop_process(self, is_recursive_call=False) -> None:
       self._stop_event.set()
       
       # 先清空队列，避免子进程阻塞
       for queue in [self._data_queue, self._config_queue]:
           if queue is not None:
               while not queue.empty():
                   try:
                       queue.get_nowait()
                   except:
                       pass
       
       # 原join逻辑...
   

3. 给相机操作单独加异常捕获：将相机捕获与图像获取逻辑包裹在单独的try-except中，确保错误触发进程退出：

   # 在Camera进程循环内替换原相机操作代码
   try:
       self.camera.capture()
       # 原图像获取逻辑...
   except Exception as e:
       full_traceback = traceback.format_exc()
       self._clear_queue(self._data_queue)
       self._data_queue_send(exception=full_traceback)
       break  # 强制退出循环
   

4. 检查Camera类的线程安全：若Camera类内部使用了线程，需确保所有子线程在关闭时能正常停止，避免线程残留导致进程无法终止

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