如何等待multiprocessing.Value的值发生变化？

我使用一个跨进程布尔值通知子进程清理并优雅退出：

self.mem_manager = multiprocessing.Manager()
self.exiting = self.mem_manager.Value(ctypes.c_byte, False)

当前我的进程run()方法末尾通过轮询实现等待：

while not self.exiting.value:
    time.sleep(1)
self.terminate_threads()
return

这种方式不够优雅，通常此类场景会使用条件变量等待。我已有一个处理线程错误的条件变量：

error_condition = threading.Condition()
... threads have access to this and will notify on error and stop ...

... more code ...

while not self.exiting.value:
    with error_condition:
        error_condition.wait(1)
        if self.errors_happened():
            ... handle errors ...
            break
self.terminate_threads()
return

我希望能实现如下逻辑，等待任一条件触发：

while not self.exiting.value:
    wait_for_either(error_condition, changed_condition(self.exiting))

请问能否实现这种等待任一条件触发的功能？

完全可以实现这种需求，下面提供两种简洁实用的方案：

方案一：复用现有错误条件变量（推荐） #

不需要新增额外的同步对象，只需在触发退出信号的代码中，同时通知现有的error_condition即可：

1. 修改退出触发逻辑：
   在设置self.exiting.value = True的地方（通常是控制进程或主进程中），添加条件变量通知：

   # 触发子进程退出时执行
   self.exiting.value = True
   with self.error_condition:
       self.error_condition.notify_all()
   

2. 保留原有等待循环：
   原有的等待逻辑无需大幅修改，每次被唤醒时同时检查退出信号和错误状态：

   while not self.exiting.value:
       with self.error_condition:
           # 等待超时或被通知唤醒
           self.error_condition.wait(1)
           if self.errors_happened():
               ...  # 执行错误处理逻辑
               break
   self.terminate_threads()
   return
   

这种方式利用现有同步结构，代码改动最小，同时实现了“错误触发”或“退出信号”任一条件唤醒的效果。

方案二：新增退出事件监听 #

如果希望退出逻辑与错误逻辑完全解耦，可以新增一个线程专门监听跨进程的exiting变量，并用threading.Event传递信号：

1. 初始化退出事件与监听线程：
   在进程run()方法开头添加如下代码：

   self.exit_event = threading.Event()
   
   def monitor_exit_signal(self):
       # 短间隔轮询跨进程退出变量，资源占用极低
       while not self.exiting.value:
           time.sleep(0.1)
       self.exit_event.set()
   
   # 启动守护线程，随主进程结束自动退出
   self.monitor_thread = threading.Thread(target=self.monitor_exit_signal, daemon=True)
   self.monitor_thread.start()
   

2. 等待任一条件触发：
   循环中同时等待退出事件和错误条件变量：

   while not self.exiting.value and not self.errors_happened():
       # 优先等待退出事件，超时1秒
       if self.exit_event.wait(1):
           break
       # 检查错误条件是否被触发
       with self.error_condition:
           if self.errors_happened():
               ...  # 执行错误处理逻辑
               break
           # 非阻塞等待错误通知，避免错过刚触发的信号
           self.error_condition.wait(0)
   self.terminate_threads()
   return
   

内容的提问来源于stack exchange，提问作者Tomáš Zato