C – IPC – Message Queues | Bits, Bytes and Brains

What ?

Message queues are a form of interprocess communication (IPC) in which processes can exchange messages with one another through a message queue provided by the operating system.

–> A message queue is a mechanism for sending and receiving messages between processes that don’t necessarily have any relationship with each other beyond the need to communicate.

Why ?

Message queues are a way for different processes or threads to communicate with each other in a reliable, asynchronous manner.

They allow data to be sent between processes without the sender needing to wait for the recipient to receive the data. This can be useful in a variety of scenarios, such as:

Interprocess communication: If you have two or more processes that need to communicate with each other, message queues provide a convenient way to do so. Each process can send messages to the queue and receive messages from the queue as needed.
Load balancing: If you have multiple worker processes that need to process tasks, you can use a message queue to distribute tasks evenly among them. Each worker process can listen to the queue and process tasks as they become available.
Asynchronous processing: If you have tasks that take a long time to complete, you can use a message queue to start the tasks and then let them run in the background while the main program continues to execute. Once the tasks are complete, they can send a message back to the main program to indicate that they’re finished.

Overall, message queues provide a simple and efficient way for processes to communicate with each other in a decoupled and asynchronous manner, making them a powerful tool for building distributed systems.

How in Linux

In Linux, a message queue is identified by a unique key, which is used by processes to access the queue. Queue contains messages sent by processes.

Each message consists of 2 parts:

message type and
message body.

The message type is an integer value that can be used to group related messages together, while the message body contains the actual data being sent between processes.

–> A process can send a message to a message queue using the msgsnd() function, which takes as input the message queue identifier, the message type, a pointer to the message body, and various flags that control how the message is sent.

–> A process can receive a message from a message queue using the msgrcv() function, which takes as input the message queue identifier, the message type (or 0 to receive any message), a pointer to a buffer to receive the message body, the size of the buffer, and various flags that control how the message is received.

How messages picked up from queue? Messages in a message queue are typically processed in a first-in, first-out (FIFO) order. If multiple processes are waiting to receive messages from the same message queue, the operating system will typically give priority to the process that has been waiting the longest.

Message queues are useful for a variety of purposes, such as coordinating activities between processes, passing data between processes, and implementing message-based protocols. They are a flexible and efficient way to implement IPC in Linux and are widely used in many different types of applications.

Implementation

Here’s an example of how to use message queues in C on Linux:

Step 1 – Header files

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

These are the required header files for working with message queues on Linux.

Step 2 – define message structure

#define MSG_SIZE 1024

struct message {
    long mtype;
    char mtext[MSG_SIZE];
};

Here, we define the maximum size of the message text and the message structure, which consists of a message type (which we can use to filter messages) and the message text. Here, the message size is fixed.

Step 3 – Generate unique key for message queue

int main() {
    key_t key;
    int msgid;
    struct message msg;
    int ret;

    // generate a key for the message queue
    if ((key = ftok(".", 'a')) == -1) {
        perror("ftok");
        exit(1);
    }

Here, we generate a unique key for the message queue using ftok(), which takes a file path and a unique identifier as arguments. In this case, we’re using the current directory and the letter ‘a’ as the unique identifier. If ftok() returns -1, there was an error generating the key, so we print an error message and exit.

Step 4 – Create message queue

    // create the message queue
    if ((msgid = msgget(key, 0666 | IPC_CREAT)) == -1) {
        perror("msgget");
        exit(1);
    }

Next, we create the message queue using msgget(), which takes the key generated by ftok() and some flags as arguments. In this case, we’re using the flags 0666 to set the permissions of the message queue (readable and writable by everyone) and IPC_CREAT to create the queue if it doesn’t already exist. If msgget() returns -1, there was an error creating the queue, so we print an error message and exit.

Step 5 – Send the message to message queue

    // send a message to the queue
    msg.mtype = 1;
    strcpy(msg.mtext, "Hello, world!");
    if ((ret = msgsnd(msgid, &msg, sizeof(struct message), 0)) == -1) {
        perror("msgsnd");
        exit(1);
    }

Here, we set the message type to 1 and the message text to “Hello, world!” in the msg structure.

We then send the message to the queue using msgsnd(), which takes the message queue ID, a pointer to the message structure, the size of the message structure, and some flags as arguments. In this case, we’re using the flag 0 to block until there’s space in the queue to send the message. If msgsnd() returns -1, there was an error sending the message, so we print an error message and exit.

Step 6 – Receive message from queue

    // receive a message from the queue
    if ((ret = msgrcv(msgid, &msg, sizeof(struct message), 1, 0)) == -1) {
        perror("msgrcv");
        exit(1);
    }
    printf("Received message: %s\n", msg.mtext);

Next, we receive a message from the queue using msgrcv(), which takes the message queue ID, a pointer to the message structure, the maximum size of the message structure, the message type to filter on (in this case, 1), and some flags as arguments. In this case, we’re using the flag 0 to block until a message with the message type is read from the queue.

Step 7 – delete the message queue

    // delete the message queue
    if ((ret = msgctl(msgid, IPC_RMID, NULL)) == -1) {
        perror("msgctl");
        exit(1);
    }

Full code

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

#define MSG_SIZE 1024

struct message {
    long mtype;
    char mtext[MSG_SIZE];
};

int main() {
    key_t key;
    int msgid;
    struct message msg;
    int ret;

    // generate a key for the message queue
    if ((key = ftok(".", 'a')) == -1) {
        perror("ftok");
        exit(1);
    }

    // create the message queue
    if ((msgid = msgget(key, 0666 | IPC_CREAT)) == -1) {
        perror("msgget");
        exit(1);
    }

    // send a message to the queue
    msg.mtype = 1;
    strcpy(msg.mtext, "Hello, world!");
    if ((ret = msgsnd(msgid, &msg, sizeof(struct message), 0)) == -1) {
        perror("msgsnd");
        exit(1);
    }

    // receive a message from the queue
    if ((ret = msgrcv(msgid, &msg, sizeof(struct message), 1, 0)) == -1) {
        perror("msgrcv");
        exit(1);
    }
    printf("Received message: %s\n", msg.mtext);

    // delete the message queue
    if ((ret = msgctl(msgid, IPC_RMID, NULL)) == -1) {
        perror("msgctl");
        exit(1);
    }

    return 0;
}

Example communication between 2 different processes using Message Queue

Here’s an another example of using message queues in Linux to communicate between two different processes:

Process A:

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

#define MSG_SIZE 1024

struct message {
    long mtype;
    char mtext[MSG_SIZE];
};

int main() {
    key_t key;
    int msgid;
    struct message msg;
    int ret;

    // generate a key for the message queue
    if ((key = ftok(".", 'a')) == -1) {
        perror("ftok");
        exit(1);
    }

    // create the message queue
    if ((msgid = msgget(key, 0666 | IPC_CREAT)) == -1) {
        perror("msgget");
        exit(1);
    }

    // send a message to the queue
    msg.mtype = 1;
    strcpy(msg.mtext, "Hello, world!");
    if ((ret = msgsnd(msgid, &msg, sizeof(struct message), 0)) == -1) {
        perror("msgsnd");
        exit(1);
    }

    printf("Message sent!\n");

    return 0;
}

Process B:

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

#define MSG_SIZE 1024

struct message {
    long mtype;
    char mtext[MSG_SIZE];
};

int main() {
    key_t key;
    int msgid;
    struct message msg;
    int ret;

    // generate a key for the message queue
    if ((key = ftok(".", 'a')) == -1) {
        perror("ftok");
        exit(1);
    }

    // get the message queue
    if ((msgid = msgget(key, 0666)) == -1) {
        perror("msgget");
        exit(1);
    }

    // receive a message from the queue
    if ((ret = msgrcv(msgid, &msg, sizeof(struct message), 1, 0)) == -1) {
        perror("msgrcv");
        exit(1);
    }

    printf("Received message: %s\n", msg.mtext);

    // remove the message queue
    if (msgctl(msgid, IPC_RMID, NULL) == -1) {
        perror("msgctl");
        exit(1);
    }

    return 0;
}

In this example, process A sends a message to the message queue, and process B receives the message from the queue.

Step 1 – Process A starts by generating a key for the message queue using ftok(), and then creates the queue using msgget(). It then sets the message type to 1, the message text to “Hello, world!”, and sends the message to the queue using msgsnd().

Step 2 – Process B starts by generating the same key for the message queue using ftok(), and then gets the existing queue using msgget(). It then receives a message from the queue using msgrcv(), and prints the message text to the console.

Step 3 – Finally, it removes the message queue using msgctl().

–> For process B to receive the message, it needs to use the same key as process A. In this case, we’re using the current directory and the letter ‘a’ as the unique identifier. If you use a different identifier, such as a file path or a numerical value, you’ll need to make sure both processes are using the same identifier.

Processing of messages from Queue : Also note that in this example, the message type is set to 1. This is an arbitrary value and could be set to any positive integer. When process B receives messages from the queue using msgrcv(), it specifies a message type of 1, which means it will only receive messages with a message type of 1. If you need to send and receive messages with different types, you can use different message type values and specify the appropriate type when sending and receiving messages.

Finally, it’s worth noting that message queues are just one of several inter-process communication mechanisms available in Linux. Depending on your needs, other options such as shared memory or sockets may be more appropriate. However, message queues can be a simple and effective way to pass data between processes in many scenarios.

Example of process listening for messages from message queue

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

#define MSG_SIZE 100

struct msgbuf {
    long mtype;
    char mtext[MSG_SIZE];
};

int main() {
    key_t key;
    int msgid;
    struct msgbuf message;

    // Generate a unique key for the message queue
    key = ftok("message_queue_example", 1);

    // Create a message queue with read/write access for the owner
    msgid = msgget(key, IPC_CREAT | 0600);

    while(1) {
        // Receive a message from the message queue with blocking flag
        if (msgrcv(msgid, &message, MSG_SIZE, 1, 0) == -1) {
            perror("msgrcv");
            exit(1);
        }

        // Print the message received from the message queue
        printf("Received message: %s\n", message.mtext);

        // Check for exit condition (e.g. "quit" message)
        if (strcmp(message.mtext, "quit") == 0) {
            break;
        }
    }

    // Clean up the message queue
    msgctl(msgid, IPC_RMID, NULL);

    return 0;
}

In this example, the process generates a unique key for the message queue using ftok(), creates a message queue using msgget() with read/write access for the owner, and enters a loop to receive messages from the message queue using msgrcv() with the blocking flag (0).

When a message is received, the process prints the message using printf(). It then checks if the message received is the exit condition (e.g. “quit” message), and breaks out of the loop if it is.

When the loop is exited, the process cleans up the message queue using msgctl() with the IPC_RMID flag to remove the message queue.

Note that in this example, the process blocks on msgrcv() waiting for a message to be received. This means that the process will be idle until a message is received.

If you need the process to do other things while waiting for messages, you can use non-blocking message queue processing like below.

Asynchronous usage on message queue

    // Generate the same key used by process A
    key = ftok("message_queue_example", 1);

    // Attach to the existing message queue with non-blocking flag
    msgid = msgget(key, IPC_NOWAIT);

    // Check if a message is available
    if (msgrcv(msgid, &message, MSG_SIZE, 1, IPC_NOWAIT) != -1) {
        // Print the message received from process A
        printf("Received message: %s\n", message.mtext);
    } else {
        if (errno == ENOMSG) {
            printf("No message available\n");
        } else {
            perror("msgrcv");
        }
    }

Process B generates the same key as Process A using ftok(), attaches to the existing message queue using msgget() with the IPC_NOWAIT flag, and checks if a message is available using msgrcv() with the IPC_NOWAIT flag.

If a message is available, Process B prints the message received from Process A. If no message is available, it prints a message indicating that no message is available. If an error occurs while checking for messages, it prints an error message using perror().

Note that in asynchronous message queue processing, it’s important to handle the case where no messages are available. This is why we use the ENOMSG error code to check if no messages are available, and print a message indicating that no message is available.