1 /* $NetBSD: regress_thread.c,v 1.4 2013/04/12 20:00:21 christos Exp $ */ 2 /* 3 * Copyright (c) 2007-2012 Niels Provos and Nick Mathewson 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions 7 * are met: 8 * 1. Redistributions of source code must retain the above copyright 9 * notice, this list of conditions and the following disclaimer. 10 * 2. Redistributions in binary form must reproduce the above copyright 11 * notice, this list of conditions and the following disclaimer in the 12 * documentation and/or other materials provided with the distribution. 13 * 3. The name of the author may not be used to endorse or promote products 14 * derived from this software without specific prior written permission. 15 * 16 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR 17 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES 18 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. 19 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, 20 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT 21 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, 22 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 23 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT 24 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF 25 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 26 */ 27 28 /* The old tests here need assertions to work. */ 29 #undef NDEBUG 30 31 #include "event2/event-config.h" 32 #include <sys/cdefs.h> 33 __RCSID("$NetBSD: regress_thread.c,v 1.4 2013/04/12 20:00:21 christos Exp $"); 34 35 #include <sys/types.h> 36 #include <stdio.h> 37 #include <stdlib.h> 38 #include <string.h> 39 #ifdef _EVENT_HAVE_UNISTD_H 40 #include <unistd.h> 41 #endif 42 #ifdef _EVENT_HAVE_SYS_WAIT_H 43 #include <sys/wait.h> 44 #endif 45 46 #ifdef _EVENT_HAVE_PTHREADS 47 #include <pthread.h> 48 #elif defined(WIN32) 49 #include <process.h> 50 #endif 51 #include <assert.h> 52 #ifdef _EVENT_HAVE_UNISTD_H 53 #include <unistd.h> 54 #endif 55 #include <time.h> 56 57 #include "sys/queue.h" 58 59 #include "event2/util.h" 60 #include "event2/event.h" 61 #include "event2/event_struct.h" 62 #include "event2/thread.h" 63 #include "evthread-internal.h" 64 #include "event-internal.h" 65 #include "defer-internal.h" 66 #include "regress.h" 67 #include "tinytest_macros.h" 68 69 #ifdef _EVENT_HAVE_PTHREADS 70 #define THREAD_T pthread_t 71 #define THREAD_FN void * 72 #define THREAD_RETURN() return (NULL) 73 #define THREAD_START(threadvar, fn, arg) \ 74 pthread_create(&(threadvar), NULL, fn, arg) 75 #define THREAD_JOIN(th) pthread_join(th, NULL) 76 #else 77 #define THREAD_T HANDLE 78 #define THREAD_FN unsigned __stdcall 79 #define THREAD_RETURN() return (0) 80 #define THREAD_START(threadvar, fn, arg) do { \ 81 uintptr_t threadhandle = _beginthreadex(NULL,0,fn,(arg),0,NULL); \ 82 (threadvar) = (HANDLE) threadhandle; \ 83 } while (/*CONSTCOND*/0) 84 #define THREAD_JOIN(th) WaitForSingleObject(th, INFINITE) 85 #endif 86 87 struct cond_wait { 88 void *lock; 89 void *cond; 90 }; 91 92 static void 93 wake_all_timeout(evutil_socket_t fd, short what, void *arg) 94 { 95 struct cond_wait *cw = arg; 96 EVLOCK_LOCK(cw->lock, 0); 97 EVTHREAD_COND_BROADCAST(cw->cond); 98 EVLOCK_UNLOCK(cw->lock, 0); 99 100 } 101 102 #if 0 103 static void 104 wake_one_timeout(evutil_socket_t fd, short what, void *arg) 105 { 106 struct cond_wait *cw = arg; 107 EVLOCK_LOCK(cw->lock, 0); 108 EVTHREAD_COND_SIGNAL(cw->cond); 109 EVLOCK_UNLOCK(cw->lock, 0); 110 } 111 #endif 112 113 #define NUM_THREADS 100 114 #define NUM_ITERATIONS 100 115 void *count_lock; 116 static int count; 117 118 static THREAD_FN 119 basic_thread(void *arg) 120 { 121 struct cond_wait cw; 122 struct event_base *base = arg; 123 struct event ev; 124 int i = 0; 125 126 EVTHREAD_ALLOC_LOCK(cw.lock, 0); 127 EVTHREAD_ALLOC_COND(cw.cond); 128 assert(cw.lock); 129 assert(cw.cond); 130 131 evtimer_assign(&ev, base, wake_all_timeout, &cw); 132 for (i = 0; i < NUM_ITERATIONS; i++) { 133 struct timeval tv; 134 evutil_timerclear(&tv); 135 tv.tv_sec = 0; 136 tv.tv_usec = 3000; 137 138 EVLOCK_LOCK(cw.lock, 0); 139 /* we need to make sure that event does not happen before 140 * we get to wait on the conditional variable */ 141 assert(evtimer_add(&ev, &tv) == 0); 142 143 assert(EVTHREAD_COND_WAIT(cw.cond, cw.lock) == 0); 144 EVLOCK_UNLOCK(cw.lock, 0); 145 146 EVLOCK_LOCK(count_lock, 0); 147 ++count; 148 EVLOCK_UNLOCK(count_lock, 0); 149 } 150 151 /* exit the loop only if all threads fired all timeouts */ 152 EVLOCK_LOCK(count_lock, 0); 153 if (count >= NUM_THREADS * NUM_ITERATIONS) 154 event_base_loopexit(base, NULL); 155 EVLOCK_UNLOCK(count_lock, 0); 156 157 EVTHREAD_FREE_LOCK(cw.lock, 0); 158 EVTHREAD_FREE_COND(cw.cond); 159 160 THREAD_RETURN(); 161 } 162 163 static int notification_fd_used = 0; 164 #ifndef WIN32 165 static int got_sigchld = 0; 166 static void 167 sigchld_cb(evutil_socket_t fd, short event, void *arg) 168 { 169 struct timeval tv; 170 struct event_base *base = arg; 171 172 got_sigchld++; 173 tv.tv_usec = 100000; 174 tv.tv_sec = 0; 175 event_base_loopexit(base, &tv); 176 } 177 178 179 static void 180 notify_fd_cb(evutil_socket_t fd, short event, void *arg) 181 { 182 ++notification_fd_used; 183 } 184 #endif 185 186 static void 187 thread_basic(void *arg) 188 { 189 THREAD_T threads[NUM_THREADS]; 190 struct event ev; 191 struct timeval tv; 192 int i; 193 struct basic_test_data *data = arg; 194 struct event_base *base = data->base; 195 196 struct event *notification_event = NULL; 197 struct event *sigchld_event = NULL; 198 199 EVTHREAD_ALLOC_LOCK(count_lock, 0); 200 tt_assert(count_lock); 201 202 tt_assert(base); 203 if (evthread_make_base_notifiable(base)<0) { 204 tt_abort_msg("Couldn't make base notifiable!"); 205 } 206 207 #ifndef WIN32 208 if (data->setup_data && !strcmp(data->setup_data, "forking")) { 209 pid_t pid; 210 int status; 211 sigchld_event = evsignal_new(base, SIGCHLD, sigchld_cb, base); 212 /* This piggybacks on the th_notify_fd weirdly, and looks 213 * inside libevent internals. Not a good idea in non-testing 214 * code! */ 215 notification_event = event_new(base, 216 base->th_notify_fd[0], EV_READ|EV_PERSIST, notify_fd_cb, 217 NULL); 218 event_add(sigchld_event, NULL); 219 event_add(notification_event, NULL); 220 221 if ((pid = fork()) == 0) { 222 event_del(notification_event); 223 if (event_reinit(base) < 0) { 224 TT_FAIL(("reinit")); 225 exit(1); 226 } 227 event_assign(notification_event, base, 228 base->th_notify_fd[0], EV_READ|EV_PERSIST, 229 notify_fd_cb, NULL); 230 event_add(notification_event, NULL); 231 goto child; 232 } 233 234 event_base_dispatch(base); 235 236 if (waitpid(pid, &status, 0) == -1) 237 tt_abort_perror("waitpid"); 238 TT_BLATHER(("Waitpid okay\n")); 239 240 tt_assert(got_sigchld); 241 tt_int_op(notification_fd_used, ==, 0); 242 243 goto end; 244 } 245 246 child: 247 #endif 248 for (i = 0; i < NUM_THREADS; ++i) 249 THREAD_START(threads[i], basic_thread, base); 250 251 evtimer_assign(&ev, base, NULL, NULL); 252 evutil_timerclear(&tv); 253 tv.tv_sec = 1000; 254 event_add(&ev, &tv); 255 256 event_base_dispatch(base); 257 258 for (i = 0; i < NUM_THREADS; ++i) 259 THREAD_JOIN(threads[i]); 260 261 event_del(&ev); 262 263 tt_int_op(count, ==, NUM_THREADS * NUM_ITERATIONS); 264 265 EVTHREAD_FREE_LOCK(count_lock, 0); 266 267 TT_BLATHER(("notifiations==%d", notification_fd_used)); 268 269 end: 270 271 if (notification_event) 272 event_free(notification_event); 273 if (sigchld_event) 274 event_free(sigchld_event); 275 } 276 277 #undef NUM_THREADS 278 #define NUM_THREADS 10 279 280 struct alerted_record { 281 struct cond_wait *cond; 282 struct timeval delay; 283 struct timeval alerted_at; 284 int timed_out; 285 }; 286 287 #if 0 288 static THREAD_FN 289 wait_for_condition(void *arg) 290 { 291 struct alerted_record *rec = arg; 292 int r; 293 294 EVLOCK_LOCK(rec->cond->lock, 0); 295 if (rec->delay.tv_sec || rec->delay.tv_usec) { 296 r = EVTHREAD_COND_WAIT_TIMED(rec->cond->cond, rec->cond->lock, 297 &rec->delay); 298 } else { 299 r = EVTHREAD_COND_WAIT(rec->cond->cond, rec->cond->lock); 300 } 301 EVLOCK_UNLOCK(rec->cond->lock, 0); 302 303 evutil_gettimeofday(&rec->alerted_at, NULL); 304 if (r == 1) 305 rec->timed_out = 1; 306 307 THREAD_RETURN(); 308 } 309 310 static void 311 thread_conditions_simple(void *arg) 312 { 313 struct timeval tv_signal, tv_timeout, tv_broadcast; 314 struct alerted_record alerted[NUM_THREADS]; 315 THREAD_T threads[NUM_THREADS]; 316 struct cond_wait cond; 317 int i; 318 struct timeval launched_at; 319 struct event wake_one; 320 struct event wake_all; 321 struct basic_test_data *data = arg; 322 struct event_base *base = data->base; 323 int n_timed_out=0, n_signal=0, n_broadcast=0; 324 325 tv_signal.tv_sec = tv_timeout.tv_sec = tv_broadcast.tv_sec = 0; 326 tv_signal.tv_usec = 30*1000; 327 tv_timeout.tv_usec = 150*1000; 328 tv_broadcast.tv_usec = 500*1000; 329 330 EVTHREAD_ALLOC_LOCK(cond.lock, EVTHREAD_LOCKTYPE_RECURSIVE); 331 EVTHREAD_ALLOC_COND(cond.cond); 332 tt_assert(cond.lock); 333 tt_assert(cond.cond); 334 for (i = 0; i < NUM_THREADS; ++i) { 335 memset(&alerted[i], 0, sizeof(struct alerted_record)); 336 alerted[i].cond = &cond; 337 } 338 339 /* Threads 5 and 6 will be allowed to time out */ 340 memcpy(&alerted[5].delay, &tv_timeout, sizeof(tv_timeout)); 341 memcpy(&alerted[6].delay, &tv_timeout, sizeof(tv_timeout)); 342 343 evtimer_assign(&wake_one, base, wake_one_timeout, &cond); 344 evtimer_assign(&wake_all, base, wake_all_timeout, &cond); 345 346 evutil_gettimeofday(&launched_at, NULL); 347 348 /* Launch the threads... */ 349 for (i = 0; i < NUM_THREADS; ++i) { 350 THREAD_START(threads[i], wait_for_condition, &alerted[i]); 351 } 352 353 /* Start the timers... */ 354 tt_int_op(event_add(&wake_one, &tv_signal), ==, 0); 355 tt_int_op(event_add(&wake_all, &tv_broadcast), ==, 0); 356 357 /* And run for a bit... */ 358 event_base_dispatch(base); 359 360 /* And wait till the threads are done. */ 361 for (i = 0; i < NUM_THREADS; ++i) 362 THREAD_JOIN(threads[i]); 363 364 /* Now, let's see what happened. At least one of 5 or 6 should 365 * have timed out. */ 366 n_timed_out = alerted[5].timed_out + alerted[6].timed_out; 367 tt_int_op(n_timed_out, >=, 1); 368 tt_int_op(n_timed_out, <=, 2); 369 370 for (i = 0; i < NUM_THREADS; ++i) { 371 const struct timeval *target_delay; 372 struct timeval target_time, actual_delay; 373 if (alerted[i].timed_out) { 374 TT_BLATHER(("%d looks like a timeout\n", i)); 375 target_delay = &tv_timeout; 376 tt_assert(i == 5 || i == 6); 377 } else if (evutil_timerisset(&alerted[i].alerted_at)) { 378 long diff1,diff2; 379 evutil_timersub(&alerted[i].alerted_at, 380 &launched_at, &actual_delay); 381 diff1 = timeval_msec_diff(&actual_delay, 382 &tv_signal); 383 diff2 = timeval_msec_diff(&actual_delay, 384 &tv_broadcast); 385 if (abs(diff1) < abs(diff2)) { 386 TT_BLATHER(("%d looks like a signal\n", i)); 387 target_delay = &tv_signal; 388 ++n_signal; 389 } else { 390 TT_BLATHER(("%d looks like a broadcast\n", i)); 391 target_delay = &tv_broadcast; 392 ++n_broadcast; 393 } 394 } else { 395 TT_FAIL(("Thread %d never got woken", i)); 396 continue; 397 } 398 evutil_timeradd(target_delay, &launched_at, &target_time); 399 test_timeval_diff_leq(&target_time, &alerted[i].alerted_at, 400 0, 150); 401 } 402 tt_int_op(n_broadcast + n_signal + n_timed_out, ==, NUM_THREADS); 403 tt_int_op(n_signal, ==, 1); 404 405 end: 406 ; 407 } 408 #endif 409 410 #define CB_COUNT 128 411 #define QUEUE_THREAD_COUNT 8 412 413 #ifdef WIN32 414 #define SLEEP_MS(ms) Sleep(ms) 415 #else 416 #define SLEEP_MS(ms) usleep((ms) * 1000) 417 #endif 418 419 struct deferred_test_data { 420 struct deferred_cb cbs[CB_COUNT]; 421 struct deferred_cb_queue *queue; 422 }; 423 424 static time_t timer_start = 0; 425 static time_t timer_end = 0; 426 static unsigned callback_count = 0; 427 static THREAD_T load_threads[QUEUE_THREAD_COUNT]; 428 static struct deferred_test_data deferred_data[QUEUE_THREAD_COUNT]; 429 430 static void 431 deferred_callback(struct deferred_cb *cb, void *arg) 432 { 433 SLEEP_MS(1); 434 callback_count += 1; 435 } 436 437 static THREAD_FN 438 load_deferred_queue(void *arg) 439 { 440 struct deferred_test_data *data = arg; 441 size_t i; 442 443 for (i = 0; i < CB_COUNT; ++i) { 444 event_deferred_cb_init(&data->cbs[i], deferred_callback, NULL); 445 event_deferred_cb_schedule(data->queue, &data->cbs[i]); 446 SLEEP_MS(1); 447 } 448 449 THREAD_RETURN(); 450 } 451 452 static void 453 timer_callback(evutil_socket_t fd, short what, void *arg) 454 { 455 timer_end = time(NULL); 456 } 457 458 static void 459 start_threads_callback(evutil_socket_t fd, short what, void *arg) 460 { 461 int i; 462 463 for (i = 0; i < QUEUE_THREAD_COUNT; ++i) { 464 THREAD_START(load_threads[i], load_deferred_queue, 465 &deferred_data[i]); 466 } 467 } 468 469 static void 470 thread_deferred_cb_skew(void *arg) 471 { 472 struct basic_test_data *data = arg; 473 struct timeval tv_timer = {4, 0}; 474 struct deferred_cb_queue *queue; 475 time_t elapsed; 476 int i; 477 478 queue = event_base_get_deferred_cb_queue(data->base); 479 tt_assert(queue); 480 481 for (i = 0; i < QUEUE_THREAD_COUNT; ++i) 482 deferred_data[i].queue = queue; 483 484 timer_start = time(NULL); 485 event_base_once(data->base, -1, EV_TIMEOUT, timer_callback, NULL, 486 &tv_timer); 487 event_base_once(data->base, -1, EV_TIMEOUT, start_threads_callback, 488 NULL, NULL); 489 event_base_dispatch(data->base); 490 491 elapsed = timer_end - timer_start; 492 TT_BLATHER(("callback count, %u", callback_count)); 493 TT_BLATHER(("elapsed time, %u", (unsigned)elapsed)); 494 /* XXX be more intelligent here. just make sure skew is 495 * within 2 seconds for now. */ 496 tt_assert(elapsed >= 4 && elapsed <= 6); 497 498 end: 499 for (i = 0; i < QUEUE_THREAD_COUNT; ++i) 500 THREAD_JOIN(load_threads[i]); 501 } 502 503 #define TEST(name) \ 504 { #name, thread_##name, TT_FORK|TT_NEED_THREADS|TT_NEED_BASE, \ 505 &basic_setup, NULL } 506 507 struct testcase_t thread_testcases[] = { 508 { "basic", thread_basic, TT_FORK|TT_NEED_THREADS|TT_NEED_BASE, 509 &basic_setup, NULL }, 510 #ifndef WIN32 511 { "forking", thread_basic, TT_FORK|TT_NEED_THREADS|TT_NEED_BASE, 512 &basic_setup, __UNCONST("forking") }, 513 #endif 514 #if 0 515 TEST(conditions_simple), 516 #endif 517 TEST(deferred_cb_skew), 518 END_OF_TESTCASES 519 }; 520 521