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

Question

当我的程序遇到断言失败时，我不希望断言库做一些导致程序比没有断言失败时更进一步的事情。但这正是内置 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() 而不是引发断言。我想知道标准包中是否已经有类似的东西。

Answer 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

Answer 2

如果我们使用 assert 并且失败它会引发断言错误。因为断言仅设计用于执行此操作（break/stop 流程）。我可以考虑将其与 Queue 同步。

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()

Answer 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 时生成异常。

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

Python 2.7 - How to prevent assertion from releasing locks

python

multithreading

assertion

python-2.7