Linux 進程間通信（一）（經典IPC：消息隊列、信號量、共享存儲）

有3種稱作XSI IPC的IPC：消息隊列、信號量、共享存儲。這種類型的IPC有如下共同的特性。

每個內核中的IPC都用一個非負整數標志。標識符是IPC對象的內部名稱，為了使多個合作進程能夠在同一IPC對象上匯聚，需要提供一個外部命名方案。因此，將每個IPC對象都與一個鍵相關聯，將這個鍵（key）作為該對象的外部名。這個鍵的數據類型是key_t，通常在頭文件<sys/types.h>中被定義為長整型，該鍵由內核變換成標識符。

 

有3種方式可以使客戶進程和服務器進程在同一IPC結構上匯聚：

（1）   服務器進程可以指定鍵IPC_PRIVATE創建一個新的IPC結構，將返回的標識符存放在某處（如一個文件）以便客戶進程取用。IPC_PRIVATE鍵也可用於父子進程，父進程指定IPC_PRIVATE創建一個新的IPC結構，所返回的標識符可供fork后的子進程使用。接着，子進程又可將該標識符作為exec函數的一個參數傳給一個新程序。

（2）   可以在一個公共頭文件中定義一個客戶進程和服務器進程都認可的鍵。然后服務器進程指定此鍵創建一個新的IPC結構。這種方法的問題是該鍵可能已於一個IPC結構相結合，在此情況下，get函數（msgget、semget、shmget）出錯返回。服務器進程必須處理這一錯誤，刪除已存在的IPC結構，然后試着再創建它。

（3）   客戶進程和服務器進程使用一個路徑名和一個id，調用ftok函數根據這兩個值生成一個鍵，然后在（2）中使用這個鍵：key_t ftok(const char* path, int id)(<sys/ipc.h>)。（注：path參數必須引用一個現有的文件，當產生鍵時，只使用id參數的低8位）。

 

XSI IPC為每個IPC結構關聯了一個ipc_perm結構，該結構規定了權限和所有者，至少包含如下成員：

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 */

    ...

};

上述結構定義在<sys/ipc.h>中，任何IPC結構都不存在執行權限，下圖顯示了每種IPC的6種權限：

 

在Linux中，可以運行ipcs –l命令來顯示IPC相關的限制：

[root@benxintuzi ipc]# ipcs -l

------ Shared Memory Limits --------

max number of segments = 4096

max seg size (kbytes) = 4194303

max total shared memory (kbytes) = 1073741824

min seg size (bytes) = 1

------ Semaphore Limits --------

max number of arrays = 128

max semaphores per array = 250

max semaphores system wide = 32000

max ops per semop call = 32

semaphore max value = 32767

------ Messages: Limits --------

max queues system wide = 996

max size of message (bytes) = 65536

default max size of queue (bytes) = 65536

 

消息隊列

消息隊列是消息的鏈接表，存儲在內核中，由消息隊列ID來標識。每個隊列都有一個msgid_ds結構與其相關聯：

struct msgid_ds

{

    struct ipc_perm   msg_perm;

    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 */

    ...

};

此結構定義了隊列的當前狀態。msgget用於創建一個新隊列或打開一個現有隊列，msgsnd將消息添加到隊列的尾端（每個消息包括一個長整型類型字段，一個非負的長度，實際的數據長度），msgrcv用於從隊列中取消息（並不一定要以先進先出次序取消息，可以按消息的類型字段取消息）。

#include <sys/msg.h>

int msgget(key_t key, int flag);

返回值：成功，返回消息隊列ID；失敗，返回-1

說明：

在創建新隊列時，要初始化msqid_ds結構的下列成員：

ipc_perm結構中的mode成員按flag中相應權限位設置。

msg_qnum、msg_lspid、msg_lrpid、msg_stime、msg_rtime設為0。

msg_ctime設為當前時間。

msg_qbytes設置為系統限制值。

 

#include <sys/msg.h>

int msgsnd(int msqid, const void* ptr, size_t nbytes, int flag);

返回值：成功，返回0；失敗，返回-1

說明：

ptr是一個指向mymesg結構的指針：

struct mymesg

{

    long   mtype;     /* positive message type */

    char   mtext[512];   /* message data, of length nbytes */

};

flag可以指定為IPC_NOWAIT。類似於文件I/O中的非阻塞標志，若消息隊列已滿，或隊列中的消息總數等於系統限制值，或隊列中的字節總數等於系統限制值，則執行IPC_NOWAIT使得msgsnd立即出錯返回EAGAIN。如果沒有指定IPC_NOWAIT，則進程會一直阻塞到：有空間可以容納要發送的消息；或者從系統中刪除了此隊列；或者捕捉到了一個信號，並從信號處理程序返回。

 

#include <sys/msg.h>

ssize_t msgrcv(int msqid, void* ptr, size_t nbytes, long type, int flag);

返回值：成功，返回消息數據部分的長度；失敗，返回-1

說明：

參數type指定了感興趣消息的類型：

type == 0: 返回隊列中的第一個消息。

type > 0: 返回隊列中消息類型為type的第一個消息。

type < 0: 返回隊列中消息類型值小於等於type絕對值的消息，如果這種消息有若干個，則取類型值最小的消息。

再次解釋一下flag參數：

flag指定為IPC_NOWAIT，如果沒有指定類型的消息可用，則msgrcv返回-1，errno設置為ENOMSG。

flag未指定為IPC_NOWAIT，則進程會一直阻塞到有了指定類型的消息可用，或者從系統中刪除了此隊列（返回-1，errno設置為EIDRM），或者捕捉到一個信號，並從信號處理程序返回（返回-1，errno設置為EINTR）。

 

#include <sys/msg.h>

int msgctl(int msqid, int cmd, struct msqid_ds* buf);

返回值：成功，返回0；失敗，返回-1

說明：

cmd參數指定了對msqid指定的隊列要執行的命令：

IPC_STAT: 取此隊列的msqid_ds結構，並將其存放到buf指向的結構中。

IPC_SET: 將字段msg_perm.uid、msg_perm.gid、msg_perm.mode、msg_qbytes從buf指向的結構復制到與這個隊列相關的msqid_ds結構中。這種命令只能由下列兩種進程執行：一種是其有效用戶ID等於msg_perm.cuid或msg_perm.uid；另一種是具有超級用戶特權的進程。只有超級用戶才能增加msg_qbytes的值。

IPC_RMID: 從系統中刪除該消息隊列以及該隊列中的所有數據。這種刪除立即生效。仍在使用這一消息隊列的進程在它們下一次試圖對此隊列進行操作時，將得到EIDRM錯誤。此命令的執行權限與IPC_SET選項等效。

上述3個命令（IPC_STAT、IPC_SET、IPC_RMID）也可用於信號量和共享存儲。 

注：

消息隊列存在的目的是用於提供高於一般速度的IPC，但是現在與其他形式的IPC相比，並沒有太大的優勢了，但是使用消息隊列還時不時地存在一些問題，因此目前程序設計中不推薦使用消息隊列來提供解決方案。

 

信號量

信號量是一個計數器，用於為多個進程提供對共享對象的訪問。為了正確地實現信號量，信號量的測試及加減1操作應當是原子操作，為此，信號量通常是在內核中實現的。

常用的信號形式是二元信號量（binary semaphore）。它控制單個資源，其初始值為1。但是，一般而言，信號量的初值也可以是任意一個正值，表明有多少個共享單位可供共享。

內核為每個信號量集合維護着一個semid_ds結構：

struct semid_ds

{

    struct ipc_perm sem_perm;

    unsigned short sem_nsems;   /* # of semaphores in set */

    time_t sem_otime;        /* last-semop() time */

    time_t sem_ctime;        /* last-change time */

    ...

};

每個信號量由一個無名結構體表示，至少包含下列成員：

struct

{

    unsigned short semval;      /* semaphore value, always >= 0 */

    pid_t sempid;               /* pid for last operation */

    unsigned short semncnt;     /* # processes awaiting semval > curval */

    unsigned short semzcnt;     /* # processes awaiting semval == 0 */

    ...

};

影響信號量集合的系統限制如下：

 

當我們想使用XSI信號量時，首先需要通過調用函數semget來獲得一個信號量ID：

#include <sys/sem.h>

int semget(key_t key, int nsems, int flag);

返回值：成功，返回信號量ID；失敗，返回-1

說明：

nsems是該集合中的信號量數。如果是創建一個新集合（一般在服務器進程中），則必須指定nsems；如果是引用現有集合（一個客戶進程），則將nsems指定為0。

 

#include <sys/sem.h>

int semctl(int semid, int semnum, int cmd, .../* union semun arg* */);

說明：

semctl函數包含了多種信號量操作。

第4個參數是可選的。如果使用該參數，則其類型是semun聯合：

union semun

{

int val;                    /* for SETVAL */

struct semid_ds* buf;      /* for IPC_STAT and IPC_SET */

unsigned short* array;     /* for GETALL and SETALL */

};

需要留意的是，這個選項是一個聯合，而非指向聯合的指針。

cmd參數指定了下列10種命令中的一種，這些命令是運行在semid指定的信號量集合上，其中有5種命令是針對一個特定的信號量值的。用semnum指定該信號量集合中的一個成員，semnum的值在0~nsems-1之間。

IPC_STAT: 取此隊列的msqid_ds結構，並將其存放到buf指向的結構中。

IPC_SET: 將字段msg_perm.uid、msg_perm.gid、msg_perm.mode、msg_qbytes從buf指向的結構復制到與這個隊列相關的msqid_ds結構中。這種命令只能由下列兩種進程執行：一種是其有效用戶ID等於msg_perm.cuid或msg_perm.uid；另一種是具有超級用戶特權的進程。只有超級用戶才能增加msg_qbytes的值。

IPC_RMID: 從系統中刪除該消息隊列以及該隊列中的所有數據。這種刪除立即生效。仍在使用這一消息隊列的進程在它們下一次試圖對此隊列進行操作時，將得到EIDRM錯誤。此命令的執行權限與IPC_SET選項等效。

GETVAL: 返回成員semnum的semval值。

SETVAL: 設置成員semnum的semval值。

GETPID: 返回成員semnum的sempid值。

GETNCNT: 返回成員semnum的semncnt值。

GETZCNT: 返回成員semnum的semzcnt值。

GETALL: 取該集合中所有的信號量值，這些值存儲在arg.array指向的數組中。

SETALL: 將該集合中所有的信號量值設置成arg.array指向的數組中的值。

 

#include <sys/sem.h>

int semop(int semid, struct sembuf semoparray[], size_t nops);

返回值：成功，返回0；失敗，返回-1

說明：

semoparray是一個指針，指向一個由sembuf結構表示的信號量操作數組：

struct sembuf

{

    unsigned short    sem_num;   /* member # in set(0, 1, ..., nsems - 1) */

    short         sem_op;       /* operation(negative, 0, or pasitive) */

    short         sem_flg;   /* IPC_NOWAIT, SEM_UNDO */

};

其中，nops說明了該數組中操作的數量。

集合中每個成員的操作由相應的sem_op值規定，可以為正值、負值、0：

（1）   sem_op為正，對應進程釋放的資源數，sem_op值會加到對應信號量的值上。如果指定了undo標志，也從該進程的信號量值中減去sem_op。

（2）   sem_op為負，表示阻塞在該信號量上的進程數。如果該信號量的值大於等於sem_op的絕對值（具有所需資源），則從信號量值中減去sem_op的絕對值。這能保證信號量的結果值大於等於0。如果指定了undo標志，則sem_op絕對值也加到該進程的此信號量調整值上；如果信號量值小於sem_op絕對值（資源不能滿足要求），則有如下情況：

1. 若指定了IPC_NOWAIT，則semop出錯返回EAGAIN。
2. 若未指定IPC_NOWAIT，則該信號量的semncnt值加1（因為調用進程將進入睡眠狀態），然后調用進程被掛起，直至下列事件之一發生：
   1. 此信號量值變為大於等於sem_op的絕對值（表示某個進程已釋放了某些資源）。此信號量的semncnt值減1（因為已結束等待），並且從信號量值中減去sem_op的絕對值。如果指定了undo標志，則sem_op的絕對值也加到該進程的此信號量調整值上。
   2. 從系統中刪除了此信號量，在這種情況下，函數出錯返回EIDRM。
   3. 進程捕捉到一個信號，並從信號的處理程序返回，在這種情況下，此信號量的semncnt值減1（因為調用進程不再等待），並且函數出錯返回EINTR。

（3）   若sem_op為0，表示調用進程希望扽帶該信號量的值變為0。

如果信號量值當前為0，函數立即返回。

如果信號量值當前不為0，則有如下情況：

1. 若指定了IPC_NOWAIT，則semop出錯返回EAGAIN。
2. 若未指定IPC_NOWAIT，則該信號量的semzcnt值加1（因為調用進程將進入睡眠狀態），然后調用進程被掛起，直至下列事件之一發生：
   1. 此信號量值變為0。此信號量的semzcnt值減1（因為已結束等待）。
   2. 從系統中刪除了此信號量，在這種情況下，函數出錯返回EIDRM。
   3. 進程捕捉到一個信號，並從信號的處理程序返回，在這種情況下，此信號量的semzcnt值減1（因為調用進程不再等待），並且函數出錯返回EINTR。

注：

semop函數具有原子性，要么執行數組中的所有操作，要么一個也不做。

 

共享存儲

共享存儲允許兩個或多個進程共享一個給定的存儲區。因為數據不需要在客戶進程和服務器進程之間復制，所以這是最快的一種的IPC。在多個進程之間同步訪問一個給定的存儲區時，若服務器進程正在將數據放入共享存儲區，則在它完成操作之前，客戶進程不應當去取這些數據（此時可以通過信號量、記錄鎖或互斥量進行同步）。

XSI共享存儲和內存映射文件之間的區別在於：前者沒有相關文件，XSI共享存儲段是內存的匿名段。

內核為每個共享存儲段維護一個結構，至少包含如下成員：

struct shmid_ds

{

    struct ipc_perm sh_perm;

    size_t        shm_segsz; /* size of segment in bytes */

    pid_t         shm_lpid;  /* pid of last shmop() */

    pid_t         shm_cpid;  /* pid of creator */

    shmatt_t      shm_nattch;   /* number of current attaches */

    time_t        shm_atime; /* last-attach time */

    time_t        shm_dtime; /* last-detach time */

    time_t        shm_ctime; /* last-change time */

    ...

};

影響共享存儲的系統限制：

 

函數shmget用來獲得一個共享存儲標識符：

#include <sys/shm.h>

int shmget(key_t key, size_t size, int flag);

返回值：成功，返回共享存儲ID；失敗，返回-1

說明：

size是該共享存儲段的長度，以字節為單位。一般而言，size長度是頁長的整數倍。如果應用程序指定的size值並非系統頁長的整數倍，那么最后一頁的剩下部分是不可使用的。如果正在創建一個新段（通常是服務器進程），則必須指定其size；如果正在引用一個現存的段（通常是客戶進程），則將size 指定為0。

 

#include <sys/shm.h>

int shmctl(int shmid, int cmd, struct shmid_ds* buf);

返回值：成功，返回0；失敗，返回-1

說明：

cmd指定下列5種命令中的一種：

IPC_STAT: 取此隊列的shmid_ds結構，並將其存放到buf指向的結構中。

IPC_SET: 按buf指向的結構中的值設置與此共享存儲段相關的shmid_ds結構中的下列3個字段：shm_perm.uid、shm_perm.gid、shm_perm.mode。這種命令只能由下列兩種進程執行：一種是其有效用戶ID等於shm_perm.cuid或shm_perm.uid；另一種是具有超級用戶特權的進程。

IPC_RMID: 從系統中刪除該共享存儲段。因為每個共享存儲段維護着一個連接計數（shmid_ds結構中的shm_nattch字段），所以除非使用該段的最后一個進程終止或與該段分離，否則不會實際上刪除該存儲段。不管此段是否仍在使用，該段標識符都會被立即刪除。此命令的執行權限與IPC_SET選項等效。

額外選項：

SHM_LOCK: 在內存中對共享存儲段加鎖，此命令只能由超級用戶執行。

SHM_UNLOCK: 解鎖共享存儲段，此命令只能由超級用戶執行。

 

一旦創建了一個共享存儲段，進程就可以調用shmat將其連接到它的地址空間中：

#include <sys/shm.h>

void* shmat(int shmid, const void* addr, int flag);

返回值：成功，返回指向共享存儲段的指針；失敗，返回-1

說明：

關於addr:

如果addr為0，則此段連接到由內核選擇的第一個可用的地址上（推薦此法）。

如果addr非0，並且沒有指定SHM_RND，則此段連接到addr指定的地址上。

如果addr非0，並且指定了SHM_RND，則此段連接到（addr – (addr mod SHMLBA)）所指定的地址上。

關於flag:

如果在flag中指定了SHM_RDONLY，則以只讀方式連接此段，否則以讀寫方式連接此段。

 

當對該共享存儲段的操作已經結束時，調用shmdt與該段分離。但這並不會從系統中刪除其標識符以及相關的數據結構，該標識符仍然存在，直至某個進程（帶IPC_RMID命令）調用shmctl刪除它們為止。

#include <sys/shm.h>

int shmdt(const void* addr);

返回值：成功，返回0；失敗，返回-1

 

如下程序打印系統存放各種類型數據的位置信息，包括未初始化數據段、棧、堆、共享存儲等：

[root@benxintuzi ipc]# cat printshm.c
#include <sys/shm.h>
#include <stdio.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 %p to %p\n", (void *)&array[0],
          (void *)&array[ARRAY_SIZE]);
        printf("stack around %p\n", (void *)&shmid);

        if ((ptr = malloc(MALLOC_SIZE)) == NULL)
                printf("malloc error\n");
        printf("malloced from %p to %p\n", (void *)ptr,
          (void *)ptr+MALLOC_SIZE);

        if ((shmid = shmget(IPC_PRIVATE, SHM_SIZE, SHM_MODE)) < 0)
                printf("shmget error\n");
        if ((shmptr = shmat(shmid, 0, 0)) == (void *)-1)
                printf("shmat error\n");
        printf("shared memory attached from %p to %p\n", (void *)shmptr,
          (void *)shmptr+SHM_SIZE);

        if (shmctl(shmid, IPC_RMID, 0) < 0)
                printf("shmctl error\n");

        return (0);
}

[root@benxintuzi ipc]# ./printshm
array[] from 0x8049140 to 0x8052d80
stack around 0xbfe0e4d4
malloced from 0x846d008 to 0x84856a8
shared memory attached from 0xb7726000 to 0xb773e6a0

注：
共享存儲段是緊靠在棧之下的。

 

實例：/dev/zero存儲映射

設備/dev/zero可以看成是字節為0的無限資源，其接收寫向它的任何數據並且忽略掉。如果我們將此設備作為IPC，那么當對其進行存儲映射時，具有如下性質：

（1）   創建一個未命名的存儲區，其長度是mmap的第二個參數，將其向上取整為系統的最近頁長。

（2）   存儲區都初始化為0。

（3）   如果多個進程的共同祖先進程對mmap指定了MAP_SHARED，則這些進程可共享此存儲區。

如下程序打開/dev/zero設備，然后指定長整型的長度調用mmap。注意，一旦存儲區映射成功，就關閉此設備。然后，進程創建了一個子進程，由於在調用mmap時指定了MAP_SHARED，所以一個進程寫到存儲映射區的數據可被另一個進程看到。父子進程交替運行，各自對共享存儲映射區中的長整型數加1。存儲映射區由mmap初始化為0。父進程先對其增1，然后子進程再對其增1，...。

#define    NLOOPS        1000
#define    SIZE        sizeof(long)    /* size of shared memory area */

static int
update(long *ptr)
{
    return((*ptr)++);    /* return value before increment */
}

int main(void)
{
    int        fd, i, counter;
    pid_t    pid;
    void    *area;

    if ((fd = open("/dev/zero", O_RDWR)) < 0)
        printf("open error\n");
    if ((area = mmap(0, SIZE, PROT_READ | PROT_WRITE, MAP_SHARED,
      fd, 0)) == MAP_FAILED)
        printf("mmap error\n");
    close(fd);        /* can close /dev/zero now that it's mapped */

    TELL_WAIT();

    if ((pid = fork()) < 0) {
        printf("fork error\n");
    } else if (pid > 0) {            /* parent */
        for (i = 0; i < NLOOPS; i += 2) {
            if ((counter = update((long *)area)) != i)
            {
                printf("parent: expected %d, got %d\n", i, counter);
                return (-1);
            }

            TELL_CHILD(pid);
            WAIT_CHILD();
        }
    } else {                        /* child */
        for (i = 1; i < NLOOPS + 1; i += 2) {
            WAIT_PARENT();

            if ((counter = update((long *)area)) != i)
            {
                printf("child: expected %d, got %d\n", i, counter);
                return (-1);
            }

            TELL_PARENT(getppid());
        }
    }

    return (0);
}

 

使用/dev/zero的優點是：在調用mmap創建映射區之前，無需存在一個實際文件。很多實現提供了一種類似於/dev/zero的設施，稱為匿名存儲映射。為了使用這種功能，在調用mmap時指定了MAP_ANON標志，並將文件描述符指定為-1。結果得到的區域是匿名的，並且創建了一個可與后代進程共享的存儲區。

對上述程序做如下3處修改即可：a.刪除/dev/zero的open語句；b.刪除fd的close語句；c.將mmap調用改為：if ((area = mmap(0, SIZE, PROT_READ | PROT_WRITE, MAP_ANON | MAP_SHARED, -1, 0)) == MAP_FAILED)。

總結：

如果要在兩個無關進程之間使用共享存儲段，有兩種替代方法：一種是應用程序使用XSI共享存儲段；另一種是使用mmap將同一文件映射到它們的地址空間，同時使用MAP_SHARED標志。

 

關於經典IPC，有如下建議：要學會使用管道和FIFO，因為這兩種基本技術可以有效地應用於大量應用程序。在新的程序設計中，盡可能地避免使用消息隊列以及信號量，而應當考慮全雙工管道和記錄鎖，它們使用起來更加簡單。共享存儲段的功能在多數情況下也可以由mmap函數替代。