进程间通信笔记（6）—POSIX信号量

开发技术作者： 2024-06-20 23:35:01

1.概述信号量（semaphore）是一种提供不同进程间或者一个给定进程不同线程之间的同步。依然分为POSIX信号量和SystemV信号量，这里先学习POSIX信号量。POSIX信号量又分为有名信号量和基于内存的信号量（无名信号量）。区别在于是否需要使用POSIX IPC名字来标识。 NOTE：

1.概述

信号量（semaphore）是1种提供不同进程间或1个给定进程不同线程之间的同步。
仍然分为POSIX信号量和SystemV信号量，这里先学习POSIX信号量。

POSIX信号量又分为着名信号量和基于内存的信号量（无名信号量）。区分在因而否需要使用POSIX IPC名字来标识。

NOTE：Linux操作系统中，POSIX着名信号量创建在虚拟文件系统中 1般挂载在/dev/shm，其名字以sem.somename的情势存在。

2.信号量操作

早在学操作系统那会，就直到信号量的PV操作，总结1下大概是这么回事：

P操作，也叫做等待（wait）1个信号量，该操作会测试信号量的值，如果其值小于或等于0，将把当前进程/线程投入眠眠，当该信号量变得大于0后就将它减1。
伪代码以下，这两步必须是原子操作。

while(sem <=0);
sem--;

V操作，挂出（post）1个信号量，该操作将信号量值加1
伪代码以下：

sem++;

信号量初始化的值的大小1般用于表示可用资源的数（例如缓冲区大小，以后代码中体现）;如果初始化为1，则称之2值信号量，2值信号量的功能就有点像互斥锁了。

不同的是：互斥锁的加锁和解锁必须在同1线程履行，而信号量的挂出却没必要由履行等待操作的线程履行。

3.POSIX信号量编程

POSIX信号量编程，离不开以下函数：

1.sem_init()初始化无名信号量

#include <semaphore.h>
int sem_init(sem_t *sem,int pshared,unsigned int value);
//pshared指定该信号量用于进程还是线程同步
//0-表示用于线程同步（所有线程可见）
//非0-表示用于进程同步（需要放在同享内存中）

2.sem_open()初始化着名信号量

#include <fcntl.h>           /* For O_* constants */
#include <sys/stat.h>        /* For mode constants */
#include <semaphore.h>
sem_t *sem_open(const char *name,int oflag);
sem_t *sem_open(const char *name,int oflag,mode_t mode,unsigned int value);
Link with -pthread.
//套路跟之前的消息队列类似
//oflag可以设置为O_CREAT | O_EXCL (如果存在则返回毛病)。
//mode可以设置为0644 自己可读写，其他用户和组内用户可读
//value表示信号量初始化的值

3.sem_wait()和sem_post()等待和挂出函数

#include <semaphore.h>
int sem_wait(sem_t *sem);//P操作
int sem_post(sem_t *sem);//V操作
Link with -pthread.

4.sem_timedwait()超时等待

 int sem_timedwait(sem_t *sem,const struct timespec *abs_timeout);
 //跟sem_wait()类似，不过，如果P操作不能立即履行，该函数将投入眠眠并等待abs_timeout中指定的时间。
 //如果超时照旧没法履行P操作，则返回timeout毛病

3.1 基于内存的信号量（无名信号量）

#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <semaphore.h>
#include <time.h>
#include <assert.h>
#include <errno.h>
#include <signal.h>

sem_t sem;

#define handle_error(msg) \
    do { perror(msg); exit(EXIT_FAILURE); } while (0)//毛病处理宏

static void handler(int sig)//信号处理函数
{//信号处理函数中将信号挂出
    printf("in signal handler \r\n");
    write(STDOUT_FILENO,"sem_post() from handler\n",24);
    if (sem_post(&sem) == -1) {
        write(STDERR_FILENO,"sem_post() Failed\n",18);
        _exit(EXIT_FAILURE);
    }
}

int main(int argc,char *argv[])
{
    struct sigaction sa;//注册信号处理函数
    struct timespec ts;//超时
    int s;

    if (argc != 3) {
        fprintf(stderr,"Usage: %s <alarm-secs> <wait-secs>\n",argv[0]);
        exit(EXIT_FAILURE);
    }//第1个函数

    if (sem_init(&sem,0,0) == -1)
    handle_error("sem_init");


    sa.sa_handler = handler;
    sigemptyset(&sa.sa_mask);
    sa.sa_flags = 0;
    if (sigaction(SIGALRM,&sa,NULL) == -1)//设置定时器信号处理
        handle_error("sigaction");
    alarm(atoi(argv[1])); //开启定时器

    if (clock_gettime(CLOCK_REALTIME,&ts) == -1)
        handle_error("clock_gettime");

    ts.tv_sec += atoi(argv[2]);


    while ((s = sem_timedwait(&sem,&ts)) == -1 && errno == EINTR)//超时等待信号量
        continue;       


    if (s == -1) {
        if (errno == ETIMEDOUT)
            printf("sem_timedwait() timed out\n");
        else
            perror("sem_timedwait");
    } else
        printf("sem_timedwait() succeeded\n");

    exit((s == 0) ? EXIT_SUCCESS : EXIT_FAILURE);
}

命令行参数有两个，第1个是定时器发出SIGALRM的时间，1个是信号量阻塞等待的时间。
并且，在接收SIGALRM信号的处理函数中，将信号量挂出，这样：
如果定时器发出SIGALRM时间大于信号阻塞等待的时间，sem_timedwait()将会返回timeout毛病。反之，如果在此之前进入信号处理函数，将信号量挂出，则将调用成功。如图：

这里写图片描述

3.2 着名信号量

着名信号量要指定1个名字somename，打开成功后将以sem.somename的情势存在于/dev/shm/目录下。
书中用2值信号量做互斥同步，这里我直接用mutex。

#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <semaphore.h>
#include <time.h>
#include <assert.h>
#include <errno.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <pthread.h>

#define MAXSIZE 10
void *produce(void *); 
void *consume(void *);

typedef void *(*handler_t)(void *);//线程函数指针

struct _shared 
{
    int buff[MAXSIZE];

    sem_t *nempty;
    sem_t *nstored;
};//同享缓冲
typedef struct _shared shared;
shared shared_;
pthread_mutex_t mutex;//互斥锁
int nitems;//生产和消费的数目

int main(int argc,char **argv)
{
    if (argc !=2)
    {
        printf("usage:namedsem <#items>\r\n");
        exit(-1);
    }
    nitems=atoi(argv[1]);
    const char *const SEM_NEMPTY = "nempty";//信号量的“名字”
    const char *const SEM_NSTORED = "nstored";//信号量的“名字”
    pthread_t tid_produce;//生产者线程 id
    pthread_t tid_consume;//消费者线程 id
    //初始化信号量和互斥锁
    pthread_mutex_init(&mutex,NULL); 
    shared_.nstored=sem_open(SEM_NSTORED,O_CREAT|O_EXCL,0644,0);
    shared_.nempty=sem_open(SEM_NEMPTY,MAXSIZE);
    memset(shared_.buff,0x00,MAXSIZE);

    //线程创建
    handler_t handler=produce;
    pthread_setconcurrency(2);
    if((pthread_create(&tid_produce,NULL,handler,NULL))<0)
    {
        printf("pthread_create error\r\n");
        exit(-1);
    }
//    sleep(5);
    handler=consume;
    if((pthread_create(&tid_consume,NULL))<0)
    {
        printf("pthread_create error\r\n");
        exit(-1);
    }

    //线程回收
    pthread_join(tid_produce,NULL);
    pthread_join(tid_consume,NULL);

    //信号量锁烧毁
    sem_unlink(SEM_NEMPTY);
    sem_unlink(SEM_NSTORED);
    pthread_mutex_destroy(&mutex);
    exit(0);
}



void *produce(void *args)
{
    int i;
    for(i=0;i<nitems;i++)
    {
        sem_wait(shared_.nempty);
        pthread_mutex_lock(&mutex);
        shared_.buff[i%MAXSIZE]=i;
        printf("add an item\r\n");
        pthread_mutex_unlock(&mutex);
        sem_post(shared_.nstored);
    }
    return NULL;
}


void *consume(void *args)
{
    int i;
    for(i=0;i<nitems;i++)
    {
        sem_wait(shared_.nstored);
        pthread_mutex_lock(&mutex);
        printf("consume an item\r\n");
        if(shared_.buff[i%MAXSIZE]!=i)
            printf("buff[%d]=%d\r\n",i,shared_.buff[i%MAXSIZE]);
        pthread_mutex_unlock(&mutex);
        sem_post(shared_.nempty);
    }
    return NULL;
}