如何避免 VxWorks 中条件变量中的竞争条件

Posted 2023-02-22

【中文标题】如何避免 VxWorks 中条件变量中的竞争条件

【英文标题】：How to avoid race conditions in a condition variable in VxWorks

【发布时间】：2013-09-30 13:58:29

【问题描述】：

我们正在基于 VxWorks 5.5 的专有嵌入式平台上进行编程。在我们的工具箱中，我们有一个条件变量，它是使用 VxWorks 二进制信号量实现的。

现在，POSIX 提供了一个等待函数，该函数也接受一个互斥体。这将解锁互斥锁（以便其他任务可能写入数据）并等待其他任务发出信号（完成写入数据）。我相信这实现了所谓的监视器，ICBWT。

我们需要这样一个等待函数，但实现它很棘手。一个简单的方法可以做到这一点：

bool condition::wait_for(mutex& mutex) const 
    unlocker ul(mutex);    // relinquish mutex
    return wait(event);
                          // ul's dtor grabs mutex again

但是，这具有竞争条件，因为它允许另一个任务在解锁之后和等待之前抢占该任务。另一个任务可以写入解锁后的日期，并在该任务开始等待信号量之前发出条件信号。 （我们已经对此进行了测试，并且确实会发生这种情况并永远阻止等待任务。）

鉴于 VxWorks 5.5 似乎没有提供 API 来在等待信号时暂时放弃信号量，有没有办法在提供的同步例程之上实现这一点？

注意： 这是一个非常古老的 VxWorks 版本，已经 编译 没有 POSIX 支持（据我了解，由专有硬件的供应商提供）。

【问题讨论】：

嗯...从未尝试在没有本机支持的情况下实现“condvar”。我一直能够单独使用信号量和互斥锁。

解锁器是做什么的？因为当我看到这个图案时，我完全期望它使用 RAII 样式。它可能会在析构函数中解锁。它可能会在有生之年解锁。这对我来说并不明显。

@sehe - 我也有点困惑。为什么不只用二进制信号量来保护它呢？好吧，语义不同，但是 WTH...

@MartinJames 可能是因为需要向所有人发出信号？无论如何，事件都有不同的语义。

@sehe, @Martin: unlocker 解锁 ctor 中的资源，并再次将其锁定在 dtor 中。 （对不起，我虽然这很明显。）

【参考方案1】：

使用原生 vxworks 这应该很容易，这里需要一个消息队列。您的 wait_for 方法可以按原样使用。

bool condition::wait_for(mutex& mutex) const 

    unlocker ul(mutex);    // relinquish mutex
    return wait(event);
                          // ul's dtor grabs mutex again

但是等待（事件）代码看起来像这样：

wait(event)

    if (msgQRecv(event->q, sigMsgBuf, sigMsgSize, timeoutTime) == OK)
    
        // got it...
    
    else
    
        // timeout, report error or something like that....
    

你的信号代码会是这样的：

signal(event)

    msgQSend(event->q, sigMsg, sigMsgSize, NO_WAIT, MSG_PRI_NORMAL);

因此，如果信号在您开始等待之前被触发，那么 msgQRecv 将在最终被调用时立即返回该信号，然后您可以如上所述在 ul dtor 中再次获取互斥锁。

event->q 是一个 MSG_Q_ID，它是在事件创建时通过调用 msgQCreate 创建的，sigMsg 中的数据由你定义……但可以只是一个随机字节的数据，或者你可以来建立一个更智能的结构，其中包含有关谁发出信号或其他可能很高兴知道的信息。

更新多个服务员，这有点棘手：所以我会做一些假设来简化事情

待处理的任务数量在事件创建时是已知的，并且是恒定的。 将始终有一个任务负责指示何时可以解锁互斥锁，所有其他任务只需要在事件发出/完成时得到通知。

这种方法使用计数信号量，与上面类似，只是有一点额外的逻辑：

wait(event)

    if (semTake(event->csm, timeoutTime) == OK)
    
        // got it...
    
    else
    
        // timeout, report error or something like that....
    

你的信号代码会是这样的：

signal(event)

    for (int x = 0; x < event->numberOfWaiters; x++)
    
        semGive(event->csm);
    

事件的创建是这样的，记住在这个例子中，服务员的数量是恒定的，并且在事件创建时是已知的。您可以使其动态化，但关键是每次事件将要发生时 numberOfWaiters 必须正确，然后解锁器才能解锁互斥锁。

createEvent(numberOfWaiters)

    event->numberOfWaiters = numberOfWaiters;
    event->csv = semCCreate(SEM_Q_FIFO, 0);
    return event;

您不能对 numberOfWaiters 抱有幻想：D 我再说一遍：在解锁器解锁互斥锁之前，numberOfWaiters 必须是正确的。要使其动态化（如果需要），您可以添加一个 setNumWaiters(numOfWaiters) 函数，并在解锁器解锁互斥锁之前在 wait_for 函数中调用它，只要它始终正确设置数字即可。

现在最后一个技巧，如上所述，假设是一个任务负责解锁互斥锁，其余的只是等待信号，这意味着只有一个任务会调用上面的 wait_for() 函数，其余任务只需调用 wait(event) 函数。

考虑到这一点，numberOfWaiters 的计算方式如下：

将调用 wait() 的任务数 为调用 wait_for() 的任务加 1

当然，如果你真的需要，你也可以让它变得更复杂，但这很有可能会起作用，因为通常 1 个任务会触发一个事件，但许多任务想知道它已经完成，这就是它所提供的。

但你的基本流程如下：

init()

    event->createEvent(3);


eventHandler()

    locker l(mutex);
    doEventProcessing();
    signal(event);


taskA()

    doOperationThatTriggersAnEvent();
    wait_for(mutex);
    eventComplete();


taskB()

    doWhateverIWant();
    // now I need to know if the event has occurred...
    wait(event);
    coolNowIKnowThatIsDone();


taskC()

    taskCIsFun();
    wait(event);
    printf("event done!\n");

当我写上面的内容时，我觉得所有的 OO 概念都已经死了，但希望你明白，实际上 wait 和 wait_for 应该采用相同的参数，或者不带参数，而是成为同一个类的成员，该类也具有所有他们需要知道的数据......但仍然是它如何工作的概述。

【讨论】：

原样的代码不适用于多个服务员，但更新它以使其与多个服务员一起工作将是微不足道的，可能最简单的方法是将 msgQ 切换为计数SEM。我会更新答案。

我想知道你是否找到一种方法来让计数信号量做到这一点。我们已经尝试过了，结果很干。但是，我们认为通过将您的消息队列想法与信号量相结合，我们找到了解决方案。我刚刚检查了这个平台上是否有消息队列（它们是另一个必须显式编译的功能），我现在会去尝试实现这个想法。

确实，我也考虑将 msgQ 用于计数 sem，但我不喜欢依赖链。这消除了中间人 msgQ，并直接通知所有等待的任务。

这两个假设都不成立。 （想象一下具有动态数量的消费者的经典生产者-消费者场景。我们有。）遗憾的是，我们发现将信号量与消息队列结合起来的想法也行不通。

确实，我提供了一些关于如何使其更具动态性的线索，可以使用这个基本算法来完成，IFF 在解锁器解锁互斥体之前知道消费者的数量。

【参考方案2】：

如果每个等待的任务都在一个单独的二进制信号量上等待，则可以避免竞争条件。 这些信号量必须在一个容器中注册，信令任务使用该容器来解除对所有等待任务的阻塞。容器必须受互斥体保护。

wait_for() 方法获取一个二进制信号量，等待它，最后将其删除。

void condition::wait_for(mutex& mutex) 
    SEM_ID sem = semBCreate(SEM_Q_PRIORITY, SEM_EMPTY);
    
        lock l(listeners_mutex);    // assure exclusive access to listeners container
        listeners.push_back(sem);       
                                   // l's dtor unlocks listeners_mutex again

    unlocker ul(mutex);             // relinquish mutex
    semTake(sem, WAIT_FOREVER);

    
        lock l(listeners_mutex);
        // remove sem from listeners
        // ...
        semDelete(sem);
    
                                   // ul's dtor grabs mutex again

signal() 方法遍历所有已注册的信号量并解锁它们。

void condition::signal() 
    lock l(listeners_mutex);
    for_each (listeners.begin(), listeners.end(), /* call semGive()... */ )

这种方法确保wait_for() 永远不会错过任何信号。缺点是需要额外的系统资源。 为了避免为每个wait_for() 调用创建和销毁信号量，可以使用池。

【讨论】：

这里的问题是 dtor 被多次调用，每次 wait_for() 调用一次，除了第一个离开 wait_for 之外的所有这一切都会阻塞互斥体......否则它非常接近我的解决方案，如果您认为二进制 sem 列表大致相当于计数 sem，您还需要在解锁器解锁之前知道侦听器的数量。除了在这种情况下，它不是硬编码的。

@Chris 关于 dtor 中的块我同意。这是从原始代码示例中复制的。显然，不应该获取互斥锁的任务必须传递一个虚拟互斥锁或在没有互斥锁参数和解锁器的情况下调用 wait_for。据我了解，侦听器的数量应该是灵活的，这意味着不与信号器共享互斥锁的侦听器将错过旧信号。计算信号量将允许以下场景：任务 A 增加 n 并解锁互斥锁，信号器任务 B 抢占 A 调用 n 次 semGive，任务 C 增加 n 并调用 semTake，最后任务 A 恢复并阻塞 semTake。

【参考方案3】：

根据描述，您可能想要实现（或使用）信号量 - 它是一种标准的 CS 算法，其语义类似于 condvar，并且有大量关于如何实现它们的教科书 (https://www.google.com/search?q=semaphore+algorithm)。

解释信号量的随机 Google 结果位于：http://www.cs.cornell.edu/courses/cs414/2007sp/lectures/08-bakery.ppt‎（参见幻灯片 32）。

【讨论】：

我建议您加倍努力，真正阅读问题。 提示：这不是关于如何实现条件变量，而是关于如何实现某种类型的等待函数。

嘿，如果提出的每个答案都是错误的，您可能需要重新考虑这个问题。正如其他人所注意到的，您正在尝试实现 condvar，如果您需要此特定行为而没有平台支持，您可能需要自己实现它，这意味着 不 使用您的平台支持的互斥锁类型。干杯。

问题清楚地表明 “但是，这是一种竞争条件，因为它允许另一个任务在解锁之后和等待之前抢占这个任务。” 如果你不能甚至连阅读问题都费心，您也无法获得cookie。 手。