第十五章：進程間通信

 

Unix系統早期一般使用多進程解決問題。有時多個進程之間需要交互數據，而進程和進程之間不能直接交互數據，因此引入了進程間通信（IPC）。

 

IPC主要包括以下方式：

1. 文件I/O

2. 信號

3. 管道

4. 共享內存

5. 消息隊列

6. 信號量集

7. 網絡socket

其中，共享內存、消息隊列和信號量集都遵循XSI IPC，因此三者編程中有很多共性。管道目前很少使用。

 

 

一、管道

管道（FIFO或PIPE）之所以目前很少使用，是由於他有兩點局限性：

1. 一般為半雙工（即同一時刻，數據只能在一個方向流動，也就是不能同時傳輸）。

2. 管道只能在兩個有共同祖先進程的兩個進程之間使用。

 

管道分為有名管道（有文件名）和無名管道（沒有名字）。有名管道就是自己創建管道文件，然后進行交互。無名管道就是系統幫我們創建管道文件，利用系統的管道文件進行交互。

也就是有名管道適用於所有的進程的通信，無名管道只適用於fork()創建的父子進程之間的通信。

 

管道也是一個文件，后綴名為.pipe。我們可以使用mkfifo命令或mkfifio()函數創建一個有名管道文件：

$ mkfifo a.pipe

mkfifo函數定義如下：

#include<sys/types.h>
#include<sys/stat.h>

int mkfifo(const char * pathname,mode_t mode);

/* 示例 */
mkfifo("a.pipe", 0777);    /* 成功返回0，失敗返回-1，錯誤代碼存在errno中 */

 

讀管道示例代碼如下： 

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>

int main()
{
    int fd = open("a.pipe", O_RDONLY);
    if (fd == -1)
        perror("open"), exit(-1);

    int x = 0;
    while (1) {
        int res = read(fd, &x, sizeof(x));
        if (res == -1) {
            perror("read");
            close(fd);
            exit(-1);
        }
        if (!res) break;
        printf("x=%d\n", x);
    }
    close(fd);

    return 0;
}

寫管道示例代碼如下：

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>

int main()
{
//    int fd = open("a.pipe", O_RDWR);    // 讀寫管道
    int fd = open("a.pipe", O_WRONLY);
    if (fd == -1)
        perror("open"), exit(-1);
    
    int x;
    for (x = 0; x < 100; x++) {
        int res = write(fd, &x, 4);
        if (res == -1) {
            perror("write");
            close(fd);
            exit(-1);
        }
    }
    close(fd);

    return 0;
}

 

在創建完成管道文件a.pipe后，先后執行兩代碼，讀管道程序會打印x=1到x=100。

 

 

二、XSI IPC

XSI IPC的固定步驟：

1. 創建或獲取IPC結構之前，必須提供一個外部提供的key

2. 每個IPC結構都有一個唯一的ID與之對應，使用key可以獲取ID

3. 外部key的類型是key_t，獲得key的方式有以下三種：

a. 使用宏IPC_PRIVATE做key。這種方式一般不使用，因為這個key只能創建，不能獲取

b. 使用ftok()函數創建key

c. 在公共的頭文件中定義每個IPC結構使用的key，key本身是一個整數

4. xxxget()函數可以使用key創建或獲得ID，比如：

shmget()/msgget()

5. 使用xxxget()函數新建IPC結構，參數flag一般為IPC_CREAT | 權限

6. 每種IPC結構都提供了一個xxxctl()函數，此函數可以修改、刪除和查詢IPC結構

7. xxxctl()函數中，cmd支持以下宏：

IPC_STAT：用於查詢

IPC_SET：用於修改權限

IPC_RMID：用於刪除

 

由上分析，ftok()是XSI IPC通用代碼，其函數定義如下：

#include <sys/types.h>
#include <sys/ipc.h>

key_t ftok(const char *pathname, int proj_id);

 

---

函數參數以及返回值：

pathname：路徑名，真實存在即可

proj_id：項目ID，就是0到255的任一整數

返回值：成功返回key；出錯返回(key_t)-1。

---

 

 

三、共享內存

共享內存允許兩個或多個進程共享一個給定的存儲區。因為數據不需要在客戶進程和服務器進程之間復制，所以這是最快的一種IPC。

其類函數定義如下：

#include <sys/ipc.h>
#include <sys/types.h>
#include <sys/shm.h>

/* 1. 創建共享內存ID */
int shmget(key_t key, size_t size, int flag);  /* size表示新建的共享內存大小，只獲取共享內存時指定為0 */

/* 2. 映射共享內存 */
void *shmat(int shmid, const void *shmaddr, int shmflg);  
　　　　　　　　　　　　    /* shmaddr，一般設為NULL */
　　　　　　　　　　　　　  /* shmflg定義為SHM_RDONLY為只讀模式，其它為讀寫模式 */

/* 3. 數據交互 */

/* 4. 解除映射 */
int shmdt(const void *shmaddr);

/* 5. 刪除共享內存，buf用於存儲共享內存信息 */
int shmctl(int shmid, int cmd, struct shmid_ds *buf);

 

讀共享內存示例代碼如下： 

#include <stdio.h>
#include <stdlib.h>
#include <sys/ipc.h>
#include <sys/shm.h>

int main()
{
    key_t key = ftok(".", 100);
    if (key == -1)
        perror("ftok"), exit(-1);
    
    /* 獲取共享內存，size和flag指定為0 */
    int shmid = shmget(key, 0, 0);
    if (shmid == -1)
        perror("shmget"), exit(-1);
    
    /* 讀取共享內存數據 */
    int* p = shmat(shmid, 0, 0);
    printf("*p=%d\n", *p);
    
    shmdt(p);

    return 0;
}

寫共享內存示例代碼如下：

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/ipc.h>
#include <sys/shm.h>

int main()
{
    key_t key = ftok(".", 100);
    if (key == -1)
        perror("ftok"), exit(-1);
    
    int shmid = shmget(key, 4, IPC_CREAT | 0666);
    if (shmid == -1)
        perror("shmget"), exit(-1);

    void* p = shmat(shmid, 0, 0);            // 讀寫模式
    int* pi = p;
    *pi = 10;
    shmdt(p);
    
    sleep(10);
    
    struct shmid_ds ds;
    shmctl(shmid, IPC_STAT, &ds);            // 查詢
    printf("shmid=%d\n", shmid);            // id
    printf("size=%d\n", ds.shm_segsz);        // 大小 
    printf("nattch=%d\n", ds.shm_nattch);    // 權限
    printf("mode=%d\n", ds.shm_perm.mode);    // 掛接數

    ds.shm_segsz = 400;                        // 不能改
    ds.shm_perm.mode = 0644;                // 能改
    shmctl(shmid, IPC_SET, &ds);            // 修改，只能修改權限
    shmctl(shmid, IPC_RMID, 0);                // 刪除

    return 0;
}

執行兩示例需要注意的是，寫示例需要先執行。如果讀示例先執行，會導致shmget()函數出錯。

 

共享內存的缺點是若多個進程同時寫，會導致讀出的數據完全混亂。

 

 

四、消息隊列

相比共享內存，消息隊列設計更加的合理。消息隊列也是采用內存做交互媒介，它先把數據傳入消息中，再把消息存入隊列。也就是從隊列中取出或放入數據。

 

為了克服共享內存的缺點，消息隊列引入了無類型消息和有類型消息。有類型消息需要定義為結構體，示例如下：

struct mymsg               // 結構名稱可以自定義
{
    long mtype;             // 消息類型，結構體的第一個成員必須是它
    char mtext[512];    // 數據區，支持任意類型的數據
};

 

消息隊列類函數定義如下： 

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

/* 1. 創建消息隊列ID */
int msgget(key_t key, int flag);

/* 2. 數據交互 */
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);

/* 3. 刪除消息隊列 */
int msgctl(int msqid, int cmd, struct msqid_ds *buf);

 

---

msgsnd()和msgrcv()函數參數以及返回值：

msqid：消息隊列ID

msgp：消息結構體指針

msgsz：消息結構體的數據部分大小

msgtyp：指定消息類型

> 0，接受特定類型的消息

= 0，接受任意類型的消息

< 0，接受類型小於等於msgtyp絕對值的消息，接收順序按消息結構體中mtype的數值從小到大

msgflg：0表示阻塞，IPC_NOWAIT非阻塞

返回值：msgsnd()成功返回0；出錯返回-1。msgrcv()成功返回消息結構體的數據部分大小；出錯返回-1。

---

 

讀消息隊列示例代碼如下：

#include <stdio.h>
#include <stdlib.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <string.h>

struct Msg
{
    long mtype;
    char buf[256];
};

int main()
{
    key_t key = ftok(".", 100);
    
    int msgid = msgget(key, 0);
    if (msgid == -1)
        perror("msgget"), exit(-1);

    struct Msg msg;

    int res = msgrcv(msgid, &msg, sizeof(msg.buf), -2, 0);
    
    if (res == -1)
        perror("msgrcv"), exit(-1);
    
    else
        printf("%s\n", msg.buf);

    return 0;
}

寫消息隊列示例代碼如下：

#include <stdio.h>
#include <stdlib.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <string.h>

struct Msg
{
    long mtype;
    char buf[256];
};

int main()
{
    key_t key = ftok(".", 100);
    
    int msgid = msgget(key, IPC_CREAT | 0666);
    if (msgid == -1)
        perror("msgget"), exit(-1);
    
    struct Msg msg1, msg2;
    msg1.mtype = 1;
    strcpy(msg1.buf, "Hello ");
    
    msg2.mtype = 2;
    strcpy(msg2.buf, "World");

    msgsnd(msgid, &msg1, sizeof(msg1.buf), 0);
    msgsnd(msgid, &msg2, sizeof(msg2.buf), 0);

    return 0;
}

 

 

五、信號量

信號量是一個計數器，負責控制訪問共享資源的最大並行進程總數。

信號量初始時設為最大值，每進入一個進程計數器-1，每離開一個進程計數器+1，當計數器到0時進程阻塞，直到計數器重新大於0則解除阻塞。

 

如果有多個共享資源需要控制最大並行進程數，則需要多個信號量。信號量集用於存儲多個信號量。IPC拿到的是信號量集，而不是單一的信號量。

 

信號量集類函數定義如下：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>

/* 1. 創建信號量集ID */
int semget(key_t key, int nsems, int flag);

/* 2. 為每個信號量設置初始的最大值 */
int semctl(int semid, int semnum, int cmd, ...);

/* 3. 計數 +1 或 -1 */
int semop(int semid, struct sembuf *sops, size_t nsops);

/* 4. 刪除信號量集 */
int semctl(int semid, int semnum, int cmd, ...);

 

semop()函數的sops定義如下：

struct sembuf
{
    unsigned short sem_num;    // 操作信號量的下標
    short sem_op;    // -1代表計數減1，1代表加1
    short sem_flg;    // 0代表阻塞，IPC_NOWAIT非阻塞
};

 

信號量集一般用於控制某類事務的數量。在此以控制子進程並發個數為例。

信號量控制示例代碼如下：

#include <stdio.h>
#include <stdlib.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <signal.h>

int semid;

void sig_exit(int signo)
{
    printf("准備刪除信號量集\n");
    semctl(semid, 0, IPC_RMID, NULL);
    exit(0);
}

int main()
{
    printf("用Ctrl + C退出\n");
    key_t key = ftok(".", 100);
    
    semid = semget(key, 1, IPC_CREAT | 0666);    // 1個元素
    if (semid == -1)
        perror("semget"),exit(-1);

    int res = semctl(semid, 0, SETVAL, 5);        // 最多支持5個並行進程
    if (!res)
        printf("信號量集成功創建\n");

    signal(SIGINT, sig_exit);

    while(1);

    return 0;
}

進程並行示例代碼如下：

#include <stdio.h>
#include <stdlib.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <unistd.h>

int main()
{
    key_t key = ftok(".", 100);
    
    int semid = semget(key, 0, 0);    // 獲取不新建
    if (semid == -1)
        perror("semget"),exit(-1);

    int i;
    for (i = 0; i < 10; i++) {
        pid_t pid = fork();
        if (!pid) {
            printf("申請一個計數\n");
            
            struct sembuf buf;
            buf.sem_num = 0;
            buf.sem_op = -1;        // 計數 -1
            buf.sem_flg = 0;        // 阻塞
            
            semop(semid, &buf, 1);    // 數組就是首元素的地址
            printf("申請成功\n");
            
            sleep(5);
            
            buf.sem_op = 1;            // 計數 +1
            printf("釋放一個計數\n");
            
            semop(semid, &buf, 1);    // 數組就是首元素的地址
            exit(0);
        }
    }
}

 

首先執行信號量控制程序，再執行進程並發程序。可以發現首先會有5個進程執行，另外5個進程等待前5個進程結束后執行。

 

 

下一章  第十六章：網絡IPC 套接字