Receiving Messages: mq_receive()
Chapter 52 · File 3 of 3 | Priority Delivery, Buffer Sizing & Patterns
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
Fails if mq_msgsize > 64
errno = EMSGSIZE
char buf[64];mq_receive(mqd, buf, 64, NULL);Fails if mq_msgsize > 64
errno = EMSGSIZE
→
✅ CORRECT
Always works regardless of mq_msgsize
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 |