Python 2.7 - 如何防止断言释放锁答案

【问题标题】：Python 2.7 - How to prevent assertion from releasing locksPython 2.7 - 如何防止断言释放锁
【发布时间】：2023-03-17 03:22:01
【问题描述】：

当我的程序遇到断言失败时，我不希望断言库执行导致程序比没有断言失败时更进一步的事情。但这正是内置的assert 似乎在做的事情：它引发了一个异常，从而释放了锁。

例如，考虑以下程序。

import threading
import time

lock = threading.Lock()

class Thread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.start()
    def run(self):
        with lock:
            time.sleep(0.2)
            print "ASSERTION ON NEXT LINE"
            assert False

            # Were it not for the assert, 
            # This thread would hold the lock for a while longer.
            time.sleep(1.0)
Thread()        


time.sleep(0.1)
with lock:
    print "*** Main Thread ***"

断言导致锁被释放，这导致主线程在断言失败和回溯之间获取锁。结果，输出为：

ASSERTION ON NEXT LINE
*** Main Thread ***
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 810, in __bootstrap_inner
    self.run()
  File "so.py", line 14, in run
    assert False
AssertionError

这是非常不可取的，因为*** Main Thread *** 是在回溯之前打印的。这很容易让人误以为*** Main Thread *** 在生成异常之前就获得了锁。

我知道我们可以编写自己的断言函数来调用os._exit() 而不是引发断言。我想知道的是标准包中是否已经有类似的东西。

【问题讨论】：

不是assert而是上下文管理器with语句会自动释放锁。
您必须区分异常的生成和堆栈跟踪的输出。堆栈跟踪打印在线程中，但主程序继续运行。
它是否只是超出了锁定块？
@Daniel。是的，它发生了——因为assert 正在释放锁。持有锁的目的应该是停止其他线程运行。
没错。 Assert 引发异常，并且上下文管理器在未捕获的异常导致它超出范围时释放锁。您需要要么不生成异常，要么不使用上下文管理器。

标签： python multithreading python-2.7 assertion

【解决方案1】：

如果您不想释放lock，则必须在with lock: 中捕获异常：

with lock:
    try:
        time.sleep(0.2)
        print "ASSERTION ON NEXT LINE"
        assert False
    except Exception:
        print "there was an exception"
    while True:
        pass

【讨论】：

那当然等同于if not (condition): print "assert failed"。我们当然可以将其放入一个函数中并称之为“my_assert”。我正在寻找已经在标准包中的东西。
看来，您遇到了 X-Y-问题。如果是这样，if-解决方案是正确的方法。
我不知道你在说什么。我是否没有充分确定assert 的具体问题？或者我没有充分说明为什么这在多线程程序中是一个严重的问题？每种语言都有一个可用的assert，我们只需要了解如何在 Python 中导入流行的那个。
问题是大多数人不认为这是断言的问题，而是您的期望。您有两个不同步的事件：在线程的默认异常处理程序中打印堆栈跟踪和 main 中的打印语句。它们发生的顺序是不确定的。顺便说一句，您在启动线程时也有竞争条件。正确的做法是不要期望在相对于主线程的特定时间打印默认堆栈跟踪，并将您想要的任何错误处理放在线程中，持有您想要的任何锁。
@埃文。这是不准确的。大多数人期望两个事件在锁块内发生时是同步的。特别是当我们知道这些事件都不包含threading.Condition.wait() 或任何可以合理释放锁的东西时。

【解决方案2】：

如果我们使用 assert 并且失败了，它会引发一个断言错误。因为断言仅用于执行此操作（中断/停止流程）。我可以考虑将其与队列同步。

import time
import threading 
from Queue import Queue



q = Queue()
# code here
class MyThread(threading.Thread):
    def run(self):
        # do stuff
        print "ASSERTION ON NEXT LINE" 
        q.put(True)
        # do other stuff

# code here
mt = MyThread()
mt.start()

try:
    assert q.get()
finally:
    mt.join()

【讨论】：

我看不出q.put() 会如何阻止释放run() 中的锁。 q.put() 似乎是非阻塞的。此外，如果我们删除断言调用，我们需要在它的位置使用其他东西来打印回溯。

【解决方案3】：

感谢大家的贡献。我将主要以丹尼尔的回答为基础。我相信最好的答案取决于锁内有多少代码。如果你有一个大的函数调用，那么

with lock:
    try:
        deep_complicated_function()
    except Exception:
        print "there was an exception"
        print '\n'.join(traceback.format_stack()[:-1])
        os._exit(1)
    while True:
        pass

最好，因为它让您有机会在一个构造中为大量异常情况提供处理程序。

但是，对于一次性断言来说，这样的输入太多了：

with lock:
   quick_update()
   assert property()

这些情况需要更方便的方法，例如：

def Assert(cond, msg = "Assertion failed."):
    if not cond:
        print msg
        print '\n'.join(traceback.format_stack()[:-1])
        os._exit(1)

这将停止该过程，从而简化调试（如问题中所述）。要在nose 和其他测试工具中使用，请通过多处理 API 启动您的主程序，并在退出代码不为 0 时生成异常。

【讨论】：