这两个中断服务程序中的竞态条件是什么？

Question

我在我的大学使用ARM微控制器进行实时系统课程。在我目前正在进行的项目中，我正在实施矢量场直方图（VFH）算法。

问题是：我需要在线程之间进行通信;更具体地说，我想要一个线程从测距仪获取传感器数据，对其进行必要的转换并将它们存入队列。他们，另一个线程必须得到这些数据和处理它等等。

目前，我使用它的一个更简单的版本 - 一个线程从ADC获取数据（SensorAcquisitionHandler），另一个线程输出第一个5项（最多）显示（ControlSignalHandler）。

/* Queue used to store data from the rangefinder sensors. */ 
static unsigned int Sensors[100]; 
static int SensorsHead = 0; 
static int SensorsTail = 0; 

void SensorAcquisitionHandler(void) { 
    /* Clear the interrupt. */ 
    ADCIntClear(ADC0_BASE, 1); 

    int i; /* Index for the measurements buffer. */ 

    /* There are only 3 rangefinders used. */ 
    if (ADCSequenceDataGet(ADC0_BASE, 1, rangeBuffer) == 3) { 
     /* Put rangeBuffer's data into SensorDataQueue. */ 
     /* Also, when using SensorDataQueue, must put what's the direction of 
     the corresponding range measurement. */ 

     /* Critical section ahead!!! Turn off interrupts!!! */ 
     IntMasterDisable(); 

     /* Temporarily using the simple FIFO... */ 
     for (i = 0; i < 3; ++i) { 
      if (SensorsHead < 100) { 
       Sensors[SensorsHead] = rangeBuffer[i]; 
       SensorsHead++; 
      } 
     } 

     /* All is fine, turn on interrupts. */ 
     IntMasterEnable(); 
    } 
} 

void ControlSignalHandler(void) { 
    /* Clear the timer interrupt. */ 
    TimerIntClear(TIMER0_BASE, TIMER_TIMA_TIMEOUT); 

    unsigned char i; /* Index for the measurements buffer. */ 
    unsigned long average = 0; 
    char buffer[20]; 

    /* Average first n (n <= 5) elements from Sensors queue. */ 
    for (i = 0; i < 5 && SensorsTail < SensorsHead; ++i) { 
     average += Sensors[SensorsTail]; 
     SensorsTail++; 
    } 

    IntMasterDisable(); 
    average /= i; 

    sprintf(buffer, "%d ", average); 

    average = 0; 

    if (SensorsTail >= SensorsHead) { 
     SensorsTail = 0; 
     SensorsHead = 0; 
    } 

    Display96x16x1StringDraw(buffer, 0, 0); 
    IntMasterEnable(); 
} 

结果在一段时间内相对稳定，但随机间隔变得非常高（几乎所有时间的结果都是〜330）。此外，当我在“非常高价值”的时刻使用符号调试器时，SensorTail和SensorHead的索引可以达到300+（队列是100个元素的阵列）。

这听起来像是某种溢出，但我无法想象它是如何发生的。有人可以帮我找到它吗？

我知道这个问题的答案是“使用线程安全的队列”，但我想了解的竞争条件是如何发生在这里，索引是如何得到搞砸了，等等。谢谢！

Answer 1

通过清除中断，可以让它再次发生。试想一下，如果（例如）在ControlSignalHandler获取它的第一个循环的中间重新进入，因为计时器设法逃脱你的代码会发生什么...

裹在IntMasterDisable + IntMasterEnable的整个功能（两者）并在禁用之后和启用之前清除中断。 （我会在启用之前立即执行。）

Answer 2

看看程序集转储。非易失性代码可以针对易失性代码重新排序。

假设你有：

assignmentA_with_volatile_operands; 
assignmentB_with_non_volatile_operands; 
assignmentC_with_volatile_operands; 

的编译是免费重新整理这：

assignmentA_with_volatile_operands; 
assignmentC_with_volatile_operands; 
assignmentB_with_non_volatile_operands; 

例如您for循环在第一处理SensorAcquisitionHandler实际上IntMasterEnable后执行的出现在for循环中的对象中没有一个是volatile合格。

编辑：

有些人觉得这些种代码的重新排序是不允许的。事实是他们是，他们是用现实生活编译器来执行的。

挥发性不作为在程序存储器中的屏障。您不应该认为易失性访问会影响非易失性访问的内存限制。

标准定义了C11，5.1.2.3什么是上符合标准的实现最低要求对于程序的观察到的行为。

gcc说例如：对非易失性对象的访问没有针对易失性访问进行排序。您不能将易失性对象用作内存屏障来对非易失性内存进行一系列写入。

http://gcc.gnu.org/onlinedocs/gcc/Volatiles.html

很多的编译器是谨慎，不要在挥发性存在执行这些类型的代码重新排序优化，但我已经看到编译器（gcc例如）执行它们。

Answer 3

您使用的是哪种特定的处理器/微控制器？什么RTOS（如果有的话）？

ARM微控制器上的中断堆栈通常非常非常小。像snprintf()这样的运行时例程可以轻松地需要数百个字节，并且可能会超出一个小堆栈。即使不考虑堆栈空间的考虑，C运行时功能在中断环境中使用通常也不安全 - 您经常会受限于哪些函数可以从中断调用。具体细节取决于您正在使用的实际RTOS和编译器工具链。

如果您违反了这些限制，它很容易导致数据损坏。

Answer 4

您可以通过使用无锁定的单读取器单写入器FIFO来避免头尾指针的争用情况 - 其中头指针只写入一个线程（或者在您的情况下是ISR）和尾巴写在其他。这意味着您在每个ISR中执行缓冲区换行测试。

如果你这样做，并在每个ISR结束时重置你的中断源，你应该不需要任何锁定 - 全局禁止中断，因为你正在做的事情是非常糟糕的举止。目前你持有的锁很长时间。

另一个原因，你需要重写你的FIFO实现：

for (i = 0; i < 3; ++i) { 
     if (SensorsHead < 100) { 

既然你一次添加3个读数，你最终会与SensorsHead==99进入SensorAcquisitionHandler() - 它保证你会扔2读数。

类似地：

/* Average first n (n <= 5) elements from Sensors queue. */ 
for (i = 0; i < 5 && SensorsTail < SensorsHead; ++i) { 
    average += Sensors[SensorsTail]; 
    SensorsTail++; 
} 

会在某些情况下上，而不是5个值以下进行计算。

根据您使用的ARM部件，没有硬件鸿沟。计算两次幂值的平均值要便宜得多，因为它是一个单周期的逻辑转换。

最后，我想Display96x16x1StringDraw(buffer, 0, 0);是一个特别昂贵的操作，它可能是线程安全的。在ISR中IO总是严格禁止的。 您可能需要在您的计时器线程和非中断上下文之间的另一个队列 - 处理输出。