如何为Python实例方法添加时间延迟

Question

我想编写好的测试来确保我的并发数据结构能够工作。但是测试甚至通过了一个显然不是线程安全的类。

class NotThreadSafe:
def __init__(self):
self.set1 = set()
self.set2 = set()
def add_to_sets(self, item):
self._add_to_set1(item)
self._add_to_set2(item)
def _add_to_set1(self, item):
self.set1.add(item)
def _add_to_set2(self, item):
self.set2.add(item)
def are_sets_equal_length(self):
return len(self.set1) == len(self.set2)

我的测试有一个读线程和一个写线程同时运行。写入线程调用add_to_sets，而读取器线程调用are_sets_equal_length。

但是读者线程总是观察到are_sets_equal_length是真的，即使作者线程理论上应该引起不平等。

我如何在add_to_set2上添加一些时间延迟，以迫使比赛条件浮出水面？

测试：

import threading
import time
def writer_fn(nts: NotThreadSafe):
for i in range(1000):
nts.add_to_sets(i)
def reader_fn(nts: NotThreadSafe, stop: list, results: list):
while not len(stop):
if not nts.are_sets_equal_length():
results.append(False)
return
results.append(True)

def test_nts():
nts = NotThreadSafe()
stop = []
results = []
reader = threading.Thread(target=reader_fn, args=[nts, stop, results])
writer = threading.Thread(target=writer_fn, args=[nts])
reader.start()
writer.start()
writer.join()
stop.append(True)
reader.join()
assert not results[0]

Answer 1

步骤1：编写一个包装器，创建一个包含时间延迟的新函数。

def slow_wrapper(method):
"""Adds a tiny delay to a method. Good for triggering race conditions that would otherwise be very rare."""
def wrapped_method(*args):
time.sleep(0.001)
return method(*args)
return wrapped_method

步骤2：在测试功能中，创建对象后，将add_to_set2更改为慢速版本：

nts = NotThreadSafe()
# change _add_to_set2 into a time-delayed version
nts._add_to_set2 = slow_wrapper(nts._add_to_set2)

第3步：运行测试。应正确触发故障。