Python 3:在多处理过程中捕获警告

Question

太长; 没看过

该warnings.catch_warnings()上下文管理是不是线程安全的.如何在并行处理环境中使用它？

背景

下面的代码使用Python multiprocessing模块的并行处理解决了最大化问题.它需要一个(不可变的)小部件列表,将它们分区(参见Python 3中大规模,强力最大化的高效多处理),找到所有分区的最大值("最终者"),然后找到最大值("冠军")那些"入围者".如果我正确地理解了我自己的代码(如果我这样做,我就不会在这里),我正在与所有子进程共享内存以给它们输入小部件,并multiprocessing使用操作系统级管道和pickle来发送当工人完成时,决赛小工具回到主要过程.

问题的根源

我希望捕获在窗口小部件从进程间管道出来时发生的取消对象之后由窗口小部件重新实例化引起的冗余窗口小部件警告.当窗口小部件对象实例化时,它们会验证自己的数据,从Python标准warnings模块发出警告,告诉应用程序的用户该窗口小部件怀疑用户的输入数据存在问题.因为unpickling导致对象实例化,所以我对代码的理解意味着每个widget对象只重新实例化一次,当且仅当它从管道出来后才是决赛者 - 请参阅下一节以了解为什么这不正确.

这些小部件在被擦除之前就已经创建了,因此用户已经痛苦地意识到他输入了什么输入并且不想再听到它.这些是我想要通过warnings模块的catch_warnings()上下文管理器捕获的警告(即with声明).

解决方案失败

在我的测试中,当多余的警告被发射到我在下面标记为A 线和B 线之间的任何地方时,我已经缩小了范围.让我感到惊讶的是,警告是在不仅仅是附近的地方发出的output_queue.get().这意味着我multiprocessing使用酸洗将小部件发送给工人.

结果是,warnings.catch_warnings()即使在从A行到B 行的所有内容中创建上下文管理器,并在此上下文中设置正确的警告过滤器也不会捕获警告.这意味着警告正在工作进程中发出.将此上下文管理器放在工作器代码周围也不会捕获警告.

代码

这个例子省略了决定,如果问题的规模太小,与派生工艺,进口多,并确定打扰的代码my_frobnal_counter,和my_load_balancer.

"Call `frobnicate(list_of_widgets)` to get the widget with the most frobnals"

def frobnicate_parallel_worker(widgets, output_queue):
    resultant_widget = max(widgets, key=my_frobnal_counter)
    output_queue.put(resultant_widget)

def frobnicate_parallel(widgets):
    output_queue = multiprocessing.Queue()
    # partitions: Generator yielding tuples of sets
    partitions = my_load_balancer(widgets)
    processes = []
    # Line A: Possible start of where the warnings are coming from.
    for partition in partitions:
        p = multiprocessing.Process(
                 target=frobnicate_parallel_worker,
                 args=(partition, output_queue))
        processes.append(p)
        p.start()
    finalists = []
    for p in processes:
        finalists.append(output_queue.get())
    # Avoid deadlocks in Unix by draining queue before joining processes
    for p in processes:
        p.join()
    # Line B: Warnings no longer possible after here.
    return max(finalists, key=my_frobnal_counter)

Answer 1

多年后，我终于有了一个解决方案（在解决一个不相关的问题时发现的）。我已经在 Python 3.7、3.8 和 3.9 上对此进行了测试。

暂时用空列表进行修补sys.warnoptions[]。您只需要在对 的调用周围执行此操作process.start()。sys.warnoptions被记录为您不应手动修改的实现细节；官方建议是使用warnings模块中的函数并PYTHONWARNINGS在os.environ. 这是行不通的。唯一有效的似乎是修补sys.warnoptions。在测试中，您可以执行以下操作：

import multiprocessing
from unittest.mock import patch
p = multiprocessing.Process(target=my_function)
with patch('sys.warnoptions', []):
    p.start()
p.join()

如果你不想使用unittest.mock，只需手动修补：

import multiprocessing
import sys
p = multiprocessing.Process(target=my_function)
old_warnoptions = sys.warnoptions
try:
    sys.warnoptions = []
    p.start()
finally:
    sys.warnoptions = old_warnoptions
p.join()