1 #ifdef T_NAMESPACE
2 #undef T_NAMESPACE
3 #endif
4
5 #include <darwintest.h>
6 #include <mach/mach.h>
7 #include <darwintest_multiprocess.h>
8
9 #include <assert.h>
10 #include <dispatch/dispatch.h>
11 #include <dispatch/private.h>
12 #include <err.h>
13 #include <errno.h>
14 #include <fcntl.h>
15 #include <poll.h>
16 #include <pthread.h>
17 #include <pthread/workqueue_private.h>
18 #include <stdio.h>
19 #include <stdlib.h>
20 #include <string.h>
21 #include <sys/event.h>
22 #include <sys/socket.h>
23 #include <sys/stat.h>
24 #include <sys/time.h>
25 #include <sys/types.h>
26 #include <sys/wait.h>
27 #include <sysexits.h>
28 #include <unistd.h>
29 #include <util.h>
30 #include <System/sys/event.h> /* kevent_qos */
31
32 T_GLOBAL_META(
33 T_META_NAMESPACE("xnu.kevent"),
34 T_META_CHECK_LEAKS(false),
35 T_META_LTEPHASE(LTE_POSTINIT));
36
37 /*
38 * Test to validate that monitoring a PTY device, FIFO, pipe, or socket pair in
39 * a dispatch source, kqueue, poll, or select delivers read events within and
40 * between processes as expected.
41 *
42 * This test catches issues with watching special devices in kqueue(),
43 * which has tricky special cases for character devices like PTYs.
44 *
45 * It also exercises the path to wake up a dispatch worker thread from the
46 * special device kqueue event, which is also a special case in kqueue().
47 *
48 * See rdar://problem/26240299&26220074&26226862&28625427 for examples and
49 * history.
50 */
51
52 #define EXPECTED_STRING "abcdefghijklmnopqrstuvwxyz. ABCDEFGHIJKLMNOPQRSTUVWXYZ. 1234567890"
53 #define EXPECTED_LEN strlen(EXPECTED_STRING)
54
55 #define READ_SETUP_TIMEOUT_SECS 2
56 #define WRITE_TIMEOUT_SECS 4
57 #define READ_TIMEOUT_SECS 4
58 #define INCREMENTAL_WRITE_SLEEP_USECS 50
59
60 static mach_timespec_t READ_SETUP_timeout = {.tv_sec = READ_SETUP_TIMEOUT_SECS, .tv_nsec = 0};
61 static mach_timespec_t READ_timeout = {.tv_sec = READ_TIMEOUT_SECS, .tv_nsec = 0};
62 static mach_timespec_t WRITE_timeout = {.tv_sec = WRITE_TIMEOUT_SECS, .tv_nsec = 0};
63
64 enum fd_pair {
65 PTY_PAIR,
66 FIFO_PAIR,
67 PIPE_PAIR,
68 SOCKET_PAIR
69 };
70
71 enum write_mode {
72 FULL_WRITE,
73 INCREMENTAL_WRITE,
74 KEVENT_INCREMENTAL_WRITE,
75 KEVENT64_INCREMENTAL_WRITE,
76 KEVENT_QOS_INCREMENTAL_WRITE,
77 WORKQ_INCREMENTAL_WRITE,
78 DISPATCH_INCREMENTAL_WRITE
79 };
80
81 enum read_mode {
82 POLL_READ,
83 SELECT_READ,
84 KEVENT_READ,
85 KEVENT64_READ,
86 KEVENT_QOS_READ,
87 WORKQ_READ,
88 DISPATCH_READ
89 };
90
91 union mode {
92 enum read_mode rd;
93 enum write_mode wr;
94 };
95
96 static struct {
97 enum fd_pair fd_pair;
98 enum write_mode wr_mode;
99 int wr_fd;
100 enum read_mode rd_mode;
101 int rd_fd;
102
103 enum writer_kind {
104 THREAD_WRITER, /* sem */
105 PROCESS_WRITER /* fd */
106 } wr_kind;
107 union {
108 semaphore_t sem;
109 struct {
110 int in_fd;
111 int out_fd;
112 };
113 } wr_wait;
114 semaphore_t wr_finished;
115 semaphore_t rd_finished;
116 } shared;
117
118 static bool handle_reading(enum fd_pair fd_pair, int fd);
119 static bool handle_writing(enum fd_pair fd_pair, int fd);
120 static void drive_kq(bool reading, union mode mode, enum fd_pair fd_pair,
121 int fd);
122
123 #pragma mark writing
124
125 static void
wake_writer(void)126 wake_writer(void)
127 {
128 T_LOG("waking writer");
129
130 switch (shared.wr_kind) {
131 case THREAD_WRITER:
132 T_LOG("signal shared.wr_wait.sem");
133 semaphore_signal(shared.wr_wait.sem);
134 break;
135 case PROCESS_WRITER: {
136 char tmp = 'a';
137 close(shared.wr_wait.out_fd);
138 T_QUIET; T_ASSERT_POSIX_SUCCESS(write(
139 shared.wr_wait.in_fd, &tmp, 1), NULL);
140 break;
141 }
142 }
143 }
144
145 static void
writer_wait(void)146 writer_wait(void)
147 {
148 switch (shared.wr_kind) {
149 case THREAD_WRITER:
150 T_LOG("wait shared.wr_wait.sem");
151 kern_return_t kret = semaphore_timedwait(shared.wr_wait.sem, READ_SETUP_timeout);
152
153 if (kret == KERN_OPERATION_TIMED_OUT) {
154 T_ASSERT_FAIL("THREAD_WRITER semaphore timedout after %d seconds", READ_SETUP_timeout.tv_sec);
155 }
156 T_QUIET;
157 T_ASSERT_MACH_SUCCESS(kret, "semaphore_timedwait shared.wr_wait.sem");
158 break;
159
160 case PROCESS_WRITER: {
161 char tmp;
162 close(shared.wr_wait.in_fd);
163 T_QUIET; T_ASSERT_POSIX_SUCCESS(read(
164 shared.wr_wait.out_fd, &tmp, 1), NULL);
165 break;
166 }
167 }
168
169 T_LOG("writer woken up, starting to write");
170 }
171
172 static bool
handle_writing(enum fd_pair __unused fd_pair,int fd)173 handle_writing(enum fd_pair __unused fd_pair, int fd)
174 {
175 static unsigned int cur_char = 0;
176 T_QUIET; T_ASSERT_POSIX_SUCCESS(write(fd,
177 &(EXPECTED_STRING[cur_char]), 1), NULL);
178 cur_char++;
179
180 return cur_char < EXPECTED_LEN;
181 }
182
183 #define EXPECTED_QOS QOS_CLASS_USER_INITIATED
184
185 static void
reenable_workq(int fd,int16_t filt)186 reenable_workq(int fd, int16_t filt)
187 {
188 struct kevent_qos_s events[] = {{
189 .ident = (uint64_t)fd,
190 .filter = filt,
191 .flags = EV_ENABLE | EV_UDATA_SPECIFIC | EV_DISPATCH,
192 .qos = (int32_t)_pthread_qos_class_encode(EXPECTED_QOS,
193 0, 0),
194 .fflags = NOTE_LOWAT,
195 .data = 1
196 }};
197
198 int kev = kevent_qos(-1, events, 1, events, 1, NULL, NULL,
199 KEVENT_FLAG_WORKQ | KEVENT_FLAG_ERROR_EVENTS);
200 T_QUIET; T_ASSERT_POSIX_SUCCESS(kev, "reenable workq in kevent_qos");
201 }
202
203 static void
workqueue_write_fn(void ** __unused buf,int * __unused count)204 workqueue_write_fn(void ** __unused buf, int * __unused count)
205 {
206 // T_MAYFAIL;
207 // T_QUIET; T_ASSERT_EFFECTIVE_QOS_EQ(EXPECTED_QOS,
208 // "writer thread should be woken up at correct QoS");
209 if (!handle_writing(shared.fd_pair, shared.wr_fd)) {
210 /* finished handling the fd, tear down the source */
211 T_LOG("signal shared.wr_finished");
212 semaphore_signal(shared.wr_finished);
213 return;
214 }
215
216 reenable_workq(shared.wr_fd, EVFILT_WRITE);
217 }
218
219 static void
workqueue_fn(pthread_priority_t __unused priority)220 workqueue_fn(pthread_priority_t __unused priority)
221 {
222 T_ASSERT_FAIL("workqueue function callback was called");
223 }
224
225 static void
drive_kq(bool reading,union mode mode,enum fd_pair fd_pair,int fd)226 drive_kq(bool reading, union mode mode, enum fd_pair fd_pair, int fd)
227 {
228 struct timespec timeout = { .tv_sec = READ_TIMEOUT_SECS };
229 int kev = -1;
230
231 struct kevent events;
232 EV_SET(&events, fd, reading ? EVFILT_READ : EVFILT_WRITE, EV_ADD,
233 NOTE_LOWAT, 1, NULL);
234 struct kevent64_s events64;
235 EV_SET64(&events64, fd, reading ? EVFILT_READ : EVFILT_WRITE, EV_ADD,
236 NOTE_LOWAT, 1, 0, 0, 0);
237 struct kevent_qos_s events_qos[] = {{
238 .ident = (uint64_t)fd,
239 .filter = reading ? EVFILT_READ : EVFILT_WRITE,
240 .flags = EV_ADD,
241 .fflags = NOTE_LOWAT,
242 .data = 1
243 }, {
244 .ident = 0,
245 .filter = EVFILT_TIMER,
246 .flags = EV_ADD,
247 .fflags = NOTE_SECONDS,
248 .data = READ_TIMEOUT_SECS
249 }};
250
251 /* determine which variant of kevent to use */
252 enum read_mode which_kevent;
253 if (reading) {
254 which_kevent = mode.rd;
255 } else {
256 if (mode.wr == KEVENT_INCREMENTAL_WRITE) {
257 which_kevent = KEVENT_READ;
258 } else if (mode.wr == KEVENT64_INCREMENTAL_WRITE) {
259 which_kevent = KEVENT64_READ;
260 } else if (mode.wr == KEVENT_QOS_INCREMENTAL_WRITE) {
261 which_kevent = KEVENT_QOS_READ;
262 } else {
263 T_ASSERT_FAIL("unexpected mode: %d", mode.wr);
264 __builtin_unreachable();
265 }
266 }
267
268 int kq_fd = kqueue();
269 T_QUIET; T_ASSERT_POSIX_SUCCESS(kq_fd, "kqueue");
270
271 switch (which_kevent) {
272 case KEVENT_READ:
273 kev = kevent(kq_fd, &events, 1, NULL, 0, NULL);
274 break;
275 case KEVENT64_READ:
276 kev = kevent64(kq_fd, &events64, 1, NULL, 0, 0, NULL);
277 break;
278 case KEVENT_QOS_READ:
279 kev = kevent_qos(kq_fd, events_qos, 2, NULL, 0, NULL, NULL, 0);
280 break;
281 case POLL_READ: /* FALLTHROUGH */
282 case SELECT_READ: /* FALLTHROUGH */
283 case DISPATCH_READ: /* FALLTHROUGH */
284 case WORKQ_READ: /* FALLTHROUGH */
285 default:
286 T_ASSERT_FAIL("unexpected mode: %d", reading ? mode.rd : mode.wr);
287 break;
288 }
289
290 if (reading) {
291 wake_writer();
292 } else {
293 writer_wait();
294 }
295
296 for (;;) {
297 switch (which_kevent) {
298 case KEVENT_READ:
299 kev = kevent(kq_fd, NULL, 0, &events, 1, &timeout);
300 break;
301 case KEVENT64_READ:
302 kev = kevent64(kq_fd, NULL, 0, &events64, 1, 0, &timeout);
303 break;
304 case KEVENT_QOS_READ:
305 kev = kevent_qos(kq_fd, NULL, 0, events_qos, 2, NULL, NULL, 0);
306
307 /* check for a timeout */
308 for (int i = 0; i < kev; i++) {
309 if (events_qos[i].filter == EVFILT_TIMER) {
310 kev = 0;
311 }
312 }
313 break;
314 case POLL_READ: /* FALLTHROUGH */
315 case SELECT_READ: /* FALLTHROUGH */
316 case DISPATCH_READ: /* FALLTHROUGH */
317 case WORKQ_READ: /* FALLTHROUGH */
318 default:
319 T_ASSERT_FAIL("unexpected mode: %d", reading ? mode.rd : mode.wr);
320 break;
321 }
322
323 if (kev == -1 && errno == EINTR) {
324 T_LOG("kevent was interrupted");
325 continue;
326 }
327 T_QUIET; T_ASSERT_POSIX_SUCCESS(kev, "kevent");
328 T_QUIET; T_ASSERT_NE(kev, 0, "kevent timed out");
329
330 if (reading) {
331 if (!handle_reading(fd_pair, fd)) {
332 break;
333 }
334 } else {
335 if (!handle_writing(fd_pair, fd)) {
336 break;
337 }
338 }
339 }
340
341 close(kq_fd);
342 }
343
344 static void *
write_to_fd(void * __unused ctx)345 write_to_fd(void * __unused ctx)
346 {
347 ssize_t bytes_wr = 0;
348
349 writer_wait();
350
351 switch (shared.wr_mode) {
352 case FULL_WRITE:
353 do {
354 if (bytes_wr == -1) {
355 T_LOG("write from child was interrupted");
356 }
357 bytes_wr = write(shared.wr_fd, EXPECTED_STRING,
358 EXPECTED_LEN);
359 } while (bytes_wr == -1 && errno == EINTR);
360 T_QUIET; T_ASSERT_POSIX_SUCCESS(bytes_wr, "write");
361 T_QUIET; T_ASSERT_EQ(bytes_wr, (ssize_t)EXPECTED_LEN,
362 "wrote enough bytes");
363 break;
364
365 case INCREMENTAL_WRITE:
366 for (unsigned int i = 0; i < EXPECTED_LEN; i++) {
367 T_QUIET;
368 T_ASSERT_POSIX_SUCCESS(write(shared.wr_fd,
369 &(EXPECTED_STRING[i]), 1), NULL);
370 usleep(INCREMENTAL_WRITE_SLEEP_USECS);
371 }
372 break;
373
374 case KEVENT_INCREMENTAL_WRITE: /* FALLTHROUGH */
375 case KEVENT64_INCREMENTAL_WRITE: /* FALLTHROUGH */
376 case KEVENT_QOS_INCREMENTAL_WRITE: {
377 union mode mode = { .wr = shared.wr_mode };
378 drive_kq(false, mode, shared.fd_pair, shared.wr_fd);
379 break;
380 }
381
382 case WORKQ_INCREMENTAL_WRITE: {
383 // prohibit ourselves from going multi-threaded see:rdar://33296008
384 _dispatch_prohibit_transition_to_multithreaded(true);
385 int changes = 1;
386
387 T_ASSERT_MACH_SUCCESS(semaphore_create(mach_task_self(), &shared.wr_finished, SYNC_POLICY_FIFO, 0),
388 "semaphore_create shared.wr_finished");
389
390 T_QUIET;
391 T_ASSERT_NE_UINT(shared.wr_finished, (unsigned)MACH_PORT_NULL, "wr_finished semaphore_create");
392
393 T_QUIET;
394 T_ASSERT_POSIX_ZERO(_pthread_workqueue_init_with_kevent(workqueue_fn, workqueue_write_fn, 0, 0), NULL);
395
396 struct kevent_qos_s events[] = {{
397 .ident = (uint64_t)shared.wr_fd,
398 .filter = EVFILT_WRITE,
399 .flags = EV_ADD | EV_UDATA_SPECIFIC | EV_DISPATCH | EV_VANISHED,
400 .fflags = NOTE_LOWAT,
401 .data = 1,
402 .qos = (int32_t)_pthread_qos_class_encode(EXPECTED_QOS,
403 0, 0)
404 }};
405
406 for (;;) {
407 int kev = kevent_qos(-1, changes == 0 ? NULL : events, changes,
408 events, 1, NULL, NULL,
409 KEVENT_FLAG_WORKQ | KEVENT_FLAG_ERROR_EVENTS);
410 if (kev == -1 && errno == EINTR) {
411 changes = 0;
412 T_LOG("kevent_qos was interrupted");
413 continue;
414 }
415
416 T_QUIET; T_ASSERT_POSIX_SUCCESS(kev, "kevent_qos");
417 break;
418 }
419 break;
420 }
421
422 case DISPATCH_INCREMENTAL_WRITE: {
423 dispatch_source_t write_src;
424
425 T_ASSERT_MACH_SUCCESS(semaphore_create(mach_task_self(), &shared.wr_finished, SYNC_POLICY_FIFO, 0),
426 "semaphore_create shared.wr_finished");
427
428 T_QUIET;
429 T_ASSERT_NE_UINT(shared.wr_finished, (unsigned)MACH_PORT_NULL, "semaphore_create");
430
431 write_src = dispatch_source_create(DISPATCH_SOURCE_TYPE_WRITE,
432 (uintptr_t)shared.wr_fd, 0, NULL);
433 T_QUIET; T_ASSERT_NOTNULL(write_src,
434 "dispatch_source_create(DISPATCH_SOURCE_TYPE_WRITE ...)");
435
436 dispatch_block_t handler = dispatch_block_create_with_qos_class(
437 DISPATCH_BLOCK_ENFORCE_QOS_CLASS, EXPECTED_QOS, 0, ^{
438 // T_MAYFAIL;
439 // T_QUIET; T_ASSERT_EFFECTIVE_QOS_EQ(EXPECTED_QOS,
440 // "write handler block should run at correct QoS");
441 if (!handle_writing(shared.fd_pair, shared.wr_fd)) {
442 /* finished handling the fd, tear down the source */
443 dispatch_source_cancel(write_src);
444 dispatch_release(write_src);
445 T_LOG("signal shared.wr_finished");
446 semaphore_signal(shared.wr_finished);
447 }
448 });
449
450 dispatch_source_set_event_handler(write_src, handler);
451 dispatch_activate(write_src);
452
453 break;
454 }
455
456 default:
457 T_ASSERT_FAIL("unrecognized write mode: %d", shared.wr_mode);
458 break;
459 }
460
461 if (shared.wr_finished) {
462 T_LOG("wait shared.wr_finished");
463 kern_return_t kret = semaphore_timedwait(shared.wr_finished, WRITE_timeout);
464 if (kret == KERN_OPERATION_TIMED_OUT) {
465 T_ASSERT_FAIL("write side semaphore timedout after %d seconds", WRITE_timeout.tv_sec);
466 }
467 T_QUIET;
468 T_ASSERT_MACH_SUCCESS(kret, "semaphore_timedwait shared.wr_finished");
469 semaphore_destroy(mach_task_self(), shared.wr_finished);
470 }
471
472 T_LOG("writer finished, closing fd");
473 T_QUIET; T_ASSERT_POSIX_SUCCESS(close(shared.wr_fd), NULL);
474 return NULL;
475 }
476
477 #pragma mark reading
478
479 #define BUF_LEN 1024
480 static char final_string[BUF_LEN];
481 static size_t final_length;
482
483 /*
484 * Read from the master PTY descriptor.
485 *
486 * Returns false if EOF is encountered, and true otherwise.
487 */
488 static bool
handle_reading(enum fd_pair fd_pair,int fd)489 handle_reading(enum fd_pair fd_pair, int fd)
490 {
491 char read_buf[BUF_LEN] = { 0 };
492 ssize_t bytes_rd = 0;
493
494 do {
495 if (bytes_rd == -1) {
496 T_LOG("read was interrupted, retrying");
497 }
498 bytes_rd = read(fd, read_buf, sizeof(read_buf) - 1);
499 } while (bytes_rd == -1 && errno == EINTR);
500
501 // T_LOG("read %zd bytes: '%s'", bytes_rd, read_buf);
502
503 T_QUIET; T_ASSERT_POSIX_SUCCESS(bytes_rd, "reading from file");
504 T_QUIET; T_ASSERT_LE(bytes_rd, (ssize_t)EXPECTED_LEN,
505 "read too much from file");
506
507 if (bytes_rd == 0) {
508 T_LOG("read EOF from file");
509 return false;
510 }
511
512 read_buf[bytes_rd] = '\0';
513 strlcpy(&(final_string[final_length]), read_buf,
514 sizeof(final_string) - final_length);
515 final_length += (size_t)bytes_rd;
516
517 T_QUIET; T_ASSERT_LE(final_length, EXPECTED_LEN,
518 "should not read more from file than what can be sent");
519
520 /* FIFOs don't send EOF when the write side closes */
521 if (final_length == strlen(EXPECTED_STRING) &&
522 (fd_pair == FIFO_PAIR)) {
523 T_LOG("read all expected bytes from FIFO");
524 return false;
525 }
526 return true;
527 }
528
529 static void
workqueue_read_fn(void ** __unused buf,int * __unused count)530 workqueue_read_fn(void ** __unused buf, int * __unused count)
531 {
532 // T_MAYFAIL;
533 // T_QUIET; T_ASSERT_EFFECTIVE_QOS_EQ(EXPECTED_QOS,
534 // "reader thread should be requested at correct QoS");
535 if (!handle_reading(shared.fd_pair, shared.rd_fd)) {
536 T_LOG("signal shared.rd_finished");
537 semaphore_signal(shared.rd_finished);
538 }
539
540 reenable_workq(shared.rd_fd, EVFILT_READ);
541 }
542
543 static void
read_from_fd(int fd,enum fd_pair fd_pair,enum read_mode mode)544 read_from_fd(int fd, enum fd_pair fd_pair, enum read_mode mode)
545 {
546 int fd_flags;
547
548 T_LOG("reader setting up");
549
550 bzero(final_string, sizeof(final_string));
551
552 fd_flags = fcntl(fd, F_GETFL, 0);
553 T_QUIET; T_ASSERT_POSIX_SUCCESS(fd_flags, "fcntl(F_GETFL)");
554
555 if (!(fd_flags & O_NONBLOCK)) {
556 T_QUIET;
557 T_ASSERT_POSIX_SUCCESS(fcntl(fd, F_SETFL,
558 fd_flags | O_NONBLOCK), NULL);
559 }
560
561 switch (mode) {
562 case POLL_READ: {
563 struct pollfd fds[] = { { .fd = fd, .events = POLLIN } };
564 wake_writer();
565
566 for (;;) {
567 fds[0].revents = 0;
568 int pol = poll(fds, 1, READ_TIMEOUT_SECS * 1000);
569 T_QUIET; T_ASSERT_POSIX_SUCCESS(pol, "poll");
570 T_QUIET; T_ASSERT_NE(pol, 0,
571 "poll should not time out after %d seconds, read %zd out "
572 "of %zu bytes",
573 READ_TIMEOUT_SECS, final_length, strlen(EXPECTED_STRING));
574 T_QUIET; T_ASSERT_FALSE(fds[0].revents & POLLERR,
575 "should not see an error on the device");
576 T_QUIET; T_ASSERT_FALSE(fds[0].revents & POLLNVAL,
577 "should not set up an invalid poll");
578
579 if (!handle_reading(fd_pair, fd)) {
580 break;
581 }
582 }
583 break;
584 }
585
586 case SELECT_READ:
587 wake_writer();
588
589 for (;;) {
590 struct timeval tv = { .tv_sec = READ_TIMEOUT_SECS };
591
592 fd_set read_fd;
593 FD_ZERO(&read_fd);
594 FD_SET(fd, &read_fd);
595 fd_set err_fd;
596 FD_ZERO(&err_fd);
597 FD_SET(fd, &err_fd);
598
599 int sel = select(fd + 1, &read_fd, NULL, NULL /*&err_fd*/, &tv);
600 if (sel == -1 && errno == EINTR) {
601 T_LOG("select interrupted");
602 continue;
603 }
604 (void)fd_pair;
605
606 T_QUIET; T_ASSERT_POSIX_SUCCESS(sel, "select");
607
608 T_QUIET; T_ASSERT_NE(sel, 0,
609 "select waited for %d seconds and timed out",
610 READ_TIMEOUT_SECS);
611
612 /* didn't fail or time out, therefore data is ready */
613 T_QUIET; T_ASSERT_NE(FD_ISSET(fd, &read_fd), 0,
614 "select should show reading fd as readable");
615
616 if (!handle_reading(fd_pair, fd)) {
617 break;
618 }
619 }
620 break;
621
622 case KEVENT_READ: /* FALLTHROUGH */
623 case KEVENT64_READ: /* FALLTHROUGH */
624 case KEVENT_QOS_READ: {
625 union mode rd_mode = { .rd = shared.rd_mode };
626 drive_kq(true, rd_mode, fd_pair, shared.rd_fd);
627 break;
628 }
629
630 case WORKQ_READ: {
631 // prohibit ourselves from going multi-threaded see:rdar://33296008
632 _dispatch_prohibit_transition_to_multithreaded(true);
633 T_ASSERT_POSIX_ZERO(_pthread_workqueue_init_with_kevent(
634 workqueue_fn, workqueue_read_fn, 0, 0), NULL);
635
636 T_ASSERT_MACH_SUCCESS(semaphore_create(mach_task_self(), &shared.rd_finished, SYNC_POLICY_FIFO, 0),
637 "semaphore_create shared.rd_finished");
638
639 T_QUIET;
640 T_ASSERT_NE_UINT(shared.rd_finished, (unsigned)MACH_PORT_NULL, "semaphore_create");
641
642 int changes = 1;
643 struct kevent_qos_s events[] = {{
644 .ident = (uint64_t)shared.rd_fd,
645 .filter = EVFILT_READ,
646 .flags = EV_ADD | EV_UDATA_SPECIFIC | EV_DISPATCH | EV_VANISHED,
647 .fflags = NOTE_LOWAT,
648 .data = 1,
649 .qos = (int32_t)_pthread_qos_class_encode(EXPECTED_QOS,
650 0, 0)
651 }};
652
653 for (;;) {
654 int kev = kevent_qos(-1, changes == 0 ? NULL : events, changes,
655 events, 1, NULL, NULL,
656 KEVENT_FLAG_WORKQ | KEVENT_FLAG_ERROR_EVENTS);
657 if (kev == -1 && errno == EINTR) {
658 changes = 0;
659 T_LOG("kevent_qos was interrupted");
660 continue;
661 }
662
663 T_QUIET; T_ASSERT_POSIX_SUCCESS(kev, "kevent_qos");
664 break;
665 }
666
667 wake_writer();
668 break;
669 }
670
671 case DISPATCH_READ: {
672 dispatch_source_t read_src;
673
674 T_ASSERT_MACH_SUCCESS(semaphore_create(mach_task_self(), &shared.rd_finished, SYNC_POLICY_FIFO, 0),
675 "semaphore_create shared.rd_finished");
676
677 T_QUIET;
678 T_ASSERT_NE_UINT(shared.rd_finished, (unsigned)MACH_PORT_NULL, "semaphore_create");
679
680 read_src = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ,
681 (uintptr_t)fd, 0, NULL);
682 T_QUIET; T_ASSERT_NOTNULL(read_src,
683 "dispatch_source_create(DISPATCH_SOURCE_TYPE_READ)");
684
685 dispatch_block_t handler = dispatch_block_create_with_qos_class(
686 DISPATCH_BLOCK_ENFORCE_QOS_CLASS, EXPECTED_QOS, 0, ^{
687 // T_MAYFAIL;
688 // T_QUIET; T_ASSERT_EFFECTIVE_QOS_EQ(EXPECTED_QOS,
689 // "read handler block should run at correct QoS");
690
691 if (!handle_reading(fd_pair, fd)) {
692 /* finished handling the fd, tear down the source */
693 dispatch_source_cancel(read_src);
694 dispatch_release(read_src);
695 T_LOG("signal shared.rd_finished");
696 semaphore_signal(shared.rd_finished);
697 }
698 });
699
700 dispatch_source_set_event_handler(read_src, handler);
701 dispatch_activate(read_src);
702
703 wake_writer();
704 break;
705 }
706
707 default:
708 T_ASSERT_FAIL("unrecognized read mode: %d", mode);
709 break;
710 }
711
712 if (shared.rd_finished) {
713 T_LOG("wait shared.rd_finished");
714 kern_return_t kret = semaphore_timedwait(shared.rd_finished, READ_timeout);
715 if (kret == KERN_OPERATION_TIMED_OUT) {
716 T_ASSERT_FAIL("reading timed out after %d seconds", READ_timeout.tv_sec);
717 }
718 T_QUIET;
719 T_ASSERT_MACH_SUCCESS(kret, "semaphore_timedwait shared.rd_finished");
720 }
721
722 T_EXPECT_EQ_STR(final_string, EXPECTED_STRING,
723 "reader should receive valid string");
724 T_QUIET; T_ASSERT_POSIX_SUCCESS(close(fd), NULL);
725 }
726
727 #pragma mark file setup
728
729 static void
fd_pair_init(enum fd_pair fd_pair,int * rd_fd,int * wr_fd)730 fd_pair_init(enum fd_pair fd_pair, int *rd_fd, int *wr_fd)
731 {
732 switch (fd_pair) {
733 case PTY_PAIR:
734 T_ASSERT_POSIX_SUCCESS(openpty(rd_fd, wr_fd, NULL, NULL, NULL),
735 NULL);
736 break;
737
738 case FIFO_PAIR: {
739 char fifo_path[] = "/tmp/async-io-fifo.XXXXXX";
740 T_QUIET; T_ASSERT_NOTNULL(mktemp(fifo_path), NULL);
741
742 T_ASSERT_POSIX_SUCCESS(mkfifo(fifo_path, 0700), "mkfifo(%s, 0700)",
743 fifo_path);
744 /*
745 * Opening the read side of a pipe will block until the write
746 * side opens -- use O_NONBLOCK.
747 */
748 *rd_fd = open(fifo_path, O_RDONLY | O_NONBLOCK);
749 T_QUIET; T_ASSERT_POSIX_SUCCESS(*rd_fd, "open(... O_RDONLY)");
750 *wr_fd = open(fifo_path, O_WRONLY | O_NONBLOCK);
751 T_QUIET; T_ASSERT_POSIX_SUCCESS(*wr_fd, "open(... O_WRONLY)");
752 break;
753 }
754
755 case PIPE_PAIR: {
756 int pipe_fds[2];
757 T_ASSERT_POSIX_SUCCESS(pipe(pipe_fds), NULL);
758 *rd_fd = pipe_fds[0];
759 *wr_fd = pipe_fds[1];
760 break;
761 }
762
763 case SOCKET_PAIR: {
764 int sock_fds[2];
765 T_ASSERT_POSIX_SUCCESS(socketpair(AF_UNIX, SOCK_STREAM, 0, sock_fds),
766 NULL);
767 *rd_fd = sock_fds[0];
768 *wr_fd = sock_fds[1];
769 break;
770 }
771
772 default:
773 T_ASSERT_FAIL("unknown descriptor pair type: %d", fd_pair);
774 break;
775 }
776
777 T_QUIET; T_ASSERT_NE(*rd_fd, -1, "reading descriptor");
778 T_QUIET; T_ASSERT_NE(*wr_fd, -1, "writing descriptor");
779 }
780
781 #pragma mark single process
782
783 static void
drive_threads(enum fd_pair fd_pair,enum read_mode rd_mode,enum write_mode wr_mode)784 drive_threads(enum fd_pair fd_pair, enum read_mode rd_mode,
785 enum write_mode wr_mode)
786 {
787 pthread_t thread;
788
789 shared.fd_pair = fd_pair;
790 shared.rd_mode = rd_mode;
791 shared.wr_mode = wr_mode;
792 fd_pair_init(fd_pair, &(shared.rd_fd), &(shared.wr_fd));
793
794 shared.wr_kind = THREAD_WRITER;
795 T_ASSERT_MACH_SUCCESS(semaphore_create(mach_task_self(), &shared.wr_wait.sem, SYNC_POLICY_FIFO, 0),
796 "semaphore_create shared.wr_wait.sem");
797
798 T_QUIET;
799 T_ASSERT_POSIX_ZERO(pthread_create(&thread, NULL, write_to_fd, NULL),
800 NULL);
801 T_LOG("created writer thread");
802
803 read_from_fd(shared.rd_fd, fd_pair, rd_mode);
804
805 T_ASSERT_POSIX_ZERO(pthread_join(thread, NULL), NULL);
806
807 T_END;
808 }
809
810 #pragma mark multiple processes
811
812 static void __attribute__((noreturn))
drive_processes(enum fd_pair fd_pair,enum read_mode rd_mode,enum write_mode wr_mode)813 drive_processes(enum fd_pair fd_pair, enum read_mode rd_mode, enum write_mode wr_mode)
814 {
815 shared.fd_pair = fd_pair;
816 shared.rd_mode = rd_mode;
817 shared.wr_mode = wr_mode;
818 fd_pair_init(fd_pair, &(shared.rd_fd), &(shared.wr_fd));
819
820 shared.wr_kind = PROCESS_WRITER;
821 int fds[2];
822 T_QUIET; T_ASSERT_POSIX_SUCCESS(pipe(fds), NULL);
823 shared.wr_wait.out_fd = fds[0];
824 shared.wr_wait.in_fd = fds[1];
825
826 T_LOG("starting subprocesses");
827 dt_helper_t helpers[2] = {
828 dt_fork_helper("reader_helper"),
829 dt_fork_helper("writer_helper")
830 };
831
832 close(shared.rd_fd);
833 close(shared.wr_fd);
834
835 dt_run_helpers(helpers, 2, 50000);
836 }
837
838 T_HELPER_DECL(reader_helper, "Read asynchronously")
839 {
840 close(shared.wr_fd);
841 read_from_fd(shared.rd_fd, shared.fd_pair, shared.rd_mode);
842 T_END;
843 }
844
845 T_HELPER_DECL(writer_helper, "Write asynchronously")
846 {
847 close(shared.rd_fd);
848 write_to_fd(NULL);
849 }
850
851 #pragma mark tests
852
853 #define WR_DECL_PROCESSES(desc_name, fd_pair, write_name, write_str, \
854 write_mode, read_name, read_mode) \
855 T_DECL(desc_name##_r##read_name##_w##write_name##_procs, "read changes to a " \
856 #desc_name " with " #read_name " and writing " #write_str \
857 " across two processes", T_META_TAG_VM_PREFERRED) \
858 { \
859 drive_processes(fd_pair, read_mode, write_mode); \
860 }
861 #define WR_DECL_THREADS(desc_name, fd_pair, write_name, write_str, \
862 write_mode, read_name, read_mode) \
863 T_DECL(desc_name##_r##read_name##_w##write_name##_thds, "read changes to a " \
864 #desc_name " with " #read_name " and writing " #write_str, \
865 T_META_TAG_VM_PREFERRED) \
866 { \
867 drive_threads(fd_pair, read_mode, write_mode); \
868 }
869
870 #define WR_DECL(desc_name, fd_pair, write_name, write_str, write_mode, \
871 read_name, read_mode) \
872 WR_DECL_PROCESSES(desc_name, fd_pair, write_name, write_str, \
873 write_mode, read_name, read_mode) \
874 WR_DECL_THREADS(desc_name, fd_pair, write_name, write_str, \
875 write_mode, read_name, read_mode)
876
877 #define RD_DECL_SAFE(desc_name, fd_pair, read_name, read_mode) \
878 WR_DECL(desc_name, fd_pair, full, "the full string", FULL_WRITE, \
879 read_name, read_mode) \
880 WR_DECL(desc_name, fd_pair, inc, "incrementally", \
881 INCREMENTAL_WRITE, read_name, read_mode)
882
883 #define RD_DECL_DISPATCH_ONLY(suffix, desc_name, fd_pair, read_name, \
884 read_mode) \
885 WR_DECL##suffix(desc_name, fd_pair, inc_dispatch, \
886 "incrementally with a dispatch source", \
887 DISPATCH_INCREMENTAL_WRITE, read_name, read_mode)
888 #define RD_DECL_WORKQ_ONLY(suffix, desc_name, fd_pair, read_name, \
889 read_mode) \
890 WR_DECL##suffix(desc_name, fd_pair, inc_workq, \
891 "incrementally with the workqueue", \
892 WORKQ_INCREMENTAL_WRITE, read_name, read_mode)
893
894 #define RD_DECL(desc_name, fd_pair, read_name, read_mode) \
895 RD_DECL_SAFE(desc_name, fd_pair, read_name, read_mode) \
896 RD_DECL_DISPATCH_ONLY(, desc_name, fd_pair, read_name, read_mode)
897 // RD_DECL_WORKQ_ONLY(, desc_name, fd_pair, read_name, read_mode)
898
899 /*
900 * dispatch_source tests cannot share the same process as other workqueue
901 * tests.
902 */
903 #define RD_DECL_DISPATCH(desc_name, fd_pair, read_name, read_mode) \
904 RD_DECL_SAFE(desc_name, fd_pair, read_name, read_mode) \
905 RD_DECL_DISPATCH_ONLY(, desc_name, fd_pair, read_name, read_mode) \
906 RD_DECL_WORKQ_ONLY(_PROCESSES, desc_name, fd_pair, read_name, \
907 read_mode)
908
909 /*
910 * Workqueue tests cannot share the same process as other workqueue or
911 * dispatch_source tests.
912 #define RD_DECL_WORKQ(desc_name, fd_pair, read_name, read_mode) \
913 * RD_DECL_SAFE(desc_name, fd_pair, read_name, read_mode) \
914 * RD_DECL_DISPATCH_ONLY(_PROCESSES, desc_name, fd_pair, read_name, \
915 * read_mode) \
916 * RD_DECL_WORKQ_ONLY(_PROCESSES, desc_name, fd_pair, read_name, \
917 * read_mode)
918 */
919
920 #define PAIR_DECL(desc_name, fd_pair) \
921 RD_DECL(desc_name, fd_pair, poll, POLL_READ) \
922 RD_DECL(desc_name, fd_pair, select, SELECT_READ) \
923 RD_DECL(desc_name, fd_pair, kevent, KEVENT_READ) \
924 RD_DECL(desc_name, fd_pair, kevent64, KEVENT64_READ) \
925 RD_DECL(desc_name, fd_pair, kevent_qos, KEVENT_QOS_READ) \
926 RD_DECL_DISPATCH(desc_name, fd_pair, dispatch_source, DISPATCH_READ)
927 // RD_DECL_WORKQ(desc_name, fd_pair, workq, WORKQ_READ)
928
929 PAIR_DECL(tty, PTY_PAIR)
930 PAIR_DECL(pipe, PIPE_PAIR)
931 PAIR_DECL(fifo, FIFO_PAIR)
932 PAIR_DECL(socket, SOCKET_PAIR)
933