【IPC】Posix信号灯

心已赠人 2022-07-27 14:47 175阅读 0赞

信号灯semaphore是一种进程间或者线程间的同步原语，有以下三种形式——

Posix有名信号灯：使用Posix IPC名字标识，相关函数为sem\_open/sem\_close/sem\_unlink，可用于进程或线程间的同步。  
Posix基于内存的信号灯：存放在共享内存区中，相关函数为sem\_init/sem\_destroy，可用于进程或线程间的同步。  
System V信号灯：在内核中维护，可用于进程或线程间的同步。

创建信号灯之后，根据信号灯具体的值，可以加锁wait或者解锁post，有时也称为PV操作，这种叫法源自荷兰语，P即减小信号灯的值，V即增加信号灯的值。二值信号灯是一种信号灯，其值为0或1，状态类似于互斥锁，要么加锁，要么解锁，不同的是互斥锁必须总是由给它加锁的线程解锁，而信号灯的V操作则不必由执行过它的P操作的同一线程执行。还有一种信号灯是计数信号灯，就是用来计数的。

下面是Posix信号灯semaphore相关的几个函数，使用链接选项“-pthread”。

#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);

sem\_open打开或创建一个有名信号灯，失败时返回SEM\_FAILED，并设置相应的errno，成功时返回指向信号灯的地址。当参数oflag指定了O\_CREAT时，需指定mode权限位和value初始值，value不超过SEM\_VALUE\_MAX，通常来说，二值信号灯初始化为1，计数信号灯初始化为大于1的值。一个进程同时打开的信号灯上限为SEM\_NSEMS\_MAX。创建信号灯的例子如下。

// semcreate.c
    #include <stdio.h>
    #include <stdlib.h>
    #include <unistd.h>
    #include <string.h>
    #include <errno.h>
    #include <fcntl.h>
    #include <semaphore.h>
    
    #define FILE_MODE (S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH)
    
    int main(int argc, char **argv)
    {
        int c, flags;
        sem_t *sem;
        unsigned int value;
        flags = O_RDWR | O_CREAT;
        value = 1;
        while ((c = getopt(argc, argv, "ei:")) != -1) {
            switch (c) {
            case 'e':
                flags |= O_EXCL;
                break;
            case 'i':
                value = atoi(optarg);
                break;
            }
        }
        if (optind != argc - 1) {
            printf("usage: %s [-e] [-i initvalue] <name>\n", argv[0]);
            exit(EXIT_FAILURE);
        }
        ;
        if ((sem = sem_open(argv[optind], flags, FILE_MODE, value)) == SEM_FAILED) {
            printf("sem_open error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        if (sem_close(sem) == -1) {
            printf("sem_close error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        exit(EXIT_SUCCESS);
    }

int sem_close(sem_t *sem);
    int sem_unlink(const char *name);

sem\_close关闭一个有名信号灯，成功时返回0，失败时返回-1并设置相应的errno。一个进程终止时，无论是自愿结束还是非自愿结束，内核都会为所有打开的有名信号灯执行这个关闭操作。然而关闭一个有名信号灯，即使当前没有任何进程使用它，也不能从系统中真正的移除，彻底移除要使用sem\_unlink。删除信号灯的例子如下。

// semunlink.c
    #include <stdio.h>
    #include <stdlib.h>
    #include <unistd.h>
    #include <string.h>
    #include <errno.h>
    #include <fcntl.h>
    #include <semaphore.h>
    
    int main(int argc, char **argv)
    {
        if (argc != 2) {
            printf("usage: %s <name>\n", argv[0]);
            exit(EXIT_FAILURE);
        }
        if (sem_unlink(argv[1]) == -1) {
            printf("sem_unlink error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        exit(EXIT_SUCCESS);
    }

int sem_wait(sem_t *sem);
    int sem_trywait(sem_t *sem);

sem\_wait对信号灯加锁，当信号灯的值大于0时，该函数将信号灯的值减1并返回，当信号灯的值等于0时，该函数的调用线程将阻塞，直到信号灯的值大于0，阻塞期间有可能被信号中断，这时返回会EINTR。sem\_trywait则不会阻塞，加锁失败时返回EAGAIN。这两个函数成功执行时返回0，错误返回-1并设置相应的errno。对信号灯加锁的例子如下。

// semwait.c
    #include <stdio.h>
    #include <stdlib.h>
    #include <unistd.h>
    #include <string.h>
    #include <errno.h>
    #include <fcntl.h>
    #include <semaphore.h>
    
    int main(int argc, char **argv)
    {
        sem_t *sem;
        int value;
        if (argc != 2) {
            printf("usage: %s <name>\n", argv[0]);
            exit(EXIT_FAILURE);
        }
        if ((sem = sem_open(argv[1], 0)) == SEM_FAILED) {
            printf("sem_open error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        if (sem_wait(sem) == -1) {
            printf("sem_wait error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        if (sem_getvalue(sem, &value) == -1) {
            printf("sem_getvalue error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        printf("pid %ld has semaphore, value = %d\n",(long)getpid(), value);
        pause(); // blocks until killed
        exit(EXIT_SUCCESS);
    }

int sem_post(sem_t *sem);
    int sem_getvalue(sem_t *sem, int *sval);

sem\_post对信号灯解锁，将信号灯的值加1，唤醒等待该信号灯的线程。sem\_getvalue用于获取某个信号灯的值。对信号灯解锁及获取信号灯值的例子如下。

// sempost.c
    #include <stdio.h>
    #include <stdlib.h>
    #include <unistd.h>
    #include <string.h>
    #include <errno.h>
    #include <fcntl.h>
    #include <semaphore.h>
    
    int main(int argc, char **argv)
    {
        sem_t *sem;
        int value;
        if (argc != 2) {
            printf("usage: %s <name>\n", argv[0]);
            exit(EXIT_FAILURE);
        }
        if ((sem = sem_open(argv[1], 0)) == SEM_FAILED) {
            printf("sem_open error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        if (sem_post(sem) == -1) {
            printf("sem_post error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        if (sem_getvalue(sem, &value) == -1) {
            printf("sem_getvalue error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        printf("value = %d\n", value);
        exit(EXIT_SUCCESS);
    }
    
    // semgetvalue.c
    #include <stdio.h>
    #include <stdlib.h>
    #include <unistd.h>
    #include <string.h>
    #include <errno.h>
    #include <fcntl.h>
    #include <semaphore.h>
    
    int main(int argc, char **argv)
    {
        sem_t *sem;
        int value;
        if (argc != 2) {
            printf("usage: %s <name>\n", argv[0]);
            exit(EXIT_FAILURE);
        }
        if ((sem = sem_open(argv[1], 0)) == SEM_FAILED) {
            printf("sem_open error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        if (sem_getvalue(sem, &value) == -1) {
            printf("sem_getvalue error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        printf("value = %d\n", value);
        exit(EXIT_SUCCESS);
    }

int sem_init(sem_t *sem, int pshared, unsigned int value);
    int sem_destroy(sem_t *sem);

相对于有名信号灯，sem\_init初始化一个匿名信号灯，参数pshared指定是在线程间还是在进程间共享，sem\_destroy销毁一个匿名信号灯。

下面举例说明Posix信号灯semaphore的用法（通过”man 7 sem\_oveview“可知，有名信号灯的名字要以一个斜线开头，后面不能包括斜线，例如”/semtest“，在Linux上，是在虚拟文件系统中创建的，位置在”/dev/shm“，例如”/dev/shm/sem.semtest“）。

// prodcons.c
    #include <stdio.h>
    #include <stdlib.h>
    #include <unistd.h>
    #include <string.h>
    #include <errno.h>
    #include <fcntl.h>
    #include <pthread.h>
    #include <semaphore.h>
    
    #define FILE_MODE (S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH)
    #define NBUFF (10) // length for shared buffer
    
    int nitems; // products number by producer
    struct
    {
        int buff[NBUFF]; // shared buffer
        sem_t *mutex; // binary semaphore, initial value is 1
        sem_t *nempty; // countable semaphore, initial value is NBUFF
        sem_t *nstored; // countable semaphore, initial value is 0
    } shared;
    
    void* producer(void*);
    void* consumer(void*);
    
    int main(int argc, char **argv)
    {
        pthread_t tid_producer, tid_consumer;
        int count_prod, count_cons;
        if (2 != argc) {
            printf("usage: %s <items>\n", argv[0]);
            exit(EXIT_FAILURE);
        }
        nitems = atoi(argv[1]);
    
        // creates three semaphores
        // "/mutex"
        if ((shared.mutex = sem_open("/mutex", O_CREAT | O_EXCL, FILE_MODE, 1)) == SEM_FAILED) {
            printf("sem_open /mutex error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        else {
            printf("semaphore /mutex created\n");
        }
        // "/nempty"
        if ((shared.nempty = sem_open("/nempty", O_CREAT | O_EXCL, FILE_MODE, NBUFF)) == SEM_FAILED) {
            printf("sem_open /nempty error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        else {
            printf("semaphore /nempty created\n");
        }
        // "/nstored"
        if ((shared.nstored = sem_open("/nstored", O_CREAT | O_EXCL, FILE_MODE, 0)) == SEM_FAILED) {
            printf("sem_open /nstored error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        else {
            printf("semaphore /nstored created\n");
        }
    
        // creates and waits for producer and consumer threads
        if (0 != pthread_setconcurrency(2)) {
            printf("pthread_setconcurrency error\n");
        }
        // producer create
        count_prod = 0;
        if (0 != pthread_create(&tid_producer, NULL, producer, (void*)&count_prod)) {
            printf("pthread_create error: producer\n");
            exit(EXIT_FAILURE);
        }
        else {
            printf("producer [%lu] created, count = %d\n", tid_producer, count_prod);
        }
        // consumer create
        count_cons = 0;
        if (0 != pthread_create(&tid_consumer, NULL, consumer, (void*)&count_cons)) {
            printf("pthread_create error: consumer\n");
            exit(EXIT_FAILURE);
        }
        else {
            printf("consumer [%lu] created, count = %d\n", tid_consumer, count_cons);
        }
        // producer join
        if (0 != pthread_join(tid_producer, NULL)) {
            printf("pthread_join error: producer\n");
            exit(EXIT_FAILURE);
        }
        else {
            printf("producer [%lu] done, count = %d\n", tid_producer, count_prod);
        }
        // consumer join
        if (0 != pthread_join(tid_consumer, NULL)) {
            printf("pthread_join error: producer\n");
            exit(EXIT_FAILURE);
        }
        else {
            printf("consumer [%lu] done, count = %d\n", tid_consumer, count_cons);
        }
    
        // removes the three created semaphores
        // "/mutex"
        if (0 != sem_unlink("/mutex")) {
            printf("sem_unlink /mutex error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        else {
            printf("semaphore /mutex removed\n");
        }
        // "/nempty"
        if (0 != sem_unlink("/nempty")) {
            printf("sem_unlink /nempty error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        else {
            printf("semaphore /nempty removed\n");
        }
        // "/nstored"
        if (0 != sem_unlink("/nstored")) {
            printf("sem_unlink /nstored error: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }
        else {
            printf("semaphore /nstored removed\n");
        }
    
        exit(EXIT_SUCCESS);
    }
    
    void* producer(void *arg)
    {
        int i;
        for (i = 0; i < nitems; ++i) {
            if (sem_wait(shared.nempty) == -1) { // wait for at least 1 empty slot
                printf("sem_wait nempty error: %s\n", strerror(errno));
            }
            if (sem_wait(shared.mutex) == -1) { // mutex lock
                printf("sem_wait mutex error: %s\n", strerror(errno));
            }
            shared.buff[i % NBUFF] = i; // store i into circular buffer
            *((int*)arg) += 1;
            if (sem_post(shared.mutex) == -1) { // mutex unlock
                printf("sem_post mutex error: %s\n", strerror(errno));
            }
            if (sem_post(shared.nstored) == -1) { // 1 more stored item
                printf("sem_post nstored error: %s\n", strerror(errno));
            }
        }
        return NULL;
    }
    
    void* consumer(void *arg)
    {
        int i;
        for (i = 0; i < nitems; ++i) {
            if (sem_wait(shared.nstored) == -1) { // wait for at least 1 stored item
                printf("sem_wait nstored error: %s\n", strerror(errno));
            }
            if (sem_wait(shared.mutex) == -1) { // mutex lock
                printf("sem_wait mutex error: %s\n", strerror(errno));
            }
            if (shared.buff[i % NBUFF] != i) { // buffer check
                printf("error: buff[%d] = %d\n", i, shared.buff[i % NBUFF]);
            }
            *((int*)arg) += 1;
            if (sem_post(shared.mutex) == -1) { // mutex unlock
                printf("sem_post mutex error: %s\n", strerror(errno));
            }
            if (sem_post(shared.nempty) == -1) { // 1 more empty slot
                printf("sem_post nempty error: %s\n", strerror(errno));
            }
        }
        return NULL;
    }

例子是一个生产者-消费者问题，共享数据为shared结构，其中buff存储了生产者生产的数据条目，初始值都为0，随后更新为与buff索引相同的值，生产者生产的数据总条目从命令行指定，buff长度和生产者生产的数据总条目可以不等，当buff写满时，返回buff的第一个索引重新填写。为了实现生产者与消费者的共享问题，shared结构定义了三个信号量，mutex是个二值信号量，初始值为1，作用类似于互斥锁，保证共享数据在同一时刻只被同一线程访问，nempty是个计数信号量，初始值为buff长度，buff满时生产者不再生产，nstored也是个计数信号量，初始值为0，buff空时消费者不再消费。

下面是程序运行结果。

$ gcc -o prodcons prodcons.c
    $ ./prodcons 16
    semaphore /mutex created
    semaphore /nempty created
    semaphore /nstored created
    producer [140660388120320] created, count = 0
    consumer [140660379727616] created, count = 0
    producer [140660388120320] done, count = 16
    consumer [140660379727616] done, count = 16
    semaphore /mutex removed
    semaphore /nempty removed
    semaphore /nstored removed