CHEATSHEETSTOOLSBOOKSHELFABOUT中文

POSIX Message Queues (mq_open, mq_send, mq_receive, mq_notify)

POSIX Message Queues provide structured, priority-ordered inter-process communication (IPC) defined by IEEE Std 1003.1 under <mqueue.h>.

Unlike byte-stream IPC primitives (such as Unix domain sockets or standard pipes) which transmit unstructured byte streams, a message queue preserves discrete message boundaries: each call to mq_send() delivers exactly one message, and each call to mq_receive() retrieves exactly one message.

Furthermore, POSIX message queues support message priority ordering, ensuring critical control or emergency messages are dequeued ahead of normal traffic.

Core System Interfaces

#include <fcntl.h>           /* For O_* constants */
#include <sys/stat.h>        /* For mode constants */
#include <mqueue.h>

mqd_t mq_open(const char *name, int oflag, ... /* mode_t mode, struct mq_attr *attr */);
int mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned int msg_prio);
ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned int *msg_prio);
int mq_notify(mqd_t mqdes, const struct sigevent *sevp);
int mq_close(mqd_t mqdes);
int mq_unlink(const char *name);

1. Queue Attributes (struct mq_attr)

When creating a new queue, you configure its capacity and maximum message size:

struct mq_attr {
    long mq_flags;       /* Flags: 0 or O_NONBLOCK */
    long mq_maxmsg;      /* Max number of messages allowed in queue (e.g. 10) */
    long mq_msgsize;     /* Max size of a single message in bytes (e.g. 1024) */
    long mq_curmsgs;     /* Number of messages currently in queue */
};

On Linux, system-wide defaults and maximums are defined in /proc/sys/fs/mqueue/msg_max (typically 10) and /proc/sys/fs/mqueue/msgsize_max (typically 8192 bytes).

2. Priority Ordering Semantics

Every message sent via mq_send() has an associated unsigned int msg_prio (from 0 to 32767 on Linux, where larger numbers represent higher priorities):

  • Messages are delivered in strict descending priority order.
  • High-priority messages (e.g. priority 10) jump ahead of lower-priority messages (e.g. priority 1) already queued.
  • Messages with identical priorities are delivered in standard First-In, First-Out (FIFO) order.

3. Producer-Consumer Code Example

The Producer (Sending Messages)

#include <fcntl.h>
#include <mqueue.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

int main() {
    struct mq_attr attr = {
        .mq_maxmsg  = 10,
        .mq_msgsize = 256,
    };

    mqd_t mq = mq_open("/task_queue", O_CREAT | O_WRONLY, 0660, &attr);
    if (mq == (mqd_t)-1) {
        perror("mq_open failed");
        exit(EXIT_FAILURE);
    }

    const char *msg_normal = "Normal background job";
    const char *msg_urgent = "CRITICAL: Flush cache immediately";

    // Send normal task (priority 1)
    mq_send(mq, msg_normal, strlen(msg_normal) + 1, 1);

    // Send urgent task (priority 10) - jumps to the front of the line
    mq_send(mq, msg_urgent, strlen(msg_urgent) + 1, 10);

    mq_close(mq);
    return 0;
}

The Consumer (Receiving Messages)

[!IMPORTANT] The buffer passed to mq_receive() must have a capacity of at least mq_attr.mq_msgsize. If msg_len < mq_msgsize, mq_receive() fails immediately with EMSGSIZE.

#include <fcntl.h>
#include <mqueue.h>
#include <stdio.h>
#include <stdlib.h>

int main() {
    mqd_t mq = mq_open("/task_queue", O_RDONLY);
    if (mq == (mqd_t)-1) {
        perror("mq_open failed");
        exit(EXIT_FAILURE);
    }

    struct mq_attr attr;
    mq_getattr(mq, &attr);

    char *buf = malloc(attr.mq_msgsize);
    unsigned int prio;

    // Receives the highest priority message first ("CRITICAL...")
    ssize_t bytes_read = mq_receive(mq, buf, attr.mq_msgsize, &prio);
    if (bytes_read >= 0) {
        printf("Received (Priority %u): %s\n", prio, buf);
    }

    free(buf);
    mq_close(mq);
    mq_unlink("/task_queue");
    return 0;
}

4. Asynchronous Notification with mq_notify()

mq_notify() allows an application to be notified asynchronously when a message arrives in a previously empty queue, avoiding polling loops:

  • Can generate a signal (e.g. SIGUSR1) or spawn a new thread to execute a notification callback.
  • One-shot behavior: Once a notification is delivered, the registration is automatically deregistered. The consumer must re-arm mq_notify() after draining the queue.

Feature Comparison: Message Queues vs. Pipes vs. Unix Sockets

Feature POSIX Message Queue Standard Pipe (pipe()) Unix Domain Socket
Data Framing Message boundaries preserved. Raw unstructured byte stream. Stream (SOCK_STREAM) or Datagram (SOCK_DGRAM).
Ordering Priority-ordered (high priority first). Strict FIFO byte stream. Strict FIFO stream/datagram.
Process Coupling Unrelated processes via name /name. Parent/child only (unless named FIFO). Unrelated processes via path or abstract name.
Event Loop Support Pollable descriptor or mq_notify(). Pollable (epoll, poll, select). Fully pollable (epoll, kqueue, poll).

Frequently Asked Questions (FAQ)

Why does mq_receive() fail with EMSGSIZE?

Because the buffer size passed to mq_receive() is smaller than the queue's maximum message size (mq_msgsize). Even if the actual message currently waiting to be read is only 10 bytes long, POSIX requires the destination buffer to be at least mq_msgsize bytes to prevent message truncation.

What happens when an application calls mq_send() on a full queue?

If opened in blocking mode, mq_send() blocks until space becomes available; in non-blocking mode (O_NONBLOCK), it returns -1 with errno == EAGAIN. A queue is considered full when the number of queued messages equals mq_maxmsg.

Where are POSIX message queues located on Linux?

Linux exposes POSIX message queues through a virtual filesystem mounted at /dev/mqueue. You can mount this filesystem with mount -t mqueue none /dev/mqueue to inspect queue attributes, view active queues with ls -l, or delete orphaned queues with rm.

Related Topics