mq_notify()
Advanced
52 – TLPI
What is mq_notify()?
All the receive functions we have looked at so far are synchronous: your process either blocks waiting for a message or actively polls the queue in a loop. Neither is efficient for processes that have other work to do.
mq_notify() solves this with asynchronous notification. A process registers itself with a queue. When a message arrives on an empty queue (i.e., the queue transitions from empty to non-empty), the kernel delivers a notification — either a signal or a thread invocation. The process does not have to watch the queue at all; it gets poked when something interesting happens.
Analogy: instead of standing in front of a letterbox checking every minute, you ask the postman to ring the bell the moment a letter arrives.
This feature is similar to the notification facility used by POSIX timers (Section 23.6 of TLPI). Both originated in POSIX.1b.
#include <mqueue.h>
int mq_notify(mqd_t mqdes, const struct sigevent *notification);
/* Returns: 0 on success, -1 on error */
/* Link with: -lrt */
| Parameter | Meaning |
|---|---|
mqdes |
Queue descriptor to watch |
notification |
Pointer to sigevent struct describing how to notify, or NULL to deregister |
The sigevent structure tells the kernel how to wake up your process when a notification is triggered.
/* Simplified view of struct sigevent (relevant fields only) */
union sigval {
int sival_int; /* integer value to pass with notification */
void *sival_ptr; /* pointer value to pass with notification */
};
struct sigevent {
int sigev_notify; /* notification method */
int sigev_signo; /* signal number (SIGEV_SIGNAL) */
union sigval sigev_value; /* data passed with notification */
void (*sigev_notify_function)(union sigval); /* thread func */
pthread_attr_t *sigev_notify_attributes; /* thread attrs */
};
| sigev_notify value | Effect |
|---|---|
SIGEV_NONE |
Register interest but deliver no actual notification (used for testing) |
SIGEV_SIGNAL |
Deliver a signal (specified by sigev_signo) to the process |
SIGEV_THREAD |
Create a new thread and invoke sigev_notify_function in it |
mq_notify(mqd, &sev). Process is now the “registered process” for this queue.mq_receive() (usually with O_NONBLOCK to drain all messages).mq_notify() again at the end of the handler. One shot = one notification.EBUSY.mq_notify() again to receive the next notification.notification = NULL to deregister (cancel) an existing registration.The simplest form: ask the kernel to deliver SIGUSR1 when a message arrives. The signal handler drains the queue.
/* mq_notify_signal.c
* Register for SIGUSR1 notification.
* Compile: gcc mq_notify_signal.c -o mq_notify_signal -lrt
* Run in background: ./mq_notify_signal /myqueue &
* Then send a message: (another terminal) ./pmsg_send /myqueue "hello" 5
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>
#include <errno.h>
#include <mqueue.h>
#include <fcntl.h>
static mqd_t g_mqd; /* global so signal handler can use it */
static size_t g_msgsize;
/* Signal handler: drains the queue, then re-registers */
static void handler(int sig)
{
/* We use async-signal-safe calls only here */
struct sigevent sev;
sev.sigev_notify = SIGEV_SIGNAL;
sev.sigev_signo = SIGUSR1;
/* Re-register BEFORE reading messages to avoid race condition */
if (mq_notify(g_mqd, &sev) == -1) {
/* Can't use perror() in signal handler, just note it */
write(STDERR_FILENO, "mq_notify re-reg failed\n", 24);
return;
}
/* Drain all available messages */
char buf[256];
unsigned int prio;
ssize_t n;
while ((n = mq_receive(g_mqd, buf, g_msgsize, &prio)) > 0) {
/* write() is async-signal-safe; printf is not */
char out[300];
int len = snprintf(out, sizeof(out),
"[handler] prio=%u bytes=%zd msg=%.*s\n",
prio, n, (int)n, buf);
write(STDOUT_FILENO, out, len);
}
/* EAGAIN means queue is empty now — normal exit from loop */
}
int main(int argc, char *argv[])
{
if (argc != 2) {
fprintf(stderr, "Usage: %s /queue-name\n", argv[0]);
exit(EXIT_FAILURE);
}
/* Open queue in non-blocking mode (for drain loop in handler) */
g_mqd = mq_open(argv[1], O_RDONLY | O_NONBLOCK);
if (g_mqd == (mqd_t)-1) { perror("mq_open"); exit(EXIT_FAILURE); }
struct mq_attr attr;
mq_getattr(g_mqd, &attr);
g_msgsize = attr.mq_msgsize;
/* Install signal handler for SIGUSR1 */
struct sigaction sa;
sigemptyset(&sa.sa_mask);
sa.sa_flags = 0;
sa.sa_handler = handler;
if (sigaction(SIGUSR1, &sa, NULL) == -1) {
perror("sigaction"); exit(EXIT_FAILURE);
}
/* Register for notification via SIGUSR1 */
struct sigevent sev;
sev.sigev_notify = SIGEV_SIGNAL;
sev.sigev_signo = SIGUSR1;
if (mq_notify(g_mqd, &sev) == -1) {
perror("mq_notify"); exit(EXIT_FAILURE);
}
printf("Registered for SIGUSR1 notification on %s\n", argv[1]);
printf("Waiting for messages... (Ctrl-C to quit)\n");
/* Main loop: do other work here; signal handler fires asynchronously */
for (;;) {
pause(); /* sleep until any signal arrives */
}
mq_close(g_mqd);
return 0;
}
Instead of a signal, the kernel spawns a new thread and calls your function directly. This is cleaner than a signal handler — you can use any library function, not just async-signal-safe ones.
/* mq_notify_thread.c
* Notification via a new thread.
* Compile: gcc mq_notify_thread.c -o mq_notify_thread -lrt -lpthread
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>
#include <errno.h>
#include <pthread.h>
#include <mqueue.h>
#include <fcntl.h>
static mqd_t g_mqd;
static size_t g_msgsize;
/* This function runs in a new thread created by the kernel */
static void notification_thread(union sigval sv)
{
(void)sv; /* sv.sival_ptr could carry custom data */
/* Re-register FIRST to avoid missing next notification */
struct sigevent sev;
memset(&sev, 0, sizeof(sev));
sev.sigev_notify = SIGEV_THREAD;
sev.sigev_notify_function = notification_thread;
sev.sigev_value.sival_ptr = NULL;
if (mq_notify(g_mqd, &sev) == -1)
perror("mq_notify re-register");
/* Drain all messages — printf is safe here (not a signal handler) */
char *buf = malloc(g_msgsize);
unsigned int prio;
ssize_t n;
while ((n = mq_receive(g_mqd, buf, g_msgsize, &prio)) > 0) {
printf("[thread] Received %zd bytes, priority=%u: %.*s\n",
n, prio, (int)n, buf);
}
free(buf);
/* Thread exits automatically when function returns */
}
int main(int argc, char *argv[])
{
if (argc != 2) {
fprintf(stderr, "Usage: %s /queue-name\n", argv[0]);
exit(EXIT_FAILURE);
}
g_mqd = mq_open(argv[1], O_RDONLY | O_NONBLOCK);
if (g_mqd == (mqd_t)-1) { perror("mq_open"); exit(EXIT_FAILURE); }
struct mq_attr attr;
mq_getattr(g_mqd, &attr);
g_msgsize = attr.mq_msgsize;
/* Set up thread notification */
struct sigevent sev;
memset(&sev, 0, sizeof(sev));
sev.sigev_notify = SIGEV_THREAD;
sev.sigev_notify_function = notification_thread;
sev.sigev_value.sival_ptr = NULL; /* no extra data for now */
sev.sigev_notify_attributes = NULL; /* use default thread attributes */
if (mq_notify(g_mqd, &sev) == -1) {
perror("mq_notify"); exit(EXIT_FAILURE);
}
printf("Registered for SIGEV_THREAD notification on %s\n", argv[1]);
printf("Main thread is free to do other work...\n");
/* Main thread does other work; notification_thread fires on message */
for (int i = 0; i < 30; i++) {
sleep(1);
printf("[main] tick %d\n", i + 1);
}
mq_close(g_mqd);
return 0;
}
Pass NULL as the second argument to cancel a previously registered notification.
/* deregister_notify.c
* Compile: gcc deregister_notify.c -o deregister_notify -lrt
*/
#include <stdio.h>
#include <stdlib.h>
#include <mqueue.h>
#include <fcntl.h>
#include <signal.h>
int main(int argc, char *argv[])
{
if (argc != 2) {
fprintf(stderr, "Usage: %s /queue-name\n", argv[0]);
exit(EXIT_FAILURE);
}
mqd_t mqd = mq_open(argv[1], O_RDONLY);
if (mqd == (mqd_t)-1) { perror("mq_open"); exit(EXIT_FAILURE); }
/* Register for SIGUSR1 first */
struct sigevent sev;
sev.sigev_notify = SIGEV_SIGNAL;
sev.sigev_signo = SIGUSR1;
if (mq_notify(mqd, &sev) == -1) {
perror("mq_notify register"); exit(EXIT_FAILURE);
}
printf("Registered for SIGUSR1.\n");
/* ... do some work ... */
/* Now deregister — pass NULL to cancel */
if (mq_notify(mqd, NULL) == -1) {
perror("mq_notify deregister"); exit(EXIT_FAILURE);
}
printf("Deregistered. No more notifications will be delivered.\n");
mq_close(mqd);
return 0;
}
The sigev_value union lets you pass a pointer or integer to the notification thread. This is useful for handing off context (e.g., the queue descriptor itself) without using a global variable.
/* mq_notify_sigval.c — pass mqd and msgsize via sival_ptr
* Compile: gcc mq_notify_sigval.c -o mq_notify_sigval -lrt -lpthread
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <mqueue.h>
#include <fcntl.h>
/* Bundle queue info into a struct to pass via sigval */
typedef struct {
mqd_t mqd;
size_t msgsize;
} QueueCtx;
static void notify_fn(union sigval sv)
{
QueueCtx *ctx = (QueueCtx *)sv.sival_ptr;
/* Re-register */
struct sigevent sev;
memset(&sev, 0, sizeof(sev));
sev.sigev_notify = SIGEV_THREAD;
sev.sigev_notify_function = notify_fn;
sev.sigev_value.sival_ptr = ctx; /* pass same context again */
mq_notify(ctx->mqd, &sev);
/* Read messages */
char *buf = malloc(ctx->msgsize);
unsigned int prio;
ssize_t n;
while ((n = mq_receive(ctx->mqd, buf, ctx->msgsize, &prio)) > 0)
printf("[notify_fn] prio=%u: %.*s\n", prio, (int)n, buf);
free(buf);
}
int main(int argc, char *argv[])
{
if (argc != 2) {
fprintf(stderr, "Usage: %s /queue-name\n", argv[0]);
exit(EXIT_FAILURE);
}
QueueCtx *ctx = malloc(sizeof(QueueCtx));
ctx->mqd = mq_open(argv[1], O_RDONLY | O_NONBLOCK);
if (ctx->mqd == (mqd_t)-1) { perror("mq_open"); exit(EXIT_FAILURE); }
struct mq_attr attr;
mq_getattr(ctx->mqd, &attr);
ctx->msgsize = attr.mq_msgsize;
struct sigevent sev;
memset(&sev, 0, sizeof(sev));
sev.sigev_notify = SIGEV_THREAD;
sev.sigev_notify_function = notify_fn;
sev.sigev_value.sival_ptr = ctx; /* pass our context */
if (mq_notify(ctx->mqd, &sev) == -1) {
perror("mq_notify"); exit(EXIT_FAILURE);
}
printf("Waiting for messages (30s)...\n");
sleep(30);
mq_close(ctx->mqd);
free(ctx);
return 0;
}
A common mistake: re-registering after reading messages. Between draining the last message and re-registering, new messages can arrive silently (the queue transitions empty→non-empty while unregistered). Always re-register before reading.
/* WRONG approach — race window between drain and re-register */
void bad_handler(int sig)
{
drain_queue(); /* reads messages */
mq_notify(g_mqd, &sev); /* re-register AFTER — TOO LATE! */
/* If a message arrived between drain and re-register: missed! */
}
/* CORRECT approach — re-register BEFORE drain */
void good_handler(int sig)
{
/* 1. Re-register first — no window for missed messages */
mq_notify(g_mqd, &sev);
/* 2. Now drain all current messages */
char buf[256];
unsigned int prio;
ssize_t n;
while ((n = mq_receive(g_mqd, buf, sizeof(buf), &prio)) > 0) {
write(STDOUT_FILENO, buf, n);
}
/* If a new message arrived during drain, queue is still non-empty.
* It will NOT re-trigger notification (queue was never empty).
* That is fine — our loop will pick it up. */
}
| Approach | CPU Usage | Latency | Best For |
|---|---|---|---|
Blocking mq_receive() |
Zero (sleeping) | Very low | Dedicated consumer thread |
| Non-blocking poll loop | High (100%) | Very low | Hard real-time, no OS latency |
mq_notify() + SIGEV_SIGNAL |
Zero between messages | Signal delivery delay | Single-threaded, I/O multiplexed programs |
mq_notify() + SIGEV_THREAD |
Zero between messages | Thread creation delay | Multi-threaded servers, can use any API |
- mq_notify() for async notification
- Signal or thread delivery
- Queue identified by name (/queuename)
- Per-message priority (0–MQ_PRIO_MAX)
- POSIX.1b origin
- No notification mechanism
- Must block or poll
- Queue identified by integer key/ID
- Message type (not priority)
- Older, less portable API
Key Terms
mq_notify() registers a process to be notified asynchronously when a message arrives on a previously empty queue. It is useful because it avoids blocking or polling — the process can do other work and will be interrupted (via signal or thread) only when there is something to read.
Only the empty-to-non-empty transition: a new message arriving on a queue that was previously empty. If the queue already had messages when you registered, no notification is sent until the queue is fully drained and a new message arrives.
Exactly one. If a process tries to register when another process is already registered, mq_notify() returns -1 with errno = EBUSY.
No. The registration is a one-shot mechanism. After the notification is delivered, the registration is automatically removed. To keep receiving notifications, the process must call mq_notify() again (ideally at the start of the handler, before reading messages).
SIGEV_SIGNAL: the kernel sends a specified signal to the process. The process handles it in a signal handler. Only async-signal-safe functions may be called inside the handler.
SIGEV_THREAD: the kernel creates a new thread and calls a specified function in it. No async-signal-safety restriction; any library function may be called.
If you drain messages before re-registering, a new message can arrive between the drain and the re-registration. The queue transitions empty→non-empty while no registration is active, so the notification is lost. The fix is to re-register first, then drain messages. Any message that arrives during the drain will still be read by the drain loop.
No. If a process is already waiting inside mq_receive(), it gets the message directly. The registered process remains registered but is not notified — notification is only triggered if no process is blocked waiting on the queue.
Call mq_notify(mqd, NULL). Passing NULL instead of a sigevent pointer cancels the current registration.
sigev_value is a union sigval that allows you to pass custom data to the notification handler — either an integer (sival_int) or a pointer (sival_ptr). For SIGEV_THREAD, the value is passed as the argument to sigev_notify_function. This is useful for passing context such as a queue descriptor or application-specific state.
System V message queues have no notification mechanism. A consumer must either block in msgrcv() or poll in a loop. POSIX message queues with mq_notify() allow a process to be idle (not consuming CPU) between messages and be woken up automatically when work arrives — a much more efficient and scalable design.
Chapter 52 Complete
You have covered mq_receive(), timed operations, and asynchronous notification.
