如何为Python实例方法添加延时以复现线程安全问题？

我想要编写可靠测试验证并发数据结构的正确性，但针对一个明显非线程安全的类，测试却始终通过。该类代码如下：

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返回True。请问如何为_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]

要强制触发竞态条件，只需在_add_to_set2方法中添加一段短暂延时，让写线程在完成set1.add(item)后、完成set2.add(item)前主动让出CPU，给读线程足够时间检测到两个集合长度不一致的状态。

修改后的NotThreadSafe类代码如下：

import time

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):
        # 添加延时，强制触发竞态条件
        time.sleep(0.001)
        self.set2.add(item)

    def are_sets_equal_length(self):
        return len(self.set1) == len(self.set2)

原理说明 #

• 写线程执行_add_to_set1后，set1已新增元素，但set2还未完成添加操作。
• time.sleep(0.001)会让当前写线程进入休眠，操作系统会调度读线程执行。
• 读线程此时调用are_sets_equal_length就能捕捉到len(set1) > len(set2)的状态，测试的断言会触发失败，验证出类的非线程安全特性。

如果单次测试仍未触发，可以适当延长延时时间，或者增加写线程的循环次数，提升竞态条件出现的概率。

内容的提问来源于stack exchange，提问作者manimino