The Pedigree Project 0.1
mqueues.c
1#define _GNU_SOURCE
2#include <errno.h>
3#include <fcntl.h>
4#include <mqueue.h>
5#include <poll.h>
6#include <pthread.h>
7#include <semaphore.h>
8#include <signal.h>
9#include <stdio.h>
10#include <stdlib.h>
11#include <string.h>
12#include <time.h>
13#include <unistd.h>
14
15#include <sys/epoll.h>
16#include <sys/mman.h>
17#include <sys/wait.h>
18
19#define CHECK(condition) \
20 do { \
21 if (!(condition)) { \
22 fprintf(stderr, "mqueues:%d: %s (errno=%d)\n", __LINE__, #condition, errno); \
23 failed = 1; \
24 goto out; \
25 } \
26 } while (0)
27
28static struct timespec deadline_after(long milliseconds) {
29 struct timespec deadline;
30 clock_gettime(CLOCK_REALTIME, &deadline);
31 deadline.tv_nsec += milliseconds * 1000000;
32 deadline.tv_sec += deadline.tv_nsec / 1000000000;
33 deadline.tv_nsec %= 1000000000;
34 return deadline;
35}
36
37static volatile sig_atomic_t signal_count, signal_code, signal_value;
38static sem_t callback_done;
39static void notified(int signal, siginfo_t* info, void* context) {
40 (void)signal;
41 (void)context;
42 signal_code = info->si_code;
43 signal_value = info->si_value.sival_int;
44 ++signal_count;
45}
46static void callback(union sigval value) {
47 signal_value = value.sival_int;
48 sem_post(&callback_done);
49}
50static volatile sig_atomic_t interrupt_calls, hold_handler;
51static struct timespec handler_until, handler_finished;
52static void interrupted(int signal) {
53 (void)signal;
54 __atomic_add_fetch(&interrupt_calls, 1, __ATOMIC_RELEASE);
55 if (hold_handler) {
56 struct timespec now, pause = {0, 10000000};
57 do {
58 nanosleep(&pause, NULL);
59 clock_gettime(CLOCK_REALTIME, &now);
60 } while (now.tv_sec < handler_until.tv_sec ||
61 (now.tv_sec == handler_until.tv_sec && now.tv_nsec < handler_until.tv_nsec));
62 clock_gettime(CLOCK_MONOTONIC, &handler_finished);
63 }
64}
65
66struct queue_wait {
67 mqd_t queue;
68 sem_t entered;
69 int sending, returned;
70 struct timespec deadline, finished;
71 ssize_t result;
72 int error;
73 char buffer[32];
74};
75static void* interrupted_queue_call(void* argument) {
76 struct queue_wait* wait = argument;
77 sem_post(&wait->entered);
78 wait->result = wait->sending ? mq_timedsend(wait->queue, "restart", 8, 1, &wait->deadline)
79 : mq_timedreceive(wait->queue, wait->buffer, sizeof(wait->buffer),
80 NULL, &wait->deadline);
81 wait->error = errno;
82 clock_gettime(CLOCK_MONOTONIC, &wait->finished);
83 __atomic_store_n(&wait->returned, 1, __ATOMIC_RELEASE);
84 return NULL;
85}
86
87static int interruption_contract(mqd_t queue, int sending, int restart, int expires) {
88 struct sigaction action = {.sa_handler = interrupted, .sa_flags = restart ? SA_RESTART : 0};
89 struct queue_wait wait = {.queue = queue, .sending = sending};
90 struct mq_attr attr;
91 char buffer[32];
92 int failed = 0;
93 hold_handler = expires;
94 __atomic_store_n(&interrupt_calls, 0, __ATOMIC_RELEASE);
95 if (sigaction(SIGUSR2, &action, NULL) || sem_init(&wait.entered, 0, 0))
96 return 1;
97 if (sending) {
98 for (int n = 0; n < 3; ++n) {
99 if (mq_send(queue, "full", 5, 0))
100 return 1;
101 }
102 }
103 wait.deadline = deadline_after(expires ? 500 : 2000);
104 handler_until = deadline_after(900);
105 pthread_t worker;
106 if (pthread_create(&worker, NULL, interrupted_queue_call, &wait))
107 return 1;
108 struct timespec gate_deadline = deadline_after(1000);
109 if (sem_timedwait(&wait.entered, &gate_deadline))
110 failed = 1;
111 // Repeated acknowledged deliveries cover the publication-to-syscall-entry
112 // window, while each call still has a bounded absolute timeout.
113 for (int round = 0; !failed && round < (expires ? 1 : 3); ++round) {
114 usleep(20000);
115 if (__atomic_load_n(&wait.returned, __ATOMIC_ACQUIRE))
116 break;
117 int previous = __atomic_load_n(&interrupt_calls, __ATOMIC_ACQUIRE);
118 if (pthread_kill(worker, SIGUSR2)) {
119 failed = 1;
120 break;
121 }
122 for (int n = 0; n < 100 && __atomic_load_n(&interrupt_calls, __ATOMIC_ACQUIRE) == previous; ++n)
123 usleep(1000);
124 if (__atomic_load_n(&interrupt_calls, __ATOMIC_ACQUIRE) == previous)
125 failed = 1;
126 if (restart && !expires && __atomic_load_n(&wait.returned, __ATOMIC_ACQUIRE))
127 failed = 1;
128 }
129 if (!expires) {
130 if (sending ? mq_receive(queue, buffer, sizeof(buffer), NULL) < 0
131 : mq_send(queue, "restart", 8, 1) != 0)
132 failed = 1;
133 }
134 if (pthread_join(worker, NULL))
135 failed = 1;
136 if (!__atomic_load_n(&interrupt_calls, __ATOMIC_ACQUIRE))
137 failed = 1;
138 if (!restart) {
139 if (wait.result != -1 || wait.error != EINTR)
140 failed = 1;
141 } else if (expires) {
142 long long after_handler = (wait.finished.tv_sec - handler_finished.tv_sec) * 1000000000LL +
143 wait.finished.tv_nsec - handler_finished.tv_nsec;
144 if (wait.result != -1 || wait.error != ETIMEDOUT || after_handler < 0 ||
145 after_handler > 300000000LL)
146 failed = 1;
147 } else if (wait.result != (sending ? 0 : 8) || (!sending && strcmp(wait.buffer, "restart"))) {
148 failed = 1;
149 }
150 hold_handler = 0;
151 sem_destroy(&wait.entered);
152 while (!mq_getattr(queue, &attr) && attr.mq_curmsgs) {
153 if (mq_receive(queue, buffer, sizeof(buffer), NULL) < 0) {
154 failed = 1;
155 break;
156 }
157 }
158 if (failed)
159 fprintf(stderr, "mq interruption send=%d restart=%d expired=%d result=%ld errno=%d\n", sending,
160 restart, expires, (long)wait.result, wait.error);
161 return failed;
162}
163
164int ipc_test_mqueue_exec(int argc, char** argv) {
165 if (argc != 3)
166 return 1;
167 int descriptor = atoi(argv[2]);
168 struct sigevent event = {.sigev_notify = SIGEV_NONE};
169 struct mq_attr attr;
170 alarm(5);
171 if (fcntl(descriptor, F_GETFD) != 0 || mq_getattr(descriptor, &attr))
172 return 2;
173 if (mq_notify(descriptor, &event))
174 return 3;
175 return mq_notify(descriptor, NULL) || mq_close(descriptor) ? 4 : 0;
176}
177
178static int child_result(pid_t child, int killed) {
179 int status;
180 for (int n = 0; n < 500; ++n) {
181 pid_t result = waitpid(child, &status, WNOHANG);
182 if (result == child)
183 return killed ? !(WIFSIGNALED(status) && WTERMSIG(status) == SIGKILL)
184 : !(WIFEXITED(status) && WEXITSTATUS(status) == 0);
185 if (result < 0 && errno != EINTR)
186 return 1;
187 usleep(10000);
188 }
189 kill(child, SIGKILL);
190 waitpid(child, &status, 0);
191 return 1;
192}
193
194int ipc_test_mqueues(void) {
195 int failed = 0, alias = -1, child_status = 0, epoll = -1;
196 mqd_t queue = (mqd_t)-1, replacement = (mqd_t)-1, read_only = (mqd_t)-1;
197 char name[80], buffer[32];
198 unsigned priority = 0;
199 struct mq_attr attr = {.mq_maxmsg = 3, .mq_msgsize = sizeof(buffer)};
200 struct mq_attr old, flags = {0};
201 struct sigaction action = {0}, previous_usr1 = {0}, previous_usr2 = {0};
202 int signals_installed = 0, callback_initialized = 0;
203 void* bad = MAP_FAILED;
204 snprintf(name, sizeof(name), "/ipc-contract-mq-%ld", (long)getpid());
205 mq_unlink(name);
206 alarm(30);
207
208 queue = mq_open(name, O_CREAT | O_EXCL | O_RDWR | O_NONBLOCK, 0600, &attr);
209 CHECK(queue != (mqd_t)-1);
210 CHECK(fcntl(queue, F_GETFD) & FD_CLOEXEC);
211 CHECK(mq_open(name, O_CREAT | O_EXCL | O_RDWR, 0600, &attr) == (mqd_t)-1 && errno == EEXIST);
212 CHECK(mq_getattr(queue, &old) == 0 && old.mq_maxmsg == 3 && old.mq_msgsize == 32 &&
213 old.mq_curmsgs == 0 && old.mq_flags == O_NONBLOCK);
214 CHECK(mq_open("/bad/name", O_CREAT | O_RDWR, 0600, &attr) == (mqd_t)-1 && errno == EACCES);
215 alias = dup(queue);
216 CHECK(alias >= 0);
217 CHECK(mq_setattr(alias, &flags, &old) == 0 && old.mq_flags == O_NONBLOCK);
218 CHECK(!(fcntl(queue, F_GETFL) & O_NONBLOCK));
219 CHECK(fcntl(alias, F_SETFL, O_NONBLOCK) == 0);
220 CHECK(mq_getattr(queue, &old) == 0 && old.mq_flags == O_NONBLOCK);
221 CHECK(mq_receive(queue, buffer, sizeof(buffer), NULL) == -1 && errno == EAGAIN);
222
223 CHECK(mq_send(queue, "low", 4, 2) == 0);
224 CHECK(mq_send(queue, "first", 6, 7) == 0);
225 CHECK(mq_send(queue, "second", 7, 7) == 0);
226 CHECK(mq_send(queue, "full", 5, 0) == -1 && errno == EAGAIN);
227 CHECK(mq_send(queue, "bad", 4, 32768) == -1 && errno == EINVAL);
228 CHECK(mq_receive(queue, buffer, sizeof(buffer) - 1, NULL) == -1 && errno == EMSGSIZE);
229 struct pollfd pollfd = {.fd = queue, .events = POLLIN | POLLOUT};
230 CHECK(poll(&pollfd, 1, 0) == 1 && (pollfd.revents & POLLIN) && !(pollfd.revents & POLLOUT));
231 epoll = epoll_create1(EPOLL_CLOEXEC);
232 CHECK(epoll >= 0);
233 struct epoll_event watch = {.events = EPOLLIN | EPOLLOUT, .data.u64 = 17}, ready;
234 CHECK(epoll_ctl(epoll, EPOLL_CTL_ADD, queue, &watch) == 0);
235 CHECK(epoll_wait(epoll, &ready, 1, 0) == 1 && ready.data.u64 == 17 && (ready.events & EPOLLIN) &&
236 !(ready.events & EPOLLOUT));
237
238 bad = mmap(NULL, 4096, PROT_NONE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
239 CHECK(bad != MAP_FAILED);
240 CHECK(mq_receive(queue, buffer, sizeof(buffer), bad) == -1 && errno == EFAULT);
241 CHECK(mq_getattr(queue, &old) == 0 && old.mq_curmsgs == 3);
242 CHECK(mq_setattr(queue, &flags, bad) == -1 && errno == EFAULT);
243 CHECK(mq_getattr(queue, &old) == 0 && old.mq_flags == O_NONBLOCK);
244 CHECK(mq_receive(queue, bad, sizeof(buffer), NULL) == -1 && errno == EFAULT);
245 CHECK(mq_receive(queue, buffer, sizeof(buffer), &priority) == 6 && priority == 7 &&
246 !strcmp(buffer, "first"));
247 CHECK(mq_receive(queue, buffer, sizeof(buffer), &priority) == 7 && priority == 7 &&
248 !strcmp(buffer, "second"));
249 CHECK(mq_receive(queue, buffer, sizeof(buffer), &priority) == 4 && priority == 2 &&
250 !strcmp(buffer, "low"));
251 CHECK(epoll_wait(epoll, &ready, 1, 0) == 1 && !(ready.events & EPOLLIN) &&
252 (ready.events & EPOLLOUT));
253 CHECK(close(epoll) == 0);
254 epoll = -1;
255 CHECK(mq_send(queue, NULL, 0, 0) == 0);
256 CHECK(mq_receive(queue, buffer, sizeof(buffer), &priority) == 0 && priority == 0);
257
258 read_only = mq_open(name, O_RDONLY | O_NONBLOCK);
259 CHECK(read_only != (mqd_t)-1);
260 CHECK(mq_send(read_only, "x", 1, 0) == -1 && errno == EBADF);
261 CHECK(mq_close(read_only) == 0);
262 read_only = (mqd_t)-1;
263 if (geteuid() == 0) {
264 pid_t denied_child = fork();
265 CHECK(denied_child >= 0);
266 if (!denied_child) {
267 if (setgid(65534) || setuid(65534))
268 _exit(1);
269 mqd_t denied = mq_open(name, O_RDONLY);
270 if (denied != (mqd_t)-1 || errno != EACCES)
271 _exit(2);
272 _exit(mq_unlink(name) == -1 && errno == EACCES ? 0 : 3);
273 }
274 CHECK(waitpid(denied_child, &child_status, 0) == denied_child && WIFEXITED(child_status) &&
275 WEXITSTATUS(child_status) == 0);
276 }
277 CHECK(mq_unlink(name) == 0);
278 CHECK(mq_open(name, O_RDWR) == (mqd_t)-1 && errno == ENOENT);
279 replacement = mq_open(name, O_CREAT | O_EXCL | O_RDWR | O_NONBLOCK, 0600, &attr);
280 CHECK(replacement != (mqd_t)-1);
281 CHECK(mq_send(queue, "unlinked", 9, 1) == 0);
282 CHECK(mq_receive(replacement, buffer, sizeof(buffer), NULL) == -1 && errno == EAGAIN);
283 CHECK(mq_receive(alias, buffer, sizeof(buffer), NULL) == 9 && !strcmp(buffer, "unlinked"));
284
285 CHECK(mq_setattr(queue, &flags, NULL) == 0);
286 struct timespec deadline = deadline_after(60);
287 CHECK(mq_timedreceive(queue, buffer, sizeof(buffer), NULL, &deadline) == -1 &&
288 errno == ETIMEDOUT);
289 deadline.tv_nsec = 1000000000;
290 CHECK(mq_timedreceive(queue, buffer, sizeof(buffer), NULL, &deadline) == -1 && errno == EINVAL);
291 CHECK(mq_send(queue, "1", 2, 0) == 0 && mq_send(queue, "2", 2, 0) == 0 &&
292 mq_send(queue, "3", 2, 0) == 0);
293 deadline = deadline_after(60);
294 CHECK(mq_timedsend(queue, "4", 2, 0, &deadline) == -1 && errno == ETIMEDOUT);
295 for (int n = 0; n < 3; ++n)
296 CHECK(mq_receive(queue, buffer, sizeof(buffer), NULL) == 2);
297
298 action.sa_sigaction = notified;
299 action.sa_flags = SA_SIGINFO;
300 sigemptyset(&action.sa_mask);
301 CHECK(sigaction(SIGUSR1, &action, &previous_usr1) == 0);
302 action.sa_handler = interrupted;
303 action.sa_flags = 0;
304 CHECK(sigaction(SIGUSR2, &action, &previous_usr2) == 0);
305 signals_installed = 1;
306 for (int sending = 0; sending < 2; ++sending) {
307 CHECK(interruption_contract(queue, sending, 0, 0) == 0);
308 CHECK(interruption_contract(queue, sending, 1, 0) == 0);
309 CHECK(interruption_contract(queue, sending, 1, 1) == 0);
310 }
311
312 struct sigevent event = {.sigev_notify = SIGEV_SIGNAL, .sigev_signo = SIGUSR1};
313 event.sigev_value.sival_int = 0x1357;
314 signal_count = 0;
315 CHECK(mq_notify(queue, &event) == 0);
316 CHECK(mq_notify(alias, &event) == -1 && errno == EBUSY);
317 CHECK(mq_send(queue, "signal", 7, 0) == 0);
318 for (int n = 0; n < 100 && !signal_count; ++n)
319 usleep(1000);
320 CHECK(signal_count == 1 && signal_code == SI_MESGQ && signal_value == 0x1357);
321 CHECK(mq_receive(queue, buffer, sizeof(buffer), NULL) == 7);
322 CHECK(mq_send(queue, "once", 5, 0) == 0 && signal_count == 1);
323 CHECK(mq_receive(queue, buffer, sizeof(buffer), NULL) == 5);
324 event.sigev_notify = SIGEV_NONE;
325 CHECK(mq_notify(queue, &event) == 0);
326 CHECK(mq_notify(alias, &event) == -1 && errno == EBUSY);
327 CHECK(mq_notify(queue, NULL) == 0);
328
329 CHECK(sem_init(&callback_done, 0, 0) == 0);
330 callback_initialized = 1;
331 event.sigev_notify = SIGEV_THREAD;
332 event.sigev_notify_function = callback;
333 event.sigev_value.sival_int = 0x2468;
334 CHECK(mq_notify(queue, &event) == 0);
335 CHECK(mq_send(queue, "thread", 7, 0) == 0);
336 deadline = deadline_after(2000);
337 CHECK(sem_timedwait(&callback_done, &deadline) == 0 && signal_value == 0x2468);
338 CHECK(mq_receive(queue, buffer, sizeof(buffer), NULL) == 7);
339 CHECK(mq_notify(queue, &event) == 0);
340 CHECK(mq_notify(queue, NULL) == 0);
341 usleep(50000);
342 CHECK(sem_trywait(&callback_done) == -1 && errno == EAGAIN);
343
344 event.sigev_notify = SIGEV_SIGNAL;
345 event.sigev_signo = SIGUSR1;
346 CHECK(mq_notify(queue, &event) == 0);
347 CHECK(close(alias) == 0);
348 alias = -1;
349 CHECK(mq_notify(queue, &event) == 0);
350 CHECK(mq_notify(queue, NULL) == 0);
351 for (int close_original = 0; close_original < 2; ++close_original) {
352 pid_t exec_child = fork();
353 CHECK(exec_child >= 0);
354 if (!exec_child) {
355 int retained = fcntl(queue, F_DUPFD, 64);
356 if (retained < 0 || (close_original && mq_close(queue)))
357 _exit(1);
358 struct sigevent none = {.sigev_notify = SIGEV_NONE};
359 if (mq_notify(retained, &none))
360 _exit(2);
361 char descriptor[32];
362 snprintf(descriptor, sizeof(descriptor), "%d", retained);
363 execl("/applications/ipc-contract-test", "ipc-contract-test", "mq-exec", descriptor, NULL);
364 _exit(127);
365 }
366 CHECK(child_result(exec_child, 0) == 0);
367 }
368
369 int gate[2];
370 CHECK(pipe(gate) == 0);
371 pid_t killed_child = fork();
372 CHECK(killed_child >= 0);
373 if (!killed_child) {
374 close(gate[0]);
375 alarm(5);
376 if (write(gate[1], "r", 1) != 1)
377 _exit(1);
378 close(gate[1]);
379 mq_receive(queue, buffer, sizeof(buffer), NULL);
380 _exit(2);
381 }
382 close(gate[1]);
383 struct pollfd gate_poll = {.fd = gate[0], .events = POLLIN};
384 CHECK(poll(&gate_poll, 1, 1000) == 1 && read(gate[0], buffer, 1) == 1);
385 close(gate[0]);
386 usleep(50000);
387 CHECK(kill(killed_child, SIGKILL) == 0 && child_result(killed_child, 1) == 0);
388 sig_atomic_t before_kill_notification = signal_count;
389 CHECK(mq_notify(queue, &event) == 0 && mq_send(queue, "alive", 6, 0) == 0);
390 for (int n = 0; n < 100 && signal_count == before_kill_notification; ++n)
391 usleep(1000);
392 CHECK(signal_count == before_kill_notification + 1);
393 CHECK(mq_receive(queue, buffer, sizeof(buffer), NULL) == 6 && !strcmp(buffer, "alive"));
394
395 pid_t child = fork();
396 CHECK(child >= 0);
397 if (!child)
398 _exit(mq_send(queue, "child", 6, 3) == 0 ? 0 : 1);
399 deadline = deadline_after(2000);
400 CHECK(mq_timedreceive(queue, buffer, sizeof(buffer), &priority, &deadline) == 6 &&
401 priority == 3 && !strcmp(buffer, "child"));
402 CHECK(waitpid(child, &child_status, 0) == child && WIFEXITED(child_status) &&
403 WEXITSTATUS(child_status) == 0);
404
405out:
406 if (epoll >= 0)
407 close(epoll);
408 if (queue != (mqd_t)-1)
409 mq_notify(queue, NULL);
410 if (alias >= 0)
411 close(alias);
412 if (read_only != (mqd_t)-1)
413 mq_close(read_only);
414 if (queue != (mqd_t)-1)
415 mq_close(queue);
416 if (replacement != (mqd_t)-1)
417 mq_close(replacement);
418 mq_unlink(name);
419 if (bad != MAP_FAILED)
420 munmap(bad, 4096);
421 if (callback_initialized)
422 sem_destroy(&callback_done);
423 if (signals_installed) {
424 sigaction(SIGUSR1, &previous_usr1, NULL);
425 sigaction(SIGUSR2, &previous_usr2, NULL);
426 }
427 alarm(0);
428 if (!failed)
429 puts("ipc-contract-test: mqueues passed");
430 return failed;
431}
queue_wait
Definition mqueues.c:66
Generated on Tue Sep 22 2026 06:05:51 for The Pedigree Project by doxygen 1.9.8