POSIX信号量解决经典同步问题

posix信号量

信号量简介

POSIX信号量是属于POSIX标准系统接口定义的实时扩展部分。在SUS（Single UNIX Specification）单一规范中，定义的XSI IPC中也同样定义了人们通常称为System V信号量的系统接口。信号量作为进程间同步的工具是很常用的一种同步IPC类型。

信号量是一种用于不同进程间进行同步的工具，当然对于进程安全的对于线程也肯定是安全的，所以信号量也理所当然可以用于同一进程内的不同线程的同步。

有了互斥量和条件变量还提供信号量的原因是：信号量的主要目的是提供一种进程间同步的方式。这种同步的进程可以共享也可以不共享内存区。虽然信号量的意图在于进程间的同步，互斥量和条件变量的意图在于线程间同步，但信号量也可用于线程间同步，互斥量和条件变量也可通过共享内存区进行进程间同步。但应该根据具体应用考虑到效率和易用性进行具体的选择。

POSIX信号量有两种：有名信号量和无名信号量，无名信号量也被称作基于内存的信号量。有名信号量通过IPC名字进行进程间的同步，而无名信号量如果不是放在进程间的共享内存区中，是不能用来进行进程间同步的，只能用来进行线程同步。本文主要是讨论无名信号量来解决经典同步问题。

无名信号量函数简介

int sem_init(sem_t *sem, int pshared, unsigned int value);  //若出错则返回-1
int sem_destroy(sem_t *sem); //成功返回0，失败返回-1
int sem_wait(sem_t *sem);   //相当于P操作
int sem_post(sem_t *sem);   //相当于V操作

函数说明

sem_init()用于无名信号量的初始化。无名信号量在初始化前一定要在内存中分配一个sem_t信号量类型的对象，这就是无名信号量又称为基于内存的信号量的原因。

sem_init()第一个参数是指向一个已经分配的sem_t变量。第二个参数pshared表示该信号量是否由于进程间通步，当pshared = 0，那么表示该信号量只能用于进程内部的线程间的同步。当pshared != 0，表示该信号量存放在共享内存区中，使使用它的进程能够访问该共享内存区进行进程同步。第三个参数value表示信号量的初始值。

这里需要注意的是，无名信号量不使用任何类似O_CREAT的标志，这表示sem_init()总是会初始化信号量的值，所以对于特定的一个信号量，我们必须保证只调用sem_init()进行初始化一次，对于一个已初始化过的信号量调用sem_init()的行为是未定义的。如果信号量还没有被某个线程调用还好，否则基本上会出现问题。

使用完一个无名信号量后，调用sem_destroy摧毁它。这里要注意的是：摧毁一个有线程阻塞在其上的信号量的行为是未定义的。

消费者问题

#include<unistd.h>
#include<semaphore.h>
#include<pthread.h>
#include<signal.h>
#include<cstring>
#include<cstdlib>
#include<cstdio>
#include<iostream>
struct Syn_info
{
Syn_info()
{
sem_init(&mutex_full,0,0);
sem_init(&mutex_empty,0,10);
sem_init(&mutex,0,1);
count=0;
}
sem_t mutex_full;
sem_t mutex_empty;
sem_t mutex;
int count;
};
/*多个生产者消费者问题*/
void* producen(void* arg)/*生产者*/
{
Syn_info* info=(Syn_info*)arg;
while(1)
{
std::cout<<"produe product "<<std::endl;
sem_wait(&info->mutex_empty);
sem_wait(&info->mutex);
if(info->count<10)
{
info->count++;
std::cout<<"producer put product in work line"<<std::endl;
std::cout<<"free product num for produce: "<<10-info->count<<std::endl;
sem_post(&info->mutex);
sem_post(&info->mutex_full);
}
else
{
std::cout<<"no free space for producer to put product"<<std::endl;
sem_post(&info->mutex);
}
sleep(1);
}
}
void* consumen(void* arg)/*消费者*/
{
Syn_info* info=(Syn_info*)arg;
while(1)
{
sem_wait(&info->mutex_full);
sem_wait(&info->mutex);
if(info->count>0)
{
info->count--;
std::cout<<"consumer get product from work line"<<std::endl;
std::cout<<"product on the work line num: "<<info->count<<std::endl;
sem_post(&info->mutex);
sem_post(&info->mutex_empty);
}
else
{
std::cout<<"no product on work line"<<std::endl;
sem_post(&info->mutex);
}
sleep(1);
}
}
/*一个生产者消费者问题*/
void* produce(void* arg)/*生产者*/
{
Syn_info* info=(Syn_info*)arg;
while(1)
{
std::cout<<"produe product "<<std::endl;
sem_wait(&info->mutex_empty);
sem_wait(&info->mutex);
std::cout<<"put product in work line"<<std::endl;
sem_post(&info->mutex);
sem_post(&info->mutex_full);
sleep(1);
}
}
void* consume(void* arg)/*消费者*/
{
Syn_info* info=(Syn_info*)arg;
while(1)
{
sem_wait(&info->mutex_full);
sem_wait(&info->mutex);
std::cout<<"product on the work line num: "<<std::endl;
sem_post(&info->mutex);
sem_post(&info->mutex_empty);
std::cout<<"consume the product"<<std::endl;
sleep(1);
}
}
int main(int argc,char** argv)
{
pthread_t* tid=new pthread_t[10]();
Syn_info* info=new Syn_info();
for(int index=0;index!=20;index++)
{
if(index<10)
{
pthread_create(tid+index,NULL,produce,(void*)info);
}
else
{
pthread_create(tid+index,NULL,consume,(void*)info);
}
}
while(1);
}

读者写者问题

#include<unistd.h>
#include<semaphore.h>
#include<pthread.h>
#include<signal.h>
#include<cstring>
#include<cstdlib>
#include<cstdio>
#include<iostream>
struct Syn_info
{
Syn_info()
{
sem_init(&mutex_r,0,1);
sem_init(&mutex_f,0,1);
sem_init(&mutex_rw,0,1);
count=0;
}
sem_t mutex_f;
sem_t mutex_r;
sem_t mutex_rw;
int count;
};
/*读者写者问题*/
void* writer(void* arg)
{
Syn_info* info=(Syn_info*)arg;
while(1)
{
sem_wait(&info->mutex_f);
sem_wait(&info->mutex_rw);
std::cout<<"write"<<std::endl;
sem_post(&info->mutex_rw);
sem_post(&info->mutex_f);
}
}
void* reader(void* arg)
{
Syn_info* info=(Syn_info*)arg;
while(1)
{
sem_wait(&info->mutex_f);
sem_wait(&info->mutex_r);
std::cout<<"one reader access"<<std::endl;
if(info->count==0)
{
sem_wait(&info->mutex_rw);
std::cout<<"prevent writer to write"<<std::endl;
}
info->count++;
sem_post(&info->mutex_r);
sem_post(&info->mutex_f);
std::cout<<"reading"<<std::endl;
sleep(3);
sem_post(&info->mutex_r);
info->count--;
if(info->count==0)
{
sem_post(&info->mutex_rw);
std::cout<<"ok ,now writer can write"<<std::endl;
}
std::cout<<"one reader leave"<<std::endl;
sem_post(&info->mutex_r);
}
}
int main(int argc,char** argv)
{
pthread_t* tid=new pthread_t[10]();
Syn_info* info=new Syn_info();
for(int index=0;index!=20;index++)
{
if(index<10)
{
pthread_create(tid+index,NULL,writer,(void*)info);
}
else
{
pthread_create(tid+index,NULL,reader,(void*)info);
}
}
while(1);
}

理发师问题

#include<unistd.h>
#include<semaphore.h>
#include<pthread.h>
#include<signal.h>
#include<cstring>
#include<cstdlib>
#include<cstdio>
#include<ctime>
#include<iostream>
struct Syn_info
{
Syn_info()
{
sem_init(&mutex,0,1);
sem_init(&customer,0,0);
sem_init(&barber,0,0);
count=10;
wait=0;
}
sem_t mutex;
sem_t customer;
sem_t barber;
int count;
int wait;
};
/*理发师问题*/
void* barber(void* arg)
{
Syn_info* info=(Syn_info*)arg;
while(1)
{
sem_wait(&info->customer);
sem_wait(&info->mutex);
info->wait--;
sem_post(&info->barber);
sem_post(&info->mutex);
std::cout<<"hair cuting....."<<std::endl;
}
}
void* customer(void* arg)
{
Syn_info* info=(Syn_info*)arg;
while(1)
{
sem_wait(&info->mutex);
if(info->wait<info->count)
{
info->wait++;
std::cout<<info->wait<<" customer waiting for haircut"<<std::endl;
sem_post(&info->customer);
sem_post(&info->mutex);
sem_wait(&info->barber);
std::cout<<"waiting for hair cut"<<std::endl;
}
else
{
sem_post(&info->mutex);
}
}
}
int main(int argc,char** argv)
{
pthread_t* tid=new pthread_t[10]();
Syn_info* info=new Syn_info();
pthread_create(tid,NULL,barber,(void*)info);
for(int index=1;index!=10;index++)
{
pthread_create(tid+index,NULL,customer,(void*)info);
}
while(1);
}

家庭吃水果问题

#include<unistd.h>
#include<semaphore.h>
#include<pthread.h>
#include<signal.h>
#include<cstring>
#include<cstdlib>
#include<cstdio>
#include<iostream>
struct Syn_info
{
Syn_info()
{
sem_init(&plate,0,1);
sem_init(&orange,0,0);
sem_init(&apple,0,0);
}
sem_t plate;
sem_t orange;
sem_t apple;
};
void* father(void* arg)
{
Syn_info* info=(Syn_info*)arg;
while(1)
{
sleep(1);
sem_wait(&info->plate);
std::cout<<"father put apple in the plate"<<std::endl;
sem_post(&info->apple);
}
}
void* monther(void* arg)
{
Syn_info* info=(Syn_info*)arg;
while(1)
{
sleep(1);
sem_wait(&info->plate);
std::cout<<"monther put orange in the plate"<<std::endl;
sem_post(&info->orange);
}
}
void* son(void* arg)
{
Syn_info* info=(Syn_info*)arg;
while(1)
{
sleep(1);
sem_wait(&info->apple);
std::cout<<"son get apple from plate"<<std::endl;
sem_post(&info->plate);
}
}
void* daughter(void* arg)
{
Syn_info* info=(Syn_info*)arg;
while(1)
{
sleep(1);
sem_wait(&info->orange);
std::cout<<"daughter get orange from plate"<<std::endl;
sem_post(&info->plate);
}
}
Syn_info* temp;
void handle(int signum)
{
std::cout<<"quit"<<std::endl;
sem_destroy(&temp->plate);
sem_destroy(&temp->apple);
sem_destroy(&temp->orange);
_exit(1);
}
int main(int argc,char** argv)
{
struct sigaction act;
act.sa_handler=handle;
act.sa_flags=0;
sigaction(SIGINT,&act,NULL);
pthread_t* tid=new pthread_t[10]();
Syn_info* info=new Syn_info();
temp=info;
pthread_create(tid+1,NULL,father,(void*)info);
pthread_create(tid+2,NULL,monther,(void*)info);
pthread_create(tid+3,NULL,son,(void*)info);
pthread_create(tid+4,NULL,daughter,(void*)info);
while(1);
}

信号量的销毁和继承

(1) 对于有名信号量在父进程中打开的任何有名信号量在子进程中仍是打开的。如果无名信号量是在单个进程内部的数据空间中，那么信号量就是进程数据段或者是堆栈上，当fork产生子进程后，该信号量只是原来的一个拷贝，和之前的信号量是独立的。

(2) 对于有名信号量，当某个持有该信号量的进程没有解锁该信号量就终止了，内核并不会将该信号量解锁。这跟记录锁不一样。对于无名信号量，如果信号量位于进程内部的内存空间中，当进程终止后，信号量也就不存在了，无所谓解锁了。如果信号量位于进程间的共享内存区中，当进程终止后，内核也不会将该信号量解锁。