进程间通信之数据传输--消息队列

Posted 2022-02-03 miaoxiong

Why do we need message queues when we already have the shared memory? It would be for multiple reasons, let us try to break this into multiple points for simplification −

• As understood, once the message is received by a process it would be no longer available for any other process. Whereas in shared memory, the data is available for multiple processes to access.

• If we want to communicate with small message formats.

• Shared memory data need to be protected with synchronization when multiple processes communicating at the same time.

• Frequency of writing and reading using the shared memory is high, then it would be very complex to implement the functionality. Not worth with regard to utilization in this kind of cases.

• What if all the processes do not need to access the shared memory but very few processes only need it, it would be better to implement with message queues.

• If we want to communicate with different data packets, say process A is sending message type 1 to process B, message type 10 to process C, and message type 20 to process D. In this case, it is simplier to implement with message queues. To simplify the given message type as 1, 10, 20, it can be either 0 or +ve or –ve as discussed below.

• Ofcourse, the order of message queue is FIFO (First In First Out). The first message inserted in the queue is the first one to be retrieved.

Using Shared Memory or Message Queues depends on the need of the application and how effectively it can be utilized.

Communication using message queues can happen in the following ways −

• Writing into the shared memory by one process and reading from the shared memory by another process. As we are aware, reading can be done with multiple processes as well.

• Writing into the shared memory by one process with different data packets and reading from it by multiple processes, i.e., as per message type.

Having seen certain information on message queues, now it is time to check for the system call (System V) which supports the message queues.

To perform communication using message queues, following are the steps −

Step 1 − Create a message queue or connect to an already existing message queue (msgget())

Step 2 − Write into message queue (msgsnd())

Step 3 − Read from the message queue (msgrcv())

Step 4 − Perform control operations on the message queue (msgctl())

Now, let us check the syntax and certain information on the above calls.

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

int msgget(key_t key, int msgflg)

This system call creates or allocates a System V message queue. Following arguments need to be passed −

• The first argument, key, recognizes the message queue. The key can be either an arbitrary value or one that can be derived from the library function ftok().

• The second argument, shmflg, specifies the required message queue flag/s such as IPC_CREAT (creating message queue if not exists) or IPC_EXCL (Used with IPC_CREAT to create the message queue and the call fails, if the message queue already exists). Need to pass the permissions as well.

Note − Refer earlier sections for details on permissions.

This call would return a valid message queue identifier (used for further calls of message queue) on success and -1 in case of failure. To know the cause of failure, check with errno variable or perror() function.

Various errors with respect to this call are EACCESS (permission denied), EEXIST (queue already exists can’t create), ENOENT (queue doesn’t exist), ENOMEM (not enough memory to create the queue), etc.

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

int msgsnd(int msgid, const void *msgp, size_t msgsz, int msgflg)

This system call sends/appends a message into the message queue (System V). Following arguments need to be passed −

• The first argument, msgid, recognizes the message queue i.e., message queue identifier. The identifier value is received upon the success of msgget()

• The second argument, msgp, is the pointer to the message, sent to the caller, defined in the structure of the following form −

struct msgbuf 
   long mtype;
   char mtext[1];
;

The variable mtype is used for communicating with different message types, explained in detail in msgrcv() call. The variable mtext is an array or other structure whose size is specified by msgsz (positive value). If the mtext field is not mentioned, then it is considered as zero size message, which is permitted.

• The third argument, msgsz, is the size of message (the message should end with a null character)

• The fourth argument, msgflg, indicates certain flags such as IPC_NOWAIT (returns immediately when no message is found in queue or MSG_NOERROR (truncates message text, if more than msgsz bytes)

This call would return 0 on success and -1 in case of failure. To know the cause of failure, check with errno variable or perror() function.

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

int msgrcv(int msgid, const void *msgp, size_t msgsz, long msgtype, int msgflg)

This system call retrieves the message from the message queue (System V). Following arguments need to be passed −

• The first argument, msgid, recognizes the message queue i.e., the message queue identifier. The identifier value is received upon the success of msgget()

• The second argument, msgp, is the pointer of the message received from the caller. It is defined in the structure of the following form −

struct msgbuf 
   long mtype;
   char mtext[1];
;

The variable mtype is used for communicating with different message types. The variable mtext is an array or other structure whose size is specified by msgsz (positive value). If the mtext field is not mentioned, then it is considered as zero size message, which is permitted.

• The third argument, msgsz, is the size of the message received (message should end with a null character)

• The fouth argument, msgtype, indicates the type of message −

  • If msgtype is 0 − Reads the first received message in the queue

  • If msgtype is +ve − Reads the first message in the queue of type msgtype (if msgtype is 10, then reads only the first message of type 10 even though other types may be in the queue at the beginning)

  • If msgtype is –ve − Reads the first message of lowest type less than or equal to the absolute value of message type (say, if msgtype is -5, then it reads first message of type less than 5 i.e., message type from 1 to 5)

• The fifth argument, msgflg, indicates certain flags such as IPC_NOWAIT (returns immediately when no message is found in the queue or MSG_NOERROR (truncates the message text if more than msgsz bytes)

This call would return the number of bytes actually received in mtext array on success and -1 in case of failure. To know the cause of failure, check with errno variable or perror() function.

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

int msgctl(int msgid, int cmd, struct msqid_ds *buf)

This system call performs control operations of the message queue (System V). Following arguments need to be passed −

• The first argument, msgid, recognizes the message queue i.e., the message queue identifier. The identifier value is received upon the success of msgget()

• The second argument, cmd, is the command to perform the required control operation on the message queue. Valid values for cmd are −

IPC_STAT − Copies information of the current values of each member of struct msqid_ds to the passed structure pointed by buf. This command requires read permission on the message queue.

IPC_SET − Sets the user ID, group ID of the owner, permissions etc pointed to by structure buf.

IPC_RMID − Removes the message queue immediately.

IPC_INFO − Returns information about the message queue limits and parameters in the structure pointed by buf, which is of type struct msginfo

MSG_INFO − Returns an msginfo structure containing information about the consumed system resources by the message queue.

• The third argument, buf, is a pointer to the message queue structure named struct msqid_ds. The values of this structure would be used for either set or get as per cmd.

This call would return the value depending on the passed command. Success of IPC_INFO and MSG_INFO or MSG_STAT returns the index or identifier of the message queue or 0 for other operations and -1 in case of failure. To know the cause of failure, check with errno variable or perror() function.

Having seen the basic information and system calls with regard to message queues, now it is time to check with a program.

Let us see the description before looking at the program −

Step 1 − Create two processes, one is for sending into message queue (msgq_send.c) and another is for retrieving from the message queue (msgq_recv.c)

Step 2 − Creating the key, using ftok() function. For this, initially file msgq.txt is created to get a unique key.

Step 3 − The sending process performs the following.

• Reads the string input from the user

• Removes the new line, if it exists

• Sends into message queue

• Repeats the process until the end of input (CTRL + D)

• Once the end of input is received, sends the message “end” to signify the end of the process

Step 4 − In the receiving process, performs the following.

• Reads the message from the queue
• Displays the output
• If the received message is “end”, finishes the process and exits

To simplify, we are not using the message type for this sample. Also, one process is writing into the queue and another process is reading from the queue. This can be extended as needed i.e., ideally one process would write into the queue and multiple processes read from the queue.

Now, let us check the process (message sending into queue) – File: msgq_send.c

/* Filename: msgq_send.c */
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define PERMS 0644
struct my_msgbuf 
   long mtype;
   char mtext[200];
;

int main(void) 
   struct my_msgbuf buf;
   int msqid;
   int len;
   key_t key;
   system("touch msgq.txt");
   
   if ((key = ftok("msgq.txt", ‘B‘)) == -1) 
      perror("ftok");
      exit(1);
   
   
   if ((msqid = msgget(key, PERMS | IPC_CREAT)) == -1) 
      perror("msgget");
      exit(1);
   
   printf("message queue: ready to send messages.\n");
   printf("Enter lines of text, ^D to quit:\n");
   buf.mtype = 1; /* we don‘t really care in this case */
   
   while(fgets(buf.mtext, sizeof buf.mtext, stdin) != NULL) 
      len = strlen(buf.mtext);
      /* remove newline at end, if it exists */
      if (buf.mtext[len-1] == ‘\n‘) buf.mtext[len-1] = ‘\0‘;
      if (msgsnd(msqid, &buf, len+1, 0) == -1) /* +1 for ‘\0‘ */
      perror("msgsnd");
   
   strcpy(buf.mtext, "end");
   len = strlen(buf.mtext);
   if (msgsnd(msqid, &buf, len+1, 0) == -1) /* +1 for ‘\0‘ */
   perror("msgsnd");
   
   if (msgctl(msqid, IPC_RMID, NULL) == -1) 
      perror("msgctl");
      exit(1);
   
   printf("message queue: done sending messages.\n");
   return 0;

Compilation and Execution Steps

message queue: ready to send messages.
Enter lines of text, ^D to quit:
this is line 1
this is line 2
message queue: done sending messages.

Following is the code from message receiving process (retrieving message from queue) – File: msgq_recv.c

/* Filename: msgq_recv.c */
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define PERMS 0644
struct my_msgbuf 
   long mtype;
   char mtext[200];
;

int main(void) 
   struct my_msgbuf buf;
   int msqid;
   int toend;
   key_t key;
   
   if ((key = ftok("msgq.txt", ‘B‘)) == -1) 
      perror("ftok");
      exit(1);
   
   
   if ((msqid = msgget(key, PERMS)) == -1)  /* connect to the queue */
      perror("msgget");
      exit(1);
   
   printf("message queue: ready to receive messages.\n");
   
   for(;;)  /* normally receiving never ends but just to make conclusion 
             /* this program ends wuth string of end */
      if (msgrcv(msqid, &buf, sizeof(buf.mtext), 0, 0) == -1) 
         perror("msgrcv");
         exit(1);
      
      printf("recvd: \"%s\"\n", buf.mtext);
      toend = strcmp(buf.mtext,"end");
      if (toend == 0)
      break;
   
   printf("message queue: done receiving messages.\n");
   system("rm msgq.txt");
   return 0;

Compilation and Execution Steps

message queue: ready to receive messages.
recvd: "this is line 1"
recvd: "this is line 2"
recvd: "end"
message queue: done receiving messages.

摘抄自：https://www.geeksforgeeks.org/ipc-using-message-queues/

也可参考：https://www.tutorialspoint.com/inter_process_communication/inter_process_communication_message_queues.htm