Message queues

Message queues are one of the interprocess communication mechanisms available under Linux. Message queues, shared memory and semaphores are normally listed as the three interprocess communication mechanisms under Linux. Semaphores, though, are really for process synchronization. In practice, shared memory, aided by semaphores, makes an interprocess communication mechanism. Message queues is the other interprocess communication mechanism.

There are two varieties of message queues, System V message queues and POSIX message queues. Both provide almost the same functionality but system calls for the two are different. System V message queues have been around for a long time, since the UNIX systems of 1980s and are a mandatory requirement of Unix-certified systems. POSIX message queues (and the complete POSIX IPC calls) were introduced in 1993 and are still an optional requirement of Unix-certified systems. This tutorial is for System V message queues.

There are three system wide limits regarding the message queues. These are, MSGMNI, maximum number of queues in the system, MSGMAX, maximum size of a message in bytes and MSGMNB, which is the maximum size of a message queue. We can see these limits with the ipcs -l command.

$ ipcs -l

------ Messages Limits --------
max queues system wide = 32000
max size of message (bytes) = 8192
default max size of queue (bytes) = 16384
...

You can query and, if required, modify the corresponding kernel parameters, kernel.msgmni, kernel.msgmax and kernel.msgmnb using the sysctl command.

To create a System V message queue, we need a System V IPC key. We can create the key with the ftok function.

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

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

The pathname must be an existing and accessible file. The contents of this file are immaterial. Only the lowest eight bits of proj_id are used, which must not be zero. A System V IPC key of type key_t is returned.

Each instance of a System V IPC mechanism, the message queue, semaphore and shared memory, has an associated system-wide identifier. A process knowing this identifier, and having the relevant permission, can straightaway use that mechanism instance. For example, if a process knows the identifier of a certain message queue, it can send and receive messages from it provided it has the write and read permissions respectively. It does not need to do msgget, an operation analogous to open for files.

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

int msgget (key_t key, int msg_flags);

The msgget system call gets the message queue identifier for the given key. The key is obtained using the ftok function. The second parameter is msg_flags. If the IPC_CREAT flag is set in msg_flags, the queue is created, if it does not already exist. If IPC_EXCL is specified along with IPC_CREAT and the message queue exists, msgget returns -1, with errno set to EEXIST. Queue permissions are expressed as nine bits comprising of read, write and execute for owner, group and others. The execute permissions are not meaningful and are not used. If the queue is created, lower order 9 bits from msg_flags are used for permissions. On success, msgget returns the message queue identifier, which is a nonnegative integer.

There is a special key, IPC_PRIVATE. If the first parameter to msgget is IPC_PRIVATE, a new queue is created. The last nine bits of msg_flags are used for permissions and the other bits are ignored. Thus, it is not necessary to specify IPC_CREAT in msg_flags. This is particularly useful for clients. A client process can create a new queue with the key IPC_PRIVATE and send the queue id in the request message to the server. The server uses that queue id for sending the response message to the client.

struct ipc_perm {
    key_t          __key;       /* Key used for msgget */
    uid_t          uid;         /* User id of owner */
    gid_t          gid;         /* Group id of owner */
    uid_t          cuid;        /* User id of creator */
    gid_t          cgid;        /* Group id of creator */
    unsigned short mode;        /* Permissions */
    unsigned short __seq;       /* Sequence number */
};

struct msqid_ds {
    struct ipc_perm msg_perm;     /* Ownership and permissions */
    time_t          msg_stime;    /* Time of last msgsnd */
    time_t          msg_rtime;    /* Time of last msgrcv */
    time_t          msg_ctime;    /* Time of last change */
    unsigned long   __msg_cbytes; /* Current number of bytes in queue */
    msgqnum_t       msg_qnum;     /* Current number of messages in queue */
    msglen_t        msg_qbytes;   /* Maximum number of bytes allowed in queue */
    pid_t           msg_lspid;    /* PID of last msgsnd */
    pid_t           msg_lrpid;    /* PID of last msgrcv */
};

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

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

With msgctl, we can do control operations on a message queue identified by msqid. The cmd parameter identifies the operation to be done. The value of cmd can be IPC_RMID, IPC_STAT and IPC_SET. When cmd is IPC_RMID, the message queue is removed. The cmd value IPC_STAT transfers the kernel information for the message queue in the msqid_ds structure pointed by buf. Similarly, the cmd value IPC_SET updates the message queue properties in the kernel for the queue from the msqid_ds structure pointed by buf. The queue properties updated are msg_qbytes, msg_perm.uid, msg_perm.gid and the last nine bits of msg_perm.mode. The value of msg_ctime for the queue is also updated.

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

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

The msgsnd system call is used to sending messages to a System V message queue. A process using msgsnd must have the write permission for the message queue. The message queue is identified by the msqid parameter in the msgsnd call. The next parameter points to the message to be sent. The structure of the message is of the form,

struct msgbuf {
    long message_type;   
    char message_text [MSG_TEXT_SIZE];
};

A message comprises of message_type, a long integer greater than zero, followed by message_text. message_text can be an array or any structure. The size of message_text is the msgsz parameter in the msgsnd call. It is OK to have messages with zero length, which means that the message just contains the message type and that there is no message_text.

In most cases, the value of msgflg would be zero. However, the maximum queue size is governed by the msg_qbytes variable in the msqid_ds structure for a queue. When a queue is created, msg_qbytes is initialized to MSGMNB bytes. This can be changed using the msgctl call. If the queue is full and adding another message would cause number of bytes in the queue to exceed msg_qbytes, msgsnd blocks till the time the message being sent can be accommodated in the queue. If you do not want msgsnd to block in such cases, you can specify msgflg as IPC_NOWAIT and msgsnd would return -1 in the case of queue being full with errno set to EAGAIN.

#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);

The msgrcv system call is for receiving messages from a System V message queue identified by msqid. The parameter msgp points to the buffer for incoming message and msgsz specifies the maximum space available for message_text member of the message. On success, msgrcv returns the number of bytes actually copied in the message_text member of the structure pointed by msgp. If the actual message to be delivered is bigger than msgsz and msgflg has MSG_NOERROR specified, the message is truncated and the first msgsz bytes of the message text are returned in mesage_text member. However, if MSG_NOERROR is not specified in msgflg, message is not taken off the queue and msgrcv returns -1 with errno set to E2BIG.

The message selected from the queue depends on the msgtyp parameter. If msgtyp is 0, the first message from the queue is read. This looks like a reasonable behavior and should suffice for most applications.

If msgtyp is greater than 0, and msgflg does not specify MSG_EXCEPT, the first message of that type is returned from the queue. Or, if MSG_EXCEPT is specified in msgflg, the first message not of that type in the queue is returned.

If msgtyp is less than zero, the first message of the lowest type, which is less than or equal to the absolute value of msgtyp, is returned.

If there is no message in the queue, msgrcv blocks till the time a message becomes available. However, if IPC_NOWAIT is specified in the msgflg argument, and there is no message in the queue, msgrcv returns -1 immediately with errno set to ENOMSG.

As an example, we will develop a server and clients which communicate using the System V message queues. The server listens for requests from clients. The clients send a line of text to server. The server counts the number of characters in the received line, appends the count to the line and sends it back to the client. In real life, a server would provide a more useful service. But, our objective is to illustrate the communication between the server and clients using message queues and the server functionality takes a backseat. The software architecture looks like this.

The server has a message queue. Its details, the parameters to the ftok function, are known to clients. The clients use these parameters to get the server queue identifier for communicating with the server. The clients create their queue with the key IPC_PRIVATE and embed their queue identifier in the request message to the server. The server does its processing, takes the queue id from the received message and sends its response message to that queue.

The messages exchanged between the clients and server have the structure,

struct message_text {
    int qid;
    char buf [200];
};

struct message {
    long message_type;
    struct message_text message_text;
};

The message comprises of a long integer message_type as required by System V message queues. We will use the integer value 1 as the message_type. message_type is followed by message_text which contains the sender queue id and a buffer for the message. To keep things simple we use the same structure for messages both ways, from clients to the server and from the server to clients. Both the client and server put their queue id in the member qid before sending a message. The server uses the client’s queue id to send response message. The clients, of course, need to know the server’s queue id beforehand as they initiate communication with the server. Or, more precisely, clients need to know the parameters to the ftok function call from which they can figure out the queue id of the server.

The server program is,

/*
 * server.c: Server program
 *           to demonstrate interprocess commnuication
 *           with System V message queues
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

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

#define SERVER_KEY_PATHNAME "/tmp/mqueue_server_key"
#define PROJECT_ID 'M'
#define QUEUE_PERMISSIONS 0660

struct message_text {
    int qid;
    char buf [200];
};

struct message {
    long message_type;
    struct message_text message_text;
};

int main (int argc, char **argv)
{
    key_t msg_queue_key;
    int qid;
    struct message message;

    if ((msg_queue_key = ftok (SERVER_KEY_PATHNAME, PROJECT_ID)) == -1) {
        perror ("ftok");
        exit (1);
    }

    if ((qid = msgget (msg_queue_key, IPC_CREAT | QUEUE_PERMISSIONS)) == -1) {
        perror ("msgget");
        exit (1);
    }

    printf ("Server: Hello, World!\n");

    while (1) {
        // read an incoming message
        if (msgrcv (qid, &message, sizeof (struct message_text), 0, 0) == -1) {
            perror ("msgrcv");
            exit (1);
        }

        printf ("Server: message received.\n");

        // process message
        int length = strlen (message.message_text.buf);
        char buf [20];
        sprintf (buf, " %d", length);
        strcat (message.message_text.buf, buf);

        int client_qid = message.message_text.qid;
        message.message_text.qid = qid;

        // send reply message to client
        if (msgsnd (client_qid, &message, sizeof (struct message_text), 0) == -1) {  
            perror ("msgget");
            exit (1);
        }

        printf ("Server: response sent to client.\n");
    }
}

The server process works with an infinite loop. It can be terminated with the kill -9 pid command. In real life, you would make it a service with scripts to start and stop it. If you kill the server, the queue is left behind, which can be removed with the ipcrm -q qid command. You can find the qid using the ipcs -q command.

The client program is,

/*
 * client.c: Client program
 *           to demonstrate interprocess commnuication
 *           with System V message queues
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

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

#define SERVER_KEY_PATHNAME "/tmp/mqueue_server_key"
#define PROJECT_ID 'M'

struct message_text {
    int qid;
    char buf [200];
};

struct message {
    long message_type;
    struct message_text message_text;
};

int main (int argc, char **argv)
{
    key_t server_queue_key;
    int server_qid, myqid;
    struct message my_message, return_message;

    // create my client queue for receiving messages from server
    if ((myqid = msgget (IPC_PRIVATE, 0660)) == -1) {
        perror ("msgget: myqid");
        exit (1);
    }

    if ((server_queue_key = ftok (SERVER_KEY_PATHNAME, PROJECT_ID)) == -1) {
        perror ("ftok");
        exit (1);
    }

    if ((server_qid = msgget (server_queue_key, 0)) == -1) {
        perror ("msgget: server_qid");
        exit (1);
    }

    my_message.message_type = 1;
    my_message.message_text.qid = myqid;

    printf ("Please type a message: ");

    while (fgets (my_message.message_text.buf, 198, stdin)) {
        // remove newline from string
        int length = strlen (my_message.message_text.buf);
        if (my_message.message_text.buf [length - 1] == '\n')
           my_message.message_text.buf [length - 1] = '\0';

        // send message to server
        if (msgsnd (server_qid, &my_message, sizeof (struct message_text), 0) == -1) {
            perror ("client: msgsnd");
            exit (1);
        }

        // read response from server
        if (msgrcv (myqid, &return_message, sizeof (struct message_text), 0, 0) == -1) {
            perror ("client: msgrcv");
            exit (1);
        }

        // process return message from server
        printf ("Message received from server: %s\n\n", return_message.message_text.buf);  

        printf ("Please type a message: ");
    }
    // remove message queue
    if (msgctl (myqid, IPC_RMID, NULL) == -1) {
            perror ("client: msgctl");
            exit (1);
    }

    printf ("Client: bye\n");

    exit (0);
}

First, we create the server key file, /tmp/mqueue_server_key. Then, the server is run in background. After that, clients can be run. A sample program execution session looks like this.

$ gcc server.c -o server
$ gcc client.c -o client
$ > /tmp/mqueue_server_key
$ ./server &
[1] 7037
$ Server: Hello, World!
$ ps
  PID TTY          TIME CMD
 3069 pts/2    00:00:00 bash
 7037 pts/2    00:00:00 server
 7038 pts/2    00:00:00 ps
$ ./client
Please type a message: Mary had a little lamb.
Server: message received.
Server: response sent to client.
Message received from server: Mary had a little lamb. 23

Please type a message: All that glitters is not gold.
Server: message received.
Server: response sent to client.
Message received from server: All that glitters is not gold. 30

Please type a message: Client: bye
$

1. interprocess-communication-using-posix-message-queues-in-linux