语境
我最近在 Code Review 上发布了一个用于审查的计时器类。我有一种直觉，因为我曾经看到 1 个单元测试失败，但无法重现该失败。因此，我发布了代码审查。
我得到了一些很好的反馈，突出了代码中的各种竞争条件。（我想）我了解问题和解决方案，但在进行任何修复之前，我想通过单元测试来暴露错误。当我尝试时，我意识到这很困难。各种堆栈交换答案表明我必须控制线程的执行以暴露错误，并且任何人为的时间不一定可以移植到不同的机器上。这似乎是我试图解决的问题之外的许多意外复杂性。
相反，我尝试使用python 的最佳静态分析 (SA) 工具PyLint，看看它是否能找出任何错误，但它不能。为什么人类可以通过代码审查（本质上是 SA）找到错误，而 SA 工具却不能？
害怕尝试让 Valgrind 与 python 一起工作（这听起来像牦牛剃须），我决定在不先复制它们的情况下修复错误。现在我在泡菜。
现在是代码。
from threading import Timer, Lock
from time import time
class NotRunningError(Exception): pass
class AlreadyRunningError(Exception): pass
class KitchenTimer(object):
'''
Loosely models a clockwork kitchen timer with the following differences:
You can start the timer with arbitrary duration (e.g. 1.2 seconds).
The timer calls back a given function when time's up.
Querying the time remaining has 0.1 second accuracy.
'''
PRECISION_NUM_DECIMAL_PLACES = 1
RUNNING = "RUNNING"
STOPPED = "STOPPED"
TIMEUP = "TIMEUP"
def __init__(self):
self._stateLock = Lock()
with self._stateLock:
self._state = self.STOPPED
self._timeRemaining = 0
def start(self, duration=1, whenTimeup=None):
'''
Starts the timer to count down from the given duration and call whenTimeup when time's up.
'''
with self._stateLock:
if self.isRunning():
raise AlreadyRunningError
else:
self._state = self.RUNNING
self.duration = duration
self._userWhenTimeup = whenTimeup
self._startTime = time()
self._timer = Timer(duration, self._whenTimeup)
self._timer.start()
def stop(self):
'''
Stops the timer, preventing whenTimeup callback.
'''
with self._stateLock:
if self.isRunning():
self._timer.cancel()
self._state = self.STOPPED
self._timeRemaining = self.duration - self._elapsedTime()
else:
raise NotRunningError()
def isRunning(self):
return self._state == self.RUNNING
def isStopped(self):
return self._state == self.STOPPED
def isTimeup(self):
return self._state == self.TIMEUP
@property
def timeRemaining(self):
if self.isRunning():
self._timeRemaining = self.duration - self._elapsedTime()
return round(self._timeRemaining, self.PRECISION_NUM_DECIMAL_PLACES)
def _whenTimeup(self):
with self._stateLock:
self._state = self.TIMEUP
self._timeRemaining = 0
if callable(self._userWhenTimeup):
self._userWhenTimeup()
def _elapsedTime(self):
return time() - self._startTime
问题
在此代码示例的上下文中，我如何公开竞争条件、修复它们并证明它们已修复？
加分
适用于其他实现和问题而不是专门针对此代码的测试框架的额外点。
带走
我的结论是，重现已识别竞态条件的技术解决方案是控制两个线程的同步性，以确保它们按照会暴露错误的顺序执行。这里重要的一点是它们已经被确定为竞争条件。我发现识别竞争条件的最佳方法是将您的代码提交代码审查并鼓励更多专家对其进行分析。