Linux——互斥锁与条件变量（一）

为容许在线程或进程间共享数据，同步一般是必需的。互斥锁和条件变量是同步的基本组成部分。

使用互斥锁和条件变量的经典例子：生产者-消费者问题。在本例中，使用多个线程而不是多个进程，由于让多个线程共享本问题采用的公共数据缓冲区很是简单，而在多个进程间共享一个公共数据缓冲区却须要某种形式的共享内存区。

1、互斥锁：上锁与解锁

互斥锁指代相互排斥（mutual exclusion），它是最基本的同步形式。互斥锁用于保护临界区（critical region），以保证任什么时候刻只有一个线程在执行其中的代码（假设互斥锁由多个线程共享），或者任什么时候刻只有一个进程在执行其中的代码（假设互斥锁由多个进程共享）。保护一个临界区的代码轮廓大致以下：

lock_the_mutex(...);
临界区
unlock_the_mutex(...);

既然任什么时候刻只有一个线程可以锁住一个给定的互斥锁，因而这样的代码保证在任什么时候刻只有一个线程在执行其临界区中的指令。

Posix互斥锁被声明为具备pthread_mutex_t数据类型的变量。若是互斥锁变量是静态分配的，那么咱们能够把它初始化成常值PTHREAD_MUTEX_INITIALIZER，例如：

static pthread_mutex_t lock =  PTHREAD_MUTEX_INITIALIZER;

若是互斥锁是动态分配的，或者分配在共享内存区中，那么咱们必须在运行之时，经过调用pthread_mutex_init函数来初始化它。后面有所介绍。

在Posix中，定义了三个函数来给一个互斥锁上锁和解锁：

#include<pthread.h>
int pthread_mutex_lock(pthread_mutex_t *mptr);
int pthread_mutex_trylock(pthread_mutex_t *mptr);
int pthread_mutex_unlock(pthread_mutex_t *mptr);
均返回：若成功则为0，若出错则为正的Exxx值

若是尝试给一个已有某个线程锁住的互斥锁上锁，那么pthread_mutex_lock将阻塞到该互斥锁解锁为止。pthread_mutex_trylock是对应的非阻塞函数，若是该互斥锁已锁住，它就返回一个EBUSY错误。

这里有一个问题：

若是有多个线程阻塞在等待同一个互斥锁上，那么当该互斥锁解锁时，哪个线程会开始运行呢？

===>>>提供一个优先级调度选项，不一样线程可被赋予不一样的优先级，同步函数（互斥锁、读写锁、信号量）将唤醒优先级最高(队列实现？)的被阻塞线程。

临界区实际上保护的是在临界区中被操纵的数据，也就是说，互斥锁一般用于保护由多个线程或多个进程分享的共享数据(shared data)。

互斥锁是协做性锁。这就是说，若是共享数据是一个链表，那么操纵该链表的全部线程都必须在实际操纵前获取该互斥锁。不过也没有办法防止某个线程不首先获取该互斥锁就操纵该链表。

2、生产者-消费者问题

同步中有一个称为“生产者-消费者(producer-consumer)”问题的经典问题，也称为有界缓冲区（bounded buffer）问题。一个或多个生产者（线程或进程）建立者一个个数据条目，而后这些条目由一个或多个消费者（线程或进程）处理。数据条目在生产者和消费者之间是使用某种类型的IPC（interprocess communication）传递的。

当共享存储区用做生产者和消费者之间的IPC形式时，生产者和消费者必须执行某种类型的显式(explicit)同步。咱们使用互斥锁展现显式同步。

在这里，咱们选择使用单个消费者线程、多个生产者线程的方法。首先，介绍一些相关信息：在单个进程中有多个生产者线程和单个消费者线程。整数数组buff含有被生产和消费的条目（也就是共享数据）。为简单起见，生产者只是把buff[0]设置为0，把buff[1]设置为1，如此等等。消费者只是沿着该数组行进，并验证每一个数组元素都是正确的。

而后，写出main函数以下：

#include “unpipc.h”
 
#define MAXNITEMS 1000000
#define MAXNTHREADS 100
 
int nitems;
 
struct{
    pthread_mutex_t mutex;
    int buff[MAXNITEMS];
    int nput;
    int nval;  
}shared= {PTHREAD_MUTEX_INITIALIZER};
 
void *produce(void *),*consumer(void *);//生产者、消费者函数
 
int main(int argc , char **argv){
    int i, nthreads, count[MAXNTHREADS];
    pthread_t tid_produce[MAXNTHREADS],tid_consume;
    if(argc!=3){
        err_quit("usage:prodcons <#items> <#threads>"); 
    }
    nitems = min(atoi(argv[1]), MAXNITEMS);
    nthreads = min(atoi(argv[2]), MAXNTHREADS);
  
    Set_concurrency(nthreads);
  
    /*start all the producer threads*/
    for(i=0;i<nthreads;i++){
    count[i] = 0;
    pthread_create(&tid_produce[i],NULL,produce,&count[i]); 
    }

    /*wait for all producer threads  */
    for(i=0;i<nthreads;i++){
    pthread_join(tid_produce[i],NULL);
    printf("count[%d]=%d",i,count[i]); 
    }
  
    /*start, then wait for the consumer thread*/
    pthread_create(&tid_consume,NULL,consume,NULL);
    pthread_join(tid_consume,NULL);
  
    exit(0);
}
 
void *produce(void *arg){
    for(;;){
        pthread_mutex_lock(&shared.mutex);
        if(shared.nput >= nitems){
        pthread_mutex_unlock(&shared.mutex);
        return NULL; 
        /* array is full,we are done*/
        }
        shared.buff[shared.nput] = shared.nval;
        shared.nput++;
        shared.nval++;
        pthread_mutex_unlock(&shared.mutex);
        *((int *)arg)+=1;
    } 
}
 
void *consume(void *arg){
    int i;
    for(i=0;i<nitems;i++){
        if(shared.buff[i] != i)
        printf("buff[%d] = %d\n",i,shared.buff[i]);
    }
    return NULL;
}

在上面的main函数，咱们须要注意这样几个方面：

一、struct shared的做用：

首先咱们分析这些成员变量，mutex是互斥锁，buff数组是数据空间，nput是buff数组中下一次存放数据的元素下标，nval是下一次存放的值。咱们将这些变量整合放在一个结构体中的目的是为了强调这些变量只应该在拥有互斥锁时访问。咱们分配这个结构，并初始化其中用于生产者线程间同步的互斥锁。

对于这里，咱们要考虑这样一个问题：将全部受互斥锁控制的临界区变量都整合放在一个结构体中是否十分完善？这是不完善的，把共享数据和它们的同步变量收集到一个结构体中，这是一个很好的编程技巧，然而在不少状况下共享数据是动态分配的，譬如说一个链表，咱们能够把一个链表的头及该链表的同步变量存放到一个结构体中，可是其余共享数据（该链表的其余部分）却不在结构体中，所以这种方法是不完善的。

二、命令行输入参数分析：形如“prodcons items threads”，其中，items指定生产者存放的条目数，threads指定待建立生产者线程的数目。

三、Set_concurrency(nthreads)函数的做用：

告诉线程系统咱们但愿并发运行多少线程。

四、建立生产者线程：

在这里，咱们用线程建立函数pthread_create(...)来建立生产者进程，每一个线程执行produce。在tid_produce数组中保存每一个生成线程的pid。传递给每一个生产者线程的参数是指向count数组中某个元素的指针。那么count数组的做用是什么？代码中，咱们首先把count计数器初始化为0，而后每一个线程在每次往缓冲区中存放一个条目时给这个计数器加1(*(int*)arg)+=1)，。当一切生产完毕时，咱们输出这个计数器数组中各元素的值，以查看每一个生产者线程分别存放了多少条目。

五、等待生产者线程，而后启动消费者线程：

等待全部生产者线程终止，同时输出每一个线程的计数器值，此后才启动单个消费者线程。此处，咱们实现了生产者与消费者之间的同步问题，接着等待消费者线程完成，而后终止进程。

六、Produce函数产生数据条目：

在这里，咱们使用互斥锁来实现对临界区的保护，同时保证在一切生产完毕的状况下给该互斥锁解锁。注意count元素的增长不属于该临界区，由于每一个线程有各自的计数器（main函数中的count）。既然如此，咱们就不把这行代码包括在由互斥锁锁住的临界区中，由于做为一个通用的编程原则，咱们老是应该努力减小由一个互斥锁锁住的代码量。

七、消费者验证数组内容：

这里不须要任何同步。

未完待续！