Signal-Based MQ Notification
SIGEV_SIGNAL | sigsuspend() | Nonblocking Queue Drain | Chapter 52.6.1 | File 2 of 3
Series Navigation: Index File 1: API & sigevent File 2: Signal Notification File 3: Thread Notification
Overview
Signal-based notification uses SIGEV_SIGNAL with mq_notify(). When a message arrives on a previously empty queue, the kernel sends a signal to the registered process. The process catches this signal and reacts accordingly.
This approach is suitable when your program already uses signal handling. The key challenge is handling the race condition between signals and the main loop safely — and this tutorial explains exactly how to do that.
Design Requirements for Signal-Based Notification
| Requirement | Why It’s Needed | How to Implement |
|---|---|---|
| Block the notification signal initially | Prevent the signal from firing while the program isn’t ready to handle it | sigprocmask(SIG_BLOCK, ...) |
| Use sigsuspend() not pause() to wait | Prevents a race condition where signal arrives between check and sleep | sigsuspend(&emptyMask) |
| Re-register BEFORE draining the queue | Avoids missing a message that arrives between drain and re-register | Call mq_notify() first, then mq_receive() |
| Open queue with O_NONBLOCK | Drain the queue fully without risk of blocking on empty queue | mq_open(..., O_RDONLY | O_NONBLOCK) |
| Use a loop to drain all messages | Multiple messages may arrive before handler runs — process all of them | while (mq_receive(...) >= 0) |
The Race Condition: Why pause() Is Wrong
Consider this flawed approach:
/* WRONG — race condition! */
for (;;) {
/* Signal arrives HERE — we miss it! */
pause(); /* Will block forever if signal already delivered */
}If a message arrives and the signal fires between the loop top and pause(), the signal is already handled before pause() is called. pause() then blocks forever waiting for a signal that has already come and gone.
The correct fix is to block the signal with sigprocmask() and use sigsuspend(), which atomically unblocks the signal and sleeps. This eliminates the window between check and sleep.
sigsuspend() is atomic: It atomically replaces the process’s signal mask with the provided mask and suspends the process. When a signal is caught, the original mask is restored. There is no gap where a signal can be lost.
Program Flow Diagram — Signal-Based Notification
| Step 1 | mq_open(name, O_RDONLY | O_NONBLOCK) — open queue in nonblocking mode |
| Step 2 | mq_getattr() — get mq_msgsize to allocate receive buffer |
| Step 3 | sigprocmask(SIG_BLOCK, ...) — block the notification signal (SIGUSR1) |
| Step 4 | sigaction() — install signal handler for SIGUSR1 |
| Step 5 | mq_notify(mqd, &sev) — initial registration |
| Step 6 (loop) | sigsuspend(&emptyMask) — atomically unblock signal and wait |
| Step 7 (loop) | mq_notify() — re-register BEFORE reading messages |
| Step 8 (loop) | while (mq_receive() >= 0) — drain all messages from queue |
| Step 9 (loop) | errno == EAGAIN — queue is empty, go back to sigsuspend() |
Complete Code Example — Signal-Based Notification
This is a complete, working example based on the TLPI reference implementation:
/* mq_notify_sig.c — POSIX MQ notification via signal (SIGEV_SIGNAL)
Compile: gcc mq_notify_sig.c -lrt -o mq_notify_sig
Usage : ./mq_notify_sig /myqueue
*/
#include <signal.h>
#include <mqueue.h>
#include <fcntl.h> /* O_NONBLOCK */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#define NOTIFY_SIG SIGUSR1
/* Signal handler — only purpose is to interrupt sigsuspend() */
static void handler(int sig)
{
/* Intentionally empty.
The real work is done in the main loop after sigsuspend() returns.
Keeping the handler minimal is best practice — avoid calling
non-async-signal-safe functions inside a signal handler. */
}
int main(int argc, char *argv[])
{
struct sigevent sev;
mqd_t mqd;
struct mq_attr attr;
void *buffer;
ssize_t numRead;
sigset_t blockMask, emptyMask;
struct sigaction sa;
if (argc != 2) {
fprintf(stderr, "Usage: %s mq-name\n", argv[0]);
return 1;
}
/* STEP 1: Open queue in nonblocking mode for reading */
mqd = mq_open(argv[1], O_RDONLY | O_NONBLOCK);
if (mqd == (mqd_t)-1) {
perror("mq_open");
return 1;
}
/* STEP 2: Get queue attributes — we need mq_msgsize for buffer */
if (mq_getattr(mqd, &attr) == -1) {
perror("mq_getattr");
return 1;
}
/* STEP 2 cont: Allocate receive buffer of maximum message size */
buffer = malloc(attr.mq_msgsize);
if (buffer == NULL) {
perror("malloc");
return 1;
}
/* STEP 3: Block the notification signal so we control when it's received */
sigemptyset(&blockMask);
sigaddset(&blockMask, NOTIFY_SIG);
if (sigprocmask(SIG_BLOCK, &blockMask, NULL) == -1) {
perror("sigprocmask");
return 1;
}
/* STEP 4: Install signal handler for the notification signal */
memset(&sa, 0, sizeof(sa));
sa.sa_handler = handler;
sigemptyset(&sa.sa_mask);
sa.sa_flags = 0;
if (sigaction(NOTIFY_SIG, &sa, NULL) == -1) {
perror("sigaction");
return 1;
}
/* STEP 5: Register for notification — first time */
sev.sigev_notify = SIGEV_SIGNAL;
sev.sigev_signo = NOTIFY_SIG;
if (mq_notify(mqd, &sev) == -1) {
perror("mq_notify");
return 1;
}
/* Empty mask — used with sigsuspend() to unblock ALL signals
while waiting. Our notification signal (blocked normally) will
therefore be deliverable during sigsuspend(). */
sigemptyset(&emptyMask);
printf("Waiting for messages on %s ...\n", argv[1]);
for (;;) {
/* STEP 6: Atomically unblock notification signal and sleep.
Returns when NOTIFY_SIG is caught. */
sigsuspend(&emptyMask);
/* STEP 7: Re-register BEFORE draining the queue.
Reason: if a new message arrives between drain-complete and
mq_notify(), we'd miss the notification for it. By registering
first, we guarantee we catch any new message that arrives
after the queue becomes empty again. */
if (mq_notify(mqd, &sev) == -1) {
perror("mq_notify reregister");
return 1;
}
/* STEP 8: Drain all available messages from the queue.
O_NONBLOCK ensures mq_receive() returns EAGAIN when empty
instead of blocking. */
while ((numRead = mq_receive(mqd, buffer,
attr.mq_msgsize, NULL)) >= 0) {
printf("Received message: %ld bytes\n", (long)numRead);
/* Process buffer here — it contains the message data */
}
/* STEP 9: EAGAIN means queue is now empty — expected end of drain */
if (errno != EAGAIN) {
perror("mq_receive unexpected error");
return 1;
}
/* Loop back to sigsuspend() to wait for next notification */
}
/* Unreachable, but good practice */
free(buffer);
mq_close(mqd);
return 0;
}How to Create the Queue and Send a Test Message
/* mq_sender.c — Create queue and send a test message */
#include <mqueue.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <stdio.h>
#include <string.h>
int main(void)
{
mqd_t mqd;
struct mq_attr attr;
const char *msg = "Hello from sender!";
/* Queue attributes */
attr.mq_flags = 0;
attr.mq_maxmsg = 10;
attr.mq_msgsize = 256;
attr.mq_curmsgs = 0;
/* Create (or open) the queue */
mqd = mq_open("/myqueue", O_CREAT | O_WRONLY, 0644, &attr);
if (mqd == (mqd_t)-1) { perror("mq_open"); return 1; }
/* Send the message with priority 1 */
if (mq_send(mqd, msg, strlen(msg) + 1, 1) == -1) {
perror("mq_send");
return 1;
}
printf("Message sent: %s\n", msg);
mq_close(mqd);
return 0;
}
/* Build and run:
Terminal 1: gcc mq_notify_sig.c -lrt -o receiver
./receiver /myqueue
Terminal 2: gcc mq_sender.c -lrt -o sender
./sender
*/Code Example 2 — Using a Realtime Signal with siginfo_t Data
Using a realtime signal (SIGRTMIN to SIGRTMAX) allows you to receive extra data (si_pid, si_uid, si_value) in the signal handler.
/* Using realtime signal with SA_SIGINFO to get sender information */
#include <signal.h>
#include <mqueue.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#define NOTIFY_SIG SIGRTMIN /* Use first realtime signal */
static volatile int notified = 0;
static pid_t sender_pid;
static uid_t sender_uid;
/* SA_SIGINFO handler receives siginfo_t with sender details */
static void rt_handler(int sig, siginfo_t *si, void *ucontext)
{
/* si_code == SI_MESGQ identifies this as MQ notification */
if (si->si_code == SI_MESGQ) {
sender_pid = si->si_pid;
sender_uid = si->si_uid;
notified = 1;
/* si->si_value.sival_int would contain sigev_value if set */
}
}
int main(int argc, char *argv[])
{
struct sigevent sev;
mqd_t mqd;
struct mq_attr attr;
void *buffer;
ssize_t numRead;
sigset_t blockMask, emptyMask;
struct sigaction sa;
if (argc != 2) { fprintf(stderr, "Usage: %s mq-name\n", argv[0]); return 1; }
mqd = mq_open(argv[1], O_RDONLY | O_NONBLOCK);
if (mqd == (mqd_t)-1) { perror("mq_open"); return 1; }
mq_getattr(mqd, &attr);
buffer = malloc(attr.mq_msgsize);
/* Block the realtime notification signal */
sigemptyset(&blockMask);
sigaddset(&blockMask, NOTIFY_SIG);
sigprocmask(SIG_BLOCK, &blockMask, NULL);
/* Use SA_SIGINFO to receive siginfo_t in the handler */
memset(&sa, 0, sizeof(sa));
sa.sa_sigaction = rt_handler;
sa.sa_flags = SA_SIGINFO;
sigemptyset(&sa.sa_mask);
sigaction(NOTIFY_SIG, &sa, NULL);
/* Register: use realtime signal, pass integer data via sigev_value */
sev.sigev_notify = SIGEV_SIGNAL;
sev.sigev_signo = NOTIFY_SIG;
sev.sigev_value.sival_int = 99; /* Custom data — accessible in handler */
mq_notify(mqd, &sev);
sigemptyset(&emptyMask);
printf("Waiting (realtime signal mode)...\n");
for (;;) {
sigsuspend(&emptyMask);
if (notified) {
printf("Notified by process PID=%d UID=%d\n",
(int)sender_pid, (int)sender_uid);
notified = 0;
}
mq_notify(mqd, &sev); /* Re-register */
while ((numRead = mq_receive(mqd, buffer,
attr.mq_msgsize, NULL)) >= 0)
printf("Read %ld bytes\n", (long)numRead);
}
free(buffer);
mq_close(mqd);
return 0;
}Critical Pitfall — Order of Re-register and Drain
WRONG Order — Will Cause Permanent Deadlock:
/* 1. Drain queue first */
while (mq_receive(mqd, buf, size, NULL) >= 0) { ... }
/* Queue is now empty */
/* 2. THEN re-register */
/* RACE: if a message arrived between step 1 and step 2,
queue is already non-empty. mq_notify() registers, but
no further notification will fire because the trigger
(empty→nonempty) already happened. Program hangs here. */
mq_notify(mqd, &sev); /* Too late! */CORRECT Order — Re-register BEFORE Drain:
/* 1. Re-register FIRST */
mq_notify(mqd, &sev);
/* 2. THEN drain queue */
while (mq_receive(mqd, buf, size, NULL) >= 0) { ... }
/* Now if a new message arrives during drain, the registration
is already in place and the next notification WILL fire. */Analogy — Edge-Triggered I/O:
This is exactly like edge-triggered mode in
This is exactly like edge-triggered mode in
epoll (EPOLLET). You get notified only on the transition (low→high), not while the condition is true. You must drain the source completely and re-arm before the next notification can fire. Failing to do either causes missed events and program hangs.Interview Questions & Answers
Q1. Why is pause() unsafe to use with mq_notify() signal notification?
pause() has a race condition: if the signal is delivered after the previous iteration’s work but before pause() is called, the signal is consumed and pause() blocks forever. sigsuspend() solves this by atomically unblocking the signal and sleeping — there is no window for the signal to be missed.Q2. Why must the message queue be opened with O_NONBLOCK for notification-based reading?
Because we loop reading all messages until the queue is empty. Without
O_NONBLOCK, the final mq_receive() call (when the queue is empty) would block the process instead of returning EAGAIN. With O_NONBLOCK, we get an immediate EAGAIN which tells us the queue is drained.Q3. Why do we re-register for notification BEFORE draining the queue?
Consider this sequence: (1) drain queue — queue becomes empty; (2) new message arrives — triggers empty→nonempty; (3) we call mq_notify(). At this point, the queue is already nonempty, so no further notification will fire. We are permanently blocked. By re-registering first, we ensure any message arriving during the drain will trigger a future notification.
Q4. What does errno == EAGAIN mean after mq_receive() with O_NONBLOCK?
It means the queue is currently empty. This is the expected termination condition for the drain loop — it is not an error. The program should loop back and wait for the next notification. Any other non-zero errno value is an unexpected error.
Q5. Why do we block the notification signal before registering with mq_notify()?
If the signal were not blocked and a message arrived immediately after
mq_notify() but before the main loop’s sigsuspend(), the signal handler would run in an unexpected context. Blocking the signal ensures it is only delivered at our controlled point (sigsuspend()), giving consistent, safe behavior.Q6. Why should the signal handler be kept minimal (nearly empty)?
Inside a signal handler, only async-signal-safe functions (like
write(), not printf()) are safe to call. Calling non-async-signal-safe functions (malloc, printf, mq_receive) from within a signal handler risks deadlocks and undefined behavior. The pattern here is: let the handler just interrupt sigsuspend(), then do the real work in the normal (non-signal-handler) context of the loop body.Q7. What is si_code == SI_MESGQ and when is it set?
SI_MESGQ is a value set in the si_code field of siginfo_t when a realtime signal is delivered as a result of a POSIX message queue notification. It allows the signal handler to distinguish this notification from other sources of the same signal number.Q8. Why would you use a realtime signal instead of SIGUSR1 for mq_notify()?
Realtime signals (SIGRTMIN–SIGRTMAX) support data delivery via
sigev_value (accessible as si_value in the handler) and also provide sender info (si_pid, si_uid). Standard signals like SIGUSR1 cannot carry this extra data. If you need to know which queue triggered the notification or pass a pointer/descriptor to the handler, a realtime signal is the right choice.