进程间通信(InterProcess Communication, IPC)
- 匿名管道
- 有名管道
- 信号
- 消息队列
- 共享内存
- 信号量
- 套接字
管道
匿名管道
- 历史上，管道是半双工的(即数据只能在一个方向上流动)。现在某些系统提供全双工管道，但是为了最佳可移植性，不应预先假定系统支持全双工管道。
- 匿名管道只能在具有公共祖先的两个进程之间使用。通常，一个管道由一个进程创建，在进程调用fork之后，这个管道就能在父进程和子进程之间使用了。
pipe函数创建管道，fd[0]为读而打开，fd[1]为写而打开。
1
2
|
#include<unsitd.h>
int pipe(int fd[2]);
|
通常，进程会先调用pipe，接着调用fork，从而创建从父进程到子进程的IPC通道。
fork之后做什么取决于我们想要的数据流的方向。对于从父进程到子进程的管道，父进程关闭管道的读端(fd[0])，子进程关闭写端(fd[1])。状态结构如下图所示。
- 当读(read)一个写端已被关闭的管道时，在所有数据都被读取后，read返回0，表示文件结束。
- 如果写(write)一个读端已被关闭的管道，则产生信号SIGPIPE。如果忽略该信号或者捕捉该信号并从其处理程序返回，则write返回-1，errno设置为EPIPE。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
#include "apue.h"
int main(void)
{
int n;
int fd[2];
pid_t pid;
char line[MAXLINE];
if (pipe(fd) < 0)
err_sys("pipe error");
if ((pid = fork()) < 0) {
err_sys("fork error");
} else if (pid > 0) { /* parent */
close(fd[0]);
write(fd[1], "hello world\n", 12);
} else { /* child */
close(fd[1]);
n = read(fd[0], line, MAXLINE);
write(STDOUT_FILENO, line, n);
}
exit(0);
}
|
popen函数：创建一个管道，fork一个子进程，关闭未使用的管道端，执行一个shell运行命令，然后等待命令终止。
1
2
3
|
#include<stdio.h>
FILE *popen(const char *cmdstring, const char *type);
int pclose(FILE *fp);
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
#include "apue.h"
#include <sys/wait.h>
#define PAGER "${PAGER:-more}" /* environment variable, or default */
int main(int argc, char *argv[])
{
char line[MAXLINE];
FILE *fpin, *fpout;
if (argc != 2)
err_quit("usage: a.out <pathname>");
if ((fpin = fopen(argv[1], "r")) == NULL)
err_sys("can't open %s", argv[1]);
if ((fpout = popen(PAGER, "w")) == NULL)
err_sys("popen error");
/* copy argv[1] to pager */
while (fgets(line, MAXLINE, fpin) != NULL) {
if (fputs(line, fpout) == EOF)
err_sys("fputs error to pipe");
}
if (ferror(fpin))
err_sys("fgets error");
if (pclose(fpout) == -1)
err_sys("pclose error");
exit(0);
}
|
FIFO(命名管道)
FIFO可以在不相关的进程间交换数据。
1
2
3
|
#include<sys/stat.h>
int mkfifo(const char *path, mode_t mode);
int mkfifoat(int fd, const char *path, mode_t mode);
|
当使用mkfifo创建一个FIFO时，需要用open来打开它。当open一个FIFO时，非阻塞标志(O_NONBLOCK)会产生如下影响：
- 没指定O_NONBLOCK是，只读open要阻塞到某个其它进程为写而打开这个FIFO为止。
- 指定了O_NONBLOCK时，只读open立即返回。如果没有进程为读而打开一个FIFO，那么只写open将返回-1，并将errno设置成ENXIO。
若write一个尚无进程为读而打开的FIFO，则产生信号SIGPIPE。若某个FIFO的最后一个写进程关闭了该FIFO，则将为该FIFO的读进程产生一个文件结束标志。
XSI IPC
每个内核中的IPC结构(消息队列，信号量或共享内存)都用一个非负整数的标识符(identifier)加以引用。
ftok函数通关一个路径名和项目id产生一个键。
1
2
|
#include<sys/ipc.h>
key_t ftok(const char *path, int id);
|
XSI IPC为每一个IPC结构关联了一个ipc_perm结构。该结构规定了权限和所有者，它至少包含下列成员：
1
2
3
4
5
6
7
8
|
struct ipc_perm {
uid_t uid; /* owner's effective user id */
gid_t gid; /* owner's effective group id */
uid_t cuid; /* creator's effective user id */
gid_t cgid; /* creator's effective group id */
mode_t mode; /* access modes */
//...
};
|
| 权限 |
位 |
| 用户读 |
0400 |
| 用户写(更改) |
0200 |
| 组读 |
0040 |
| 组写(更改) |
0020 |
| 其它读 |
0004 |
| 其它写(更改) |
0002 |
在linux中可以运行ipcs -l来显示IPC相关的限制。
消息队列
消息队列是消息的链接表，存储在内核中，由消息队列标识符标识。
smgget用于创建一个新队列或打开一个现有队列。msgsnd将新消息添加到队列尾端。每个消息包含一个正的长整型类型的字段，一个非负的长度以及实际数据字节数，所有这些都在将消息添加到队列时，传递给msgsnd。msgrcv用于从队列中读取消息。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
#include<sys/msg.h>
struct msqid_ds {
struct ipc_perm msg_perm; /* see Section 15.6.2 */
msgqnum_t msg_qnum; /* # of messages on queue */
msglen_t msg_qbytes; /* max # of bytes on queue */
pid_t msg_lspid; /* pid of last msgsnd() */
pid_t msg_lrpid; /* pid of last msgrcv() */
time_t msg_stime; /* last-msgsnd() time */
time_t msg_rtime; /* last-msgrcv() time */
time_t msg_ctime; /* last-change time */
//...
};
int msgget(key_t key, int flag);
int msgctl(int msqid, int cmd, struct msqid_ds *buf);
int msgsnd(int msqid, const void *ptr, size_t nbytes, int flag);
ssize_t msgrcv(int msqid, void *ptr, size_t nbytes, long type, int flag);
|
信号量
信号量是一个计数器，用于为多个进程提供对共享数据对象的访问。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
#include<sys/sem.h>
int semget(key_t key, int nsems, int flag);
int semctl(int semid, int semnum, int cmd, .../* union semun arg */);
union semun{
int val;
struct semid_ds *buf;
unsigned short *array;
};
int semop(int semid, struct sembuf semoparray[], size_t nops);
struct sembuf{
unsigned short sem_num;
short sem_op;
short sem_flg;
};
|
共享内存
1
2
3
4
5
|
#include<sys/shm.h>
int shmget(key_t key, size_t size, int flag);
int shmctl(int shmid, int cmd, struct shmid_ds* buf);
void *shmat(int shmid, const void *addr, int flag);
int shmdt(const void *addr);
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
#include "apue.h"
#include <sys/shm.h>
#define ARRAY_SIZE 40000
#define MALLOC_SIZE 100000
#define SHM_SIZE 100000
#define SHM_MODE 0600 /* user read/write */
char array[ARRAY_SIZE]; /* uninitialized data = bss */
int main(void)
{
int shmid;
char *ptr, *shmptr;
printf("array[] from %lx to %lx\n", (unsigned long)&array[0],
(unsigned long)&array[ARRAY_SIZE]);
printf("stack around %lx\n", (unsigned long)&shmid);
if ((ptr = malloc(MALLOC_SIZE)) == NULL)
err_sys("malloc error");
printf("malloced from %lx to %lx\n", (unsigned long)ptr,
(unsigned long)ptr+MALLOC_SIZE);
if ((shmid = shmget(IPC_PRIVATE, SHM_SIZE, SHM_MODE)) < 0)
err_sys("shmget error");
if ((shmptr = shmat(shmid, 0, 0)) == (void *)-1)
err_sys("shmat error");
printf("shared memory attached from %lx to %lx\n",
(unsigned long)shmptr, (unsigned long)shmptr+SHM_SIZE);
if (shmctl(shmid, IPC_RMID, 0) < 0)
err_sys("shmctl error");
exit(0);
}
|
POSIX信号量
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
#include<semaphore.h>
sem_t *sem_open(const char *name, int oflag, .../* mode_t mode,
unsigned int value */);
int sem_close(sem_t *sem);
int sem_unlink(const char *name);
int sem_trywait(sem_t *sem);
int sem_wait(sem_t *sem);
#include<time.h>
int sem_timedwait(sem_t *restrict sem,
const struct timespec *restrict tsptr);
int sem_post(sem_t *sem);
int sem_init(sem_t *sem, int pshared, unsigned int value);
int sem_destroy(sem_t *sem);
int sem_getvalue(sem_t *restrict sem, int *restrict valp);
|
