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
wake_all_timeout(evutil_socket_t fd,short what,void * arg)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
basic_thread(void * arg)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
sigchld_cb(evutil_socket_t fd,short event,void * arg)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
notify_fd_cb(evutil_socket_t fd,short event,void * arg)180 notify_fd_cb(evutil_socket_t fd, short event, void *arg)
181 {
182 ++notification_fd_used;
183 }
184 #endif
185
186 static void
thread_basic(void * arg)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
deferred_callback(struct deferred_cb * cb,void * arg)431 deferred_callback(struct deferred_cb *cb, void *arg)
432 {
433 SLEEP_MS(1);
434 callback_count += 1;
435 }
436
437 static THREAD_FN
load_deferred_queue(void * arg)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
timer_callback(evutil_socket_t fd,short what,void * arg)453 timer_callback(evutil_socket_t fd, short what, void *arg)
454 {
455 timer_end = time(NULL);
456 }
457
458 static void
start_threads_callback(evutil_socket_t fd,short what,void * arg)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
thread_deferred_cb_skew(void * arg)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