如何在这个python代码中可靠地重现竞争条件？

Question

上下文

我最近发了一个timer class for review on Code Review。我曾有过一种直觉，觉得有并发错误，因为我曾经看到过一个单元测试失败，但无法重现失败。因此，我的代码后审查。

我得到了一些很好的反馈，强调了代码中的各种比赛条件。（我想）我理解这个问题和解决方案，但是在做任何修复之前，我想用单元测试来暴露错误。当我努力的时候，我意识到这很难。各种堆栈交换答案表明，我必须控制线程的执行以暴露错误，任何人为的计时都不一定能移植到不同的机器上。这似乎是我试图解决的问题之外的许多偶然的复杂性。

相反，我尝试使用the best static analysis (SA) tool for python，PyLint，看看它是否能找出任何错误，但它不能。为什么人类可以通过代码审查（本质上是SA）找到错误，但是SA工具却不能？

由于害怕尝试get Valgrind working with 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

问题

在这个代码示例的上下文中，如何公开竞争条件，修复它们，并证明它们已修复？

加分

适合于其他实现和问题的测试框架的额外点，而不是专门针对此代码。

外卖

我的收获是，重现已识别的争用条件的技术解决方案是控制两个线程的同步性，以确保它们的执行顺序将暴露一个bug。这里重要的一点是，他们已经确定了比赛条件。我发现识别竞争条件的最好方法是将代码提交给代码评审，并鼓励更多的专家对其进行分析。

Answer 1

你可以用很多线程来测试它：

import sys, random, thread
def timeup():
    sys.stdout.write("Timer:: Up %f" % time())

def trdfunc(kt, tid):
    while True :
        sleep(1)
        if not kt.isRunning():
            if kt.start(1, timeup):
                sys.stdout.write("[%d]: started\n" % tid)
        else:
            if random.random() < 0.1:
                kt.stop()
                sys.stdout.write("[%d]: stopped\n" % tid)
        sys.stdout.write("[%d] remains %f\n" % ( tid, kt.timeRemaining))

kt = KitchenTimer()
kt.start(1, timeup)
for i in range(1, 100):
    thread.start_new_thread ( trdfunc, (kt, i) )
trdfunc(kt, 0)

我看到了几个问题：

• 当一个线程看到计时器没有运行并试图启动它时 代码通常会引发一个异常，原因是在 测试并启动。我觉得破例太多了。或者你可以 具有原子测试和启动功能

• stop也会出现类似的问题。您可以实现testAndStop 功能。

• 甚至这个来自timeRemaining函数的代码：

  if self.isRunning():
     self._timeRemaining = self.duration - self._elapsedTime()
  

  需要某种原子性，也许你需要在 测试正在运行。

如果计划在线程之间共享该类，则需要解决这些问题。