进程间通信笔记（5）—互斥锁和条件变量

1.简介

互斥与等待用于线程之间的同步，这将允许多线程或多进程之间共享数据。

生产者消费者模型是一个典型的例子，考虑两种情况：

1.生产者的生产的速度快于消费者。

2.消费者的消费的速度快于生产者。

这样通信双方有一方必然阻塞，比如说，上述生产者生产速度快于消费者消费速度的情况，如果使用管道来处理这个问题，当管道被塞满后，内核就在生产者调用

write

时把它投入睡眠，直到管道中有空余的空间。这就是同步，不过内核帮咱们把这些事情做了，这种类型的同步是隐式的（implicit），类似的使用消息队列的IPC形式，内核依旧会处理同步。

然而，当我们使用共享内存这种IPC形式的时候，就必须执行某种类型的显示（explicit）同步。

互斥锁和条件变量能够帮我们实现显示的同步：

1.互斥锁保护持有锁的线程进入临界区后安全的修改线程共享的内容。

2.条件变量可用于等待条件的发生。

使用的互斥锁和条件变量的时候就是注意不要死锁和虚假唤醒即可。

2.示例

结合条件变量和互斥锁的应用，可以参考封装条件变量，介绍了《Linux多线程服务端编程》中一个简单的BlockingQueue（消费>生产的情况）。

因为生产者消费者同时启动线程，这里用到了互斥和等待两种同步。

代码见

~/unpv22e/mutex/prodcons6.c

/* include globals */
#include    "unpipc.h"

#define MAXNITEMS       1000000
#define MAXNTHREADS         100

int     nitems;
int     buff[MAXNITEMS];
struct {
pthread_mutex_t   mutex;
int               nput;   /* next index to store */
int               nval;   /* next value to store */
} put = { PTHREAD_MUTEX_INITIALIZER };//花括号初始化结构体。
//用于同步buffer的序号和值

struct {
pthread_mutex_t   mutex;
pthread_cond_t    cond;
int               nready; /* number ready for consumer */
} nready = { PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER };
//用于同步计数

void    *produce(void *), *consume(void *);//线程声明

/* include main */
int
main(int argc, char **argv)
{
int         i, nthreads, count[MAXNTHREADS];
pthread_t   tid_produce[MAXNTHREADS], tid_consume;

if (argc != 3)
err_quit("usage: prodcons6 <#items> <#threads>");
//多个生产者，1个消费者
nitems = min(atoi(argv[1]), MAXNITEMS);
nthreads = min(atoi(argv[2]), MAXNTHREADS);

Set_concurrency(nthreads + 1);
/* 4create all producers and one consumer */
for (i = 0; i < nthreads; i++) {
count[i] = 0;//count统计每个生产者的“产量”
Pthread_create(&tid_produce[i], NULL, produce, &count[i]);
}
Pthread_create(&tid_consume, NULL, consume, NULL);//生产者

/* wait for all producers and the consumer */
//线程回收
for (i = 0; i < nthreads; i++) {
Pthread_join(tid_produce[i], NULL);
printf("count[%d] = %d\n", i, count[i]);
}
//线程回收
Pthread_join(tid_consume, NULL);

exit(0);
}
/* end main */

/* include prodcons */
void *
produce(void *arg)
{
for ( ; ; ) {
Pthread_mutex_lock(&put.mutex);
if (put.nput >= nitems) {
Pthread_mutex_unlock(&put.mutex);
return(NULL);       /* array is full, we're done */
}
buff[put.nput] = put.nval;
put.nput++;
put.nval++;
Pthread_mutex_unlock(&put.mutex);

Pthread_mutex_lock(&nready.mutex);
if (nready.nready == 0)
Pthread_cond_signal(&nready.cond);//通知等待线程。
nready.nready++;
Pthread_mutex_unlock(&nready.mutex);

*((int *) arg) += 1;
}
}

void *
consume(void *arg)
{
int     i;

for (i = 0; i < nitems; i++) {
Pthread_mutex_lock(&nready.mutex);
while (nready.nready == 0)//防止虚假唤醒
Pthread_cond_wait(&nready.cond, &nready.mutex);
nready.nready--;
Pthread_mutex_unlock(&nready.mutex);

//这里只打印出错
//没有同步的多线程会产生混乱
if (buff[i] != i)
printf("buff[%d] = %d\n", i, buff[i]);
}
return(NULL);
}
/* end prodcons */

书中提到一种修改，避免上锁冲突，可以将唤醒等待线程的操作放到临界区外面。

int dosignal;
lock(&mutex);//临界区
dosignal=(ready==0);
ready++;
unlock(&mutex);

if (dosignal)
con_signal(&cond);//可以移除临界区。