Receiving Messages: mq_receive() Priority Delivery, Buffer Sizing & Patterns

Receiving Messages: mq_receive()

Chapter 52 · File 3 of 3 | Priority Delivery, Buffer Sizing & Patterns

← Index | ← File 2: mq_send | File 3: mq_receive

How mq_receive() Works

mq_receive() retrieves one message from a POSIX message queue and removes it. It always retrieves the oldest message with the highest priority — not simply the first message that was sent. This is the key difference from a plain FIFO pipe or System V FIFO-ordered message queue.

If the queue is empty, mq_receive() blocks (sleeps) until a message arrives, unless the queue was opened with O_NONBLOCK, in which case it fails immediately with errno = EAGAIN.

mq_receive() Function Signature

#include<mqueue.h>
ssize_t mq_receive(

mqd_t mqdes, /* message queue descriptor */

char *msg_ptr, /* buffer to receive message into */

size_t msg_len, /* size of buffer (must be ≥ mq_msgsize) */

unsigned int *msg_prio /* if non-NULL, receives priority of message */

);
/* Returns number of bytes in received message, or -1 on error */

Parameter	Type	Description
`mqdes`	mqd_t	Queue descriptor — must be open for reading (O_RDONLY or O_RDWR)
`msg_ptr`	char *	Buffer where the received message will be written
`msg_len`	size_t	Must be ≥ mq_msgsize of the queue. If smaller, EMSGSIZE is returned.
`msg_prio`	unsigned int *	If non-NULL, the priority of the received message is stored here. Pass NULL if you don’t need it.

Return Value: On success, mq_receive() returns the number of bytes in the received message (which may be 0 for a zero-length message). On error, it returns -1 and sets errno.

The Buffer Sizing Rule — Critical!

The msg_len you pass to mq_receive() must be at least mq_msgsize bytes. If your buffer is too small, the call fails with EMSGSIZE. The safe pattern is to always read the queue’s mq_msgsize via mq_getattr() and allocate your buffer accordingly.

❌ WRONG char buf[64];

mq_receive(mqd, buf, 64, NULL);

 Fails if mq_msgsize > 64

errno = EMSGSIZE 
→
✅ CORRECT mq_getattr(mqd, &attr);

char *buf = malloc(attr.mq_msgsize);

mq_receive(mqd, buf, attr.mq_msgsize, NULL);

 Always works regardless of mq_msgsize 

Error Conditions

Error	Cause	Fix
`EMSGSIZE`	msg_len < mq_msgsize of the queue	Use mq_getattr() to size your buffer correctly
`EAGAIN`	Queue empty & O_NONBLOCK set	Retry later or use blocking mode
`EBADF`	mqdes not open for reading	Open with O_RDONLY or O_RDWR
`EINTR`	Blocked call interrupted by a signal	Retry the call in a loop

Code Examples

Example 1: Safe Basic Receiver (Auto-sized Buffer)

/* receiver.c — safely receives one message from a queue */
/* compile: gcc -o receiver receiver.c -lrt */
#include <stdio.h>
#include <stdlib.h>
#include <mqueue.h>
#include <fcntl.h>

#define QUEUE_NAME "/demo_queue"

int main(void)
{
    mqd_t mqd;
    struct mq_attr attr;
    char *buf;
    ssize_t bytes_read;
    unsigned int priority;

    /* Open queue for reading */
    mqd = mq_open(QUEUE_NAME, O_RDONLY);
    if (mqd == (mqd_t) -1) { perror("mq_open"); exit(1); }

    /* Step 1: find out the max message size */
    if (mq_getattr(mqd, &attr) == -1) { perror("mq_getattr"); exit(1); }

    /* Step 2: allocate a buffer exactly that size */
    buf = malloc(attr.mq_msgsize);
    if (buf == NULL) { perror("malloc"); exit(1); }

    /* Step 3: receive — blocks until a message arrives */
    bytes_read = mq_receive(mqd, buf, attr.mq_msgsize, &priority);
    if (bytes_read == -1) {
        perror("mq_receive");
        free(buf);
        exit(1);
    }

    /* null-terminate if treating as text */
    buf[bytes_read] = '\0';
    printf("Received %zd bytes (priority %u): \"%s\"\n",
           bytes_read, priority, buf);

    free(buf);
    mq_close(mqd);
    return 0;
}

Example 2: Drain All Messages from a Queue

/* drain_queue.c — receive all messages until queue is empty */
/* compile: gcc -o drain_queue drain_queue.c -lrt */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <mqueue.h>
#include <fcntl.h>

#define QUEUE_NAME "/demo_queue"

int main(void)
{
    mqd_t mqd;
    struct mq_attr attr;
    char *buf;
    ssize_t bytes;
    unsigned int prio;
    int count = 0;

    /* Open non-blocking so we can detect empty queue */
    mqd = mq_open(QUEUE_NAME, O_RDONLY | O_NONBLOCK);
    if (mqd == (mqd_t) -1) { perror("mq_open"); exit(1); }

    if (mq_getattr(mqd, &attr) == -1) { perror("mq_getattr"); exit(1); }

    buf = malloc(attr.mq_msgsize + 1);   /* +1 for null terminator */
    if (!buf) { perror("malloc"); exit(1); }

    printf("Draining queue (capacity: %ld messages)...\n", attr.mq_maxmsg);

    /* Keep reading until EAGAIN (queue empty) */
    while (1) {
        bytes = mq_receive(mqd, buf, attr.mq_msgsize, &prio);

        if (bytes == -1) {
            if (errno == EAGAIN) {
                /* Queue is now empty */
                printf("Queue empty. Total messages received: %d\n", count);
                break;
            }
            perror("mq_receive");
            break;
        }

        buf[bytes] = '\0';
        printf("[%d] prio=%u len=%zd msg=\"%s\"\n", ++count, prio, bytes, buf);
    }

    free(buf);
    mq_close(mqd);
    return 0;
}

Example 3: Verify Priority-Based Delivery Order

/* priority_demo.c — send messages in mixed order, observe priority delivery */
/* compile: gcc -o priority_demo priority_demo.c -lrt */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <fcntl.h>

#define QNAME "/prio_demo"

/* Helper to send a labelled message */
static void send_msg(mqd_t mqd, const char *text, unsigned int prio) {
    if (mq_send(mqd, text, strlen(text), prio) == -1)
        perror("mq_send");
    else
        printf("  SENT prio=%-3u : %s\n", prio, text);
}

int main(void)
{
    mqd_t mqd;
    struct mq_attr attr;
    char buf[256];
    ssize_t bytes;
    unsigned int prio;

    attr.mq_flags   = 0;
    attr.mq_maxmsg  = 10;
    attr.mq_msgsize = 256;
    attr.mq_curmsgs = 0;

    /* Create queue */
    mq_unlink(QNAME);  /* remove if it exists */
    mqd = mq_open(QNAME, O_CREAT | O_RDWR, 0644, &attr);
    if (mqd == (mqd_t) -1) { perror("mq_open"); exit(1); }

    printf("--- Sending messages ---\n");
    send_msg(mqd, "Task: low urgency",    2);
    send_msg(mqd, "Task: high urgency",  10);
    send_msg(mqd, "Task: normal",         5);
    send_msg(mqd, "Task: critical",      20);
    send_msg(mqd, "Task: also normal",    5);

    printf("\n--- Receiving messages (by priority) ---\n");
    int order = 1;
    while (1) {
        /* Use O_NONBLOCK — queue already has all messages loaded */
        bytes = mq_receive(mqd, buf, sizeof(buf), &prio);
        if (bytes == -1) break;   /* queue empty */

        buf[bytes] = '\0';
        printf("  RECV #%d prio=%-3u : %s\n", order++, prio, buf);
    }

    mq_close(mqd);
    mq_unlink(QNAME);
    return 0;
}

Expected output:



      RECV #1 prio=20 : Task: critical

RECV #2 prio=10 : Task: high urgency

RECV #3 prio=5  : Task: normal

RECV #4 prio=5  : Task: also normal

RECV #5 prio=2  : Task: low urgency

Example 4: Receiving a Struct Message

/* struct_receiver.c — receives a binary struct message */
/* compile: gcc -o struct_receiver struct_receiver.c -lrt */
#include <stdio.h>
#include <stdlib.h>
#include <mqueue.h>
#include <fcntl.h>

#define QUEUE_NAME "/struct_queue"

typedef struct {
    int   sensor_id;
    float temperature;
    int   alarm_level;
} SensorData;

int main(void)
{
    mqd_t mqd;
    SensorData reading;
    ssize_t bytes;
    unsigned int prio;

    mqd = mq_open(QUEUE_NAME, O_RDONLY);
    if (mqd == (mqd_t) -1) { perror("mq_open"); exit(1); }

    /* Receive directly into the struct buffer */
    bytes = mq_receive(mqd, (char *)&reading, sizeof(reading), &prio);
    if (bytes == -1) {
        perror("mq_receive");
        mq_close(mqd);
        exit(1);
    }

    if (bytes != sizeof(SensorData)) {
        fprintf(stderr, "Unexpected message size: %zd\n", bytes);
    } else {
        printf("Sensor ID   : %d\n",   reading.sensor_id);
        printf("Temperature : %.1f°C\n", reading.temperature);
        printf("Alarm Level : %d\n",   reading.alarm_level);
        printf("Priority    : %u\n",   prio);
    }

    mq_close(mqd);
    return 0;
}

Example 5: Handle EINTR Correctly (Signal-Safe Loop)

/* signal_safe_recv.c — mq_receive() restarts after signal interruption */
/* compile: gcc -o signal_safe_recv signal_safe_recv.c -lrt */
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <mqueue.h>
#include <fcntl.h>

#define QUEUE_NAME "/demo_queue"

ssize_t reliable_receive(mqd_t mqd, char *buf, size_t buflen, unsigned int *prio)
{
    ssize_t ret;

    /* Retry the call if it is interrupted by a signal */
    do {
        ret = mq_receive(mqd, buf, buflen, prio);
    } while (ret == -1 && errno == EINTR);

    return ret;
}

int main(void)
{
    mqd_t mqd;
    struct mq_attr attr;
    char *buf;
    ssize_t bytes;
    unsigned int prio;

    mqd = mq_open(QUEUE_NAME, O_RDONLY);
    if (mqd == (mqd_t) -1) { perror("mq_open"); exit(1); }

    mq_getattr(mqd, &attr);
    buf = malloc(attr.mq_msgsize + 1);
    if (!buf) { perror("malloc"); exit(1); }

    bytes = reliable_receive(mqd, buf, attr.mq_msgsize, &prio);
    if (bytes > 0) {
        buf[bytes] = '\0';
        printf("Received (prio=%u): %s\n", prio, buf);
    }

    free(buf);
    mq_close(mqd);
    return 0;
}

Example 6: Full Producer-Consumer Pattern

/* producer_consumer.c — producer thread sends, consumer thread receives */
/* compile: gcc -o prod_cons producer_consumer.c -lrt -lpthread */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>
#include <mqueue.h>
#include <fcntl.h>

#define QUEUE_NAME  "/pc_queue"
#define MAX_MSG     5
#define MSG_SIZE    64
#define NUM_MSGS    5

/* Producer: sends 5 messages with ascending priorities */
void *producer(void *arg)
{
    mqd_t mqd;
    char buf[MSG_SIZE];
    int i;

    mqd = mq_open(QUEUE_NAME, O_WRONLY);
    if (mqd == (mqd_t) -1) { perror("producer: mq_open"); return NULL; }

    for (i = 1; i <= NUM_MSGS; i++) {
        snprintf(buf, sizeof(buf), "Job #%d", i);
        unsigned int prio = (unsigned int)(i * 2);   /* 2, 4, 6, 8, 10 */
        if (mq_send(mqd, buf, strlen(buf), prio) == -1) {
            perror("producer: mq_send"); break;
        }
        printf("[Producer] Sent: \"%s\" (prio=%u)\n", buf, prio);
        usleep(50000);   /* 50ms between sends */
    }

    mq_close(mqd);
    return NULL;
}

/* Consumer: receives messages and processes in priority order */
void *consumer(void *arg)
{
    mqd_t mqd;
    struct mq_attr attr;
    char *buf;
    ssize_t bytes;
    unsigned int prio;
    int received = 0;

    mqd = mq_open(QUEUE_NAME, O_RDONLY);
    if (mqd == (mqd_t) -1) { perror("consumer: mq_open"); return NULL; }

    mq_getattr(mqd, &attr);
    buf = malloc(attr.mq_msgsize + 1);
    if (!buf) return NULL;

    while (received < NUM_MSGS) {
        bytes = mq_receive(mqd, buf, attr.mq_msgsize, &prio);
        if (bytes == -1) { perror("consumer: mq_receive"); break; }
        buf[bytes] = '\0';
        printf("[Consumer] Recv: \"%s\" (prio=%u)\n", buf, prio);
        received++;
    }

    free(buf);
    mq_close(mqd);
    return NULL;
}

int main(void)
{
    pthread_t prod_tid, cons_tid;
    struct mq_attr attr = {
        .mq_flags   = 0,
        .mq_maxmsg  = MAX_MSG,
        .mq_msgsize = MSG_SIZE,
        .mq_curmsgs = 0
    };

    /* Create queue */
    mq_unlink(QUEUE_NAME);
    mqd_t mqd = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, &attr);
    if (mqd == (mqd_t) -1) { perror("mq_open"); exit(1); }
    mq_close(mqd);

    pthread_create(&prod_tid, NULL, producer, NULL);
    pthread_create(&cons_tid, NULL, consumer, NULL);

    pthread_join(prod_tid, NULL);
    pthread_join(cons_tid, NULL);

    mq_unlink(QUEUE_NAME);
    return 0;
}

mq_receive vs read() from a Pipe

Feature	mq_receive()	read() from Pipe/FIFO
Message boundaries	Preserved — one call = one message	No — byte stream, must frame yourself
Priority ordering	Yes — highest priority first	No — FIFO only
Priority value returned	Yes (via msg_prio parameter)	No
Buffer size constraint	Must be ≥ mq_msgsize	Any size (may get partial data)
Named (accessible by path)	Yes (/name in filesystem)	FIFO yes, anonymous pipe no

Interview Questions & Answers

Q1. What message does mq_receive() retrieve from the queue?

It retrieves the oldest message with the highest priority currently in the queue. If multiple messages have the same highest priority, the one that was sent first (FIFO within that priority level) is returned. The message is atomically removed from the queue.

Q2. What size must the buffer passed to mq_receive() be?

The buffer must be at least mq_msgsize bytes — the maximum message size configured for that queue. If the buffer is smaller, mq_receive() fails with EMSGSIZE. The safe practice is to call mq_getattr() and dynamically allocate attr.mq_msgsize bytes.

Q3. What does mq_receive() do when the queue is empty?

In blocking mode (no O_NONBLOCK): it blocks until a message arrives or is interrupted by a signal. In non-blocking mode (O_NONBLOCK set): it returns immediately with -1 and errno = EAGAIN.

Q4. What is the return value of mq_receive() and what does it represent?

On success, mq_receive() returns the number of bytes in the received message as an ssize_t. This may be 0 for zero-length messages. On error, it returns -1 and sets errno. Do not assume the message is null-terminated — always use the returned byte count.

Q5. How do you find out the priority of the received message?

Pass a pointer to an unsigned int as the last argument (msg_prio). After the call, that variable holds the priority of the message that was just received. If you don’t need the priority, pass NULL.

Q6. Why should you handle EINTR in a mq_receive() loop?

If the process receives a signal while blocked in mq_receive(), the call is interrupted and returns -1 with errno = EINTR. This does not mean an error occurred — the queue is still intact and no message was consumed. The correct pattern is to check for EINTR and restart the call in a loop: do { ret = mq_receive(...); } while (ret == -1 && errno == EINTR);

Q7. Is mq_receive() a destructive operation?

Yes. mq_receive() atomically removes the message from the queue while reading it. There is no “peek” operation in POSIX message queues. Once a message is received, it is gone from the queue. If you need to route the message to multiple consumers, your application must forward it after receiving, or use a different IPC mechanism.

Q8. How do you use mq_receive() to implement a simple task dispatcher?

Create a queue where senders assign priorities based on task urgency. The dispatcher calls mq_receive() in a blocking loop — it always wakes up with the most urgent pending task. High-priority tasks (e.g., emergency shutdown: priority 20) will always be dispatched before low-priority ones (e.g., background logging: priority 1), regardless of arrival order. This is much simpler to implement than maintaining a sorted task list manually.

← File 2: mq_send() Back to Index →

embeddedpathashala.com