IPC in Unix Using Message Queues

Message Queue
A message queue contains a message type and message data, and is either private or public. If it is private, only the creating process and its child process can access it. If it is public, any process with the key of the queue can access it. Message queues, which is an IPC mechanism used by Unix, is better than pipes. This is because, in pipes, you have to decide the protocol used for communication. A stream is taken from a pipe and you have to parse the bytes and separate them into packets. The data sent through pipes is in FIFO order. This means that you need to use all the bytes sent before requesting for any other stream.

In a message queue, the sending process calls a system routine for placing the message in a queue that can be read by another process. Each process is provided identification or a type, which allows other processes to identify and select it. A common key is shared between processes accessing the queue. In Unix, all the functions related to messages are defined in the system file, <sys/msg.h>.

C Code For Adding Messages to the Message Queue

Code:

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

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

struct mebufmsg
{
    long msgtype;
    char msgtext[200];
};

int main(void)
{
   struct mebufmsg buffr;
   int msgqid;
   key_t kiy;

   if ((kiy = ftok("sam.c", 'B')) == -1)
   {
      perror("Call to ftok() is failed!");
      exit(1);
   }

   if ((msgqid = msgget(kiy, 0644 | IPC_CREAT)) == -1)
   {
      perror("Call to msgget() is failed!");
      exit(1);
   }
   printf("Enter few lines of text. To quit, press ^D: \n");
   buffr.msgtype = 1;
   while(gets(buff.msgtext), !feof(stdin)) 
   {
   if (msgsnd(msgqid, (struct msgbuf *)&buffr, sizeof(buffr), 0) == -1)
        perror("msgsnd");
   }

   if (msgctl(msgqid, IPC_RMID, NULL) == -1)
   {
      perror("msgctl");
      exit(1);
   }
    return 0;
}

In the above code each line entered by the user is bundled into a message and added to the message queue.

C Code For Retrieving Messages from the Message Queue

Code:

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

struct mebufmsg
{
   long msgtype;
   char msgtext[200];
};

int main(void)
{
    struct memsgbuf buffr;
    int msgqid;
    key_t kiy;

    if ((kiy = ftok("sam.c", 'B')) == -1)
    {/* same key as sam.c */
        perror("call to ftok() function is failed");
        exit(1);
    }

    if ((msgqid = msgget(kiy, 0644)) == -1)
    { /* Connecting to the queue */
        perror("msgget() system functionfailed");
        exit(1);
    }
    printf("samuel.c: This program is ready to receive messages.\n");

    for(;;)
    { /* Samuel do not quits ever*/
        if (msgrcv(msgqid, (struct mmebufmsg *)&buffr, sizeof(buffr), 0, 0) == -1)
        {
            perror("msgrcv");
            exit(1);
        }
        printf("samuel: \"%s\"\n", buffr.msgtext);
    }
    return 0;
}

Message Passing Technique

The message queuing mechanism allows processes to send and receive messages, and to queue messages in random order for processing. The IPC message is assigned a type, which helps a server process to direct a message queue between clients.

To send and receive messages, the queue is initialized by using the msgget() function call. The msgsnd() function is used to send messages, and the msgrcv() function is used to receive messages. When a message is sent, the entire message is copied to the message queue.

The functions, msgsnd() and msgrcv(), are used for either blocking or non-blocking operations. In the blocking or synchronous message passing technique, the sending process cannot continue with the next message unless the previous message is transferred or acknowledged by a receiver. The IPC signal and other mechanisms are used for the transfer of messages. Non-blocking operations are used for asynchronous message transfer. This means that the process is not suspended during the sending or receiving of messages. A blocked message operation remains suspended until the call succeeds, the process receives the signal, and the queue is removed.

Initializing and Controlling Messages
A message queue is initialized by using the msgget()system function. The syntax of this function is:

int msgget(key_t key, int msgflg)

The msgget() function returns a message ID of the queue corresponding to the first argument, the key of the function. The second argument, msgflg, passed to the function is an octal integer.

The Use of the msgget() Function

Code:

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

key_t kiy; /*This key is passed to msgget() */ 
int msgflag /*This msgflag, flag is passed to msgget() */ 
int msgqid; /*This stores the return value from msgget() */ 
kiy =    
msgflag =    
if ((msgqid = msgget(kiy, msgflag)) == &ndash;1)
{
    perror("Call to msgget() system function is failed");
    exit(1);
}
else
    (void) fprintf(stderr, &ldquo;msgget succeeded");
/* Code may go here*/

The Unix system has a function call, msgctl(), to control message queues. This function changes the permissions and other features of a message queue. The syntax for this function is:

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

This function requires three arguments. The first argument identifies the existing message queue. The second argument may be IPC_STAT, IPC_SET, or IPC_RMID. IPC_STAT. This argument stores the queue status information in the buf struct variable, which can be set only if the process has read permission. IPC_SET assigns the owner’s user and group ID, permissions, and the size of the message queue. The settings are changed only if the process has the user ID of the owner. IPC_RMID removes the message queue denoted by the first argument of the msgctl() function.

Use of the msgctl() Function

Code:

#include<sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
...
if (msgctl(msgqid, IPC_STAT, &buffr) == -1)
{
   perror("Call to msgctl() system function is failed");
   exit(1);
}
...
if (msgctl(msgqid, IPC_SET, &buffr) == -1)
{
   perror("Call to msgctl() is failed");
   exit(1);
}

Sending and Receiving Messages

The syntax of the msgsnd() and the msgrcv() system functions is:
int msgsnd(int msgqid, const void *msgptr, size_t msgsize, int msgflag); int msgrcv(int msgqid, void *msgptr, size_t msgsize, long msgtype,int msgflag);
The first argument in both the functions is the ID of an existing message queue. The second argument, msgptr, is a pointer to a structure that contains the type of the message and the text. An example of this user-defined buffer is:

Code:

 
struct msgmine
{
   long   msgtype;    /* This stores the message type value */
   char msgtext[MSGSZ];
/*This indicates the message text of length,MSGSZ */
}

The third argument, msgsize, specifies the length of the message in bytes.
The structure member, msgtype, is the type of the received message specified by the sending process. The msgflg argument specifies the the value of the flag, which decides which action will be taken by the msgsnd() and msgrcv() function. The various actions that can take place are:

When msgflag and IPC_NOWAIT have a non-zero value, the message is not sent.
When msgflag and IPC_NOWAIT have a 0 value, the calling process suspends its execution unless:
- The condition responsible for the suspension no longer exists. This allows the message to be sent.
- The message queue identifier, msgqid, is removed from the system. When this happens, errno is reset to EIDRM and -1 is returned.
- The calling process receives a signal. When the process receives this signal, the message is not sent and the calling process resumes its execution.

When execution is successful, the actions based on the data structure associated with msgqid include:

msg_qnum is incremented by 1.
msg_lspid is set as equal to the process ID of the calling process.
msg_stime is set as equal to the current time.

The Use of the msgsnd() and the msgrcv() Functions

Code:

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

int msgflag; /*This flag is used for the operation */
struct msgbuffr *msgptr; /*This is a pointer to the message buffer */
int msgsize; /* This stores the message size */
long msgtype; /* This gets the desired message type */
int msgqid /*This gets the message queue ID.*/

msgptr = (struct mebufmsg
*)malloc((unsigned)(sizeof(struct mebufmsg)
- sizeof msgptr->msgtext + maxmlimit));

if (msgptr == NULL)
{
   if (msgptr == NULL)
   {
      (void) fprintf(stderr, "msgop: %s %d byte messages.\n","Unable to allocate message buffer for", maxmlimit);
      exit(1);
      }
}

msgsize = ...
msgflag = ...

if (msgsnd(msgqid, msgptr, msgsize, msgflag) == -1)
perror("msgop: call to msgsnd failed");
...
msgsize = ...
msgtype = first_on_queue;
msgflag = ...
if (rtrn = msgrcv(msqid, msgp, msgsz, msgtyp, msgflg) == -1)
perror("Call to msgrcv() system function is failed!");
.../* Code may continue here*/