c - Pthread 条件信号 - 未按预期工作

Question

我正在做一个项目并尝试使用 pthread_cond_wait()和 pthread_cond_signal()同步两个线程。

我的代码看起来像这样：

pthread_mutex_t lock_it = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t write_it = PTHREAD_COND_INITIALIZER;

   int main(int argc, char**argv)
  {

     pthread_t t_send_segments, t_recv_acks;

     pthread_create(&t_send_segments, NULL, send_segments, (void*)NULL);
     pthread_create(&t_recv_acks,     NULL, recv_acks,     (void*)NULL);

     pthread_join(t_recv_acks, (void**)NULL);

     pthread_mutex_destroy(&lock_it);
     pthread_cond_destroy(&write_it);
}

 void* send_segments(void *v) {
    for(;;) {
       pthread_mutex_lock(&lock_it);
       printf("s1\n");
       printf("s2\n");
       pthread_cond_wait(&write_it, &lock_it);
       printf("s3\n");
       printf("s4\n");
       printf("s5\n"); 
       pthread_mutex_unlock(&lock_it);
    }
    return 0;
 }

 void* recv_acks(void *v) {
    for(;;) {
       pthread_mutex_lock(&lock_it);
       printf("r1\n");
       pthread_cond_signal(&write_it);
       printf("r2\n");
       pthread_mutex_unlock(&lock_it);
    }  
    return 0;
 }

预期的输出是：

s1
s2
r1
s3
s4
s5
s1
s2
r2
r1
s3
s4
s5

(etc)

我的输出根本不遵循这种模式。显然我在某处有逻辑错误，但我不明白在哪里。为什么recv_acks()线程在遇到- 因为总是首先执行（因为我创建线程的顺序）并且总是在临界区之后执行，所以它在命中时并不总是屈服？pthread_cond_signal()pthread_cond_wait()cond_wait()

Answer 1

该pthread_cond_signal函数不会导致当前线程屈服，也不会释放互斥锁。它所做的只是重新启动一个在条件下挂起的线程pthread_cond_wait。这只是意味着被唤醒的线程可用于调度，它不会立即执行。线程调度程序将在将来的某个时间安排它。

此外，仅仅因为 s-thread 已被唤醒并正在竞争互斥锁，这并不意味着它将获得下一个互斥锁。互斥锁不一定对所有请求它的线程都是公平的。根据pthread_mutex手册页：“pthread_mutex_lock锁定给定的互斥锁。如果互斥锁当前被解锁，则它被锁定并由调用线程拥有，并pthread_mutex_lock立即返回。” 因此，r-thread 可以在其循环中旋转多次，在被调度程序换出之前多次愉快地解锁和重新锁定互斥锁。这意味着如果调度程序在释放互斥锁的短暂时间内碰巧中断了 r 线程，则 s 线程将只有在互斥锁上获得机会。

为了实现您想要的输出，两个线程都需要使用条件来控制它们的执行，并在暂停自己之前互相发出信号。但是，这可能是也可能不是您真正想要对您的实际项目执行的操作。

另一个注意事项：创建线程的顺序并不重要。创建线程不会产生创建线程。因此，主线程可能会在任何一个线程被调度之前创建两个线程，并且线程调度程序可以自由地调度其中一个线程以供接下来执行。如果 s-thread 确实首先在您的平台上运行，那恰好是您平台上的实现行为，而不是应该依赖的东西。

Answer 2

我似乎记得在某处读过pthread_cond_signal()实际上不会导致线程立即屈服的地方。由于pthread_cond_signal()不释放锁，调用它的线程必须继续执行，直到它释放锁，然后它才会屈服并允许发出信号的线程从pthread_cond_wait().

如果是这样的话，那么我认为你的输出看起来像

s1
s2
r1
r2
s3
s4
s5

等等......如果不是这样，您可能需要将实际输出编辑到您的问题中以获得准确的答案。

Answer 3

你在错误地使用条件。我不完全理解你想要什么，但你想要做的看起来很像一个典型的同步问题，称为生产者/消费者问题，可以像我之后那样在条件下实现。

您应该查看一下以了解如何使用条件。问题如下：我有两个线程，一个将数据写入固定大小的缓冲区，另一个从缓冲区读取数据。如果缓冲区已满，则第一个线程无法写入更多数据，如果缓冲区为空，则第二个线程无法读取。

#include <stdlib.h>
#include <stdio.h>

#include <pthread.h>

#define BUFFER_SIZE 4 

int buffer_nb_entries = 0; 

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; 
pthread_cond_t cond = PTHREAD_COND_INITIALIZER; 

void send(){
for(;;){
  pthread_mutex_lock(&mutex);
  while( buffer_nb_entries == BUFFER_SIZE) /* If buffer is full, then wait */
    pthread_cond_wait(&cond, &mutex); 

  /* Here I am sure that buffer is not full */
  printf("sending\n"); 
  buffer_nb_entries++;

  pthread_cond_signal(&cond); // signal that the condition has changed. 
  pthread_mutex_unlock(&mutex); 
 }
}

void receive(){
  for(;;){
    pthread_mutex_lock(&mutex); 
    while(buffer_nb_entries == 0)
      pthread_cond_wait(&cond, &mutex);
    /* Here I am sure that buffer is not empty */
    printf("receiving\n");
    buffer_nb_entries--; 
    pthread_cond_signal(&cond); 
    pthread_mutex_unlock(&mutex); 
  }

}

int main(){
  pthread_t s, r; 

  pthread_create(&s, NULL, (void *(*)(void*))send, NULL); 
  pthread_create(&r, NULL, (void *(*)(void*))receive, NULL); 

  pthread_join(s, NULL); 

}

请注意，您必须在调用 pthread_cond_wait() 之前测试某些内容，如果您不这样做并且如果您之前已调用过信号函数，那么您可能会永远休眠。

Answer 4

我认为你的麻烦来自试图用一把锁做太多事情。你应该只为一件事使用锁，这样就不会混淆你在等待什么。我建议为写信号添加第二个锁。此外，您应该为第二组消息添加第二个 cond_wait。如果你不这样做，事情运行的顺序将是随机的。这是我编辑的程序版本：

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

#define NUM_MESSAGES 3
int messages = 0;
void * send_segments(void *);
void * recv_acks(void *v);
pthread_mutex_t lock_it = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t write_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t write_cond = PTHREAD_COND_INITIALIZER;

   int main(int argc, char**argv)
  {

     pthread_t t_send_segments, t_recv_acks;

     pthread_create(&t_send_segments, NULL, send_segments, (void*)NULL);
     pthread_create(&t_recv_acks,     NULL, recv_acks,     (void*)NULL);

     pthread_join(t_recv_acks, (void**)NULL);
     pthread_join(t_send_segments, (void**)NULL);

     //pthread_mutex_destroy(&lock_it);
     //pthread_cond_destroy(&write_it);
}

 void* send_segments(void *v) {
    for(;;) {
       pthread_mutex_lock(&lock_it);
       printf("s1\n");
       printf("s2\n");
       pthread_mutex_unlock(&lock_it);

       // wait for other thread
       pthread_mutex_lock(&write_mutex);
       pthread_cond_wait(&write_cond, &write_mutex);
       pthread_mutex_unlock(&write_mutex);

       pthread_mutex_lock(&lock_it);
       printf("s3\n");
       printf("s4\n");
       printf("s5\n"); 
       pthread_mutex_unlock(&lock_it);

       // wait for other thread
       pthread_mutex_lock(&write_mutex);
       pthread_cond_wait(&write_cond, &write_mutex);
       pthread_mutex_unlock(&write_mutex);

       if (messages > NUM_MESSAGES) return( NULL );
    }
    return 0;
 }


 void* recv_acks(void *v) {
    for(;;) {
       // write first response
       pthread_mutex_lock(&lock_it);
       printf("r1\n");
       pthread_mutex_unlock(&lock_it);

       // signal other thread
       pthread_mutex_lock(&write_mutex);
       pthread_cond_signal(&write_cond);
       pthread_mutex_unlock(&write_mutex);

       // write second response
       pthread_mutex_lock(&lock_it);
       printf("r2\n\n");
       // increment count before releasing lock, otherwise the other thread
       // will be stuck waiting for a write_cond signal
       messages++;
       pthread_mutex_unlock(&lock_it);

       // signal other thread
       pthread_mutex_lock(&write_mutex);
       pthread_cond_signal(&write_cond);
       pthread_mutex_unlock(&write_mutex);

       if (messages > NUM_MESSAGES) return(NULL);
    } 
    return 0;
 }

如果我运行这个程序，我会得到以下输出（请注意，为了清楚起见，我在 r2 之后添加了第二个换行符）：

leif@indurain:~/tmp$ ./test
s1
s2
r1
s3
s4
s5
r2

s1
s2
r1
s3
s4
s5
r2

s1
s2
r1
s3
s4
s5
r2

s1
s2
r1
s3
s4
s5
r2

leif@indurain:~/tmp$