进程间通信 - 信号量
我想到的第一个问题是，为什么我们需要信号量？一个简单的答案是，保护多个进程之间共享的关键/公共区域。
让我们假设，多个进程正在使用相同的代码区域，如果所有进程都想并行访问，那么结果就会重叠。例如，多个用户仅使用一台打印机（公共/关键部分），假设 3 个用户同时执行 3 个作业，如果所有作业并行启动，则一个用户的输出会与另一个用户的输出重叠。因此，我们需要使用信号量来保护这一点，即在一个进程运行时锁定关键部分，并在进程完成时解锁。这将为每个用户/进程重复，以便一项作业不会与另一项作业重叠。
信号量基本上分为两种类型 -
二进制信号量- 只有两种状态 0 和 1，即锁定/解锁或可用/不可用，互斥实现。
计数信号量- 允许任意资源计数的信号量称为计数信号量。
假设我们有 5 台打印机（假设 1 台打印机只接受 1 个作业），并且我们有 3 个作业要打印。现在将为 3 台打印机提供 3 个作业（每台 1 个）。在此过程中又出现了 4 个工作岗位。现在，在 2 台可用打印机中，已安排了 2 项作业，我们还剩下 2 项作业，这些作业只有在其中一台资源/打印机可用后才能完成。这种根据资源可用性进行的调度可以被视为计数信号量。
要使用信号量执行同步，请执行以下步骤 -
步骤 1 - 创建信号量或连接到已存在的信号量 (semget())
步骤 2 - 对信号量执行操作，即分配、释放或等待资源 (semop())
步骤 3 - 对消息队列执行控制操作 (semctl())
现在，让我们用我们拥有的系统调用来检查这一点。
#include <sys/types.h> #include <sys/ipc.h> #include <sys/sem.h> int semget(key_t key, int nsems, int semflg)
此系统调用创建或分配 System V 信号量集。需要传递以下参数 -
第一个参数 key 识别消息队列。密钥可以是任意值，也可以是从库函数 ftok() 导出的值。
第二个参数 nsems 指定信号量的数量。如果是二进制，则为 1，意味着需要 1 个信号量集，否则根据所需的信号量集数量计数。
第三个参数 semflg 指定所需的信号量标志，例如 IPC_CREAT（如果信号量不存在则创建信号量）或 IPC_EXCL（与 IPC_CREAT 一起使用来创建信号量，如果信号量已存在，则调用失败）。还需要通过权限。
注意- 有关权限的详细信息，请参阅前面的部分。
如果成功，此调用将返回有效的信号量标识符（用于进一步调用信号量），如果失败则返回 -1。要了解失败的原因，请使用 errno 变量或 perror() 函数进行检查。
与此调用相关的各种错误有 EACCESS（权限被拒绝）、EEXIST（队列已存在，无法创建）、ENOENT（队列不存在）、ENOMEM（没有足够的内存来创建队列）、ENOSPC（最大设置限制）超过）等
#include <sys/types.h> #include <sys/ipc.h> #include <sys/sem.h> int semop(int semid, struct sembuf *semops, size_t nsemops)
该系统调用对 System V 信号量集执行操作，即分配资源、等待资源或释放资源。需要传递以下参数 -
第一个参数 semid 表示由 semget() 创建的信号量集标识符。
第二个参数 semops 是指向要对信号量集执行的操作数组的指针。结构如下 -
struct sembuf {
unsigned short sem_num; /* Semaphore set num */
short sem_op; /* Semaphore operation */
short sem_flg; /* Operation flags, IPC_NOWAIT, SEM_UNDO */
};
上述结构中的元素 sem_op 表示需要执行的操作 -
如果sem_op为-ve，则分配或获取资源。阻塞调用进程，直到其他进程释放足够的资源，以便该进程可以分配。
如果 sem_op 为零，则调用进程将等待或Hibernate，直到信号量值达到 0。
如果sem_op为+ve，则释放资源。
例如 -
结构 sembuf sem_lock = { 0, -1, SEM_UNDO };
结构 sembuf sem_unlock = {0, 1, SEM_UNDO };
第三个参数 nsemops 是该数组中的操作数。
#include <sys/types.h> #include <sys/ipc.h> #include <sys/sem.h> int semctl(int semid, int semnum, int cmd, …)
该系统调用执行 System V 信号量的控制操作。需要传递以下参数 -
第一个参数 semid 是信号量的标识符。这个id是信号量标识符，它是semget()系统调用的返回值。
第二个参数 semnum 是信号量的数量。信号量从 0 开始编号。
第三个参数 cmd 是对信号量执行所需控制操作的命令。
第四个参数的类型为 union semun，取决于 cmd。对于少数情况，第四个参数不适用。
让我们检查并集 semun -
union semun {
int val; /* val for SETVAL */
struct semid_ds *buf; /* Buffer for IPC_STAT and IPC_SET */
unsigned short *array; /* Buffer for GETALL and SETALL */
struct seminfo *__buf; /* Buffer for IPC_INFO and SEM_INFO*/
};
sys/sem.h 中定义的 semid_ds 数据结构如下 -
struct semid_ds {
struct ipc_perm sem_perm; /* Permissions */
time_t sem_otime; /* Last semop time */
time_t sem_ctime; /* Last change time */
unsigned long sem_nsems; /* Number of semaphores in the set */
};
注意- 请参阅手册页了解其他数据结构。
联合塞蒙 arg；cmd 的有效值为 -
IPC_STAT - 将 struct semid_ds 每个成员的当前值信息复制到 arg.buf 指向的传递结构。该命令需要信号量的读取权限。
IPC_SET - 设置结构 semid_ds 指向的用户 ID、所有者组 ID、权限等。
IPC_RMID - 删除信号量集。
IPC_INFO - 返回有关 arg.__buf 指向的结构 semid_ds 中的信号量限制和参数的信息。
SEM_INFO - 返回一个 semininfo 结构，其中包含有关信号量消耗的系统资源的信息。
此调用将根据传递的命令返回值（非负值）。成功后，IPC_INFO 和 SEM_INFO 或 SEM_STAT 根据信号量返回最高使用条目的索引或标识符，或者返回 semncnt 的值（对于 GETNCNT）或者 sempid 的值（对于 GETPID）或者 semval 的值（对于成功的其他操作） GETVAL 0 1 发生故障时。要了解失败的原因，请使用 errno 变量或 perror() 函数进行检查。
在查看代码之前，让我们先了解其实现 -
创建两个进程，例如子进程和父进程。
创建共享内存主要需要存储计数器和其他标志，以指示共享内存中读/写过程的结束。
计数器按父进程和子进程的计数递增。该计数要么作为命令行参数传递，要么作为默认值（如果未作为命令行参数传递或值小于 10000）。以一定的睡眠时间调用，以确保父级和子级同时（即并行）访问共享内存。
由于父级和子级都以 1 为步长递增计数器，因此最终值应为计数器的两倍。由于父进程和子进程同时执行操作，因此计数器不会按要求递增。因此，我们需要确保一个进程完成后其他进程的完整性。
以上所有实现均在shm_write_cntr.c文件中执行
检查计数器值是否在文件 shm_read_cntr.c 中实现
为了确保完成，信号量程序在文件 shm_write_cntr_with_sem.c 中实现。整个过程完成后删除信号量（从其他程序读取完成后）
由于我们有单独的文件来读取共享内存中计数器的值，并且写入不会产生任何影响，因此读取程序保持不变（shm_read_cntr.c）
在一个终端执行写入程序并从另一个终端执行读取程序总是更好。由于只有写入和读取过程完成后程序才算完成执行，因此写入程序执行完毕后运行程序就可以了。写入程序将等待读取程序运行，并且只有在完成后才完成。
没有信号量的程序。
/* Filename: shm_write_cntr.c */
#include<stdio.h>
#include<sys/ipc.h>
#include<sys/shm.h>
#include<sys/types.h>
#include<string.h>
#include<errno.h>
#include<stdlib.h>
#include<unistd.h>
#include<string.h>
#define SHM_KEY 0x12345
struct shmseg {
int cntr;
int write_complete;
int read_complete;
};
void shared_memory_cntr_increment(int pid, struct shmseg *shmp, int total_count);
int main(int argc, char *argv[]) {
int shmid;
struct shmseg *shmp;
char *bufptr;
int total_count;
int sleep_time;
pid_t pid;
if (argc != 2)
total_count = 10000;
else {
total_count = atoi(argv[1]);
if (total_count < 10000)
total_count = 10000;
}
printf("Total Count is %d\n", total_count);
shmid = shmget(SHM_KEY, sizeof(struct shmseg), 0644|IPC_CREAT);
if (shmid == -1) {
perror("Shared memory");
return 1;
}
// Attach to the segment to get a pointer to it.
shmp = shmat(shmid, NULL, 0);
if (shmp == (void *) -1) {
perror("Shared memory attach");
return 1;
}
shmp->cntr = 0;
pid = fork();
/* Parent Process - Writing Once */
if (pid > 0) {
shared_memory_cntr_increment(pid, shmp, total_count);
} else if (pid == 0) {
shared_memory_cntr_increment(pid, shmp, total_count);
return 0;
} else {
perror("Fork Failure\n");
return 1;
}
while (shmp->read_complete != 1)
sleep(1);
if (shmdt(shmp) == -1) {
perror("shmdt");
return 1;
}
if (shmctl(shmid, IPC_RMID, 0) == -1) {
perror("shmctl");
return 1;
}
printf("Writing Process: Complete\n");
return 0;
}
/* Increment the counter of shared memory by total_count in steps of 1 */
void shared_memory_cntr_increment(int pid, struct shmseg *shmp, int total_count) {
int cntr;
int numtimes;
int sleep_time;
cntr = shmp->cntr;
shmp->write_complete = 0;
if (pid == 0)
printf("SHM_WRITE: CHILD: Now writing\n");
else if (pid > 0)
printf("SHM_WRITE: PARENT: Now writing\n");
//printf("SHM_CNTR is %d\n", shmp->cntr);
/* Increment the counter in shared memory by total_count in steps of 1 */
for (numtimes = 0; numtimes < total_count; numtimes++) {
cntr += 1;
shmp->cntr = cntr;
/* Sleeping for a second for every thousand */
sleep_time = cntr % 1000;
if (sleep_time == 0)
sleep(1);
}
shmp->write_complete = 1;
if (pid == 0)
printf("SHM_WRITE: CHILD: Writing Done\n");
else if (pid > 0)
printf("SHM_WRITE: PARENT: Writing Done\n");
return;
}
编译和执行步骤
Total Count is 10000 SHM_WRITE: PARENT: Now writing SHM_WRITE: CHILD: Now writing SHM_WRITE: PARENT: Writing Done SHM_WRITE: CHILD: Writing Done Writing Process: Complete
现在，让我们检查共享内存读取程序。
/* Filename: shm_read_cntr.c */
#include<stdio.h>
#include<sys/ipc.h>
#include<sys/shm.h>
#include<sys/types.h>
#include<string.h>
#include<errno.h>
#include<stdlib.h>
#include<unistd.h>
#define SHM_KEY 0x12345
struct shmseg {
int cntr;
int write_complete;
int read_complete;
};
int main(int argc, char *argv[]) {
int shmid, numtimes;
struct shmseg *shmp;
int total_count;
int cntr;
int sleep_time;
if (argc != 2)
total_count = 10000;
else {
total_count = atoi(argv[1]);
if (total_count < 10000)
total_count = 10000;
}
shmid = shmget(SHM_KEY, sizeof(struct shmseg), 0644|IPC_CREAT);
if (shmid == -1) {
perror("Shared memory");
return 1;
}
// Attach to the segment to get a pointer to it.
shmp = shmat(shmid, NULL, 0);
if (shmp == (void *) -1) {
perror("Shared memory attach");
return 1;
}
/* Read the shared memory cntr and print it on standard output */
while (shmp->write_complete != 1) {
if (shmp->cntr == -1) {
perror("read");
return 1;
}
sleep(3);
}
printf("Reading Process: Shared Memory: Counter is %d\n", shmp->cntr);
printf("Reading Process: Reading Done, Detaching Shared Memory\n");
shmp->read_complete = 1;
if (shmdt(shmp) == -1) {
perror("shmdt");
return 1;
}
printf("Reading Process: Complete\n");
return 0;
}
编译和执行步骤
Reading Process: Shared Memory: Counter is 11000 Reading Process: Reading Done, Detaching Shared Memory Reading Process: Complete
如果观察上面的输出，计数器应该为 20000，但是，由于在一个进程任务完成之前，其他进程也在并行处理，因此计数器值与预期不符。输出会因系统而异，并且每次执行也会有所不同。为了保证两个进程在完成一项任务后执行该任务，应该使用同步机制来实现。
现在，让我们使用信号量检查同一应用程序。
注意- 阅读程序保持不变。
/* Filename: shm_write_cntr_with_sem.c */
#include<stdio.h>
#include<sys/types.h>
#include<sys/ipc.h>
#include<sys/shm.h>
#include<sys/sem.h>
#include<string.h>
#include<errno.h>
#include<stdlib.h>
#include<unistd.h>
#include<string.h>
#define SHM_KEY 0x12345
#define SEM_KEY 0x54321
#define MAX_TRIES 20
struct shmseg {
int cntr;
int write_complete;
int read_complete;
};
void shared_memory_cntr_increment(int, struct shmseg*, int);
void remove_semaphore();
int main(int argc, char *argv[]) {
int shmid;
struct shmseg *shmp;
char *bufptr;
int total_count;
int sleep_time;
pid_t pid;
if (argc != 2)
total_count = 10000;
else {
total_count = atoi(argv[1]);
if (total_count < 10000)
total_count = 10000;
}
printf("Total Count is %d\n", total_count);
shmid = shmget(SHM_KEY, sizeof(struct shmseg), 0644|IPC_CREAT);
if (shmid == -1) {
perror("Shared memory");
return 1;
}
// Attach to the segment to get a pointer to it.
shmp = shmat(shmid, NULL, 0);
if (shmp == (void *) -1) {
perror("Shared memory attach: ");
return 1;
}
shmp->cntr = 0;
pid = fork();
/* Parent Process - Writing Once */
if (pid > 0) {
shared_memory_cntr_increment(pid, shmp, total_count);
} else if (pid == 0) {
shared_memory_cntr_increment(pid, shmp, total_count);
return 0;
} else {
perror("Fork Failure\n");
return 1;
}
while (shmp->read_complete != 1)
sleep(1);
if (shmdt(shmp) == -1) {
perror("shmdt");
return 1;
}
if (shmctl(shmid, IPC_RMID, 0) == -1) {
perror("shmctl");
return 1;
}
printf("Writing Process: Complete\n");
remove_semaphore();
return 0;
}
/* Increment the counter of shared memory by total_count in steps of 1 */
void shared_memory_cntr_increment(int pid, struct shmseg *shmp, int total_count) {
int cntr;
int numtimes;
int sleep_time;
int semid;
struct sembuf sem_buf;
struct semid_ds buf;
int tries;
int retval;
semid = semget(SEM_KEY, 1, IPC_CREAT | IPC_EXCL | 0666);
//printf("errno is %d and semid is %d\n", errno, semid);
/* Got the semaphore */
if (semid >= 0) {
printf("First Process\n");
sem_buf.sem_op = 1;
sem_buf.sem_flg = 0;
sem_buf.sem_num = 0;
retval = semop(semid, &sem_buf, 1);
if (retval == -1) {
perror("Semaphore Operation: ");
return;
}
} else if (errno == EEXIST) { // Already other process got it
int ready = 0;
printf("Second Process\n");
semid = semget(SEM_KEY, 1, 0);
if (semid < 0) {
perror("Semaphore GET: ");
return;
}
/* Waiting for the resource */
sem_buf.sem_num = 0;
sem_buf.sem_op = 0;
sem_buf.sem_flg = SEM_UNDO;
retval = semop(semid, &sem_buf, 1);
if (retval == -1) {
perror("Semaphore Locked: ");
return;
}
}
sem_buf.sem_num = 0;
sem_buf.sem_op = -1; /* Allocating the resources */
sem_buf.sem_flg = SEM_UNDO;
retval = semop(semid, &sem_buf, 1);
if (retval == -1) {
perror("Semaphore Locked: ");
return;
}
cntr = shmp->cntr;
shmp->write_complete = 0;
if (pid == 0)
printf("SHM_WRITE: CHILD: Now writing\n");
else if (pid > 0)
printf("SHM_WRITE: PARENT: Now writing\n");
//printf("SHM_CNTR is %d\n", shmp->cntr);
/* Increment the counter in shared memory by total_count in steps of 1 */
for (numtimes = 0; numtimes < total_count; numtimes++) {
cntr += 1;
shmp->cntr = cntr;
/* Sleeping for a second for every thousand */
sleep_time = cntr % 1000;
if (sleep_time == 0)
sleep(1);
}
shmp->write_complete = 1;
sem_buf.sem_op = 1; /* Releasing the resource */
retval = semop(semid, &sem_buf, 1);
if (retval == -1) {
perror("Semaphore Locked\n");
return;
}
if (pid == 0)
printf("SHM_WRITE: CHILD: Writing Done\n");
else if (pid > 0)
printf("SHM_WRITE: PARENT: Writing Done\n");
return;
}
void remove_semaphore() {
int semid;
int retval;
semid = semget(SEM_KEY, 1, 0);
if (semid < 0) {
perror("Remove Semaphore: Semaphore GET: ");
return;
}
retval = semctl(semid, 0, IPC_RMID);
if (retval == -1) {
perror("Remove Semaphore: Semaphore CTL: ");
return;
}
return;
}
编译和执行步骤
Total Count is 10000 First Process SHM_WRITE: PARENT: Now writing Second Process SHM_WRITE: PARENT: Writing Done SHM_WRITE: CHILD: Now writing SHM_WRITE: CHILD: Writing Done Writing Process: Complete
现在，我们将通过读取过程来检查计数器值。
执行步骤
Reading Process: Shared Memory: Counter is 20000 Reading Process: Reading Done, Detaching Shared Memory Reading Process: Complete