Operating SystemsInter-Process Communication

Message Queues

LevelIntermediate

Duration60 mins

TopicInter-Process Communication

2 / 5

POSIX Message Queues

The Modern Standard for Message Passing

While System V message queues pioneered structured IPC in Unix, they carry historical baggage—awkward APIs, numeric keys, and limited features. POSIX message queues, standardized in IEEE Std 1003.1b (POSIX.1b, the Real-Time Extensions), provide a cleaner, more powerful alternative.

POSIX message queues address many System V limitations:

File-like naming: Queues are identified by names (like /myqueue) rather than numeric keys
Reference counting: Queues are cleaned up automatically when no process has them open
Asynchronous notification: Processes can be notified when messages arrive without polling
Per-message priority: Built-in priority support with guaranteed ordering
Better portability: Part of POSIX standard, available on Linux, FreeBSD, macOS, and others

Despite these improvements, the fundamental model remains the same: message queues provide typed, structured, asynchronous communication between processes.

What You Will Master

By the end of this page, you will understand the complete POSIX message queue API: creating and opening queues with mq_open(), sending with mq_send(), receiving with mq_receive(), configuring queue attributes, setting up asynchronous notifications, and the crucial differences from System V that inform your choice of mechanism.

POSIX vs System V Message Queues

Before diving into the API, let's understand the architectural differences between these two message queue implementations:

Fundamental Design Philosophy

System V IPC was designed as a monolithic kernel facility. Objects exist purely in kernel memory, identified by numeric keys, with no filesystem representation.

POSIX IPC was designed with a more Unix-like philosophy. Objects have names that look like filesystem paths, can be created/opened/closed like files, and on many systems are actually represented in a virtual filesystem (/dev/mqueue/).

System V vs POSIX Message Queue Comparison
Aspect	System V	POSIX
Identification	Numeric key (`key_t`)	Name string (`/name`)
Creation	`msgget(key, flags)`	`mq_open("/name", flags, mode, attr)`
Handle Type	`int msqid`	`mqd_t` (like file descriptor)
Priority Support	Via message type workaround	Native per-message priority
Notification	None (must poll)	Async notification (signal/thread)
Cleanup	Manual `msgctl(IPC_RMID)`	Auto when last reference closes + unlink
Filesystem	None	`/dev/mqueue/` on Linux
System Limits	`/proc/sys/kernel/msg*`	`/proc/sys/fs/mqueue/*`
Link Library	None (built into libc)	`-lrt` required on Linux

When to Choose POSIX

Prefer POSIX message queues when: (1) You need asynchronous notification without polling, (2) Message priority is a natural fit for your protocol, (3) You want cleaner, more maintainable code, (4) You're writing new code without legacy constraints. Stick with System V when maintaining existing code or when portability to very old Unix systems matters.

Queue Naming Convention

POSIX message queue names follow specific rules:

Naming Requirements

Must begin with / — The name must start with a forward slash
No additional slashes — Only one leading slash; no internal slashes
Maximum length — Typically NAME_MAX (usually 255) characters including the slash
Portable characters — Alphanumeric, underscore, and hyphen are safe

// Valid names:
"/myqueue"
"/task_queue_1"
"/ipc-channel-main"

// Invalid names:
"myqueue"        // Missing leading slash
"/my/queue"      // Internal slash
""               // Empty

Filesystem Representation

On Linux, POSIX message queues appear in a virtual filesystem. To view and manage them:

# Mount the message queue filesystem (usually already mounted)
mount -t mqueue none /dev/mqueue

# List existing queues
ls -la /dev/mqueue/

# View queue contents/attributes
cat /dev/mqueue/myqueue

# Remove a queue directly
rm /dev/mqueue/myqueue

This filesystem representation is incredibly useful for debugging—you can inspect queue states, see message counts, and even remove stuck queues without writing code.

queue_naming.c
C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/stat.h>
 
// Build with: gcc queue_naming.c -o queue_naming -lrt
 
int test_queue_name(const char *name) {
    printf("Testing queue name: '%s'\n", name);
    
    mqd_t mq = mq_open(name, O_CREAT | O_RDWR, 0644, NULL);
    
    if (mq == (mqd_t)-1) {
        printf("  FAILED: %s\n\n", strerror(errno));
        return -1;
    }
    
    printf("  SUCCESS: Queue created\n");
    mq_close(mq);
    mq_unlink(name);
    printf("  Cleaned up\n\n");
    return 0;
}
 
int main() {
    printf("=== POSIX Message Queue Naming Tests ===\n\n");
    
    // Valid names
    test_queue_name("/simple");
    test_queue_name("/task_queue_1");
    test_queue_name("/my-app-channel");
    
    // Invalid names
    test_queue_name("no_leading_slash");  // EINVAL
    test_queue_name("/path/with/slashes"); // EINVAL on most systems
    test_queue_name("/");                  // EINVAL
    
    return EXIT_SUCCESS;
}

Creating and Opening Queues with mq_open()

The mq_open() function creates a new message queue or opens an existing one. Its semantics closely mirror the open() system call for files.

The mq_open() Function

#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>

mqd_t mq_open(const char *name, int oflag);
mqd_t mq_open(const char *name, int oflag, mode_t mode, 
              struct mq_attr *attr);

Parameters:

Parameter	Description
`name`	Queue name (must start with `/`)
`oflag`	Open flags (similar to `open()`)
`mode`	Permissions when creating (like file mode)
`attr`	Queue attributes (NULL for defaults)

Returns:

Message queue descriptor (mqd_t) on success
(mqd_t)-1 on error

Open Flags

The oflag parameter combines access mode and creation flags:

Access Mode (exactly one required):

O_RDONLY — Open for receiving only
O_WRONLY — Open for sending only
O_RDWR — Open for both sending and receiving

Creation Flags (optional):

O_CREAT — Create queue if it doesn't exist
O_EXCL — With O_CREAT, fail if queue exists
O_NONBLOCK — Non-blocking send/receive operations

mq_open_examples.c
C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <errno.h>
 
// Compile: gcc mq_open_examples.c -o mq_open_examples -lrt
 
#define QUEUE_NAME "/demo_queue"
 
int main() {
    mqd_t mq;
    
    // ========================================
    // Example 1: Create with default attributes
    // ========================================
    printf("=== Example 1: Create with defaults ===\n");
    mq = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, NULL);
    if (mq == (mqd_t)-1) {
        perror("mq_open (defaults)");
        return EXIT_FAILURE;
    }
    printf("Queue created with default attributes\n");
    mq_close(mq);
    mq_unlink(QUEUE_NAME);
    
    // ========================================
    // Example 2: Create with custom attributes
    // ========================================
    printf("\n=== Example 2: Create with custom attributes ===\n");
    
    struct mq_attr attr;
    attr.mq_flags = 0;          // Blocking mode
    attr.mq_maxmsg = 10;        // Max 10 messages in queue
    attr.mq_msgsize = 256;      // Max 256 bytes per message
    attr.mq_curmsgs = 0;        // Read-only (ignored on create)
    
    mq = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, &attr);
    if (mq == (mqd_t)-1) {
        perror("mq_open (custom)");
        return EXIT_FAILURE;
    }
    
    // Verify attributes
    struct mq_attr actual;
    mq_getattr(mq, &actual);
    printf("Queue created:\n");
    printf("  Max messages: %ld\n", actual.mq_maxmsg);
    printf("  Max msg size: %ld bytes\n", actual.mq_msgsize);
    
    mq_close(mq);
    mq_unlink(QUEUE_NAME);
    
    // ========================================
    // Example 3: Exclusive creation
    // ========================================
    printf("\n=== Example 3: Exclusive creation ===\n");
    
    // First creation succeeds
    mq = mq_open(QUEUE_NAME, O_CREAT | O_EXCL | O_RDWR, 0644, NULL);
    if (mq != (mqd_t)-1) {
        printf("First creation: SUCCESS\n");
        
        // Second creation with O_EXCL fails
        mqd_t mq2 = mq_open(QUEUE_NAME, O_CREAT | O_EXCL | O_RDWR, 0644, NULL);
        if (mq2 == (mqd_t)-1 && errno == EEXIST) {
            printf("Second creation: Failed (EEXIST) as expected\n");
        }
        
        mq_close(mq);
        mq_unlink(QUEUE_NAME);
    }
    
    // ========================================
    // Example 4: Open existing queue
    // ========================================
    printf("\n=== Example 4: Open existing queue ===\n");
    
    // Create first
    mq = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, NULL);
    mq_close(mq);
    
    // Open for reading only
    mqd_t reader = mq_open(QUEUE_NAME, O_RDONLY);
    if (reader != (mqd_t)-1) {
        printf("Reader opened queue successfully\n");
        mq_close(reader);
    }
    
    // Open for writing only
    mqd_t writer = mq_open(QUEUE_NAME, O_WRONLY);
    if (writer != (mqd_t)-1) {
        printf("Writer opened queue successfully\n");
        mq_close(writer);
    }
    
    mq_unlink(QUEUE_NAME);
    
    printf("\nAll examples completed\n");
    return EXIT_SUCCESS;
}

Queue Attributes Are Fixed at Creation

The mq_maxmsg and mq_msgsize attributes can ONLY be set when creating a queue. If the queue already exists, the attr parameter is ignored (existing attributes are preserved). This means you cannot resize an existing queue—you must delete and recreate it.

Queue Attributes and Configuration

POSIX message queues have configurable attributes that control their behavior and capacity.

The mq_attr Structure

struct mq_attr {
    long mq_flags;    // Queue flags (0 or O_NONBLOCK)
    long mq_maxmsg;   // Maximum number of messages
    long mq_msgsize;  // Maximum message size (bytes)
    long mq_curmsgs;  // Current messages in queue (read-only)
};

Attribute Functions

int mq_getattr(mqd_t mqdes, struct mq_attr *attr);
int mq_setattr(mqd_t mqdes, const struct mq_attr *newattr, 
               struct mq_attr *oldattr);

Note: mq_setattr() can only modify mq_flags (to toggle O_NONBLOCK). The size attributes are immutable after creation.

queue_attributes.c
C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <errno.h>
 
#define QUEUE_NAME "/attr_demo"
 
int main() {
    // Create queue with specific attributes
    struct mq_attr create_attr = {
        .mq_flags = 0,
        .mq_maxmsg = 5,      // Small queue for demo
        .mq_msgsize = 128,   // 128-byte messages
        .mq_curmsgs = 0      // Ignored on create
    };
    
    mqd_t mq = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, &create_attr);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        return EXIT_FAILURE;
    }
    
    // ========================================
    // Read current attributes
    // ========================================
    struct mq_attr current;
    mq_getattr(mq, &current);
    
    printf("=== Queue Attributes ===\n");
    printf("Flags: %ld (%s)\n", current.mq_flags,
           (current.mq_flags & O_NONBLOCK) ? "non-blocking" : "blocking");
    printf("Max messages: %ld\n", current.mq_maxmsg);
    printf("Max message size: %ld bytes\n", current.mq_msgsize);
    printf("Current messages: %ld\n", current.mq_curmsgs);
    
    // ========================================
    // Send some messages to show curmsgs change
    // ========================================
    const char *msg = "Test message";
    for (int i = 0; i < 3; i++) {
        mq_send(mq, msg, strlen(msg), 0);
    }
    
    mq_getattr(mq, &current);
    printf("\nAfter sending 3 messages:\n");
    printf("Current messages: %ld\n", current.mq_curmsgs);
    
    // ========================================
    // Toggle non-blocking mode
    // ========================================
    printf("\n=== Toggling Non-Blocking Mode ===\n");
    
    struct mq_attr new_attr, old_attr;
    new_attr.mq_flags = O_NONBLOCK;  // Only this field matters
    
    if (mq_setattr(mq, &new_attr, &old_attr) == 0) {
        printf("Previous flags: %ld\n", old_attr.mq_flags);
        
        mq_getattr(mq, &current);
        printf("New flags: %ld (non-blocking: %s)\n", 
               current.mq_flags,
               (current.mq_flags & O_NONBLOCK) ? "YES" : "NO");
    }
    
    // Demonstration: non-blocking receive when queue is empty
    char buf[128];
    // First, drain the queue
    while (mq_receive(mq, buf, sizeof(buf), NULL) != -1);
    
    // Now try to receive from empty queue
    ssize_t ret = mq_receive(mq, buf, sizeof(buf), NULL);
    if (ret == -1 && errno == EAGAIN) {
        printf("\nNon-blocking receive: EAGAIN (queue empty)\n");
    }
    
    // Clean up
    mq_close(mq);
    mq_unlink(QUEUE_NAME);
    
    return EXIT_SUCCESS;
}

System Limits

•MSGMAX — Maximum message size (default often 8192 bytes). Check: cat /proc/sys/fs/mqueue/msgsize_max
•MSGMNB — Maximum queue capacity. Check: cat /proc/sys/fs/mqueue/msg_max
•MSGMNI — Maximum queues system-wide. Check: cat /proc/sys/fs/mqueue/queues_max
•Per-user limits — Resource limits via ulimit or /etc/security/limits.conf

Sending Messages with mq_send()

The mq_send() function adds a message to a POSIX queue with an optional priority.

The mq_send() Function

#include <mqueue.h>

int mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, 
            unsigned int msg_prio);

// Timed version
int mq_timedsend(mqd_t mqdes, const char *msg_ptr, size_t msg_len,
                 unsigned int msg_prio, 
                 const struct timespec *abs_timeout);

Parameters:

Parameter	Description
`mqdes`	Queue descriptor from `mq_open()`
`msg_ptr`	Pointer to message data
`msg_len`	Size of message in bytes
`msg_prio`	Priority (0 = lowest, typically max ~32767)

Returns:

0 on success
-1 on error

Priority Semantics

Unlike System V (where priority is encoded in message type), POSIX provides native priority support:

Higher numeric values = higher priority
Messages are ordered by priority first, then FIFO within priority
Priority 0 is the lowest priority
Maximum priority is implementation-defined (sysconf(_SC_MQ_PRIO_MAX))

mq_send_demo.c
C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <fcntl.h>
#include <errno.h>
#include <time.h>
 
#define QUEUE_NAME "/priority_demo"
#define MAX_MSG_SIZE 256
 
int main() {
    // Create queue
    struct mq_attr attr = {
        .mq_flags = 0,
        .mq_maxmsg = 10,
        .mq_msgsize = MAX_MSG_SIZE,
        .mq_curmsgs = 0
    };
    
    mqd_t mq = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, &attr);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        return EXIT_FAILURE;
    }
    
    // ========================================
    // Send messages with different priorities
    // ========================================
    printf("=== Sending Messages ===\n");
    
    // Send in non-priority order
    mq_send(mq, "Low priority message", 20, 1);
    printf("Sent: priority=1 (low)\n");
    
    mq_send(mq, "Medium priority message", 23, 5);
    printf("Sent: priority=5 (medium)\n");
    
    mq_send(mq, "High priority message", 21, 10);
    printf("Sent: priority=10 (high)\n");
    
    mq_send(mq, "Another low priority", 20, 1);
    printf("Sent: priority=1 (low, second)\n");
    
    // ========================================
    // Receive messages (will come in priority order)
    // ========================================
    printf("\n=== Receiving Messages ===\n");
    
    char buffer[MAX_MSG_SIZE];
    unsigned int priority;
    ssize_t bytes;
    
    while ((bytes = mq_receive(mq, buffer, MAX_MSG_SIZE, &priority)) > 0) {
        buffer[bytes] = '\0';
        printf("Received (prio=%u): %s\n", priority, buffer);
    }
    
    // ========================================
    // Timed send demonstration
    // ========================================
    printf("\n=== Timed Send ===\n");
    
    // Fill the queue
    for (int i = 0; i < 10; i++) {
        mq_send(mq, "filler", 6, 0);
    }
    printf("Queue is now full (10 messages)\n");
    
    // Try timed send with 1-second timeout
    struct timespec timeout;
    clock_gettime(CLOCK_REALTIME, &timeout);
    timeout.tv_sec += 1;  // 1 second from now
    
    printf("Attempting timed send (1 second timeout)...\n");
    if (mq_timedsend(mq, "overflow", 8, 0, &timeout) == -1) {
        if (errno == ETIMEDOUT) {
            printf("Send timed out (queue still full)\n");
        } else {
            perror("mq_timedsend");
        }
    }
    
    // Clean up
    mq_close(mq);
    mq_unlink(QUEUE_NAME);
    
    return EXIT_SUCCESS;
}

Message Size Requirement

Unlike System V, where you can send less than the structure size, POSIX requires msg_len ≤ mq_msgsize. If msg_len > mq_msgsize, the call fails with EMSGSIZE. The receiver's buffer must also be at least mq_msgsize bytes.

Receiving Messages with mq_receive()

The mq_receive() function retrieves the highest-priority message from a queue.

The mq_receive() Function

#include <mqueue.h>

ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len,
                   unsigned int *msg_prio);

// Timed version
ssize_t mq_timedreceive(mqd_t mqdes, char *msg_ptr, size_t msg_len,
                        unsigned int *msg_prio,
                        const struct timespec *abs_timeout);

Parameters:

Parameter	Description
`mqdes`	Queue descriptor
`msg_ptr`	Buffer to receive message
`msg_len`	Buffer size (must be ≥ `mq_msgsize`)
`msg_prio`	Receives the message's priority (can be NULL)

Returns:

Number of bytes in the received message on success
-1 on error

Critical Buffer Size Rule

The msg_len parameter must be at least as large as the queue's mq_msgsize attribute. This is different from System V, where you could receive partial messages with MSG_NOERROR. POSIX enforces buffer adequacy at receive time.

mq_receive_patterns.c
C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <fcntl.h>
#include <errno.h>
#include <time.h>
 
#define QUEUE_NAME "/receive_demo"
 
int main() {
    struct mq_attr attr = {
        .mq_maxmsg = 10,
        .mq_msgsize = 256
    };
    
    mqd_t mq = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, &attr);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        return EXIT_FAILURE;
    }
    
    // Pre-populate with messages
    mq_send(mq, "Message A", 9, 5);
    mq_send(mq, "Message B", 9, 10);
    mq_send(mq, "Message C", 9, 5);
    
    // ========================================
    // Pattern 1: Basic receive with priority
    // ========================================
    printf("=== Pattern 1: Basic Receive ===\n");
    
    char buffer[256];
    unsigned int priority;
    ssize_t len;
    
    len = mq_receive(mq, buffer, sizeof(buffer), &priority);
    if (len > 0) {
        buffer[len] = '\0';
        printf("Received: '%s' (priority: %u, len: %zd)\n", 
               buffer, priority, len);
    }
    
    // ========================================
    // Pattern 2: Receive without caring about priority
    // ========================================
    printf("\n=== Pattern 2: Ignore Priority ===\n");
    
    len = mq_receive(mq, buffer, sizeof(buffer), NULL);  // NULL for priority
    if (len > 0) {
        buffer[len] = '\0';
        printf("Received: '%s' (len: %zd)\n", buffer, len);
    }
    
    // ========================================
    // Pattern 3: Non-blocking receive
    // ========================================
    printf("\n=== Pattern 3: Non-Blocking ===\n");
    
    // Set non-blocking mode
    struct mq_attr new_attr = { .mq_flags = O_NONBLOCK };
    mq_setattr(mq, &new_attr, NULL);
    
    // Drain remaining messages
    while ((len = mq_receive(mq, buffer, sizeof(buffer), NULL)) > 0) {
        buffer[len] = '\0';
        printf("Received: '%s'\n", buffer);
    }
    
    // Try to receive from empty queue
    len = mq_receive(mq, buffer, sizeof(buffer), NULL);
    if (len == -1 && errno == EAGAIN) {
        printf("Queue empty (EAGAIN)\n");
    }
    
    // ========================================
    // Pattern 4: Timed receive
    // ========================================
    printf("\n=== Pattern 4: Timed Receive ===\n");
    
    // Reset to blocking mode for timed receive
    new_attr.mq_flags = 0;
    mq_setattr(mq, &new_attr, NULL);
    
    struct timespec timeout;
    clock_gettime(CLOCK_REALTIME, &timeout);
    timeout.tv_sec += 2;  // 2-second timeout
    
    printf("Waiting for message (2 second timeout)...\n");
    len = mq_timedreceive(mq, buffer, sizeof(buffer), NULL, &timeout);
    
    if (len == -1) {
        if (errno == ETIMEDOUT) {
            printf("Timed out waiting for message\n");
        } else {
            perror("mq_timedreceive");
        }
    }
    
    mq_close(mq);
    mq_unlink(QUEUE_NAME);
    
    return EXIT_SUCCESS;
}

Asynchronous Message Notification

One of the most powerful features of POSIX message queues is asynchronous notification—the ability to be notified when a message arrives without polling.

The mq_notify() Function

#include <mqueue.h>

int mq_notify(mqd_t mqdes, const struct sigevent *sevp);

The sigevent structure specifies how notification should occur:

struct sigevent {
    int          sigev_notify;    // Notification method
    int          sigev_signo;     // Signal number (for SIGEV_SIGNAL)
    union sigval sigev_value;     // Data passed to handler
    void       (*sigev_notify_function)(union sigval);  // Thread function
    pthread_attr_t *sigev_notify_attributes;  // Thread attributes
};

Notification Methods

Method	Description
`SIGEV_NONE`	Disable notification
`SIGEV_SIGNAL`	Deliver a signal when message arrives
`SIGEV_THREAD`	Spawn a new thread to handle arrival

Critical Semantics

One-shot: Notification is delivered only once, then automatically unregistered. You must re-register after each notification.
Empty-to-non-empty transition: Notification occurs only when the queue goes from empty to having a message—not for every message.
Single registration: Only one process can register for notification at a time.

mq_notify_demo.c
C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <signal.h>
#include <pthread.h>
#include <fcntl.h>
#include <unistd.h>
 
#define QUEUE_NAME "/notify_demo"
#define MAX_MSG_SIZE 256
 
static mqd_t mq_global;
 
// ========================================
// Signal-based notification handler
// ========================================
void signal_handler(int signo, siginfo_t *info, void *context) {
    char buffer[MAX_MSG_SIZE];
    unsigned int priority;
    ssize_t len;
    
    printf("\n[Signal Handler] Received SIGUSR1\n");
    
    // Read message(s)
    while ((len = mq_receive(mq_global, buffer, MAX_MSG_SIZE, &priority)) > 0) {
        buffer[len] = '\0';
        printf("[Signal Handler] Message: '%s' (prio=%u)\n", buffer, priority);
    }
    
    // Re-register for notification (one-shot!)
    struct sigevent sev;
    sev.sigev_notify = SIGEV_SIGNAL;
    sev.sigev_signo = SIGUSR1;
    mq_notify(mq_global, &sev);
}
 
// ========================================
// Thread-based notification handler
// ========================================
void thread_handler(union sigval sv) {
    mqd_t mq = *((mqd_t *)sv.sival_ptr);
    char buffer[MAX_MSG_SIZE];
    ssize_t len;
    
    printf("\n[Thread Handler] Notification received\n");
    
    while ((len = mq_receive(mq, buffer, MAX_MSG_SIZE, NULL)) > 0) {
        buffer[len] = '\0';
        printf("[Thread Handler] Message: '%s'\n", buffer);
    }
    
    // Re-register (must pass sival_ptr for context)
    struct sigevent sev;
    sev.sigev_notify = SIGEV_THREAD;
    sev.sigev_notify_function = thread_handler;
    sev.sigev_value.sival_ptr = sv.sival_ptr;
    sev.sigev_notify_attributes = NULL;
    mq_notify(mq, &sev);
}
 
int main() {
    struct mq_attr attr = {
        .mq_maxmsg = 10,
        .mq_msgsize = MAX_MSG_SIZE
    };
    
    mq_global = mq_open(QUEUE_NAME, O_CREAT | O_RDWR | O_NONBLOCK, 
                        0644, &attr);
    if (mq_global == (mqd_t)-1) {
        perror("mq_open");
        return EXIT_FAILURE;
    }
    
    printf("=== Signal-Based Notification Demo ===\n");
    
    // Setup signal handler
    struct sigaction sa;
    sa.sa_flags = SA_SIGINFO;
    sa.sa_sigaction = signal_handler;
    sigemptyset(&sa.sa_mask);
    sigaction(SIGUSR1, &sa, NULL);
    
    // Register for signal notification
    struct sigevent sev;
    sev.sigev_notify = SIGEV_SIGNAL;
    sev.sigev_signo = SIGUSR1;
    
    if (mq_notify(mq_global, &sev) == -1) {
        perror("mq_notify");
        mq_close(mq_global);
        mq_unlink(QUEUE_NAME);
        return EXIT_FAILURE;
    }
    
    printf("Notification registered. Sending messages...\n");
    
    // Send messages from main thread
    mq_send(mq_global, "First message", 13, 0);
    sleep(1);
    mq_send(mq_global, "Second message", 14, 5);
    sleep(1);
    
    // Cleanup
    mq_close(mq_global);
    mq_unlink(QUEUE_NAME);
    
    return EXIT_SUCCESS;
}

Thread vs Signal Notification

SIGEV_THREAD is generally preferred: (1) No signal handler restrictions on what functions can be called; (2) No need to set up signal infrastructure; (3) Cleaner async programming model. Use SIGEV_SIGNAL when you need to wake a blocking main loop or integrate with existing signal-driven code.

Summary: POSIX Message Queues

We've covered the complete POSIX message queue API. Let's consolidate the key concepts:

Key Takeaways

•Name-based identification — Queues use /name strings, not numeric keys. Simpler to manage and debug.
•File-like API — mq_open(), mq_close(), mq_unlink() mirror file operations. Reference counting enables automatic cleanup.
•Native priority support — Per-message priority with guaranteed ordering (highest first, then FIFO).
•Fixed attributes — mq_maxmsg and mq_msgsize are set at creation and cannot be changed.
•Buffer size enforcement — Receiver's buffer must be at least mq_msgsize bytes.
•Async notification — mq_notify() enables event-driven design without polling. One-shot, must re-register.

Page Complete

You now understand POSIX message queues—the modern, standards-compliant approach to message-based IPC. Next, we'll explore message types and how to design effective message protocols for inter-process communication.

2 / 5

Loading learning content...

Operating SystemsInter-Process Communication

Message Queues

LevelIntermediate

Duration60 mins

TopicInter-Process Communication

2 / 5

POSIX Message Queues

The Modern Standard for Message Passing

POSIX message queues address many System V limitations:

File-like naming: Queues are identified by names (like /myqueue) rather than numeric keys
Reference counting: Queues are cleaned up automatically when no process has them open
Asynchronous notification: Processes can be notified when messages arrive without polling
Per-message priority: Built-in priority support with guaranteed ordering
Better portability: Part of POSIX standard, available on Linux, FreeBSD, macOS, and others

Despite these improvements, the fundamental model remains the same: message queues provide typed, structured, asynchronous communication between processes.

What You Will Master

POSIX vs System V Message Queues

Before diving into the API, let's understand the architectural differences between these two message queue implementations:

Fundamental Design Philosophy

System V IPC was designed as a monolithic kernel facility. Objects exist purely in kernel memory, identified by numeric keys, with no filesystem representation.

System V vs POSIX Message Queue Comparison
Aspect	System V	POSIX
Identification	Numeric key (`key_t`)	Name string (`/name`)
Creation	`msgget(key, flags)`	`mq_open("/name", flags, mode, attr)`
Handle Type	`int msqid`	`mqd_t` (like file descriptor)
Priority Support	Via message type workaround	Native per-message priority
Notification	None (must poll)	Async notification (signal/thread)
Cleanup	Manual `msgctl(IPC_RMID)`	Auto when last reference closes + unlink
Filesystem	None	`/dev/mqueue/` on Linux
System Limits	`/proc/sys/kernel/msg*`	`/proc/sys/fs/mqueue/*`
Link Library	None (built into libc)	`-lrt` required on Linux

When to Choose POSIX

Queue Naming Convention

POSIX message queue names follow specific rules:

Naming Requirements

Must begin with / — The name must start with a forward slash
No additional slashes — Only one leading slash; no internal slashes
Maximum length — Typically NAME_MAX (usually 255) characters including the slash
Portable characters — Alphanumeric, underscore, and hyphen are safe

// Valid names:
"/myqueue"
"/task_queue_1"
"/ipc-channel-main"

// Invalid names:
"myqueue"        // Missing leading slash
"/my/queue"      // Internal slash
""               // Empty

Filesystem Representation

On Linux, POSIX message queues appear in a virtual filesystem. To view and manage them:

# Mount the message queue filesystem (usually already mounted)
mount -t mqueue none /dev/mqueue

# List existing queues
ls -la /dev/mqueue/

# View queue contents/attributes
cat /dev/mqueue/myqueue

# Remove a queue directly
rm /dev/mqueue/myqueue

This filesystem representation is incredibly useful for debugging—you can inspect queue states, see message counts, and even remove stuck queues without writing code.

queue_naming.c
C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/stat.h>
 
// Build with: gcc queue_naming.c -o queue_naming -lrt
 
int test_queue_name(const char *name) {
    printf("Testing queue name: '%s'\n", name);
    
    mqd_t mq = mq_open(name, O_CREAT | O_RDWR, 0644, NULL);
    
    if (mq == (mqd_t)-1) {
        printf("  FAILED: %s\n\n", strerror(errno));
        return -1;
    }
    
    printf("  SUCCESS: Queue created\n");
    mq_close(mq);
    mq_unlink(name);
    printf("  Cleaned up\n\n");
    return 0;
}
 
int main() {
    printf("=== POSIX Message Queue Naming Tests ===\n\n");
    
    // Valid names
    test_queue_name("/simple");
    test_queue_name("/task_queue_1");
    test_queue_name("/my-app-channel");
    
    // Invalid names
    test_queue_name("no_leading_slash");  // EINVAL
    test_queue_name("/path/with/slashes"); // EINVAL on most systems
    test_queue_name("/");                  // EINVAL
    
    return EXIT_SUCCESS;
}

Creating and Opening Queues with mq_open()

The mq_open() function creates a new message queue or opens an existing one. Its semantics closely mirror the open() system call for files.

The mq_open() Function

#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>

mqd_t mq_open(const char *name, int oflag);
mqd_t mq_open(const char *name, int oflag, mode_t mode, 
              struct mq_attr *attr);

Parameters:

Parameter	Description
`name`	Queue name (must start with `/`)
`oflag`	Open flags (similar to `open()`)
`mode`	Permissions when creating (like file mode)
`attr`	Queue attributes (NULL for defaults)

Returns:

Message queue descriptor (mqd_t) on success
(mqd_t)-1 on error

Open Flags

The oflag parameter combines access mode and creation flags:

Access Mode (exactly one required):

O_RDONLY — Open for receiving only
O_WRONLY — Open for sending only
O_RDWR — Open for both sending and receiving

Creation Flags (optional):

O_CREAT — Create queue if it doesn't exist
O_EXCL — With O_CREAT, fail if queue exists
O_NONBLOCK — Non-blocking send/receive operations

mq_open_examples.c
C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <errno.h>
 
// Compile: gcc mq_open_examples.c -o mq_open_examples -lrt
 
#define QUEUE_NAME "/demo_queue"
 
int main() {
    mqd_t mq;
    
    // ========================================
    // Example 1: Create with default attributes
    // ========================================
    printf("=== Example 1: Create with defaults ===\n");
    mq = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, NULL);
    if (mq == (mqd_t)-1) {
        perror("mq_open (defaults)");
        return EXIT_FAILURE;
    }
    printf("Queue created with default attributes\n");
    mq_close(mq);
    mq_unlink(QUEUE_NAME);
    
    // ========================================
    // Example 2: Create with custom attributes
    // ========================================
    printf("\n=== Example 2: Create with custom attributes ===\n");
    
    struct mq_attr attr;
    attr.mq_flags = 0;          // Blocking mode
    attr.mq_maxmsg = 10;        // Max 10 messages in queue
    attr.mq_msgsize = 256;      // Max 256 bytes per message
    attr.mq_curmsgs = 0;        // Read-only (ignored on create)
    
    mq = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, &attr);
    if (mq == (mqd_t)-1) {
        perror("mq_open (custom)");
        return EXIT_FAILURE;
    }
    
    // Verify attributes
    struct mq_attr actual;
    mq_getattr(mq, &actual);
    printf("Queue created:\n");
    printf("  Max messages: %ld\n", actual.mq_maxmsg);
    printf("  Max msg size: %ld bytes\n", actual.mq_msgsize);
    
    mq_close(mq);
    mq_unlink(QUEUE_NAME);
    
    // ========================================
    // Example 3: Exclusive creation
    // ========================================
    printf("\n=== Example 3: Exclusive creation ===\n");
    
    // First creation succeeds
    mq = mq_open(QUEUE_NAME, O_CREAT | O_EXCL | O_RDWR, 0644, NULL);
    if (mq != (mqd_t)-1) {
        printf("First creation: SUCCESS\n");
        
        // Second creation with O_EXCL fails
        mqd_t mq2 = mq_open(QUEUE_NAME, O_CREAT | O_EXCL | O_RDWR, 0644, NULL);
        if (mq2 == (mqd_t)-1 && errno == EEXIST) {
            printf("Second creation: Failed (EEXIST) as expected\n");
        }
        
        mq_close(mq);
        mq_unlink(QUEUE_NAME);
    }
    
    // ========================================
    // Example 4: Open existing queue
    // ========================================
    printf("\n=== Example 4: Open existing queue ===\n");
    
    // Create first
    mq = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, NULL);
    mq_close(mq);
    
    // Open for reading only
    mqd_t reader = mq_open(QUEUE_NAME, O_RDONLY);
    if (reader != (mqd_t)-1) {
        printf("Reader opened queue successfully\n");
        mq_close(reader);
    }
    
    // Open for writing only
    mqd_t writer = mq_open(QUEUE_NAME, O_WRONLY);
    if (writer != (mqd_t)-1) {
        printf("Writer opened queue successfully\n");
        mq_close(writer);
    }
    
    mq_unlink(QUEUE_NAME);
    
    printf("\nAll examples completed\n");
    return EXIT_SUCCESS;
}

Queue Attributes Are Fixed at Creation

Queue Attributes and Configuration

POSIX message queues have configurable attributes that control their behavior and capacity.

The mq_attr Structure

struct mq_attr {
    long mq_flags;    // Queue flags (0 or O_NONBLOCK)
    long mq_maxmsg;   // Maximum number of messages
    long mq_msgsize;  // Maximum message size (bytes)
    long mq_curmsgs;  // Current messages in queue (read-only)
};

Attribute Functions

int mq_getattr(mqd_t mqdes, struct mq_attr *attr);
int mq_setattr(mqd_t mqdes, const struct mq_attr *newattr, 
               struct mq_attr *oldattr);

Note: mq_setattr() can only modify mq_flags (to toggle O_NONBLOCK). The size attributes are immutable after creation.

queue_attributes.c
C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <errno.h>
 
#define QUEUE_NAME "/attr_demo"
 
int main() {
    // Create queue with specific attributes
    struct mq_attr create_attr = {
        .mq_flags = 0,
        .mq_maxmsg = 5,      // Small queue for demo
        .mq_msgsize = 128,   // 128-byte messages
        .mq_curmsgs = 0      // Ignored on create
    };
    
    mqd_t mq = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, &create_attr);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        return EXIT_FAILURE;
    }
    
    // ========================================
    // Read current attributes
    // ========================================
    struct mq_attr current;
    mq_getattr(mq, &current);
    
    printf("=== Queue Attributes ===\n");
    printf("Flags: %ld (%s)\n", current.mq_flags,
           (current.mq_flags & O_NONBLOCK) ? "non-blocking" : "blocking");
    printf("Max messages: %ld\n", current.mq_maxmsg);
    printf("Max message size: %ld bytes\n", current.mq_msgsize);
    printf("Current messages: %ld\n", current.mq_curmsgs);
    
    // ========================================
    // Send some messages to show curmsgs change
    // ========================================
    const char *msg = "Test message";
    for (int i = 0; i < 3; i++) {
        mq_send(mq, msg, strlen(msg), 0);
    }
    
    mq_getattr(mq, &current);
    printf("\nAfter sending 3 messages:\n");
    printf("Current messages: %ld\n", current.mq_curmsgs);
    
    // ========================================
    // Toggle non-blocking mode
    // ========================================
    printf("\n=== Toggling Non-Blocking Mode ===\n");
    
    struct mq_attr new_attr, old_attr;
    new_attr.mq_flags = O_NONBLOCK;  // Only this field matters
    
    if (mq_setattr(mq, &new_attr, &old_attr) == 0) {
        printf("Previous flags: %ld\n", old_attr.mq_flags);
        
        mq_getattr(mq, &current);
        printf("New flags: %ld (non-blocking: %s)\n", 
               current.mq_flags,
               (current.mq_flags & O_NONBLOCK) ? "YES" : "NO");
    }
    
    // Demonstration: non-blocking receive when queue is empty
    char buf[128];
    // First, drain the queue
    while (mq_receive(mq, buf, sizeof(buf), NULL) != -1);
    
    // Now try to receive from empty queue
    ssize_t ret = mq_receive(mq, buf, sizeof(buf), NULL);
    if (ret == -1 && errno == EAGAIN) {
        printf("\nNon-blocking receive: EAGAIN (queue empty)\n");
    }
    
    // Clean up
    mq_close(mq);
    mq_unlink(QUEUE_NAME);
    
    return EXIT_SUCCESS;
}

System Limits

•MSGMAX — Maximum message size (default often 8192 bytes). Check: cat /proc/sys/fs/mqueue/msgsize_max
•MSGMNB — Maximum queue capacity. Check: cat /proc/sys/fs/mqueue/msg_max
•MSGMNI — Maximum queues system-wide. Check: cat /proc/sys/fs/mqueue/queues_max
•Per-user limits — Resource limits via ulimit or /etc/security/limits.conf

Sending Messages with mq_send()

The mq_send() function adds a message to a POSIX queue with an optional priority.

The mq_send() Function

#include <mqueue.h>

int mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, 
            unsigned int msg_prio);

// Timed version
int mq_timedsend(mqd_t mqdes, const char *msg_ptr, size_t msg_len,
                 unsigned int msg_prio, 
                 const struct timespec *abs_timeout);

Parameters:

Parameter	Description
`mqdes`	Queue descriptor from `mq_open()`
`msg_ptr`	Pointer to message data
`msg_len`	Size of message in bytes
`msg_prio`	Priority (0 = lowest, typically max ~32767)

Returns:

0 on success
-1 on error

Priority Semantics

Unlike System V (where priority is encoded in message type), POSIX provides native priority support:

Higher numeric values = higher priority
Messages are ordered by priority first, then FIFO within priority
Priority 0 is the lowest priority
Maximum priority is implementation-defined (sysconf(_SC_MQ_PRIO_MAX))

mq_send_demo.c
C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <fcntl.h>
#include <errno.h>
#include <time.h>
 
#define QUEUE_NAME "/priority_demo"
#define MAX_MSG_SIZE 256
 
int main() {
    // Create queue
    struct mq_attr attr = {
        .mq_flags = 0,
        .mq_maxmsg = 10,
        .mq_msgsize = MAX_MSG_SIZE,
        .mq_curmsgs = 0
    };
    
    mqd_t mq = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, &attr);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        return EXIT_FAILURE;
    }
    
    // ========================================
    // Send messages with different priorities
    // ========================================
    printf("=== Sending Messages ===\n");
    
    // Send in non-priority order
    mq_send(mq, "Low priority message", 20, 1);
    printf("Sent: priority=1 (low)\n");
    
    mq_send(mq, "Medium priority message", 23, 5);
    printf("Sent: priority=5 (medium)\n");
    
    mq_send(mq, "High priority message", 21, 10);
    printf("Sent: priority=10 (high)\n");
    
    mq_send(mq, "Another low priority", 20, 1);
    printf("Sent: priority=1 (low, second)\n");
    
    // ========================================
    // Receive messages (will come in priority order)
    // ========================================
    printf("\n=== Receiving Messages ===\n");
    
    char buffer[MAX_MSG_SIZE];
    unsigned int priority;
    ssize_t bytes;
    
    while ((bytes = mq_receive(mq, buffer, MAX_MSG_SIZE, &priority)) > 0) {
        buffer[bytes] = '\0';
        printf("Received (prio=%u): %s\n", priority, buffer);
    }
    
    // ========================================
    // Timed send demonstration
    // ========================================
    printf("\n=== Timed Send ===\n");
    
    // Fill the queue
    for (int i = 0; i < 10; i++) {
        mq_send(mq, "filler", 6, 0);
    }
    printf("Queue is now full (10 messages)\n");
    
    // Try timed send with 1-second timeout
    struct timespec timeout;
    clock_gettime(CLOCK_REALTIME, &timeout);
    timeout.tv_sec += 1;  // 1 second from now
    
    printf("Attempting timed send (1 second timeout)...\n");
    if (mq_timedsend(mq, "overflow", 8, 0, &timeout) == -1) {
        if (errno == ETIMEDOUT) {
            printf("Send timed out (queue still full)\n");
        } else {
            perror("mq_timedsend");
        }
    }
    
    // Clean up
    mq_close(mq);
    mq_unlink(QUEUE_NAME);
    
    return EXIT_SUCCESS;
}

Message Size Requirement

Receiving Messages with mq_receive()

The mq_receive() function retrieves the highest-priority message from a queue.

The mq_receive() Function

#include <mqueue.h>

ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len,
                   unsigned int *msg_prio);

// Timed version
ssize_t mq_timedreceive(mqd_t mqdes, char *msg_ptr, size_t msg_len,
                        unsigned int *msg_prio,
                        const struct timespec *abs_timeout);

Parameters:

Parameter	Description
`mqdes`	Queue descriptor
`msg_ptr`	Buffer to receive message
`msg_len`	Buffer size (must be ≥ `mq_msgsize`)
`msg_prio`	Receives the message's priority (can be NULL)

Returns:

Number of bytes in the received message on success
-1 on error

Critical Buffer Size Rule

mq_receive_patterns.c
C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <fcntl.h>
#include <errno.h>
#include <time.h>
 
#define QUEUE_NAME "/receive_demo"
 
int main() {
    struct mq_attr attr = {
        .mq_maxmsg = 10,
        .mq_msgsize = 256
    };
    
    mqd_t mq = mq_open(QUEUE_NAME, O_CREAT | O_RDWR, 0644, &attr);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        return EXIT_FAILURE;
    }
    
    // Pre-populate with messages
    mq_send(mq, "Message A", 9, 5);
    mq_send(mq, "Message B", 9, 10);
    mq_send(mq, "Message C", 9, 5);
    
    // ========================================
    // Pattern 1: Basic receive with priority
    // ========================================
    printf("=== Pattern 1: Basic Receive ===\n");
    
    char buffer[256];
    unsigned int priority;
    ssize_t len;
    
    len = mq_receive(mq, buffer, sizeof(buffer), &priority);
    if (len > 0) {
        buffer[len] = '\0';
        printf("Received: '%s' (priority: %u, len: %zd)\n", 
               buffer, priority, len);
    }
    
    // ========================================
    // Pattern 2: Receive without caring about priority
    // ========================================
    printf("\n=== Pattern 2: Ignore Priority ===\n");
    
    len = mq_receive(mq, buffer, sizeof(buffer), NULL);  // NULL for priority
    if (len > 0) {
        buffer[len] = '\0';
        printf("Received: '%s' (len: %zd)\n", buffer, len);
    }
    
    // ========================================
    // Pattern 3: Non-blocking receive
    // ========================================
    printf("\n=== Pattern 3: Non-Blocking ===\n");
    
    // Set non-blocking mode
    struct mq_attr new_attr = { .mq_flags = O_NONBLOCK };
    mq_setattr(mq, &new_attr, NULL);
    
    // Drain remaining messages
    while ((len = mq_receive(mq, buffer, sizeof(buffer), NULL)) > 0) {
        buffer[len] = '\0';
        printf("Received: '%s'\n", buffer);
    }
    
    // Try to receive from empty queue
    len = mq_receive(mq, buffer, sizeof(buffer), NULL);
    if (len == -1 && errno == EAGAIN) {
        printf("Queue empty (EAGAIN)\n");
    }
    
    // ========================================
    // Pattern 4: Timed receive
    // ========================================
    printf("\n=== Pattern 4: Timed Receive ===\n");
    
    // Reset to blocking mode for timed receive
    new_attr.mq_flags = 0;
    mq_setattr(mq, &new_attr, NULL);
    
    struct timespec timeout;
    clock_gettime(CLOCK_REALTIME, &timeout);
    timeout.tv_sec += 2;  // 2-second timeout
    
    printf("Waiting for message (2 second timeout)...\n");
    len = mq_timedreceive(mq, buffer, sizeof(buffer), NULL, &timeout);
    
    if (len == -1) {
        if (errno == ETIMEDOUT) {
            printf("Timed out waiting for message\n");
        } else {
            perror("mq_timedreceive");
        }
    }
    
    mq_close(mq);
    mq_unlink(QUEUE_NAME);
    
    return EXIT_SUCCESS;
}

Asynchronous Message Notification

One of the most powerful features of POSIX message queues is asynchronous notification—the ability to be notified when a message arrives without polling.

The mq_notify() Function

#include <mqueue.h>

int mq_notify(mqd_t mqdes, const struct sigevent *sevp);

The sigevent structure specifies how notification should occur:

struct sigevent {
    int          sigev_notify;    // Notification method
    int          sigev_signo;     // Signal number (for SIGEV_SIGNAL)
    union sigval sigev_value;     // Data passed to handler
    void       (*sigev_notify_function)(union sigval);  // Thread function
    pthread_attr_t *sigev_notify_attributes;  // Thread attributes
};

Notification Methods

Method	Description
`SIGEV_NONE`	Disable notification
`SIGEV_SIGNAL`	Deliver a signal when message arrives
`SIGEV_THREAD`	Spawn a new thread to handle arrival

Critical Semantics

One-shot: Notification is delivered only once, then automatically unregistered. You must re-register after each notification.
Empty-to-non-empty transition: Notification occurs only when the queue goes from empty to having a message—not for every message.
Single registration: Only one process can register for notification at a time.

mq_notify_demo.c
C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mqueue.h>
#include <signal.h>
#include <pthread.h>
#include <fcntl.h>
#include <unistd.h>
 
#define QUEUE_NAME "/notify_demo"
#define MAX_MSG_SIZE 256
 
static mqd_t mq_global;
 
// ========================================
// Signal-based notification handler
// ========================================
void signal_handler(int signo, siginfo_t *info, void *context) {
    char buffer[MAX_MSG_SIZE];
    unsigned int priority;
    ssize_t len;
    
    printf("\n[Signal Handler] Received SIGUSR1\n");
    
    // Read message(s)
    while ((len = mq_receive(mq_global, buffer, MAX_MSG_SIZE, &priority)) > 0) {
        buffer[len] = '\0';
        printf("[Signal Handler] Message: '%s' (prio=%u)\n", buffer, priority);
    }
    
    // Re-register for notification (one-shot!)
    struct sigevent sev;
    sev.sigev_notify = SIGEV_SIGNAL;
    sev.sigev_signo = SIGUSR1;
    mq_notify(mq_global, &sev);
}
 
// ========================================
// Thread-based notification handler
// ========================================
void thread_handler(union sigval sv) {
    mqd_t mq = *((mqd_t *)sv.sival_ptr);
    char buffer[MAX_MSG_SIZE];
    ssize_t len;
    
    printf("\n[Thread Handler] Notification received\n");
    
    while ((len = mq_receive(mq, buffer, MAX_MSG_SIZE, NULL)) > 0) {
        buffer[len] = '\0';
        printf("[Thread Handler] Message: '%s'\n", buffer);
    }
    
    // Re-register (must pass sival_ptr for context)
    struct sigevent sev;
    sev.sigev_notify = SIGEV_THREAD;
    sev.sigev_notify_function = thread_handler;
    sev.sigev_value.sival_ptr = sv.sival_ptr;
    sev.sigev_notify_attributes = NULL;
    mq_notify(mq, &sev);
}
 
int main() {
    struct mq_attr attr = {
        .mq_maxmsg = 10,
        .mq_msgsize = MAX_MSG_SIZE
    };
    
    mq_global = mq_open(QUEUE_NAME, O_CREAT | O_RDWR | O_NONBLOCK, 
                        0644, &attr);
    if (mq_global == (mqd_t)-1) {
        perror("mq_open");
        return EXIT_FAILURE;
    }
    
    printf("=== Signal-Based Notification Demo ===\n");
    
    // Setup signal handler
    struct sigaction sa;
    sa.sa_flags = SA_SIGINFO;
    sa.sa_sigaction = signal_handler;
    sigemptyset(&sa.sa_mask);
    sigaction(SIGUSR1, &sa, NULL);
    
    // Register for signal notification
    struct sigevent sev;
    sev.sigev_notify = SIGEV_SIGNAL;
    sev.sigev_signo = SIGUSR1;
    
    if (mq_notify(mq_global, &sev) == -1) {
        perror("mq_notify");
        mq_close(mq_global);
        mq_unlink(QUEUE_NAME);
        return EXIT_FAILURE;
    }
    
    printf("Notification registered. Sending messages...\n");
    
    // Send messages from main thread
    mq_send(mq_global, "First message", 13, 0);
    sleep(1);
    mq_send(mq_global, "Second message", 14, 5);
    sleep(1);
    
    // Cleanup
    mq_close(mq_global);
    mq_unlink(QUEUE_NAME);
    
    return EXIT_SUCCESS;
}

Thread vs Signal Notification

Summary: POSIX Message Queues

We've covered the complete POSIX message queue API. Let's consolidate the key concepts:

Key Takeaways

•Name-based identification — Queues use /name strings, not numeric keys. Simpler to manage and debug.
•File-like API — mq_open(), mq_close(), mq_unlink() mirror file operations. Reference counting enables automatic cleanup.
•Native priority support — Per-message priority with guaranteed ordering (highest first, then FIFO).
•Fixed attributes — mq_maxmsg and mq_msgsize are set at creation and cannot be changed.
•Buffer size enforcement — Receiver's buffer must be at least mq_msgsize bytes.
•Async notification — mq_notify() enables event-driven design without polling. One-shot, must re-register.

Page Complete

2 / 5