线程间同步与互斥:生产者消费者问题

总结一下线程间同步与互斥生 产者消费者问题

一 互斥锁(mutex)

对于多线程的程序,访问冲突的问题是很普遍的,解决的办法是引⼊入互斥锁(Mutex,MutualExclusive Lock),获得锁的线程可以完成“读-修改-写”的操作,然后释放锁给其它线程,没有获得锁的线程只能等待而不能访问共享数据,这样“读-修改-写”三步操作组成一个原子操作,要么都执行,要么都不执行,不会执行到中间被打断,也不会在其它处理器上并行做这个操作。

 互斥锁的两个基本操作lock 和unlock的实现：为了实现互斥锁操作,大多数体系结构都提供了swap或exchange指令,该指令的作用是把寄存器和内存单元的数据相交换,由于只有一条指令,保证了原子性,即使是多处理器平台,访问内存的 总线周期也有先后,一个处理器上的交换指令执行时另一个处理器的交换指令只能等待总线周期。

 以x86为例的lock和unlock的伪代码

lock：
movb $0, %al
xchgb %al, mutex
if(al寄存器的内容 >0)
{ return 0;
}else
挂起等待;
goto lock;
unlock：
movb $1, mutex
唤醒等待mutex的线程;
retrun 0;

也许还有读者好奇,“挂起等待”和“唤醒等待线程”的操作如何实现?每个Mutex有⼀一个等待队列,一个线程要在Mutex上挂起等待,首先在把自己加入等待队列中,然后置线程状态为睡眠,然后调用调度器函数切换到别的线程。一个线程要唤醒等待队列中的其它线程,只需从等待队列中取出一 项,把它的状态从睡眠改为就绪,加入就绪队列,那么下次调度器函数执行时就有可能切换到被唤醒的线程。

介绍下相关函数：

(1)互斥锁的创建和销毁



返回值:成功返回0,失败返回错误号。

pthread_mutex_init函数对Mutex做初始化,参数attr设定Mutex的属性,如果attr为NULL则表示缺省属性，用pthread_mutex_init函 数初始化的Mutex可以⽤用pthread_mutex_destroy销毁。如果Mutex变量是静态分配的(全局变量 或static变量),也可以⽤用宏定义PTHREAD_MUTEX_INITIALIZER来初始化,相当于用pthread_mutex_init初始化并且attr参数为NULL。

(2)加锁和解锁函数



返回值：成功返回0，错误返回错误码

一个线程可以调⽤用pthread_mutex_lock获得Mutex,如果这时另一个线程已经调pthread_mutex_lock获得了该Mutex,则当前线程需要挂起等待,直到另一个线程调用pthread_mutex_unlock释放Mutex,当前线程被唤醒,才能获得该Mutex并继续执行。

如果一个线程既想获得锁,又不想挂起等待,可以调⽤用pthread_mutex_trylock,如果Mutex已经被 另一个线程获得,这个函数会失败返回错误码,而不会使线程挂起等待。

二 条件变量Condition Variable



和Mutex的初始化和销毁类似,pthread_cond_init函数初始化一个Condition Variable,attr参数为NULL则表示缺省属性,pthread_cond_destroy函数销毁一个Condition Variable。如果ConditionVariable是静态分配的,也可以用宏定义PTHEAD_COND_INITIALIZER初始化,相当于用pthread_cond_init函数初始化并且attr参数为NULL。



可见，一个Condition Variable总是和一个Mutex搭配使用的。一个线程可以调用pthread_cond_wait在一个Condition Variable上阻塞等待, 这个函数做以下三步操作:
1. 释放Mutex，解锁

2. 阻塞等待

3. 当被唤醒时,重新获得Mutex并返回

pthread_cond_timedwait函数还有一个额外的参数可以设定等待超时,如果到达了abstime所指定的 时刻仍然没有别的线程来唤醒当前线程,就返回ETIMEDOUT。



pthread_cond_signal 唤醒在Condition Variable上等待的单个进程 

pthread_cond_broadcast唤醒在Condition Variable上等待的所有线程

生成者消费者问题：生产者之间是互斥关系，消费者之间是互斥关系，生成者和消费者之间是互斥同步关系

代码：

#include<stdio.h>
#include<unistd.h>
#include<stdlib.h>
#include<assert.h>
#include<pthread.h>
typedef struct Node
{
int _data;
struct Node* _next;
}Node_t,*Node_p,**Node_pp;

static Node_p malloc_Node(int data)
{
Node_p p = (Node_p)malloc(sizeof(Node_t));
if(p == NULL)
{
perror("malloc");
exit(1);
}
p->_data = data;
p->_next = NULL;
return p;
}
static int is_empty(Node_p head)
{
assert(head);
return (head->_next == NULL) ? 1 : 0;
}
static void delete_node(Node_p p)
{
if(p)
{
free(p);
}
}
void init_list(Node_pp head)
{
assert(head);
*head = malloc_Node(0);
}
void push_list(Node_p head,int data) // head insert
{
assert(head);
Node_p p =(Node_p)malloc_Node(data);
p->_next = head->_next;
head->_next = p;
}
void pop_list(Node_p head,int *out) //head pop
{
assert(head);
assert(out);
if(!is_empty(head))
{
Node_p del = head->_next;
head->_next = del->_next;
*out = del->_data;
delete_node(del);
}
}
void destroy_list(Node_p head)
{
assert(head);
while(head->_next)
{
Node_p del = head->_next;
head->_next = del->_next;
delete_node(del);
}
}
void show_list(Node_p head)
{
assert(head);
Node_p p = head->_next;
while(p)
{
printf("%d ",p->_data);
p = p->_next;
}
printf("\n");
}

pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
void* consumer(void *arg)
{
Node_p head = (Node_p)arg;
int data = 0;
while(1)
{
pthread_mutex_lock(&lock);
while(is_empty(head))
{
printf("no data can be consumed\n");
pthread_cond_wait(&cond,&lock);
}
pop_list(head,&data);
printf("consume data: %d \n",data);
pthread_mutex_unlock(&lock);
//sleep(1);
}
return NULL;
}
void* producter(void *arg)
{
Node_p head = (Node_p)arg;
int data = 0;
while(1)
{
pthread_mutex_lock(&lock);
data = rand()%1234;
push_list(head,data);
printf("product data: %d \n",data);
pthread_mutex_unlock(&lock);
sleep(1);
pthread_cond_signal(&cond);
printf("data is signal to consumer !\n");
}
return NULL;
}

int main()
{
Node_p head = NULL;
init_list(&head);
pthread_t id1,id2;

pthread_create(&id1,NULL,consumer,head);
pthread_create(&id2,NULL,producter,head);

pthread_join(id1,NULL);
pthread_join(id2,NULL);
pthread_mutex_destroy(&lock);
pthread_cond_destroy(&cond);
destroy_list(head);

/*
Node_p head = NULL;
init_list(&head);
int i = 0;
for(;i < 10;++i)
{
push_list(head,i);
show_list(head);
sleep(1);
}
int data = 0;
for(i = 0; i < 10; ++i)
{
pop_list(head,&data);
show_list(head);
sleep(1);
}
destroy_list(head);
*/

return 0;
}