『 Linux 』简要介绍System V IPC 消息队列 信号量

---

消息队列

System V 消息队列是操作系统所提供的一种进程间通信的方式;

能够使多个进程或线程以队列的方式将数据传入至操作系统内核所提供的一个消息队列中从而进行进程间通信;

消息队列是一种面向信息而不是面向字节流的一种通信方式;

消息队列的主要特性为:

• 异步通信

  发送者将消息发送至消息队列后可以立即返回,而接受者可以在任何时候从队列中读取消息;

  允许发送者和接受者在时间上进行"解耦合"从而提高系统的灵活性和响应性;

• 队列管理

  消息队列在内核中实现,消息按照发送顺序保存在队列中,接收者可以按照顺序读取这些消息队列;

• 消息格式

  消息被阻止具有头部和数据的结构;

  头部一般包含消息的类型长度等信息,而数据部分包含实际的消息内容;

• 访问控制

  System V消息队列具有访问控制机制,可通过权限设置控制可访问队列的进程;

进程间通信的方式必须是让不同的进程能够看到同一份资源;

资源的存在形式一般可以为:

• 文件缓冲区

  类比于管道;

• 内存块

  类比于共享内存;

• 队列

  此处的队列与用户自定义生成的队列不同,该消息队列由操作系统内核进行描述与管理;

消息队列允许不同进程间进行双向通信,即数据流是双向的;

• 允许不同的进程向内核发送带类型的数据块;

  该处的类型用于向不同进程标明哪段消息是由哪个进程发送至哪个进程的;

System V IPC所制定的一些列进程间通信所使用的接口都通过System V标准化定义;

其对应的接口具有相似性,可以举一反三;

在命令行中查看消息队列使用ipcs -q msqid,删除消息队列则使用ipcrm -q msqid;

---

System V 消息队列接口介绍

• 获取/创建 一个消息队列

  获取/创建一个消息队列通常使用msgget()系统调用接口;

  NAMEmsgget - get a System V message queue identifierSYNOPSIS#include <sys/types.h>#include <sys/ipc.h>#include <sys/msg.h>int msgget(key_t key, int msgflg);RETURN VALUEIf successful, the return value will be the message queue identifier (a nonnegative integer), otherwise -1 with errno indicating the error.DESCRIPTIONThe  msgget()  system  call  returns the System V message queue identifier associated with the value of the key argument.  A new message queue is created if key has thevalue IPC_PRIVATE or key isn't IPC_PRIVATE, no message queue with the given key key exists, and IPC_CREAT is specified in msgflg.If msgflg specifies both IPC_CREAT and IPC_EXCL and a message queue already exists for key, then msgget() fails with errno set to EEXIST.  (This  is  analogous  to  theeffect of the combination O_CREAT | O_EXCL for open(2).)

  返回值为int类型,与System V共享内存相同,当该系统调用接口调用成功时将返回一个msqid用于对消息队列的具体操作;

  调用失败时返回-1并设置errno;

  调用参数如下:

  • key_t key

    与共享内存相同,传入一个key值使多个进程可通过该key值看到同一份资源(消息队列);

    该参数可用户自定或是通过调用系统调用接口ftok()进行传入(ftok()系统调用接口不赘述);

  • int msgflg

    该参数用于创建/获取消息队列的具体选项;

    选项分别为O_CREAT,O_EXCL;

    其中单独使用O_CREAT为:创建一个消息队列,若对应key的消息队列存在则返回该消息队列的msqid;

    O_CREAT,O_EXCL同时使用位:创建一个消息队列,若对应key的消息队列存在则返回错误,用来保证所创建的消息队列是全新的;

    O_EXCL不单独使用(无作用);

• 消息队列的释放/操作

  消息队列的释放/操作使用系统调用接口msgctl();

  NAMEmsgctl - System V message control operationsSYNOPSIS#include <sys/types.h>#include <sys/ipc.h>#include <sys/msg.h>int msgctl(int msqid, int cmd, struct msqid_ds *buf);
  RETURN VALUEOn success, IPC_STAT, IPC_SET, and IPC_RMID return 0.  A successful IPC_INFO or MSG_INFO operation returns the index of the highest used entry in the kernel's  internalarray  recording  information  about  all message queues.  (This information can be used with repeated MSG_STAT operations to obtain information about all queues on thesystem.)  A successful MSG_STAT operation returns the identifier of the queue whose index was given in msqid.On error, -1 is returned with errno indicating the error.

  调用参数如下:

  • int msqid

    该参数为需要具体释放(操作)消息队列的msqid;

  • int cmd

    该参数为向该系统调用接口传入一个需要对该消息队列具体操作的选项;

           IPC_STATCopy information from the kernel data structure associated with msqid into the msqid_ds structure pointed to by buf.  The caller must have read permission on themessage queue.IPC_SETWrite the values of some members of the msqid_ds structure pointed to by buf to the kernel data structure associated with this message queue, updating  also  itsmsg_ctime  member.   The  following  members  of  the  structure  are  updated:  msg_qbytes,  msg_perm.uid,  msg_perm.gid,  and (the least significant 9 bits of)msg_perm.mode.  The effective UID of the calling process must match the owner (msg_perm.uid) or creator (msg_perm.cuid) of the message queue, or the caller  mustbe privileged.  Appropriate privilege (Linux: the CAP_IPC_RESOURCE capability) is required to raise the msg_qbytes value beyond the system parameter MSGMNB.IPC_RMIDImmediately  remove the message queue, awakening all waiting reader and writer processes (with an error return and errno set to EIDRM).  The calling process musthave appropriate privileges or its effective user ID must be either that of the creator or owner of the message queue.IPC_INFO (Linux-specific)Returns information about system-wide message queue limits and parameters in the structure pointed to by buf.  This structure is of type msginfo (thus, a cast isrequired), defined in <sys/msg.h> if the _GNU_SOURCE feature test macro is defined:MSG_INFO (Linux-specific)Returns  a  msginfo  structure  containing  the  same  information  as  for IPC_INFO, except that the following fields are returned with information about systemresources consumed by message queues: the msgpool field returns the number of message queues that currently exist on the system; the  msgmap  field  returns  thetotal number of messages in all queues on the system; and the msgtql field returns the total number of bytes in all messages in all queues on the system.MSG_STAT (Linux-specific)Returns  a msqid_ds structure as for IPC_STAT.  However, the msqid argument is not a queue identifier, but instead an index into the kernel's internal array thatmaintains information about all message queues on the system.

    其中释放消息队列时所使用的选项为IPC_RMID;

  • struct msqid_ds *buf

    该参数为向该系统调用接口传入该消息队列的内核数据结构,并配合第二个参数对属性进行处理;

    其中对应的结构属性为:

    msgctl() performs the control operation specified by cmd on the System V message queue with identifier msqid.The msqid_ds data structure is defined in <sys/msg.h> as follows:struct msqid_ds {struct ipc_perm msg_perm;     /* Ownership and permissions */time_t          msg_stime;    /* Time of last msgsnd(2) */time_t          msg_rtime;    /* Time of last msgrcv(2) */time_t          msg_ctime;    /* Time of last change */unsigned long   __msg_cbytes; /* Current number of bytes inqueue (nonstandard) */msgqnum_t       msg_qnum;     /* Current number of messagesin queue */msglen_t        msg_qbytes;   /* Maximum number of bytesallowed in queue */pid_t           msg_lspid;    /* PID of last msgsnd(2) */pid_t           msg_lrpid;    /* PID of last msgrcv(2) */};The ipc_perm structure is defined as follows (the highlighted fields are settable using IPC_SET):struct ipc_perm {key_t          __key;       /* Key supplied to msgget(2) */uid_t          uid;         /* Effective UID of owner */gid_t          gid;         /* Effective GID of owner */uid_t          cuid;        /* Effective UID of creator */gid_t          cgid;        /* Effective GID of creator */unsigned short mode;        /* Permissions */unsigned short __seq;       /* Sequence number */};
    

    释放消息队列所采用的IPC_RMID选项不涉及属性,即释放时传入nullptr即可;

• 向消息队列发送数据

  调用系统调用接口msgsnd()向消息队列发送数据;

  NAMEmsgrcv, msgsnd - System V message queue operationsSYNOPSIS#include <sys/types.h>#include <sys/ipc.h>#include <sys/msg.h>int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);RETURN VALUEOn  failure  both  functions  return  -1 with errno indicating the error, otherwise msgsnd() returns 0 and msgrcv() returns the number of bytes actually copied into themtext array.

  调用成功时返回0,调用失败时返回-1;

  其对应参数为:

  • int msqid

    需要发送的消息队列的msqid;

  • const void *msgp

    所发送数据的数据块位置;

    在使用消息队列发送消息时用户需要自定义一个结构体,其类型为struct msgbuf;

               struct msgbuf {// 数据块的类型 必须存在且不能为0long mtype;       /* message type, must be > 0 */// 数据消息的内容 可以数组的形式传递char mtext[1];    /* message data */};
    

  • size_t msgsz

    所发送数据的数据块大小;

  • int msgflg

    设置发送的方式,一般为阻塞发送与非阻塞发送;

           IPC_NOWAITReturn immediately if no message of the requested type is in the queue.  The system call fails with errno set to ENOMSG.MSG_EXCEPTUsed with msgtyp greater than 0 to read the first message in the queue with message type that differs from msgtyp.MSG_NOERRORTo truncate the message text if longer than msgsz bytes.
    

• 从消息队列中读取数据

  调用系统调用接口msgrcv()从消息队列中读取数据;

  
  NAMEmsgrcv, msgsnd - System V message queue operationsSYNOPSIS#include <sys/types.h>#include <sys/ipc.h>#include <sys/msg.h>ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp,int msgflg);
  RETURN VALUEOn  failure  both  functions  return  -1 with errno indicating the error, otherwise msgsnd() returns 0 and msgrcv() returns the number of bytes actually copied into themtext array.

  调用成功返回0,失败返回-1;

  调用参数:

  • int msqid

    读取的消息队列的msqid;

  • void* msgp

    消息队列的缓冲区(同上以struct msgbuf的形式存在);

  • size_t msgsz

    需要读取的大小;

  • long msgtyp

    所读数据的类型;

  • int msgflg

    所读数据的方式(阻塞,非阻塞等);

---

System V IPC 在内核数据结构中的管理

• 组织方式

  System V IPC进程间通信的所有通信方式都将被操作系统内核管理,以模块的形式组织在一起;

  在这些进程间通信方式中必定存在属于一个struct_XXX_ds的数据结构;

  这些数据结构中的第一个成员普遍为sturct ipc_perm XXX _perm的数据结构;

  操作系统内核将通过管理数组的方式将多个sturct ipc_perm XXX _perm结构体进行管理;

  由于sturct ipc_perm XXX _perm数据结构为struct_XXX_ds数据结构的首位即首地址,可通过强制类型转换的方式将对应的地址转换为struct_XXX_ds类型的地址从而访问到struct_XXX_ds结构中的其他属性;

  当需要访问一个通信方式时将访问该数组的下标;

  而该数组的下标即为xxxid,例如共享内存的shmid,消息队列的msqid等等;

  而在产生XXXid时需要使用key,若key相同时则可能会导致两个不同种类的进程间通信方式对应的下标可能冲突;

  这个数组是一个线性且具有轮转机制的数组;

• 类型的强转

  在进行类型强转时操作系统将率先通过数组下标访问到struct ipc_perm XXX_perm结构体,在该结构体中实际直接或间接存放了该进程通信方式的类型;

  当内核识别到该类型时可通过识别将其强转为对应类型的地址从而对struct_XXX_ds结构体变量进行访问;

• 行为类比

  对于强转并绑定的行为本质上是一种多态的行为;

  而其中对应的struct ipc_perm XXX_perm结构体即可看成基类,struct_XXX_ds看成派生类;

  

---

临界资源

• 数据不一致问题

  当一个共享资源同时被多个执行流共享时,其可能出现两个不同的执行流同时分别向该共享资源中进行读和写;

  当写端未完全将数据写入完成时读端就已经将已经写入的资源进行读取;

  从而导致数据错乱即为数据不一致的问题;

  一般这种情况存在于不存在同步互斥机制的共享资源中,如共享内存;

  在使用命令行运行一些程序并向程序进行打印时可能会发生错乱的问题,而错乱的问题本质上就是数据不一致的问题即多个进程可能同时向命令行显示器打印信息,标明了在某些命令行显示器中其可能是一种不存在同步互斥保护机制的资源,需要用户自行上锁;

  可通过加锁的方式为其提供互斥属性;

• 临界资源

  临界资源指任何时刻只允许一个执行流访问的共享资源,一般该共享资源为用户或操作系统内核所维护的一段内存空间,如管道;

• 临界区

  临界区指用户层当中访问并操作临界资源的代码被称为临界区(部分代码);

  #include <stdio.h>
  #include <stdlib.h>
  #include <unistd.h>
  #include <fcntl.h>
  #include <sys/stat.h>
  #include <sys/wait.h>#define FIFO_NAME "/tmp/my_fifo"// 函数用于读取和更新共享数据
  void increment() {int fifo_fd;for (int i = 0; i < 100000; ++i) {// 1. 进入共享区：打开FIFOfifo_fd = open(FIFO_NAME, O_RDWR);if (fifo_fd < 0) {perror("open");exit(1);}// 2. 进入共享区：读取当前值int value;read(fifo_fd, &value, sizeof(value));// 3. 修改共享数据value++;// 4. 进入共享区：将修改后的值写回FIFOlseek(fifo_fd, 0, SEEK_SET);write(fifo_fd, &value, sizeof(value));// 5. 离开共享区：关闭FIFOclose(fifo_fd);}
  }int main() {// 创建FIFOif (mkfifo(FIFO_NAME, 0666) == -1) {perror("mkfifo");exit(1);}// 初始化共享数据int fifo_fd = open(FIFO_NAME, O_WRONLY);if (fifo_fd < 0) {perror("open");exit(1);}int initial_value = 0;write(fifo_fd, &initial_value, sizeof(initial_value));close(fifo_fd);// 创建子进程pid_t pid = fork();if (pid < 0) {perror("fork");exit(1);}if (pid == 0) {// 子进程increment();exit(0);} else {// 父进程increment();// 等待子进程完成wait(NULL);// 读取最终结果fifo_fd = open(FIFO_NAME, O_RDONLY);int final_value;read(fifo_fd, &final_value, sizeof(final_value));printf("Final value: %d\n", final_value);close(fifo_fd);// 删除FIFOunlink(FIFO_NAME);}return 0;
  }
  

---

System V 信号量

信号量又被称为信号灯,本质上信号量可以类比为一种计数器,用来描述临界资源中的资源数量;

一个资源可能会被分为n份,若是存在n+个执行流对资源进行访问操作时必定存在两个或以上的执行流同时访问一份资源从而可能造成数据不一致的问题;

信号量则可以通过对临界资源的计数从而限制访问资源的执行流;

即引入一个计数器cnt,当计数器cnt--时即表示申请一份资源;

此处申请不代表已经使用而是代表预定,而是代表在申请成功后具有访问该资源的权限,申请计数器资源本质上是对资源的预定机制;

当计数器cnt<=0时表示资源被申请完了,若是还有执行流想申请对应资源将返回错误;

计数器可保证进入共享资源的执行流的数量;

当执行流想要访问对应的资源时需要先申请计数器资源而不是直接访问;

• 二元信号量

  当一个资源不需要被分成n份而是以一个整体进行访问时需要的计数器值为1;

  描述该资源的计数器(信号量)即被称为 二元信号量 ;

  二元信号量 的值只有0与1,当信号量为0时表示不可访问,1则为可以访问;

  本质 二元信号量 就是一个锁;

• 信号量的原子性

  信号量本身是一种共享资源,当多个执行流需要访问同一个共享资源时需要申请信号量计数器,而信号量计数器本身是一种共享资源需要被保护;

  保护信号量本身的是其具有的原子性;

  原子性的特点如下:

  • 不可分割

    原子操作是不可分割的,要么完全执行完毕,要么完全不执行,不存在中间状态(即不存在正在进行);

  • 排他性

    原子操作在执行期间不会被其他操作打断;

    这通常通过硬件指令或同步机制来实现;

  • 一致性

    原子性保证了在操作执行过程中数据的一致性,不会出现部分更新的情况从而避免数据不一致;

  信号量的申请与释放操作分为 P操作 与 V操作:

  • P操作(wait)

    当执行流需要访问资源时执行P操作:

    如果信号量大于0,则-1并继续执行;

    如果信号量值为0则进入等待状态直到信号量>0;

  • V操作(signal)

    当执行流释放资源时执行V操作:

    增加信号量的值;

    如果有其他执行流在等待信号量则唤醒其中一个;

  而信号量的申请与释放操作本身就是原子的,属于两态的,并不会在中途被其他执行流所给打断;

---

System V 信号量的接口介绍

• 创建信号量

  调用系统调用接口semget()来创建/获取信号量;

  NAMEsemget - get a System V semaphore set identifierSYNOPSIS#include <sys/types.h>#include <sys/ipc.h>#include <sys/sem.h>int semget(key_t key, int nsems, int semflg);RETURN VALUEIf successful, the return value will be the semaphore set identifier (a nonnegative integer), otherwise -1 is returned, with errno indicating the error.
  

  参数与返回值:

  • key_t key

    传入一个key值,该key值可以使不同的进程看到同一个信号量;

  • int nsems

    传入所需信号量的数量;

  • int semflg

    表示选项,选项同样是O_CREAT与O_EXCL,此处不赘述;

  • 返回值

    其返回值与共享内存,消息队列的返回值类似;

    调用成功时返回对应key的信号量idsemid,调用失败时返回-1;

• 信号量的删除与属性操作

  调用系统调用接口semctl()对信号量进行操作与删除;

  NAMEsemctl - System V semaphore control operationsSYNOPSIS#include <sys/types.h>#include <sys/ipc.h>#include <sys/sem.h>int semctl(int semid, int semnum, int cmd, ...);DESCRIPTIONsemctl()  performs the control operation specified by cmd on the System V semaphore set identified by semid, or on the semnum-th semaphore of that set.  (The semaphoresin a set are numbered starting at 0.)RETURN VALUEOn failure semctl() returns -1 with errno indicating the error.Otherwise the system call returns a nonnegative value depending on cmd as follows:GETNCNT     the value of semncnt.GETPID      the value of sempid.GETVAL      the value of semval.GETZCNT     the value of semzcnt.IPC_INFO    the index of the highest used entry in the kernel's internal array recording information about all semaphore sets.   (This  information  can  be  used  withrepeated SEM_STAT operations to obtain information about all semaphore sets on the system.)SEM_INFO    As for IPC_INFO.SEM_STAT    the identifier of the semaphore set whose index was given in semid.All other cmd values return 0 on success.
  

  删除信号量与共享内存/消息队列相同传入IPC_RMID即可;

  其他不进行赘述;

• 对信号量进行基本操作

  通常调用系统调用接口semop()对信号量进行基本操作;

  NAMEsemop, semtimedop - System V semaphore operationsSYNOPSIS#include <sys/types.h>#include <sys/ipc.h>#include <sys/sem.h>int semop(int semid, struct sembuf *sops, unsigned nsops);RETURN VALUEIf successful semop() and semtimedop() return 0; otherwise they return -1 with errno indicating the error.
  

  • int semid

    该参数为信号量集的标识符;

  • struct sembuf* sops

    指向slembuf结构体数组的指针,每个sembuf结构体指定一个信号量操作;

    其中sembuf结构体的定义为:

    struct sembuf {unsigned short sem_num;  // 信号量集中的信号量编号short sem_op;            // 要执行的操作short sem_flg;           // 操作标志
    };
    

  • unsigned nsops

    sops数组中的操作数;

信号量不用做数据传输但仍属于通信的一种方式是因为:

通信不仅仅是通信数据,互相协同也是通信方式的一种;

而在进行协同时需要使得不同的执行流看到同一个信号量,信号量本身不是传输数据而是传输一个"信号"使得限制其他执行流;