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

Question

我们正在基于 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 支持(据我了解，由专有硬件的供应商提供)。

Answer 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 中的数据由你定义……但可以只是一个随机字节的数据，或者你可以来建立一个更智能的结构，其中包含有关谁发出信号或其他可能很高兴知道的信息。

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

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

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

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 应该采用相同的参数，或者不带参数，而是成为同一个类的成员，该类也具有所有他们需要知道的数据......但仍然是它如何工作的概述。