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

红太狼 2022-09-26 03:03 205阅读 0赞

## **1.概述** ##

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

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

> **NOTE：Linux操作系统中，POSIX有名信号量创建在虚拟文件系统中 一般挂载在/dev/shm，其名字以sem.somename的形式存在。**

--------------------

## **2.信号量操作** ##

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

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

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

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

sem++;

信号量初始化的值的大小一般用于表示可用资源的数（例如缓冲区大小，之后代码中体现）;如果初始化为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);
        }//第一个函数
    
        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);
    }

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

![这里写图片描述][20160919095318098]

--------------------

### **3.2 有名信号量** ###

有名信号量要指定一个名字somename，打开成功后将以sem.somename的形式存在于/dev/shm/目录下。  
书中用二值信号量做互斥同步，这里我直接用`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,O_CREAT|O_EXCL,0644,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,handler,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;
    }

--------------------

## **4.参考** ##

1.UNP卷二  
2.man page `(man sem_overview)`

[20160919095318098]: /images/20220717/4b088054d7b84e508533cadf374528fd.png