tests/select_stress.c

   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 = 60;
  42
  43 static inline uint64_t
  44 ns_to_abs(uint64_t ns)
  45 {
  46         return ns * g_timebase.denom / g_timebase.numer;
  47 }
  48
  49 static inline uint64_t
  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 *
  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
  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
 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
 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 *
 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 *
 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
 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
 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 }