Inter Process Communication-Message Queue

Interprocess communication (IPC) is a set of programming interfaces that allow a programmer to coordinate activities among different program processes that can run concurrently in an operating system. This allows a program to handle many user requests at the same time. Since even a single user request may result in multiple processes running in the operating system on the user's behalf, the processes need to communicate with each other. The IPC interfaces make this possible. Each IPC method has its own advantages and limitations so it is not unusual for a single program to use all of the IPC methods.

Message Based Communication

Messages are a very general form of communication. Messages can be used to send and receive formatted data streams between arbitrary processes. Messages may have types. This helps in message interpretation. The type may specify appropriate permissions for processes. Usually at the receiver end, messages are put in a queue. Messages may also be formatted in their structure. This again is determined by the application process. Messages are also the choice for many parallel computers such as Intel's hyper-cube. The following four system calls achieve message transfers amongst processes.

 msgget() returns (and possibly creates) message descriptor(s) to designate a message queue for use in other systems calls.

msgctl() has options to set and return parameters associated with a message descriptor. It also has an option to remove descriptors.

msgsnd() sends a message using a message queue.

msgrcv() receives a message using a message queue.

Let us now study some details of these system calls. msgget() system call : The syntax of this call is as follows:

int msgget(key_t key, int flag);

The msgget() system call has one primary argument, the key, a second argument which is a flag. It returns an integer called a qid which is the id of a queue. The returned qid is an index to the kernel's message queue data-structure table. The call returns -1 if there is an error. This call gets the resource, a message queue. The first argument key_t, is defined in sys/types.h file as being a long. The second argument uses the following flags: ™

MSG_R : The process has read permission ™

 MSG_W : The process has write permission ™

 MSG_RWAIT : A reader is waiting to read a message from message queue ™

MSG_WWAIT : A writer is waiting to write a message to message queue ™

MSD_LOCKED : The msg queue is locked ™

MSG_LOCKWAIT : The msg queue is waiting for a lock ™

 IPC_NOWAIT : Described earlier ™

 IPC_EXCL : ....

In most cases these options can be used in bit-ored manner. It is important to have the readers and writers of a message identify the relevant queue for message exchange. This is done by associating and using the correct qid or key. The key can be kept relatively private between processes by using a makekey() function (also used for data encryption). For simple programs it is probably sufficient to use the process id of the creator process (assuming that other processes wishing to access the queue know it). Usually, kernel uses some algorithm to translate the key into qid. The access permissions for the IPC methods are stored in IPC permissions structure which is a simple table. Entries in kernel's message queue data structures are C structures. These resemble tables and have several fields to describe permissions, size of queue, and other information. The message queue data structure is as follows.

struct meqid_ds {

struct ipc_perm meg_perm; /* permission structure */

struct msg *msg_first; /* pointer to first message */

struct msg *msg_last; /* ........... last ..........*/

ushort msg_cbytes; /* no. of bytes in queue */

ushort msg_qnum; /* no. of messages on queue */

ushort msg_qbytes; /* Max. no. of bytes on queue */

ushort msg_lspid; /* pid of last msgsnd */

ushort msg_lrpid; /* pid of the last msgrcv */

time_t msg_stime; /* last msgsnd time */

time_t msg_rtime; /* .....msgrcv................*/

time_t msg_ctime; /* last change time */

 }

There is one message structure for each message that may be in the system.

struct msg

{

struct msg *msg_next; /* pointer to next message */

long msg_type; /* message type */

ushort msg_ts; /* message text size */

ushort msg_spot; /* internal address */

}

Note that several processes may send messages to the same message queue. The “type" of message is used to determine which process amongst the processes is the originator of the message received by some other process. This can be done by hard coding a particular number for type or using process-id of the sender as the msg_type.

The msgctl() function call: This system call enables three basic actions. The most obvious one is to remove message queue data structure from the kernel. The second action allows a user to examine the contents of a message queue data structure by copying them into a buffer in user's data area. The third action allows a user to set the contents of a message queue data structure in the kernel by copying them from a buffer in the user's data area. The system call has the following syntax.

int msgctl(int qid, int command, struct msqid_ds *ptr);

This system call is used to control the resource (a message queue). The first argument is the qid which is assumed to exist before call to msgctl(). Otherwise the system is in error state. Note that if msgget() and msgctl() are called by two different processes then there is a potential for a “race" condition to occur. The second argument command is an integer which must be one of the following constants (defined in the header file sys/msg.h).

 IPC STAT: Places the contents of the kernel structure indexed by the first argument, qid, into a data structure pointed to by the third argument, ptr. This enables the user to examine and change the contents of a copy of the kernel's data structure, as this is in user space.

 IPC SET: Places the contents of the data structures in user space pointed to by the third argument, ptr, into the kernel's data structure indexed by first argument qid, thus enabling a user to change the contents of the kernel's data structure. The only fields that a user can change are msg_perm.uid, msg_perm.gid, msg_perm.mode, and msg_qbytes.

IPC RMID : Removes the kernel data structure entry indexed by qid

The msgsnd() and msgrcv() system calls have the following syntax.

 int msgsnd(int qid, struct msgbuf *msg_ptr, int message_size, int flag );

int msgrcv(int qid, struct msgbuf *msg_ptr, int message_size, int msgtype, int flag );

Both of these calls operate on a message queue by sending and receiving messages respectively. The first three arguments are the same for both of these functions. The syntax of the buffer structure is as follows.

 struct msgbuf{ long mtype; char mtext[1]; }.

 This captures the message type and text. The flags specify the actions to be taken if the queue is full, or if the total number of messages on all the message queues exceeds a prescribed limit. With the flags the following actions take place.

If IPC_NOWAIT is set, no message is sent and the calling process returns without any error action. If IPC_NOWAIT is set to 0, then the calling process suspends until any of the following two events occur.

1. A message is removed from this or from other queue.

2. The queue is removed by another process. If the message data structure indexed by qid is removed when the flag argument is 0, an error occurs (msgsnd() returns -1).

The fourth arg to msgrcv() is a message type. It is a long integer. The type argument is used as follows.

·         If the value is 0, the first message on the queue is received.

·         If the value is positive, the queue is scanned till the first message of this type is received. The pointer is then set to the first message of the queue.

·         If the value is -ve, the message queue is scanned to find the first message with a type whose value is less than, or equal to, this argument.

The flags in the msgrcv() are treated the same way as for msgsnd(). A successful execution of either msgsnd(), or msgrcv() always updates the appropriate entries in msgid_ds data structure. With the above explanation, let us examine the message passing program which follows

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/msg.h>

main(argc, argv) int argc; char *argv[];

{ int status, pid, pid1;

if (( pid=fork())==0) execlp("./messender", "messender", argv[1], argv[2], 0);

 if (( pid1=fork())==0) execlp("./mesrec", "mesrec", argv[1], 0);

wait(&status);  /* wait for some child to terminate */

 wait(&status); /* wait for some child to terminate */

}

message sender program.

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/msg.h>

main(argc, argv)

 int argc; char *argv[];

/* This is the sender. It sends messages using IPC system V messages queues.*/ 

/* It takes two arguments : */

/* No. of messages and no. of bytes */

/* key_t MSGKEY = 100; */

/* struct msgformat {long mtype; int mpid; char mtext[256]} msg; */

{ key_t MSGKEY = 100;

 struct msgformat { long mtype; int mpid; char mtext[256]; } msg;

 int i ;

 int msgid;

int loop, bytes;

extern cleanup();

loop = atoi(argv[1]);

bytes = atoi(argv[2]);

printf("In the sender child \n");

for ( i = 0; i < bytes; i++ ) msg.mtext[i] = 'm';

printf("the number of 'm' s is : %6d \n", i);

msgid = msgget(MSGKEY, 0660 | IPC_CREAT);

 msg.mtype = 1;

msg.mpid = getpid(); /* Send number of messages specified by user argument */

for (i=0; i<32; i++) signal(i, cleanup);

 } /* end of main */

cleanup()

 { int msgid;

msgctl(msgid, IPC_RMID, 0);

exit(0);

}

 receiver program listing.

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/msg.h>

main(argc, argv) int argc; char *argv[];

/* The receiver of the two processes communicating message using */

/* IPC system V messages queues. */

/* It takes two arguments: No. of messages and no. of bytes */

/* key_t MSGKEY = 100; */

/* struct msgformat {long mtype; int mpid; char mtext[256]} msg; */

{ key_t MSGKEY = 100;

struct msgformat { long mtype; int mpid; char mtext[256]; } msg;

int i, pid, *pint;

 int msgid;

 int loop, bytes;

 msgid = msgget(MSGKEY, 0777);

loop = atoi(argv[1]);

bytes = atoi(argv[2]);

for ( i = 0; i <= bytes; i++ ) { printf("receiving a message \n");

 msgrcv(msgid, &msg, 256, 2, 0);

 }

 }