带中断的互斥锁安全性（嵌入式固件）

Question

带中断的互斥锁安全性（嵌入式固件）

编辑 @Mike指出，下面代码中的try_lock函数是不安全的，访问器的创建也可能产生竞争条件。（来自所有人的）建议使我确信，我走错了路。

原始问题

锁定在嵌入式微控制器上的要求与多线程有足够的不同，以至于我无法将多线程示例转换为嵌入式应用程序。通常，我没有任何类型的OS或线程，只是main没有硬件定期调用的任何中断函数。

我通常需要从中断中填充缓冲区，然后在中进行处理，这很常见main。我创建了IrqMutex下面的类以尝试安全地实现此目的。每个试图访问缓冲区的人都会被分配一个唯一的ID IrqMutexAccessor，然后他们可以try_lock()和unlock()。阻塞lock()功能的想法不适用于中断，因为除非您允许中断完成，否则其他代码将无法执行，因此该unlock()代码将永远无法运行。但是，我确实main()偶尔使用代码中的阻塞锁。

但是，我知道没有C ++ 11内存屏障（在许多嵌入式平台上不可用），双重检查锁将无法工作。老实说，尽管阅读了很多，但我并不太真正理解为什么/为什么内存访问重新排序会导致问题。我认为使用volatile sig_atomic_t（可能与使用唯一ID结合使用）使此操作不同于双重检查锁。但是，我希望有人可以：确认以下代码正确无误，解释为什么它不安全，或者提供一种更好的方法来完成此操作。

class IrqMutex {
friend class IrqMutexAccessor;

private:
    std::sig_atomic_t accessorIdEnum;
    volatile std::sig_atomic_t owner;
protected:
    std::sig_atomic_t nextAccessor(void) { return ++accessorIdEnum; }
    bool have_lock(std::sig_atomic_t accessorId) {
        return (owner == accessorId);
    }
    bool try_lock(std::sig_atomic_t accessorId) {
        // Only try to get a lock, while it isn't already owned.
        while (owner == SIG_ATOMIC_MIN) {
            // <-- If an interrupt occurs here, both attempts can get a lock at the same time.

            // Try to take ownership of this Mutex.
            owner = accessorId; // SET

            // Double check that we are the owner.
            if (owner == accessorId) return true;

            // Someone else must have taken ownership between CHECK and SET.
            // If they released it after CHECK, we'll loop back and try again.
            // Otherwise someone else has a lock and we have failed.
        }        

        // This shouldn't happen unless they called try_lock on something they already owned.
        if (owner == accessorId) return true;

        // If someone else owns it, we failed.
        return false;
    }
    bool unlock(std::sig_atomic_t accessorId) {
        // Double check that the owner called this function (not strictly required)
        if (owner == accessorId) {
            owner = SIG_ATOMIC_MIN;
            return true;
        }

        // We still return true if the mutex was unlocked anyway.
        return (owner == SIG_ATOMIC_MIN);
    }
public:
    IrqMutex(void) : accessorIdEnum(SIG_ATOMIC_MIN), owner(SIG_ATOMIC_MIN) {}
};

// This class is used to manage our unique accessorId.
class IrqMutexAccessor {
friend class IrqMutex;
private:
    IrqMutex& mutex;
    const std::sig_atomic_t accessorId;
public:
    IrqMutexAccessor(IrqMutex& m) : mutex(m), accessorId(m.nextAccessor()) {}
    bool have_lock(void) { return mutex.have_lock(accessorId); }
    bool try_lock(void) { return mutex.try_lock(accessorId); }
    bool unlock(void) { return mutex.unlock(accessorId); }
};

Run Code Online (Sandbox Code Playgroud)

因为只有一个处理器，并且没有线程互斥锁可以达到我认为与正常情况稍有不同的目的。我经常遇到两个主要用例。

中断是生产者，它拥有空闲缓冲区的所有权，并用数据包加载它。中断/生产者可能会在多个中断调用之间长时间保持所有权锁定。主要功能是使用者，在准备好处理完整缓冲区时将其拥有。竞争条件很少发生，但是如果中断/生产者完成了一个数据包并需要一个新的缓冲区，但是它们都已满，它将尝试使用最早的缓冲区（这是一个丢弃的数据包事件）。如果主服务器/消费者在同一时间开始读取和处理最旧的缓冲区，它们将相互踩踏。
中断只是某种事物的快速改变或增加（例如计数器）。但是，如果要重置计数器或通过main（）代码调用跳转到某个新值，则我们不希望在计数器发生变化时尝试对其进行写入。main实际上在这里进行了阻塞循环来获得锁，但是我认为几乎不必在这里等待两次以上的尝试几乎是不可能的。一旦拥有锁，对计数器中断的任何调用都将被跳过，但是对于诸如计数器之类的东西来说，这通常并不重要。然后，我更新计数器值并将其解锁，以便它可以再次开始递增。

我意识到这两个示例有些笨拙，但是这些模式的某些版本出现在我从事的每个项目的许多外围设备中，我希望有一个可重用的代码可以在各种嵌入式平台上安全地处理这些代码。我包括了C标签，因为所有这些都可以直接转换为C代码，并且在某些嵌入式编译器上就可以使用全部。因此，我试图找到一种可以保证在C和C ++中都能正常工作的通用方法。

struct ExampleCounter {
    volatile long long int value;
    IrqMutex mutex;
} exampleCounter;

struct ExampleBuffer {
    volatile char data[256];
    volatile size_t index;
    IrqMutex mutex; // One mutex per buffer.
} exampleBuffers[2];

const volatile char * const REGISTER;

// This accessor shouldn't be created in an interrupt or a race condition can occur.
static IrqMutexAccessor myMutex(exampleCounter.mutex);
void __irqQuickFunction(void) {
    // Obtain a lock, add the data then unlock all within one function call.
    if (myMutex.try_lock()) {
        exampleCounter.value++;
        myMutex.unlock();
    } else {
        // If we failed to obtain a lock, we skipped this update this one time.
    }
}

// These accessors shouldn't be created in an interrupt or a race condition can occur.
static IrqMutexAccessor myMutexes[2] = {
    IrqMutexAccessor(exampleBuffers[0].mutex),
    IrqMutexAccessor(exampleBuffers[1].mutex)
};
void __irqLongFunction(void) {
    static size_t bufferIndex = 0;

    // Check if we have a lock.
    if (!myMutex[bufferIndex].have_lock() and !myMutex[bufferIndex].try_lock()) {
        // If we can't get a lock try the other buffer
        bufferIndex = (bufferIndex + 1) % 2;

        // One buffer should always be available so the next line should always be successful.
        if (!myMutex[bufferIndex].try_lock()) return;
    }

    // ... at this point we know we have a lock ...

    // Get data from the hardware and modify the buffer here.
    const char c = *REGISTER;
    exampleBuffers[bufferIndex].data[exampleBuffers[bufferIndex].index++] = c;

    // We may keep the lock for multiple function calls until the end of packet.
    static const char END_PACKET_SIGNAL = '\0';    
    if (c == END_PACKET_SIGNAL) {
        // Unlock this buffer so it can be read from main.
        myMutex[bufferIndex].unlock();

        // Switch to the other buffer for next time.
        bufferIndex = (bufferIndex + 1) % 2;
    }
}

int main(void) {
    while (true) {
        // Mutex for counter
        static IrqMutexAccessor myCounterMutex(exampleCounter.mutex);

        // Change counter value
        if (EVERY_ONCE_IN_A_WHILE) {
            // Skip any updates that occur while we are updating the counter.
            while(!myCounterMutex.try_lock()) {
                // Wait for the interrupt to release its lock.
            }

            // Set the counter to a new value.
            exampleCounter.value = 500;

            // Updates will start again as soon as we unlock it.
            myCounterMutex.unlock();
        }

        // Mutexes for __irqLongFunction.
        static IrqMutexAccessor myBufferMutexes[2] = {
            IrqMutexAccessor(exampleBuffers[0].mutex),
            IrqMutexAccessor(exampleBuffers[1].mutex)
        };

        // Process buffers from __irqLongFunction.
        for (size_t i = 0; i < 2; i++)  {
            // Obtain a lock so we can read the data.
            if (!myBufferMutexes[i].try_lock()) continue;
                // Check that the buffer isn't empty.
                if (exampleBuffers[i].index == 0) {
                    myBufferMutexes[i].unlock(); // Don't forget to unlock.
                    continue;
                }

                // ... read and do something with the data here ...
                exampleBuffer.index = 0;

                myBufferMutexes[i].unlock();
            }
        }
    }
}

Run Code Online (Sandbox Code Playgroud)

另请注意，我volatile在中断例程可读取或写入的任何变量上使用了（除非仅从中断中访问该变量，如中的static bufferIndex值__irqLongFunction）。我读过互斥锁消除了volatile多线程代码中的一些需求，但我认为这不适用于这里。 我使用了适量的volatile吗？ 我用它在：ExampleBuffer[].data[256]，ExampleBuffer[].index，和ExampleCounter.value。

Answer 1

Mik*_*ike 5

我为长答案道歉，但也许它适合一个长问题。

要回答你的第一个问题，我会说你的 IrqMutex 实现是不安全的。让我试着解释我在哪里看到问题。

功能 `nextAccessor`

std::sig_atomic_t nextAccessor(void) { return ++accessorIdEnum; }

Run Code Online (Sandbox Code Playgroud)

该函数具有竞争条件，因为增量运算符不是原子的，尽管它位于标记为的原子值上volatile。它涉及 3 个操作：读取的当前值accessorIdEnum，增加它，并将结果写回。如果IrqMutexAccessor同时创建两个s，则它们可能获得相同的 ID。

功能 `try_lock`

该try_lock函数还具有竞争条件。一个线程（例如主线程）可以进入while循环，然后在获得所有权之前，另一个线程（例如中断）也可以进入while循环并获得锁的所有权（返回true）。然后第一个线程可以继续，移动到owner = accessorId，因此“也”获取锁。所以两个线程（或者你的main线程和一个中断）可以同时try_lock在一个无主互斥锁上并且都返回true.

通过 RAII 禁用中断

我们可以通过使用 RAII 禁用中断来实现某种程度的简单性和封装，例如以下类：

std::sig_atomic_t nextAccessor(void) { return ++accessorIdEnum; }

Run Code Online (Sandbox Code Playgroud)

我建议禁用中断以获得互斥体实现本身所需的原子性。例如类似的东西：

class InterruptLock {
public:
    InterruptLock() { 
        prevInterruptState = currentInterruptState();
        disableInterrupts();
    }

    ~InterruptLock() { 
        restoreInterrupts(prevInterruptState);
    }
private:
    int prevInterruptState; // Whatever type this should be for the platform
    InterruptLock(const InterruptLock&); // Not copy-constructable
};

Run Code Online (Sandbox Code Playgroud)

根据您的平台，这可能看起来不同，但您明白了。

正如您所说，这提供了一个平台，可以将通用代码中的禁用和启用中断抽象出来，并将其封装到这一类中。

互斥和中断

说了我将如何考虑实现互斥类之后，我实际上不会为您的用例使用互斥类。正如您所指出的，互斥锁与中断并不能很好地配合使用，因为中断无法“阻止”尝试获取互斥锁。出于这个原因，对于直接与中断交换数据的代码，我会强烈考虑直接禁用中断（在主“线程”接触数据的很短的时间内）。

所以你的计数器可能看起来像这样：

bool try_lock(std::sig_atomic_t accessorId) {
    InterruptLock lock;
    if (owner == SIG_ATOMIC_MIN) {
        owner = accessorId;
        return true;
    }
    return false;
}
bool unlock(std::sig_atomic_t accessorId) {
    InterruptLock lock;
    if (owner == accessorId) {
        owner = SIG_ATOMIC_MIN;
        return true;
    }
    return false;
}

Run Code Online (Sandbox Code Playgroud)

在我看来，这更容易阅读，更容易推理，并且在出现争用（即错过计时器节拍）时不会“滑倒”。

关于缓冲区用例，我强烈建议不要为多个中断周期持有锁。锁/互斥锁应该只保留“接触”一块内存所需的最轻微的时刻——刚好足以读取或写入它。进来，出去。

所以这就是缓冲示例的样子：

volatile long long int exampleCounter;

void __irqQuickFunction(void) {
    exampleCounter++;
}
...
// Change counter value
if (EVERY_ONCE_IN_A_WHILE) {
    InterruptLock lock;
    exampleCounter = 500;
}

Run Code Online (Sandbox Code Playgroud)

这段代码可以说更容易推理，因为中断和主循环之间只有两个共享内存位置：一个将数据包从中断传递到主循环，另一个将空缓冲区传递回中断。我们也只接触“锁定”下的那些变量，并且只在“移动”值所需的最短时间。（为简单起见，当主循环需要很长时间才能释放缓冲区时，我跳过了缓冲区溢出逻辑）。

确实，在这种情况下甚至可能不需要锁，因为我们只是读取和写入简单的值，但是禁用中断的成本并不高，否则犯错误的风险在我看来是不值得的观点。

编辑

正如评论中所指出的，上述解决方案旨在仅解决多线程问题，并省略了溢出检查。这是更完整的解决方案，在溢出条件下应该是健壮的：

const size_t BUFFER_COUNT = 2; 

struct ExampleBuffer {
    char data[256];
    ExampleBuffer* next;
} exampleBuffers[BUFFER_COUNT];

volatile size_t overflowCount = 0;

class BufferList {
public:
    BufferList() : first(nullptr), last(nullptr) { }

    // Atomic enqueue
    void enqueue(ExampleBuffer* buffer) {
        InterruptLock lock;
        if (last)
            last->next = buffer;
        else {
            first = buffer;
            last = buffer;
        }
    }

    // Atomic dequeue (or returns null)
    ExampleBuffer* dequeueOrNull() {
        InterruptLock lock;
        ExampleBuffer* result = first;
        if (first) {
            first = first->next;
            if (!first)
                last = nullptr;
        }
        return result;
    }
private:
    ExampleBuffer* first;
    ExampleBuffer* last;
} freeBuffers, buffersAwaitingConsumption;

const volatile char * const REGISTER;

void __irqLongFunction(void) {

    static const char END_PACKET_SIGNAL = '\0';    
    static size_t index = 0;
    static ExampleBuffer* receiveBuffer = &exampleBuffers[0];

    // Recovery from overflow?
    if (!receiveBuffer) {
        // Try get another free buffer
        receiveBuffer = freeBuffers.dequeueOrNull();
        // Still no buffer?
        if (!receiveBuffer) {
            overflowCount++;
            return; 
        }
    }

    // Get data from the hardware and modify the buffer here.
    const char c = *REGISTER;

    if (index < sizeof(receiveBuffer->data))
        receiveBuffer->data[index++] = c;

    // End of packet, or out of space?
    if (c == END_PACKET_SIGNAL) {
        // Make the packet available to the consumer
        buffersAwaitingConsumption.enqueue(receiveBuffer);
        // Move on to the next free buffer
        receiveBuffer = freeBuffers.dequeueOrNull();
        index = 0;
    }
}

size_t getAndResetOverflowCount() {
    InterruptLock lock;
    size_t result = overflowCount;
    overflowCount = 0;
    return result;
}


int main(void) {

    // All buffers are free at the start
    for (int i = 0; i < BUFFER_COUNT; i++)
        freeBuffers.enqueue(&exampleBuffers[i]);

    while (true) {

        // Fetch packet from shared variable
        ExampleBuffer* packet = dequeueOrNull();

        if (packet) {
            // ... read and do something with the data here ...

            // Once we're done with the buffer, we need to release it back to the producer
            freeBuffers.enqueue(packet);
        }

        size_t overflowBytes = getAndResetOverflowCount();
        if (overflowBytes) {
            // ...
        }
    }
}

Run Code Online (Sandbox Code Playgroud)

关键变化：

如果中断用完空闲缓冲区，它将恢复
如果中断在没有接收缓冲区的情况下接收数据，它将通过以下方式将其传达给主线程 getAndResetOverflowCount
如果您不断收到缓冲区溢出，您可以简单地增加缓冲区计数
我已经将多线程访问封装到一个队列类中，实现为一个链表（BufferList），它支持原子出队和入队。前面的示例也使用了队列，但长度为 0-1（一个项目已入队或未入队），因此队列的实现只是一个变量。在空闲缓冲区用完的情况下，接收队列可能有 2 个项目，因此我将其升级为适当的队列，而不是添加更多共享变量。

归档时间：	11 年前
查看次数：	1602 次
最近记录：	11 年前

带中断的互斥锁安全性（嵌入式固件）

原始问题

功能 nextAccessor

功能 try_lock

通过 RAII 禁用中断

互斥和中断

编辑

功能 `nextAccessor`

功能 `try_lock`