为什么pthreads的条件变量函数需要互斥锁？

Question

我正在读书pthread.h; 条件变量相关函数(如pthread_cond_wait(3))需要互斥量作为参数.为什么？据我所知,我将创建一个互斥体只是用作该参数？那个互斥锁应该做什么？

Answer 1

它只是条件变量实现(或最初)的方式.

互斥锁用于保护条件变量本身.这就是为什么你在等待之前需要它锁定的原因.

等待将"原子地"解锁互斥锁,允许其他人访问条件变量(用于信令).然后,当发信号通知或广播条件变量时,等待列表中的一个或多个线程将被唤醒,并且互斥锁将再次为该线程神奇地锁定.

您通常会看到以下有条件变量的操作,说明它们的工作原理.以下示例是一个工作线程,通过信号向条件变量提供工作.

thread:
    initialise.
    lock mutex.
    while thread not told to stop working:
        wait on condvar using mutex.
        if work is available to be done:
            do the work.
    unlock mutex.
    clean up.
    exit thread.

如果等待返回时有一些可用,则在此循环内完成工作.当线程被标记为停止工作时(通常由另一个线程设置退出条件然后踢条件变量以唤醒该线程),循环将退出,互斥锁将被解锁并且该线程将退出.

上面的代码是单一消费者模型,因为在工作完成时互斥锁保持锁定状态.对于多消费者变体,您可以使用,例如:

thread:
    initialise.
    lock mutex.
    while thread not told to stop working:
        wait on condvar using mutex.
        if work is available to be done:
            copy work to thread local storage.
            unlock mutex.
            do the work.
            lock mutex.
    unlock mutex.
    clean up.
    exit thread.

这允许其他消费者在这个工作时接收工作.

条件变量减轻了轮询某些条件的负担,反而允许另一个线程在需要发生某些事情时通知您.另一个线程可以告诉该线程工作是否可用,如下所示:

lock mutex.
flag work as available.
signal condition variable.
unlock mutex.

通常错误地称为虚假唤醒的绝大多数通常总是因为多个线程已在其pthread_cond_wait呼叫(广播)中发出信号,一个人将使用互斥锁返回,执行工作,然后重新等待.

然后,当没有工作要做时,第二个发出信号的线程可能会出现.所以你必须有一个额外的变量来表明应该完成工作(这里固有地使用condvar/mutex对进行互斥保护 - 但是在更改之前需要锁定互斥锁的其他线程).

这是技术上是可行的一个线程从条件等待而不被其他进程被踢回(这是一个真正的虚假唤醒),但是,在我所有的多年工作的并行线程,无论是在代码的开发/服务,并为用户其中,我从来没有收到其中的一个.也许那只是因为惠普有一个不错的实施:-)

在任何情况下,处理错误情况的相同代码也处理真正的虚假唤醒,因为不会为那些设置工作可用标志.

Answer 2

如果您只能发出条件信号,则条件变量非常有限,通常您需要处理与发出信号的条件相关的一些数据.信号/唤醒必须以原子方式完成,以实现不引入竞争条件或过于复杂的情况

由于技术原因,pthreads也会给你一个虚假的唤醒.这意味着您需要检查一个谓词,这样您就可以确定实际上已经发出信号 - 并将其与虚假唤醒区分开来.检查等待它的这种情况需要加以保护 - 所以条件变量需要一种方法来原子地等待/唤醒,同时锁定/解锁保护该条件的互斥锁.

考虑一个简单的示例,其中通知您生成了一些数据.也许另一个线程制作了你想要的一些数据,并设置了指向该数据的指针.

想象一下生产者线程通过'some_data'指针将一些数据提供给另一个消费者线程.

while(1) {
    pthread_cond_wait(&cond); //imagine cond_wait did not have a mutex
    char *data = some_data;
    some_data = NULL;
    handle(data);
}

你自然会得到很多竞争条件,如果另一个线程some_data = new_data在你醒来之后就做了什么,但是在你之前做了什么data = some_data

你不能真正创建自己的互斥锁来保护这种情况.例如

while(1) {

    pthread_cond_wait(&cond); //imagine cond_wait did not have a mutex
    pthread_mutex_lock(&mutex);
    char *data = some_data;
    some_data = NULL;
    pthread_mutex_unlock(&mutex);
    handle(data);
}

无法工作,在唤醒和抓住互斥锁之间仍然存在竞争条件的可能性.在pthread_cond_wait之前放置互斥锁对您没有帮助,因为您现在将在等待时持有互斥锁 - 即生产者将永远无法获取互斥锁.(注意,在这种情况下,您可以创建第二个条件变量来通知生产者您已完成some_data- 尽管这将变得复杂,尤其是如果您想要许多生产者/消费者.)

因此,您需要一种在等待/从状态唤醒时以原子方式释放/获取互斥锁的方法.这就是pthread条件变量的作用,这就是你要做的:

while(1) {
    pthread_mutex_lock(&mutex);
    while(some_data == NULL) { // predicate to acccount for spurious wakeups,would also 
                               // make it robust if there were several consumers
       pthread_cond_wait(&cond,&mutex); //atomically lock/unlock mutex
    }

    char *data = some_data;
    some_data = NULL;
    pthread_mutex_unlock(&mutex);
    handle(data);
}

(生产者自然需要采取相同的预防措施,始终使用相同的互斥锁保护'some_data',并确保如果some_data当前不会覆盖some_data!= NULL)

Answer 3

POSIX条件变量是无状态的.因此,维护国家是你的责任.由于等待的线程和告诉其他线程停止等待的线程将访问状态,因此它必须受互斥锁保护.如果您认为可以在没有互斥锁的情况下使用条件变量,那么您还没有意识到条件变量是无状态的.

条件变量是围绕条件构建的.等待条件变量的线程正在等待某些条件.发出条件变量信号的线程会改变这种情况.例如,线程可能正在等待某些数据到达.其他一些线程可能会注意到数据已经到达."数据到达"是条件.

这是条件变量的经典用法,简化:

while(1)
{
    pthread_mutex_lock(&work_mutex);

    while (work_queue_empty())       // wait for work
       pthread_cond_wait(&work_cv, &work_mutex);

    work = get_work_from_queue();    // get work

    pthread_mutex_unlock(&work_mutex);

    do_work(work);                   // do that work
}

看看线程如何等待工作.这项工作受互斥锁保护.等待释放互斥锁,以便另一个线程可以给这个线程一些工作.这是如何发出信号:

void AssignWork(WorkItem work)
{
    pthread_mutex_lock(&work_mutex);

    add_work_to_queue(work);           // put work item on queue

    pthread_cond_signal(&work_cv);     // wake worker thread

    pthread_mutex_unlock(&work_mutex);
}

请注意,您需要使用互斥锁来保护工作队列.请注意,条件变量本身不知道是否有工作.也就是说,条件变量必须与条件相关联,该条件必须由代码维护,并且由于它在线程之间共享,因此必须由互斥锁保护.

Answer 4

并非所有条件变量函数都需要互斥锁:只有等待操作才能执行.信号和广播操作不需要互斥锁.条件变量也不与特定互斥锁永久关联; 外部互斥锁不保护条件变量.如果条件变量具有内部状态,例如等待线程的队列,则必须通过条件变量内部的内部锁来保护它.

等待操作将条件变量和互斥锁组合在一起,因为:

• 一个线程锁定了互斥锁,在共享变量上评估了一些表达式并发现它是假的,这样它就需要等待.
• 线程必须原子地从拥有互斥锁转移到等待条件.

由于这个原因,wait操作将互斥和条件作为参数:这样它就可以管理线程从拥有互斥锁到等待的原子转移,这样线程就不会成为丢失唤醒竞争条件的牺牲品.

如果一个线程放弃一个互斥锁,然后等待一个无状态同步对象,但是以一种非原子的方式,将会出现一个丢失的唤醒竞争条件:当该线程不再具有该锁时,存在一个时间窗口,并且尚未开始等待对象.在此窗口期间,另一个线程可以进入,使等待条件成立,发出无状态同步信号然后消失.无状态对象不记得它被发出信号(它是无状态的).那么原始线程就会在无状态同步对象上进入休眠状态,并且不会唤醒,即使它需要的条件已经成为现实:丢失唤醒.

条件变量等待函数通过确保调用线程被注册以在放弃互斥锁之前可靠地捕获唤醒来避免丢失唤醒.如果条件变量wait函数没有将互斥锁作为参数,则这是不可能的.

Answer 5

我没有发现其他答案与本页一样简洁易读.通常,等待代码看起来像这样:

mutex.lock()
while(!check())
    condition.wait()
mutex.unlock()

wait()包含互斥锁的原因有三个:

1. 在没有互斥的情况下,另一个线程可以signal()在wait()我们错过这个唤醒之前.
2. 通常check()依赖于来自另一个线程的修改,所以无论如何你都需要相互排斥.
3. 确保优先级最高的线程首先进行(互斥锁的队列允许调度程序决定谁下一个).

第三点并不总是一个问题 - 历史背景从文章到这个对话都有联系.

关于这种机制经常提到虚假唤醒(即等待线程在没有signal()被调用的情况下被唤醒).但是,这些事件由循环处理check().

Answer 6

条件变量与互斥体相关联，因为它是避免竞争的唯一方法，它旨在避免。

// incorrect usage:
// thread 1:
while (notDone) {
    pthread_mutex_lock(&mutex);
    bool ready = protectedReadyToRunVariable
    pthread_mutex_unlock(&mutex);
    if (ready) {
        doWork();
    } else {
        pthread_cond_wait(&cond1); // invalid syntax: this SHOULD have a mutex
    }
}

// signalling thread
// thread 2:
prepareToRunThread1();
pthread_mutex_lock(&mutex);
   protectedReadyToRuNVariable = true;
pthread_mutex_unlock(&mutex);
pthread_cond_signal(&cond1);

Now, lets look at a particularly nasty interleaving of these operations

pthread_mutex_lock(&mutex);
bool ready = protectedReadyToRunVariable;
pthread_mutex_unlock(&mutex);
                                 pthread_mutex_lock(&mutex);
                                 protectedReadyToRuNVariable = true;
                                 pthread_mutex_unlock(&mutex);
                                 pthread_cond_signal(&cond1);
if (ready) {
pthread_cond_wait(&cond1); // uh o!

此时，没有线程会向条件变量发出信号，因此线程 1 将永远等待，即使 protectedReadyToRunVariable 表示它已准备就绪！

解决这个问题的唯一方法是让条件变量原子地释放互斥锁，同时开始等待条件变量。这就是 cond_wait 函数需要互斥锁的原因

// correct usage:
// thread 1:
while (notDone) {
    pthread_mutex_lock(&mutex);
    bool ready = protectedReadyToRunVariable
    if (ready) {
        pthread_mutex_unlock(&mutex);
        doWork();
    } else {
        pthread_cond_wait(&mutex, &cond1);
    }
}

// signalling thread
// thread 2:
prepareToRunThread1();
pthread_mutex_lock(&mutex);
   protectedReadyToRuNVariable = true;
   pthread_cond_signal(&mutex, &cond1);
pthread_mutex_unlock(&mutex);