linux进程间通信快速入门【三】：信号量(XSI、POSIX以及PV原语)

文章目录

XSI
- semget
- semop、semtimedop
- semctl
- 基于共享内存demo修改
- XSI信号量的限制
PV原语
- PV控制并发进程数
POSIX信号量
- 使用posix命名信号量
- 使用posix匿名信号量
参考

在前两篇文章中我们使用的racingdemo都没有对临界区代码进行加锁，这里我们介绍以下信号量的使用。
Linux环境下主要实现的信号量有两种。根据标准的不同，它们跟共享内存类似，一套XSI的信号量，一套POSIX的信号量。下面我们分别使用它们实现一套类似文件锁的方法，来简单看看它们的使用。

XSI

XSI信号量就是内核实现的一个计数器，可以对计数器做甲减操作，并且操作时遵守一些基本操作原则，即：对计数器做加操作立即返回，做减操作要检查计数器当前值是否够减？（减被减数之后是否小于0）如果够，则减操作不会被阻塞；如果不够，则阻塞等待到够减为止。
调用API如下:

semget

#include <sys/sem.h>int semget(key_t key, int nsems, int semflg);

可以使用semget创建或者打开一个已经创建的信号量数组。
key用来标识系统内的信号量。这里除了可以使用ftok产生以外，还可以使用IPC_PRIVATE创建一个没有key的信号量。
如果指定的key已经存在，则意味着打开这个信号量，这时nsems参数指定为0，semflg参数也指定为0。
nsems参数表示在创建信号量数组的时候，这个数组中的信号量个数是几个。
semflg参数用来指定标志位，主要有：IPC_CREAT，IPC_EXCL和权限mode。

semop、semtimedop

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>int semop(int semid, struct sembuf *sops, size_t nsops);int semtimedop(int semid, struct sembuf *sops, size_t nsops, const struct timespec *timeout);

使用semop调用来对信号量数组进行操作。nsops指定对数组中的几个元素进行操作，如数组中只有一个信号量就指定为1。操作的所有参数都定义在一个sembuf结构体里:

unsigned short sem_num;  /* semaphore number */
short          sem_op;   /* semaphore operation */
short          sem_flg;  /* operation flags */

sem_flg可以指定的参数包括IPC_NOWAIT和SEM_UNDO当制定了SEM_UNDO，进程退出的时候会自动UNDO它对信号量的操作。
对信号量的操作会作用在指定的第sem_num个信号量。一个信号量集合中的第1个信号量的编号从0开始。所以，对于只有一个信号量的信号集，这个sem_num应指定为0。
sem_op用来指定对信号量的操作，可以有的操作有三种：

正值操作：对信号量计数器的值（semval）进行加操作。
0值操作：对计数器的值没有影响，而且要求对进程对信号量必须有读权限。实际上这个行为是一个“等待计数器为0”的操作：
- 如果计数器的值为0，则操作可以立即返回。如果不是0并且sem_flg被设置为IPC_NOWAIT的情况下，0值操作也不会阻塞，而是会立即返回，并且errno被设置为EAGAIN。
- 如果不是0，且没设置IPC_NOWAIT时，操作会阻塞，直到计数器值变成0为止，此时相关信号量的semncnt值会加1，这个值用来记录有多少个进程（线程）在此信号量上等待。
除了计数器变为0会导致阻塞停止以外，还有其他情况也会导致停止等待：信号量被删除，semop操作会失败，并且errno被置为EIDRM。进程被信号（signal）打断，errno会被置为EINTR，切semzcnt会被正常做减处理。
负值操作：对计数器做减操作，且进程对信号量必须有写权限。
- 如果当前计数器的值大于或等于指定负值的绝对值，则semop可以立即返回，并且计数器的值会被置为减操作的结果。
- 如果sem_op的绝对值大于计数器的值semval，则说明目前不够减。如果sem_flg设置了IPC_NOWAIT，semop操作依然会立即返回并且errno被置为EAGAIN。如果没设置IPC_NOWAIT，则会阻塞，直到以下几种情况发生为止：
  - semval的值大于或等于sem_op的绝对值，这时表示有足够的值做减法了。
  - 信号量被删除，semop返回EIDRM。
  - 进程（线程）被信号打断，semop返回EINTR。
    semtimedop提供了一个带超时机制的结构，以便实现等待超时。

semctl

观察semop的行为我们会发现，有必要在一个信号量创建之后对其默认的计数器semval进行赋值。所以，我们需要在semop之前，使用semctl进行赋值操作。

int semctl(int semid, int semnum, int cmd, ...);

这个调用是一个可变参实现，具体参数要根据cmd的不同而变化。在一般的使用中，我们主要要学会使用它改变semval的值和查看、修改sem的属性。相关的cmd为：SETVAL、IPC_RMID、IPC_STAT。
修改：

semctl(semid, 0, SETVAL, 1);

这个调用可以将指定的sem的semval值设置为1。更具体的参数解释大家可以参考man 2 semctl。

基于共享内存demo修改

参考：https://blog.csdn.net/qq_42604176/article/details/123449737?spm=1001.2014.3001.5501的XSI示例

#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <fcntl.h>
#include <string.h>
#include <sys/file.h>
#include <wait.h>
#include <sys/mman.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/types.h>
#include <sys/sem.h>#define COUNT 100
#define PATHNAME "/etc/passwd"static int lockid;// 初始化信号量
int mylock_init(void)
{int semid;semid = semget(IPC_PRIVATE, 1, IPC_CREAT|0600);if (semid < 0) {perror("semget()");return -1;}if (semctl(semid, 0, SETVAL, 1) < 0) {perror("semctl()");return -1;}return semid;
}
// 销毁信号量
void mylock_destroy(int lockid)
{semctl(lockid, 0, IPC_RMID);
}// 信号量值--，表示锁住
int mylock(int lockid)
{struct sembuf sbuf;sbuf.sem_num = 0;sbuf.sem_op = -1;sbuf.sem_flg = 0;// 为什么是while循环？// 防止进程（线程）被信号打断，semop返回EINTRwhile (semop(lockid, &sbuf, 1) < 0) {if (errno == EINTR) {continue;}perror("semop()");return -1;}return 0;
}// 信号量++, 表示解锁
int myunlock(int lockid)
{struct sembuf sbuf;sbuf.sem_num = 0;sbuf.sem_op = 1;sbuf.sem_flg = 0;if (semop(lockid, &sbuf, 1) < 0) {perror("semop()");return -1;}return 0;
}
// 参考https://blog.csdn.net/qq_42604176/article/details/123449737?spm=1001.2014.3001.5501 的XSI的do_child 看看两者有何不同呢
int do_child(int proj_id)
{int interval;int *shm_p, shm_id;key_t shm_key;/* 使用ftok产生shmkey */if ((shm_key = ftok(PATHNAME, proj_id)) == -1) {perror("ftok()");exit(1);}/* 在子进程中使用shmget取到已经在父进程中创建好的共享内存id，注意shmget的第三个参数的使用。 */shm_id = shmget(shm_key, sizeof(int), 0);if (shm_id < 0) {perror("shmget()");exit(1);}/* 使用shmat将相关共享内存段映射到本进程的内存地址。 */shm_p = (int *)shmat(shm_id, NULL, 0);if ((void *)shm_p == (void *)-1) {perror("shmat()");exit(1);}/* critical section */// 对于临界区代码进行加锁解锁if (mylock(lockid) == -1) {exit(1);}interval = *shm_p;interval++;usleep(1);*shm_p = interval;if (myunlock(lockid) == -1) {exit(1);}/* critical section *//* 使用shmdt解除本进程内对共享内存的地址映射，本操作不会删除共享内存。 */if (shmdt(shm_p) < 0) {perror("shmdt()");exit(1);}exit(0);
}int main()
{pid_t pid;int count;int *shm_p;int shm_id, proj_id;key_t shm_key;// 初始化信号量lockid = mylock_init();if (lockid == -1) {exit(1);}proj_id = 1234;/* 使用约定好的文件路径和proj_id产生shm_key。 */if ((shm_key = ftok(PATHNAME, proj_id)) == -1) {perror("ftok()");exit(1);}/* 使用shm_key创建一个共享内存，如果系统中已经存在此共享内存则报错退出，创建出来的共享内存权限为0600。 */shm_id = shmget(shm_key, sizeof(int), IPC_CREAT|IPC_EXCL|0600);if (shm_id < 0) {perror("shmget()");exit(1);}/* 将创建好的共享内存映射进父进程的地址以便访问。 */shm_p = (int *)shmat(shm_id, NULL, 0);if ((void *)shm_p == (void *)-1) {perror("shmat()");exit(1);}/* 共享内存赋值为0。 */*shm_p = 0;/*  打开100个子进程并发读写共享内存。 */for (count=0;count<COUNT;count++) {pid = fork();if (pid < 0) {perror("fork()");exit(1);}if (pid == 0) {do_child(proj_id);}}/* 等待所有子进程执行完毕。 */for (count=0;count<COUNT;count++) {wait(NULL);}/* 显示当前共享内存的值。 */printf("shm_p: %d\n", *shm_p);/* 解除共享内存地质映射。 */if (shmdt(shm_p) < 0) {perror("shmdt()");exit(1);}/* 删除共享内存。 */if (shmctl(shm_id, IPC_RMID, NULL) < 0) {perror("shmctl()");exit(1);}// 销毁信号量mylock_destroy(lockid);exit(0);
}

编译运行结果：

[root@VM-90-225-centos /home/hanhan/SocketTest/LocalSocketDemo]# g++ ./racing_posix_shm.cpp -lrt -o racing_posix_shm
[root@VM-90-225-centos /home/hanhan/SocketTest/LocalSocketDemo]# ./racing_posix_shm
shm_p: 100

XSI信号量的限制

系统中对于XSI信号量的限制都放在一个文件中，路径为：/proc/sys/kernel/sem。文件中包涵4个限制值，它们分别的含义是：

[root@VM-90-225-centos /]# cat /proc/sys/kernel/sem
32000   1024000000      500     32000

SEMMSL：一个信号量集（semaphore set）中，最多可以有多少个信号量。这个限制实际上就是semget调用的第二个参数的个数上限。

SEMMNS：系统中在所有信号量集中最多可以有多少个信号量。

SEMOPM：可以使用semop系统调用指定的操作数限制。这个实际上是semop调用中，第二个参数的结构体中的sem_op的数字上限。

SEMMNI：系统中信号量的id标示数限制。就是信号量集的个数上限。

PV原语

PV操作是操作系统原理中的重点内容之一，而根据上述的互斥锁功能的描述来看，实际上我们的互斥锁就是一个典型的PV操作。加锁行为就是P操作，解锁就是V操作。PV操作是计算机操作系统需要提供的基本功能之一。我们都知道现在的计算机基本都是多核甚至多CPU的场景，所以很多计算任务如果可以并发执行，那么无疑可以增加计算能力。假设我们使用多进程的方式进行并发运算，那么并发多少个进程合适呢？虽然说这个问题会根据不同的应用场景发生变化，但是如果假定是一个极度消耗CPU的运算的话，那么无疑有几个CPU就应该并发几个进程。此时并发个数如果过多，则会增加调度开销导致整体吞度量下降，而过少则无法利用多个CPU核心。
下面我们将用PV操作源于控制同时进行运算的进程个数。对于互斥锁来说，计数器的初值为1，而对于这个PV操作，计数器的初值设置为当前计算机的核心个数。应用采用并发的方式找到10010001到10020000数字范围内质数，并控制并发的进程数为计算机核心数。

整个进程组的执行逻辑可以描述为，父进程需要运算判断10010001到10020000数字范围内所有出现的质数，采用每算一个数打开一个子进程的方式。为控制同时进行运算的子进程个数不超过CPU个数，所以申请了一个值为CPU个数的信号量计数器，每创建一个子进程，就对计数器做P操作，子进程运算完推出对计数器做V操作。由于P操作在计数器是0的情况下会阻塞，直到有其他子进程退出时使用V操作使计数器加1，所以整个进程组不会产生大于CPU个数的子进程进行任务的运算。

PV控制并发进程数

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <unistd.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <sys/shm.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <signal.h>#define START 10010001
#define END 10020000
#define NPROC 4static int pv_id;int mysem_init(int n)
{int semid;semid = semget(IPC_PRIVATE, 1, IPC_CREAT|0600);if (semid < 0) {perror("semget()");return -1;}if (semctl(semid, 0, SETVAL, n) < 0) {perror("semctl()");return -1;}return semid;
}void mysem_destroy(int pv_id)
{semctl(pv_id, 0, IPC_RMID);
}int P(int pv_id)
{struct sembuf sbuf;sbuf.sem_num = 0;sbuf.sem_op = -1;sbuf.sem_flg = 0;while (semop(pv_id, &sbuf, 1) < 0) {if (errno == EINTR) {continue;}perror("semop(p)");return -1;}return 0;
}int V(int pv_id)
{struct sembuf sbuf;sbuf.sem_num = 0;sbuf.sem_op = 1;sbuf.sem_flg = 0;if (semop(pv_id, &sbuf, 1) < 0) {perror("semop(v)");return -1;}return 0;
}int prime_proc(int n)
{int i, j, flag;flag = 1;for (i=2;i<n/2;++i) {if (n%i == 0) {flag = 0;break;}}if (flag == 1) {printf("%d is a prime\n", n);}/* 子进程判断完当前数字退出之前进行V操作 */V(pv_id);exit(0);
}void sig_child(int sig_num)
{while (waitpid(-1, NULL, WNOHANG) > 0);
}int main(void)
{pid_t pid;int i;/* 当子进程退出的时候使用信号处理进行回收，以防止产生很多僵尸进程 */if (signal(SIGCHLD, sig_child) == SIG_ERR) {perror("signal()");exit(1);}pv_id = mysem_init(NPROC);/* 每个需要运算的数字都打开一个子进程进行判断 */for (i=START;i<END;i+=2) {/* 创建子进程的时候进行P操作。 */P(pv_id);pid = fork();if (pid < 0) {/* 如果创建失败则应该V操作 */V(pv_id);perror("fork()");exit(1);}if (pid == 0) {/* 创建子进程进行这个数字的判断 */prime_proc(i);}}/* 在此等待所有数都运算完，以防止运算到最后父进程先mysem_destroy，导致最后四个子进程进行V操作时报错 */while (1) {sleep(1);};mysem_destroy(pv_id);exit(0);
}

这段代码使用了信号处理的方式回收子进程，以防产生过多的僵尸进程。使用这个方法引出的问题在于，如果父进程不在退出前等所有子进程回收完毕，那么父进程将在最后几个子进程执行完之前就将信号量删除了，导致最后几个子进程进行V操作的时候会报错。

POSIX信号量

POSIX提供了一套新的信号量原语

#include <fcntl.h>
#include <sys/stat.h>
#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来创建或访问一个已经创建的POSIX信号量。创建时，可以使用value参数对其直接赋值。

int sem_wait(sem_t *sem);
int sem_trywait(sem_t *sem);
int sem_timedwait(sem_t *sem, const struct timespec *abs_timeout);

sem_wait会对指定信号量进行减操作，如果信号量原值大于0，则减操作立即返回。如果当前值为0，则sem_wait会阻塞，直到能减为止。

int sem_post(sem_t *sem);

sem_post用来对信号量做加操作。这会导致某个已经使用sem_wait等在这个信号量上的进程返回。

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

sem_getvalue用来返回当前信号量的值到sval指向的内存地址中。如果当前有进程使用sem_wait等待此信号量，POSIX可以允许有两种返回，一种是返回0，另一种是返回一个负值，这个负值的绝对值就是等待进程的个数。Linux默认的实现是返回0。

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

使用sem_close可以在进程内部关闭一个信号量，sem_unlink可以在系统中删除信号量。
POSIX信号量实现的更清晰简洁，相比之下，XSI信号量更加复杂，但是却更佳灵活，应用场景更加广泛。在XSI信号量中，对计数器的加和减操作都是通过semop方法和一个sembuff的结构体来实现的，但是在POSIX中则给出了更清晰的定义：使用sem_post函数可以增加信号量计数器的值，使用sem_wait可以减少计数器的值。如果计数器的值当前是0，则sem_wait操作会阻塞到值大于0。

POSIX信号量也提供了两种方式的实现，命名信号量和匿名信号量。这有点类似XSI方式使用ftok文件路径创建和IPC_PRIVATE方式创建的区别。但是表现形式不太一样：
命名信号量：
命名信号量实际上就是有一个文件名的信号量。跟POSIX共享内存类似，信号量也会在/dev/shm目录下创建一个文件，如果有这个文件名就是一个命名信号量。其它进程可以通过这个文件名来通过sem_open方法使用这个信号量。除了访问一个命名信号量以外，sem_open方法还可以创建一个信号量。创建之后，就可以使用sem_wait、sem_post等方法进行操作了。这里要注意的是，一个命名信号量在用sem_close关闭之后，还要使用sem_unlink删除其文件名，才算彻底被删除。
匿名信号量：
一个匿名信号量仅仅就是一段内存区，并没有一个文件名与之对应。匿名信号量使用sem_init进行初始化，使用sem_destroy()销毁。操作方法跟命名信号量一样。匿名内存的初始化方法跟sem_open不一样，sem_init要求对一段已有内存进行初始化，而不是在/dev/shm下产生一个文件。这就要求：如果信号量是在一个进程中的多个线程中使用，那么它所在的内存区应该是这些线程应该都能访问到的全局变量或者malloc分配到的内存。如果是在多个进程间共享，那么这段内存应该本身是一段共享内存（使用mmap、shmget或shm_open申请的内存）

使用posix命名信号量

#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <fcntl.h>
#include <string.h>
#include <sys/file.h>
#include <wait.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <semaphore.h>#define COUNT 100
#define SHMPATH "/shm"
#define SEMPATH "/sem"static sem_t *sem;sem_t *mylock_init(void)
{sem_t * ret;ret = sem_open(SEMPATH, O_CREAT|O_EXCL, 0600, 1);if (ret == SEM_FAILED) {perror("sem_open()");return NULL;}return ret;
}void mylock_destroy(sem_t *sem)
{sem_close(sem);sem_unlink(SEMPATH);
}int mylock(sem_t *sem)
{while (sem_wait(sem) < 0) {if (errno == EINTR) {continue;}perror("sem_wait()");return -1;}return 0;
}int myunlock(sem_t *sem)
{if (sem_post(sem) < 0) {perror("semop()");return -1;}
}int do_child(char * shmpath)
{int interval, shmfd, ret;int *shm_p;shmfd = shm_open(shmpath, O_RDWR, 0600);if (shmfd < 0) {perror("shm_open()");exit(1);}shm_p = (int *)mmap(NULL, sizeof(int), PROT_WRITE|PROT_READ, MAP_SHARED, shmfd, 0);if (MAP_FAILED == shm_p) {perror("mmap()");exit(1);}/* critical section */mylock(sem);interval = *shm_p;interval++;usleep(1);*shm_p = interval;myunlock(sem);/* critical section */munmap(shm_p, sizeof(int));close(shmfd);exit(0);
}int main()
{pid_t pid;int count, shmfd, ret;int *shm_p;sem = mylock_init();if (sem == NULL) {fprintf(stderr, "mylock_init(): error!\n");exit(1);}shmfd = shm_open(SHMPATH, O_RDWR|O_CREAT|O_TRUNC, 0600);if (shmfd < 0) {perror("shm_open()");exit(1);}ret = ftruncate(shmfd, sizeof(int));if (ret < 0) {perror("ftruncate()");exit(1);}shm_p = (int *)mmap(NULL, sizeof(int), PROT_WRITE|PROT_READ, MAP_SHARED, shmfd, 0);if (MAP_FAILED == shm_p) {perror("mmap()");exit(1);}*shm_p = 0;for (count=0;count<COUNT;count++) {pid = fork();if (pid < 0) {perror("fork()");exit(1);}if (pid == 0) {do_child(SHMPATH);}}for (count=0;count<COUNT;count++) {wait(NULL);}printf("shm_p: %d\n", *shm_p);munmap(shm_p, sizeof(int));close(shmfd);shm_unlink(SHMPATH);sleep(3000);mylock_destroy(sem);exit(0);
}

编译：

g++ demo.cpp -lrt -lpthread -o demo

使用posix匿名信号量

#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <fcntl.h>
#include <string.h>
#include <sys/file.h>
#include <wait.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <semaphore.h>#define COUNT 100
#define SHMPATH "/shm"static sem_t *sem;void mylock_init(void)
{sem_init(sem, 1, 1);
}void mylock_destroy(sem_t *sem)
{sem_destroy(sem);
}int mylock(sem_t *sem)
{while (sem_wait(sem) < 0) {if (errno == EINTR) {continue;}perror("sem_wait()");return -1;}return 0;
}int myunlock(sem_t *sem)
{if (sem_post(sem) < 0) {perror("semop()");return -1;}
}int do_child(char * shmpath)
{int interval, shmfd, ret;int *shm_p;shmfd = shm_open(shmpath, O_RDWR, 0600);if (shmfd < 0) {perror("shm_open()");exit(1);}shm_p = (int *)mmap(NULL, sizeof(int), PROT_WRITE|PROT_READ, MAP_SHARED, shmfd, 0);if (MAP_FAILED == shm_p) {perror("mmap()");exit(1);}/* critical section */mylock(sem);interval = *shm_p;interval++;usleep(1);*shm_p = interval;myunlock(sem);/* critical section */munmap(shm_p, sizeof(int));close(shmfd);exit(0);
}int main()
{pid_t pid;int count, shmfd, ret;int *shm_p;sem = (sem_t *)mmap(NULL, sizeof(sem_t), PROT_WRITE|PROT_READ, MAP_SHARED|MAP_ANONYMOUS, -1, 0);if ((void *)sem == MAP_FAILED) {perror("mmap()");exit(1);}mylock_init();shmfd = shm_open(SHMPATH, O_RDWR|O_CREAT|O_TRUNC, 0600);if (shmfd < 0) {perror("shm_open()");exit(1);}ret = ftruncate(shmfd, sizeof(int));if (ret < 0) {perror("ftruncate()");exit(1);}shm_p = (int *)mmap(NULL, sizeof(int), PROT_WRITE|PROT_READ, MAP_SHARED, shmfd, 0);if (MAP_FAILED == shm_p) {perror("mmap()");exit(1);}*shm_p = 0;for (count=0;count<COUNT;count++) {pid = fork();if (pid < 0) {perror("fork()");exit(1);}if (pid == 0) {do_child(SHMPATH);}}for (count=0;count<COUNT;count++) {wait(NULL);}printf("shm_p: %d\n", *shm_p);munmap(shm_p, sizeof(int));close(shmfd);shm_unlink(SHMPATH);sleep(3000);mylock_destroy(sem);exit(0);
}

参考

https://zorrozou.github.io/docs/books/linuxde-jin-cheng-jian-tong-4fe1-xin-hao-liang.html