1 #include <darwintest.h>
2
3 #include <errno.h>
4 #include <pthread.h>
5 #include <stdio.h>
6 #include <stdlib.h>
7 #include <unistd.h>
8
9 #include <mach/mach.h>
10 #include <mach/mach_time.h>
11 #include <mach/semaphore.h>
12 #include <sys/select.h>
13
14 /* Select parameters */
15 #define TIMEOUT_CHANCE 17 /* one in this many times, timeout */
16 #define TIMEOUT_POLLCHANCE 11 /* one in this many is a poll */
17 #define TIMEOUT_SCALE 5 /* microseconds multiplier */
18
19 static semaphore_t g_thread_sem;
20 static semaphore_t g_sync_sem;
21
22 struct endpoint {
23 int fd[4];
24 pthread_t pth;
25 };
26
27 typedef void * (*thread_func)(struct endpoint *ep);
28 typedef void (*setup_func)(struct endpoint *ep);
29
30 struct thread_sync_arg {
31 struct endpoint ep;
32 setup_func setup;
33 thread_func work;
34 };
35
36 static mach_timebase_info_data_t g_timebase;
37
38 static int g_sleep_iterations = 150000;
39 static int g_sleep_usecs = 30;
40 static int g_stress_nthreads = 100;
41 static uint64_t g_stress_duration = 5;
42
43 static inline uint64_t
ns_to_abs(uint64_t ns)44 ns_to_abs(uint64_t ns)
45 {
46 return ns * g_timebase.denom / g_timebase.numer;
47 }
48
49 static inline uint64_t
abs_to_ns(uint64_t abs)50 abs_to_ns(uint64_t abs)
51 {
52 return abs * g_timebase.numer / g_timebase.denom;
53 }
54
55
56
57 /*
58 * Synchronize the startup / initialization of a set of threads
59 */
60 static void *
thread_sync(void * ctx)61 thread_sync(void *ctx)
62 {
63 struct thread_sync_arg *a = (struct thread_sync_arg *)ctx;
64 T_QUIET;
65 T_ASSERT_TRUE(((a != NULL) && (a->work != NULL)), "thread setup error");
66
67 if (a->setup) {
68 (a->setup)(&a->ep);
69 }
70
71 semaphore_wait_signal(g_thread_sem, g_sync_sem);
72 return (a->work)(&a->ep);
73 }
74
75 struct select_stress_args {
76 struct endpoint *ep;
77 int nthreads;
78 };
79
80 static void
setup_stress_event(struct endpoint * ep)81 setup_stress_event(struct endpoint *ep)
82 {
83 T_QUIET;
84 T_WITH_ERRNO;
85 T_ASSERT_POSIX_SUCCESS(pipe(&ep->fd[0]), "pipe()");
86
87 T_LOG("th[0x%lx]: fd:{%d,%d}, ep@%p",
88 (uintptr_t)pthread_self(), ep->fd[0], ep->fd[1], (void *)ep);
89 }
90
91 /*
92 * Cause file descriptors to be reused/replaced. We expect that it will at
93 * least take the lowest fd as part of the descriptor list. This may be
94 * optimistic, but it shows replacing an fd out from under a select() if it
95 * happens.
96 *
97 * We potentially delay the open for a random amount of time so that another
98 * thread can come in and wake up the fd_set with a bad (closed) fd in the set.
99 */
100 static void
recycle_fds(struct endpoint * ep)101 recycle_fds(struct endpoint *ep)
102 {
103 /* close endpoint descriptors in random order */
104 if (random() % 1) {
105 close(ep->fd[0]);
106 close(ep->fd[1]);
107 } else {
108 close(ep->fd[1]);
109 close(ep->fd[0]);
110 }
111
112 /* randomize a delay */
113 if ((random() % ep->fd[0]) == 0) {
114 usleep(((random() % ep->fd[1]) + 1) * ep->fd[1]);
115 }
116
117 /* reopen the FDs, hopefully in the middle of select() */
118 T_QUIET;
119 T_WITH_ERRNO;
120 T_ASSERT_POSIX_SUCCESS(pipe(&ep->fd[0]), "pipe");
121 }
122
123
124 /*
125 * Send a byte of data down the thread end of a pipe to wake up the select
126 * on the other end of it. Select will wake up normally because of this,
127 * and read the byte out. Hopefully, another thread has closed/reopened its FDs.
128 */
129 static void
write_data(struct endpoint * ep)130 write_data(struct endpoint *ep)
131 {
132 T_QUIET;
133 T_WITH_ERRNO;
134 T_ASSERT_POSIX_SUCCESS(write(ep->fd[1], "X", 1), "th[0x%lx] write_data(fd=%d)",
135 (uintptr_t)pthread_self(), ep->fd[1]);
136 }
137
138 static void *
do_stress_events(struct endpoint * ep)139 do_stress_events(struct endpoint *ep)
140 {
141 unsigned write_freq = (unsigned)(((uintptr_t)pthread_self() & 0xff0000) >> 16);
142
143 /* some default */
144 if (write_freq == 0) {
145 write_freq = 31;
146 }
147
148 T_LOG("th[0x%lx] write_freq:%d", (uintptr_t)pthread_self(), write_freq);
149
150 for (;;) {
151 /* randomized delay between events */
152 usleep(((random() % ep->fd[1]) + 1) * ep->fd[1]);
153
154 if ((random() % write_freq) == 0) {
155 write_data(ep);
156 } else {
157 recycle_fds(ep);
158 }
159 }
160 }
161
162 struct selarg {
163 struct thread_sync_arg *th;
164 fd_set def_readfds;
165 int max_fd;
166 int nthreads;
167 int ret;
168
169 pthread_t pth;
170 };
171
172 /*
173 * Put the actual call to select in its own thread so we can catch errors that
174 * occur only the first time a thread calls select.
175 */
176 static void *
do_select(void * arg)177 do_select(void *arg)
178 {
179 struct selarg *sarg = (struct selarg *)arg;
180 struct timeval timeout;
181 struct timeval *tp = NULL;
182 fd_set readfds;
183 int nfd;
184
185 sarg->ret = 0;
186
187 FD_COPY(&sarg->def_readfds, &readfds);
188
189 /* Add a timeout probablistically */
190 if ((random() % TIMEOUT_CHANCE) == 0) {
191 timeout.tv_sec = random() % 1;
192 timeout.tv_usec = ((random() % TIMEOUT_POLLCHANCE) * TIMEOUT_SCALE);
193 tp = &timeout;
194 }
195
196 /* Do the select */
197 nfd = select(sarg->max_fd + 1, &readfds, 0, 0, tp);
198 if (nfd < 0) {
199 /* EBADF: fd_set has changed */
200 if (errno == EBADF) {
201 sarg->ret = EBADF;
202 return NULL;
203 }
204
205 /* Other errors are fatal */
206 T_QUIET;
207 T_WITH_ERRNO;
208 T_ASSERT_POSIX_SUCCESS(nfd, "select:stress");
209 }
210
211 /* Fast: handle timeouts */
212 if (nfd == 0) {
213 return NULL;
214 }
215
216 /* Slower: discard read input thrown at us from threads */
217 for (int i = 0; i < sarg->nthreads; i++) {
218 struct endpoint *ep = &sarg->th[i].ep;
219
220 if (FD_ISSET(ep->fd[0], &readfds)) {
221 char c;
222 (void)read(ep->fd[0], &c, 1);
223 }
224 }
225
226 return NULL;
227 }
228
229
230 static void
test_select_stress(int nthreads,uint64_t duration_seconds)231 test_select_stress(int nthreads, uint64_t duration_seconds)
232 {
233 uint64_t deadline;
234 uint64_t seconds_remain, last_print_time;
235
236 struct selarg sarg;
237
238 int started_threads = 0;
239 struct thread_sync_arg *th;
240
241 if (nthreads < 2) {
242 T_LOG("forcing a minimum of 2 threads");
243 nthreads = 2;
244 }
245
246 /*
247 * Allocate memory for endpoint data
248 */
249 th = calloc(nthreads, sizeof(*th));
250 T_QUIET;
251 T_ASSERT_NOTNULL(th, "select_stress: No memory for thread endpoints");
252
253 T_LOG("Select stress test: %d threads, for %lld seconds", nthreads, duration_seconds);
254
255 /*
256 * Startup all the threads
257 */
258 T_LOG("\tcreating threads...");
259 for (int i = 0; i < nthreads; i++) {
260 struct endpoint *e = &th[i].ep;
261 th[i].setup = setup_stress_event;
262 th[i].work = do_stress_events;
263 T_QUIET;
264 T_WITH_ERRNO;
265 T_ASSERT_POSIX_ZERO(pthread_create(&e->pth, 0, thread_sync, &th[i]),
266 "pthread_create:do_stress_events");
267 }
268
269 /*
270 * Wait for all the threads to start up
271 */
272 while (started_threads < nthreads) {
273 if (semaphore_wait(g_sync_sem) == KERN_SUCCESS) {
274 ++started_threads;
275 }
276 }
277
278 /*
279 * Kick everyone off
280 */
281 semaphore_signal_all(g_thread_sem);
282
283 /*
284 * Calculate a stop time
285 */
286 deadline = mach_absolute_time() + ns_to_abs(duration_seconds * NSEC_PER_SEC);
287 seconds_remain = duration_seconds;
288 last_print_time = seconds_remain + 1;
289
290 /*
291 * Perform the select and read any data that comes from the
292 * constituent thread FDs.
293 */
294
295 T_LOG("\ttest running!");
296 handle_ebadf:
297 /* (re) set up the select fd set */
298 sarg.max_fd = 0;
299 FD_ZERO(&sarg.def_readfds);
300 for (int i = 0; i < nthreads; i++) {
301 struct endpoint *ep = &th[i].ep;
302
303 FD_SET(ep->fd[0], &sarg.def_readfds);
304 if (ep->fd[0] > sarg.max_fd) {
305 sarg.max_fd = ep->fd[0];
306 }
307 }
308
309 sarg.th = th;
310 sarg.nthreads = nthreads;
311
312 while (mach_absolute_time() < deadline) {
313 void *thret = NULL;
314
315 seconds_remain = abs_to_ns(deadline - mach_absolute_time()) / NSEC_PER_SEC;
316 if (last_print_time > seconds_remain) {
317 T_LOG(" %6lld...", seconds_remain);
318 last_print_time = seconds_remain;
319 }
320
321 sarg.ret = 0;
322 T_QUIET;
323 T_WITH_ERRNO;
324 T_ASSERT_POSIX_ZERO(pthread_create(&sarg.pth, 0, do_select, &sarg),
325 "pthread_create:do_select");
326
327 T_QUIET;
328 T_WITH_ERRNO;
329 T_ASSERT_POSIX_ZERO(pthread_cancel(sarg.pth), "pthread_cancel");
330 T_QUIET;
331 T_WITH_ERRNO;
332 T_ASSERT_POSIX_ZERO(pthread_join(sarg.pth, &thret), "pthread_join");
333
334 if (sarg.ret == EBADF) {
335 goto handle_ebadf;
336 }
337 T_QUIET;
338 T_ASSERT_GE(sarg.ret, 0, "threaded do_select returned an \
339 error: %d!", sarg.ret);
340 }
341
342 T_PASS("select stress test passed");
343 }
344
345
346 /*
347 * TEST: use select as sleep()
348 */
349 static void
test_select_sleep(uint32_t niterations,unsigned long usecs)350 test_select_sleep(uint32_t niterations, unsigned long usecs)
351 {
352 int ret;
353 struct timeval tv;
354 tv.tv_sec = 0;
355 tv.tv_usec = usecs;
356
357 if (!niterations) {
358 T_FAIL("select sleep test skipped");
359 return;
360 }
361
362 T_LOG("Testing select as sleep (n=%d, us=%ld)...", niterations, usecs);
363
364 while (niterations--) {
365 ret = select(0, NULL, NULL, NULL, &tv);
366 if (ret < 0 && errno != EINTR) {
367 T_QUIET;
368 T_WITH_ERRNO;
369 T_ASSERT_POSIX_SUCCESS(ret, "select:sleep");
370 }
371 }
372
373 T_PASS("select sleep test passed");
374 }
375
376 #define get_env_arg(NM, sval, val) \
377 do { \
378 sval = getenv(#NM); \
379 if (sval) { \
380 long v = atol(sval); \
381 if (v <= 0) \
382 v =1 ; \
383 val = (typeof(val))v; \
384 } \
385 } while (0)
386
387 T_DECL(select_sleep, "select sleep test for rdar://problem/20804876 Gala: select with no FDs leaks waitq table objects (causes asserts/panics)")
388 {
389 char *env_sval = NULL;
390
391 get_env_arg(SELSLEEP_ITERATIONS, env_sval, g_sleep_iterations);
392 get_env_arg(SELSLEEP_INTERVAL, env_sval, g_sleep_usecs);
393
394 test_select_sleep((uint32_t)g_sleep_iterations, (unsigned long)g_sleep_usecs);
395 }
396
397 T_DECL(select_stress, "select stress test for rdar://problem/20804876 Gala: select with no FDs leaks waitq table objects (causes asserts/panics)")
398 {
399 char *env_sval = NULL;
400
401 T_QUIET;
402 T_ASSERT_MACH_SUCCESS(mach_timebase_info(&g_timebase),
403 "Can't get mach_timebase_info!");
404
405 get_env_arg(SELSTRESS_THREADS, env_sval, g_stress_nthreads);
406 get_env_arg(SELSTRESS_DURATION, env_sval, g_stress_duration);
407
408 T_QUIET;
409 T_ASSERT_MACH_SUCCESS(semaphore_create(mach_task_self(), &g_sync_sem, SYNC_POLICY_FIFO, 0),
410 "semaphore_create(g_sync_sem)");
411 T_QUIET;
412 T_ASSERT_MACH_SUCCESS(semaphore_create(mach_task_self(), &g_thread_sem, SYNC_POLICY_FIFO, 0),
413 "semaphore_create(g_thread_sem)");
414
415 test_select_stress(g_stress_nthreads, g_stress_duration);
416 }
417