1 /* $NetBSD: regress_bufferevent.c,v 1.4 2021/04/07 03:36:48 christos Exp $ */
2
3 /*
4 * Copyright (c) 2003-2007 Niels Provos <provos@citi.umich.edu>
5 * Copyright (c) 2007-2012 Niels Provos and Nick Mathewson
6 *
7 * Redistribution and use in source and binary forms, with or without
8 * modification, are permitted provided that the following conditions
9 * are met:
10 * 1. Redistributions of source code must retain the above copyright
11 * notice, this list of conditions and the following disclaimer.
12 * 2. Redistributions in binary form must reproduce the above copyright
13 * notice, this list of conditions and the following disclaimer in the
14 * documentation and/or other materials provided with the distribution.
15 * 3. The name of the author may not be used to endorse or promote products
16 * derived from this software without specific prior written permission.
17 *
18 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
19 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
20 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
21 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
22 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
23 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
24 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
25 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
26 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
27 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
28 */
29 #include "util-internal.h"
30
31 /* The old tests here need assertions to work. */
32 #undef NDEBUG
33
34 /**
35 * - clang supports __has_feature
36 * - gcc supports __SANITIZE_ADDRESS__
37 *
38 * Let's set __SANITIZE_ADDRESS__ if __has_feature(address_sanitizer)
39 */
40 #ifndef __has_feature
41 #define __has_feature(x) 0
42 #endif
43 #if !defined(__SANITIZE_ADDRESS__) && __has_feature(address_sanitizer)
44 #define __SANITIZE_ADDRESS__
45 #endif
46
47 #ifdef _WIN32
48 #include <winsock2.h>
49 #include <windows.h>
50 #endif
51
52 #include "event2/event-config.h"
53 #include <sys/cdefs.h>
54 __RCSID("$NetBSD: regress_bufferevent.c,v 1.4 2021/04/07 03:36:48 christos Exp $");
55
56 #include <sys/types.h>
57 #include <sys/stat.h>
58 #ifdef EVENT__HAVE_SYS_TIME_H
59 #include <sys/time.h>
60 #endif
61 #include <sys/queue.h>
62 #ifndef _WIN32
63 #include <sys/socket.h>
64 #include <sys/wait.h>
65 #include <signal.h>
66 #include <unistd.h>
67 #include <netdb.h>
68 #include <netinet/in.h>
69 #endif
70 #include <fcntl.h>
71 #include <signal.h>
72 #include <stdlib.h>
73 #include <stdio.h>
74 #include <string.h>
75 #include <errno.h>
76 #include <assert.h>
77
78 #ifdef EVENT__HAVE_ARPA_INET_H
79 #include <arpa/inet.h>
80 #endif
81
82 #include "event2/event-config.h"
83 #include "event2/event.h"
84 #include "event2/event_struct.h"
85 #include "event2/event_compat.h"
86 #include "event2/tag.h"
87 #include "event2/buffer.h"
88 #include "event2/bufferevent.h"
89 #include "event2/bufferevent_compat.h"
90 #include "event2/bufferevent_struct.h"
91 #include "event2/listener.h"
92 #include "event2/util.h"
93
94 #include "bufferevent-internal.h"
95 #include "evthread-internal.h"
96 #include "util-internal.h"
97 #ifdef _WIN32
98 #include "iocp-internal.h"
99 #endif
100
101 #include "regress.h"
102 #include "regress_testutils.h"
103
104 /*
105 * simple bufferevent test
106 */
107
108 static void
readcb(struct bufferevent * bev,void * arg)109 readcb(struct bufferevent *bev, void *arg)
110 {
111 if (evbuffer_get_length(bev->input) == 8333) {
112 struct evbuffer *evbuf = evbuffer_new();
113 assert(evbuf != NULL);
114
115 /* gratuitous test of bufferevent_read_buffer */
116 bufferevent_read_buffer(bev, evbuf);
117
118 bufferevent_disable(bev, EV_READ);
119
120 if (evbuffer_get_length(evbuf) == 8333) {
121 test_ok++;
122 }
123
124 evbuffer_free(evbuf);
125 }
126 }
127
128 static void
writecb(struct bufferevent * bev,void * arg)129 writecb(struct bufferevent *bev, void *arg)
130 {
131 if (evbuffer_get_length(bev->output) == 0) {
132 test_ok++;
133 }
134 }
135
136 static void
errorcb(struct bufferevent * bev,short what,void * arg)137 errorcb(struct bufferevent *bev, short what, void *arg)
138 {
139 test_ok = -2;
140 }
141
142 static void
test_bufferevent_impl(int use_pair,int flush)143 test_bufferevent_impl(int use_pair, int flush)
144 {
145 struct bufferevent *bev1 = NULL, *bev2 = NULL;
146 char buffer[8333];
147 int i;
148 int expected = 2;
149
150 if (use_pair) {
151 struct bufferevent *xpair[2];
152 tt_assert(0 == bufferevent_pair_new(NULL, 0, xpair));
153 bev1 = xpair[0];
154 bev2 = xpair[1];
155 bufferevent_setcb(bev1, readcb, writecb, errorcb, bev1);
156 bufferevent_setcb(bev2, readcb, writecb, errorcb, NULL);
157 tt_fd_op(bufferevent_getfd(bev1), ==, EVUTIL_INVALID_SOCKET);
158 tt_ptr_op(bufferevent_get_underlying(bev1), ==, NULL);
159 tt_ptr_op(bufferevent_pair_get_partner(bev1), ==, bev2);
160 tt_ptr_op(bufferevent_pair_get_partner(bev2), ==, bev1);
161 } else {
162 bev1 = bufferevent_new(pair[0], readcb, writecb, errorcb, NULL);
163 bev2 = bufferevent_new(pair[1], readcb, writecb, errorcb, NULL);
164 tt_fd_op(bufferevent_getfd(bev1), ==, pair[0]);
165 tt_ptr_op(bufferevent_get_underlying(bev1), ==, NULL);
166 tt_ptr_op(bufferevent_pair_get_partner(bev1), ==, NULL);
167 tt_ptr_op(bufferevent_pair_get_partner(bev2), ==, NULL);
168 }
169
170 {
171 /* Test getcb. */
172 bufferevent_data_cb r, w;
173 bufferevent_event_cb e;
174 void *a;
175 bufferevent_getcb(bev1, &r, &w, &e, &a);
176 tt_ptr_op(r, ==, readcb);
177 tt_ptr_op(w, ==, writecb);
178 tt_ptr_op(e, ==, errorcb);
179 tt_ptr_op(a, ==, use_pair ? bev1 : NULL);
180 }
181
182 bufferevent_disable(bev1, EV_READ);
183 bufferevent_enable(bev2, EV_READ);
184
185 tt_int_op(bufferevent_get_enabled(bev1), ==, EV_WRITE);
186 tt_int_op(bufferevent_get_enabled(bev2), ==, EV_WRITE|EV_READ);
187
188 for (i = 0; i < (int)sizeof(buffer); i++)
189 buffer[i] = i;
190
191 bufferevent_write(bev1, buffer, sizeof(buffer));
192 if (flush >= 0) {
193 tt_int_op(bufferevent_flush(bev1, EV_WRITE, flush), >=, 0);
194 }
195
196 event_dispatch();
197
198 bufferevent_free(bev2);
199 tt_ptr_op(bufferevent_pair_get_partner(bev1), ==, NULL);
200 bufferevent_free(bev1);
201
202 /** Only pair call errorcb for BEV_FINISHED */
203 if (use_pair && flush == BEV_FINISHED) {
204 expected = -1;
205 }
206 if (test_ok != expected)
207 test_ok = 0;
208 end:
209 ;
210 }
211
test_bufferevent(void)212 static void test_bufferevent(void) { test_bufferevent_impl(0, -1); }
test_bufferevent_pair(void)213 static void test_bufferevent_pair(void) { test_bufferevent_impl(1, -1); }
214
test_bufferevent_flush_normal(void)215 static void test_bufferevent_flush_normal(void) { test_bufferevent_impl(0, BEV_NORMAL); }
test_bufferevent_flush_flush(void)216 static void test_bufferevent_flush_flush(void) { test_bufferevent_impl(0, BEV_FLUSH); }
test_bufferevent_flush_finished(void)217 static void test_bufferevent_flush_finished(void) { test_bufferevent_impl(0, BEV_FINISHED); }
218
test_bufferevent_pair_flush_normal(void)219 static void test_bufferevent_pair_flush_normal(void) { test_bufferevent_impl(1, BEV_NORMAL); }
test_bufferevent_pair_flush_flush(void)220 static void test_bufferevent_pair_flush_flush(void) { test_bufferevent_impl(1, BEV_FLUSH); }
test_bufferevent_pair_flush_finished(void)221 static void test_bufferevent_pair_flush_finished(void) { test_bufferevent_impl(1, BEV_FINISHED); }
222
223 #if defined(EVTHREAD_USE_PTHREADS_IMPLEMENTED) && !defined(__SANITIZE_ADDRESS__)
224 /**
225 * Trace lock/unlock/alloc/free for locks.
226 * (More heavier then evthread_debug*)
227 */
228 typedef struct
229 {
230 void *lock;
231 enum {
232 ALLOC, FREE,
233 } status;
234 size_t locked /** allow recursive locking */;
235 } lock_wrapper;
236 struct lock_unlock_base
237 {
238 /* Original callbacks */
239 struct evthread_lock_callbacks cbs;
240 /* Map of locks */
241 lock_wrapper *locks;
242 size_t nr_locks;
243 } lu_base = {
244 .locks = NULL,
245 };
246
lu_find(void * lock_)247 static lock_wrapper *lu_find(void *lock_)
248 {
249 size_t i;
250 for (i = 0; i < lu_base.nr_locks; ++i) {
251 lock_wrapper *lock = &lu_base.locks[i];
252 if (lock->lock == lock_)
253 return lock;
254 }
255 return NULL;
256 }
257
trace_lock_alloc(unsigned locktype)258 static void *trace_lock_alloc(unsigned locktype)
259 {
260 void *lock;
261 ++lu_base.nr_locks;
262 lu_base.locks = realloc(lu_base.locks,
263 sizeof(lock_wrapper) * lu_base.nr_locks);
264 lock = lu_base.cbs.alloc(locktype);
265 lu_base.locks[lu_base.nr_locks - 1] = (lock_wrapper){ lock, ALLOC, 0 };
266 return lock;
267 }
trace_lock_free(void * lock_,unsigned locktype)268 static void trace_lock_free(void *lock_, unsigned locktype)
269 {
270 lock_wrapper *lock = lu_find(lock_);
271 if (!lock || lock->status == FREE || lock->locked) {
272 TT_FAIL(("lock: free error"));
273 } else {
274 lock->status = FREE;
275 lu_base.cbs.free(lock_, locktype);
276 }
277 }
trace_lock_lock(unsigned mode,void * lock_)278 static int trace_lock_lock(unsigned mode, void *lock_)
279 {
280 lock_wrapper *lock = lu_find(lock_);
281 if (!lock || lock->status == FREE) {
282 TT_FAIL(("lock: lock error"));
283 return -1;
284 } else {
285 ++lock->locked;
286 return lu_base.cbs.lock(mode, lock_);
287 }
288 }
trace_lock_unlock(unsigned mode,void * lock_)289 static int trace_lock_unlock(unsigned mode, void *lock_)
290 {
291 lock_wrapper *lock = lu_find(lock_);
292 if (!lock || lock->status == FREE || !lock->locked) {
293 TT_FAIL(("lock: unlock error"));
294 return -1;
295 } else {
296 --lock->locked;
297 return lu_base.cbs.unlock(mode, lock_);
298 }
299 }
lock_unlock_free_thread_cbs(void)300 static void lock_unlock_free_thread_cbs(void)
301 {
302 event_base_free(NULL);
303
304 if (libevent_tests_running_in_debug_mode)
305 libevent_global_shutdown();
306
307 /** drop immutable flag */
308 evthread_set_lock_callbacks(NULL);
309 /** avoid calling of event_global_setup_locks_() for new cbs */
310 libevent_global_shutdown();
311 /** drop immutable flag for non-debug ops (since called after shutdown) */
312 evthread_set_lock_callbacks(NULL);
313 }
314
use_lock_unlock_profiler(void)315 static int use_lock_unlock_profiler(void)
316 {
317 struct evthread_lock_callbacks cbs = {
318 EVTHREAD_LOCK_API_VERSION,
319 EVTHREAD_LOCKTYPE_RECURSIVE,
320 trace_lock_alloc,
321 trace_lock_free,
322 trace_lock_lock,
323 trace_lock_unlock,
324 };
325 memcpy(&lu_base.cbs, evthread_get_lock_callbacks(),
326 sizeof(lu_base.cbs));
327 {
328 lock_unlock_free_thread_cbs();
329
330 evthread_set_lock_callbacks(&cbs);
331 /** re-create debug locks correctly */
332 evthread_enable_lock_debugging();
333
334 event_init();
335 }
336 return 0;
337 }
free_lock_unlock_profiler(struct basic_test_data * data)338 static void free_lock_unlock_profiler(struct basic_test_data *data)
339 {
340 /** fix "held_by" for kqueue */
341 evthread_set_lock_callbacks(NULL);
342
343 lock_unlock_free_thread_cbs();
344 free(lu_base.locks);
345 data->base = NULL;
346 }
347
test_bufferevent_pair_release_lock(void * arg)348 static void test_bufferevent_pair_release_lock(void *arg)
349 {
350 struct basic_test_data *data = arg;
351 use_lock_unlock_profiler();
352 {
353 struct bufferevent *xpair[2];
354 if (!bufferevent_pair_new(NULL, BEV_OPT_THREADSAFE, xpair)) {
355 bufferevent_free(xpair[0]);
356 bufferevent_free(xpair[1]);
357 } else
358 tt_abort_perror("bufferevent_pair_new");
359 }
360 free_lock_unlock_profiler(data);
361 end:
362 ;
363 }
364 #endif
365
366 /*
367 * test watermarks and bufferevent
368 */
369
370 static void
wm_readcb(struct bufferevent * bev,void * arg)371 wm_readcb(struct bufferevent *bev, void *arg)
372 {
373 struct evbuffer *evbuf = evbuffer_new();
374 int len = (int)evbuffer_get_length(bev->input);
375 static int nread;
376
377 assert(len >= 10 && len <= 20);
378
379 assert(evbuf != NULL);
380
381 /* gratuitous test of bufferevent_read_buffer */
382 bufferevent_read_buffer(bev, evbuf);
383
384 nread += len;
385 if (nread == 65000) {
386 bufferevent_disable(bev, EV_READ);
387 test_ok++;
388 }
389
390 evbuffer_free(evbuf);
391 }
392
393 static void
wm_writecb(struct bufferevent * bev,void * arg)394 wm_writecb(struct bufferevent *bev, void *arg)
395 {
396 assert(evbuffer_get_length(bev->output) <= 100);
397 if (evbuffer_get_length(bev->output) == 0) {
398 evbuffer_drain(bev->output, evbuffer_get_length(bev->output));
399 test_ok++;
400 }
401 }
402
403 static void
wm_errorcb(struct bufferevent * bev,short what,void * arg)404 wm_errorcb(struct bufferevent *bev, short what, void *arg)
405 {
406 test_ok = -2;
407 }
408
409 static void
test_bufferevent_watermarks_impl(int use_pair)410 test_bufferevent_watermarks_impl(int use_pair)
411 {
412 struct bufferevent *bev1 = NULL, *bev2 = NULL;
413 char buffer[65000];
414 size_t low, high;
415 int i;
416 test_ok = 0;
417
418 if (use_pair) {
419 struct bufferevent *xpair[2];
420 tt_assert(0 == bufferevent_pair_new(NULL, 0, xpair));
421 bev1 = xpair[0];
422 bev2 = xpair[1];
423 bufferevent_setcb(bev1, NULL, wm_writecb, errorcb, NULL);
424 bufferevent_setcb(bev2, wm_readcb, NULL, errorcb, NULL);
425 } else {
426 bev1 = bufferevent_new(pair[0], NULL, wm_writecb, wm_errorcb, NULL);
427 bev2 = bufferevent_new(pair[1], wm_readcb, NULL, wm_errorcb, NULL);
428 }
429 tt_assert(bev1);
430 tt_assert(bev2);
431 bufferevent_disable(bev1, EV_READ);
432 bufferevent_enable(bev2, EV_READ);
433
434 /* By default, low watermarks are set to 0 */
435 bufferevent_getwatermark(bev1, EV_READ, &low, NULL);
436 tt_int_op(low, ==, 0);
437 bufferevent_getwatermark(bev2, EV_WRITE, &low, NULL);
438 tt_int_op(low, ==, 0);
439
440 for (i = 0; i < (int)sizeof(buffer); i++)
441 buffer[i] = (char)i;
442
443 /* limit the reading on the receiving bufferevent */
444 bufferevent_setwatermark(bev2, EV_READ, 10, 20);
445
446 bufferevent_getwatermark(bev2, EV_READ, &low, &high);
447 tt_int_op(low, ==, 10);
448 tt_int_op(high, ==, 20);
449
450 /* Tell the sending bufferevent not to notify us till it's down to
451 100 bytes. */
452 bufferevent_setwatermark(bev1, EV_WRITE, 100, 2000);
453
454 bufferevent_getwatermark(bev1, EV_WRITE, &low, &high);
455 tt_int_op(low, ==, 100);
456 tt_int_op(high, ==, 2000);
457
458 {
459 int r = bufferevent_getwatermark(bev1, EV_WRITE | EV_READ, &low, &high);
460 tt_int_op(r, !=, 0);
461 }
462
463 bufferevent_write(bev1, buffer, sizeof(buffer));
464
465 event_dispatch();
466
467 tt_int_op(test_ok, ==, 2);
468
469 /* The write callback drained all the data from outbuf, so we
470 * should have removed the write event... */
471 tt_assert(!event_pending(&bev2->ev_write, EV_WRITE, NULL));
472
473 end:
474 if (bev1)
475 bufferevent_free(bev1);
476 if (bev2)
477 bufferevent_free(bev2);
478 }
479
480 static void
test_bufferevent_watermarks(void)481 test_bufferevent_watermarks(void)
482 {
483 test_bufferevent_watermarks_impl(0);
484 }
485
486 static void
test_bufferevent_pair_watermarks(void)487 test_bufferevent_pair_watermarks(void)
488 {
489 test_bufferevent_watermarks_impl(1);
490 }
491
492 /*
493 * Test bufferevent filters
494 */
495
496 /* strip an 'x' from each byte */
497
498 static enum bufferevent_filter_result
bufferevent_input_filter(struct evbuffer * src,struct evbuffer * dst,ev_ssize_t lim,enum bufferevent_flush_mode state,void * ctx)499 bufferevent_input_filter(struct evbuffer *src, struct evbuffer *dst,
500 ev_ssize_t lim, enum bufferevent_flush_mode state, void *ctx)
501 {
502 const unsigned char *buffer;
503 unsigned i;
504
505 buffer = evbuffer_pullup(src, evbuffer_get_length(src));
506 for (i = 0; i < evbuffer_get_length(src); i += 2) {
507 if (buffer[i] == '-')
508 continue;
509
510 assert(buffer[i] == 'x');
511 evbuffer_add(dst, buffer + i + 1, 1);
512 }
513
514 evbuffer_drain(src, i);
515 return (BEV_OK);
516 }
517
518 /* add an 'x' before each byte */
519
520 static enum bufferevent_filter_result
bufferevent_output_filter(struct evbuffer * src,struct evbuffer * dst,ev_ssize_t lim,enum bufferevent_flush_mode state,void * ctx)521 bufferevent_output_filter(struct evbuffer *src, struct evbuffer *dst,
522 ev_ssize_t lim, enum bufferevent_flush_mode state, void *ctx)
523 {
524 const unsigned char *buffer;
525 unsigned i;
526 struct bufferevent **bevp = ctx;
527
528 ++test_ok;
529
530 if (test_ok == 1) {
531 buffer = evbuffer_pullup(src, evbuffer_get_length(src));
532 for (i = 0; i < evbuffer_get_length(src); ++i) {
533 evbuffer_add(dst, "x", 1);
534 evbuffer_add(dst, buffer + i, 1);
535 }
536 evbuffer_drain(src, evbuffer_get_length(src));
537 } else {
538 return BEV_ERROR;
539 }
540
541 if (bevp && test_ok == 1) {
542 int prev = ++test_ok;
543 bufferevent_write(*bevp, "-", 1);
544 /* check that during this bufferevent_write()
545 * bufferevent_output_filter() will not be called again */
546 assert(test_ok == prev);
547 --test_ok;
548 }
549
550 return (BEV_OK);
551 }
552
553 static void
test_bufferevent_filters_impl(int use_pair,int disable)554 test_bufferevent_filters_impl(int use_pair, int disable)
555 {
556 struct bufferevent *bev1 = NULL, *bev2 = NULL;
557 struct bufferevent *bev1_base = NULL, *bev2_base = NULL;
558 char buffer[8333];
559 int i;
560
561 test_ok = 0;
562
563 if (use_pair) {
564 struct bufferevent *xpair[2];
565 tt_assert(0 == bufferevent_pair_new(NULL, 0, xpair));
566 bev1 = xpair[0];
567 bev2 = xpair[1];
568 } else {
569 bev1 = bufferevent_socket_new(NULL, pair[0], 0);
570 bev2 = bufferevent_socket_new(NULL, pair[1], 0);
571 }
572 bev1_base = bev1;
573 bev2_base = bev2;
574
575 for (i = 0; i < (int)sizeof(buffer); i++)
576 buffer[i] = i;
577
578 bev1 = bufferevent_filter_new(bev1, NULL, bufferevent_output_filter,
579 BEV_OPT_CLOSE_ON_FREE, NULL,
580 disable ? &bev1 : NULL);
581
582 bev2 = bufferevent_filter_new(bev2, bufferevent_input_filter,
583 NULL, BEV_OPT_CLOSE_ON_FREE, NULL, NULL);
584 bufferevent_setcb(bev1, NULL, writecb, errorcb, NULL);
585 bufferevent_setcb(bev2, readcb, NULL, errorcb, NULL);
586
587 tt_ptr_op(bufferevent_get_underlying(bev1), ==, bev1_base);
588 tt_ptr_op(bufferevent_get_underlying(bev2), ==, bev2_base);
589 tt_fd_op(bufferevent_getfd(bev1), ==, bufferevent_getfd(bev1_base));
590 tt_fd_op(bufferevent_getfd(bev2), ==, bufferevent_getfd(bev2_base));
591
592 bufferevent_disable(bev1, EV_READ);
593 bufferevent_enable(bev2, EV_READ);
594 /* insert some filters */
595 bufferevent_write(bev1, buffer, sizeof(buffer));
596
597 event_dispatch();
598
599 if (test_ok != 3 + !!disable)
600 test_ok = 0;
601
602 end:
603 if (bev1)
604 bufferevent_free(bev1);
605 if (bev2)
606 bufferevent_free(bev2);
607
608 }
609
test_bufferevent_filters(void)610 static void test_bufferevent_filters(void)
611 { test_bufferevent_filters_impl(0, 0); }
test_bufferevent_pair_filters(void)612 static void test_bufferevent_pair_filters(void)
613 { test_bufferevent_filters_impl(1, 0); }
test_bufferevent_filters_disable(void)614 static void test_bufferevent_filters_disable(void)
615 { test_bufferevent_filters_impl(0, 1); }
test_bufferevent_pair_filters_disable(void)616 static void test_bufferevent_pair_filters_disable(void)
617 { test_bufferevent_filters_impl(1, 1); }
618
619
620 static void
sender_writecb(struct bufferevent * bev,void * ctx)621 sender_writecb(struct bufferevent *bev, void *ctx)
622 {
623 if (evbuffer_get_length(bufferevent_get_output(bev)) == 0) {
624 bufferevent_disable(bev,EV_READ|EV_WRITE);
625 TT_BLATHER(("Flushed %d: freeing it.", (int)bufferevent_getfd(bev)));
626 bufferevent_free(bev);
627 }
628 }
629
630 static void
sender_errorcb(struct bufferevent * bev,short what,void * ctx)631 sender_errorcb(struct bufferevent *bev, short what, void *ctx)
632 {
633 TT_FAIL(("Got sender error %d",(int)what));
634 }
635
636 static int bufferevent_connect_test_flags = 0;
637 static int bufferevent_trigger_test_flags = 0;
638 static int n_strings_read = 0;
639 static int n_reads_invoked = 0;
640 static int n_events_invoked = 0;
641
642 #define TEST_STR "Now is the time for all good events to signal for " \
643 "the good of their protocol"
644 static void
listen_cb(struct evconnlistener * listener,evutil_socket_t fd,struct sockaddr * sa,int socklen,void * arg)645 listen_cb(struct evconnlistener *listener, evutil_socket_t fd,
646 struct sockaddr *sa, int socklen, void *arg)
647 {
648 struct event_base *base = arg;
649 struct bufferevent *bev;
650 const char s[] = TEST_STR;
651 TT_BLATHER(("Got a request on socket %d", (int)fd ));
652 bev = bufferevent_socket_new(base, fd, bufferevent_connect_test_flags);
653 tt_assert(bev);
654 bufferevent_setcb(bev, NULL, sender_writecb, sender_errorcb, NULL);
655 bufferevent_write(bev, s, sizeof(s));
656 end:
657 ;
658 }
659
660 static evutil_socket_t
fake_listener_create(struct sockaddr_in * localhost)661 fake_listener_create(struct sockaddr_in *localhost)
662 {
663 struct sockaddr *sa = (struct sockaddr *)localhost;
664 evutil_socket_t fd = -1;
665 ev_socklen_t slen = sizeof(*localhost);
666
667 memset(localhost, 0, sizeof(*localhost));
668 localhost->sin_port = 0; /* have the kernel pick a port */
669 localhost->sin_addr.s_addr = htonl(0x7f000001L);
670 localhost->sin_family = AF_INET;
671
672 /* bind, but don't listen or accept. should trigger
673 "Connection refused" reliably on most platforms. */
674 fd = socket(localhost->sin_family, SOCK_STREAM, 0);
675 tt_assert(fd >= 0);
676 tt_assert(bind(fd, sa, slen) == 0);
677 tt_assert(getsockname(fd, sa, &slen) == 0);
678
679 return fd;
680
681 end:
682 return -1;
683 }
684
685 static void
reader_eventcb(struct bufferevent * bev,short what,void * ctx)686 reader_eventcb(struct bufferevent *bev, short what, void *ctx)
687 {
688 struct event_base *base = ctx;
689 if (what & BEV_EVENT_ERROR) {
690 perror("foobar");
691 TT_FAIL(("got connector error %d", (int)what));
692 return;
693 }
694 if (what & BEV_EVENT_CONNECTED) {
695 TT_BLATHER(("connected on %d", (int)bufferevent_getfd(bev)));
696 bufferevent_enable(bev, EV_READ);
697 }
698 if (what & BEV_EVENT_EOF) {
699 char buf[512];
700 size_t n;
701 n = bufferevent_read(bev, buf, sizeof(buf)-1);
702 tt_int_op(n, >=, 0);
703 buf[n] = '\0';
704 tt_str_op(buf, ==, TEST_STR);
705 if (++n_strings_read == 2)
706 event_base_loopexit(base, NULL);
707 TT_BLATHER(("EOF on %d: %d strings read.",
708 (int)bufferevent_getfd(bev), n_strings_read));
709 }
710 end:
711 ;
712 }
713
714 static void
reader_eventcb_simple(struct bufferevent * bev,short what,void * ctx)715 reader_eventcb_simple(struct bufferevent *bev, short what, void *ctx)
716 {
717 TT_BLATHER(("Read eventcb simple invoked on %d.",
718 (int)bufferevent_getfd(bev)));
719 n_events_invoked++;
720 }
721
722 static void
reader_readcb(struct bufferevent * bev,void * ctx)723 reader_readcb(struct bufferevent *bev, void *ctx)
724 {
725 TT_BLATHER(("Read invoked on %d.", (int)bufferevent_getfd(bev)));
726 n_reads_invoked++;
727 }
728
729 static void
test_bufferevent_connect(void * arg)730 test_bufferevent_connect(void *arg)
731 {
732 struct basic_test_data *data = arg;
733 struct evconnlistener *lev=NULL;
734 struct bufferevent *bev1=NULL, *bev2=NULL;
735 struct sockaddr_in localhost;
736 struct sockaddr_storage ss;
737 struct sockaddr *sa;
738 ev_socklen_t slen;
739
740 int be_flags=BEV_OPT_CLOSE_ON_FREE;
741
742 if (strstr((char*)data->setup_data, "defer")) {
743 be_flags |= BEV_OPT_DEFER_CALLBACKS;
744 }
745 if (strstr((char*)data->setup_data, "unlocked")) {
746 be_flags |= BEV_OPT_UNLOCK_CALLBACKS;
747 }
748 if (strstr((char*)data->setup_data, "lock")) {
749 be_flags |= BEV_OPT_THREADSAFE;
750 }
751 bufferevent_connect_test_flags = be_flags;
752 #ifdef _WIN32
753 if (!strcmp((char*)data->setup_data, "unset_connectex")) {
754 struct win32_extension_fns *ext =
755 (struct win32_extension_fns *)
756 event_get_win32_extension_fns_();
757 ext->ConnectEx = NULL;
758 }
759 #endif
760
761 memset(&localhost, 0, sizeof(localhost));
762
763 localhost.sin_port = 0; /* pick-a-port */
764 localhost.sin_addr.s_addr = htonl(0x7f000001L);
765 localhost.sin_family = AF_INET;
766 sa = (struct sockaddr *)&localhost;
767 lev = evconnlistener_new_bind(data->base, listen_cb, data->base,
768 LEV_OPT_CLOSE_ON_FREE|LEV_OPT_REUSEABLE,
769 16, sa, sizeof(localhost));
770 tt_assert(lev);
771
772 sa = (struct sockaddr *)&ss;
773 slen = sizeof(ss);
774 if (regress_get_listener_addr(lev, sa, &slen) < 0) {
775 tt_abort_perror("getsockname");
776 }
777
778 tt_assert(!evconnlistener_enable(lev));
779 bev1 = bufferevent_socket_new(data->base, -1, be_flags);
780 bev2 = bufferevent_socket_new(data->base, -1, be_flags);
781 tt_assert(bev1);
782 tt_assert(bev2);
783 bufferevent_setcb(bev1, reader_readcb,NULL, reader_eventcb, data->base);
784 bufferevent_setcb(bev2, reader_readcb,NULL, reader_eventcb, data->base);
785
786 bufferevent_enable(bev1, EV_READ);
787 bufferevent_enable(bev2, EV_READ);
788
789 tt_want(!bufferevent_socket_connect(bev1, sa, sizeof(localhost)));
790 tt_want(!bufferevent_socket_connect(bev2, sa, sizeof(localhost)));
791
792 event_base_dispatch(data->base);
793
794 tt_int_op(n_strings_read, ==, 2);
795 tt_int_op(n_reads_invoked, >=, 2);
796 end:
797 if (lev)
798 evconnlistener_free(lev);
799
800 if (bev1)
801 bufferevent_free(bev1);
802
803 if (bev2)
804 bufferevent_free(bev2);
805 }
806
807 static void
close_socket_cb(evutil_socket_t fd,short what,void * arg)808 close_socket_cb(evutil_socket_t fd, short what, void *arg)
809 {
810 evutil_socket_t *fdp = arg;
811 if (*fdp >= 0) {
812 evutil_closesocket(*fdp);
813 *fdp = -1;
814 }
815 }
816
817 static void
test_bufferevent_connect_fail_eventcb(void * arg)818 test_bufferevent_connect_fail_eventcb(void *arg)
819 {
820 struct basic_test_data *data = arg;
821 int flags = BEV_OPT_CLOSE_ON_FREE | (long)data->setup_data;
822 struct event close_listener_event;
823 struct bufferevent *bev = NULL;
824 struct evconnlistener *lev = NULL;
825 struct sockaddr_in localhost;
826 struct timeval close_timeout = { 0, 300000 };
827 ev_socklen_t slen = sizeof(localhost);
828 evutil_socket_t fake_listener = -1;
829 int r;
830
831 fake_listener = fake_listener_create(&localhost);
832
833 tt_int_op(n_events_invoked, ==, 0);
834
835 bev = bufferevent_socket_new(data->base, -1, flags);
836 tt_assert(bev);
837 bufferevent_setcb(bev, reader_readcb, reader_readcb,
838 reader_eventcb_simple, data->base);
839 bufferevent_enable(bev, EV_READ|EV_WRITE);
840 tt_int_op(n_events_invoked, ==, 0);
841 tt_int_op(n_reads_invoked, ==, 0);
842
843 /** @see also test_bufferevent_connect_fail() */
844 r = bufferevent_socket_connect(bev, (struct sockaddr *)&localhost, slen);
845 /* XXXX we'd like to test the '0' case everywhere, but FreeBSD tells
846 * detects the error immediately, which is not really wrong of it. */
847 tt_want(r == 0 || r == -1);
848
849 tt_int_op(n_events_invoked, ==, 0);
850 tt_int_op(n_reads_invoked, ==, 0);
851
852 /* Close the listener socket after a delay. This should trigger
853 "connection refused" on some other platforms, including OSX. */
854 evtimer_assign(&close_listener_event, data->base, close_socket_cb,
855 &fake_listener);
856 event_add(&close_listener_event, &close_timeout);
857
858 event_base_dispatch(data->base);
859 tt_int_op(n_events_invoked, ==, 1);
860 tt_int_op(n_reads_invoked, ==, 0);
861
862 end:
863 if (lev)
864 evconnlistener_free(lev);
865 if (bev)
866 bufferevent_free(bev);
867 if (fake_listener >= 0)
868 evutil_closesocket(fake_listener);
869 }
870
871 static void
want_fail_eventcb(struct bufferevent * bev,short what,void * ctx)872 want_fail_eventcb(struct bufferevent *bev, short what, void *ctx)
873 {
874 struct event_base *base = ctx;
875 const char *err;
876 evutil_socket_t s;
877
878 if (what & BEV_EVENT_ERROR) {
879 s = bufferevent_getfd(bev);
880 err = evutil_socket_error_to_string(evutil_socket_geterror(s));
881 TT_BLATHER(("connection failure on "EV_SOCK_FMT": %s",
882 EV_SOCK_ARG(s), err));
883 test_ok = 1;
884 } else {
885 TT_FAIL(("didn't fail? what %hd", what));
886 }
887
888 event_base_loopexit(base, NULL);
889 }
890
891 static void
test_bufferevent_connect_fail(void * arg)892 test_bufferevent_connect_fail(void *arg)
893 {
894 struct basic_test_data *data = (struct basic_test_data *)arg;
895 struct bufferevent *bev=NULL;
896 struct event close_listener_event;
897 int close_listener_event_added = 0;
898 struct timeval close_timeout = { 0, 300000 };
899 struct sockaddr_in localhost;
900 ev_socklen_t slen = sizeof(localhost);
901 evutil_socket_t fake_listener = -1;
902 int r;
903
904 test_ok = 0;
905
906 fake_listener = fake_listener_create(&localhost);
907 bev = bufferevent_socket_new(data->base, -1,
908 BEV_OPT_CLOSE_ON_FREE | BEV_OPT_DEFER_CALLBACKS);
909 tt_assert(bev);
910 bufferevent_setcb(bev, NULL, NULL, want_fail_eventcb, data->base);
911
912 r = bufferevent_socket_connect(bev, (struct sockaddr *)&localhost, slen);
913 /* XXXX we'd like to test the '0' case everywhere, but FreeBSD tells
914 * detects the error immediately, which is not really wrong of it. */
915 tt_want(r == 0 || r == -1);
916
917 /* Close the listener socket after a delay. This should trigger
918 "connection refused" on some other platforms, including OSX. */
919 evtimer_assign(&close_listener_event, data->base, close_socket_cb,
920 &fake_listener);
921 event_add(&close_listener_event, &close_timeout);
922 close_listener_event_added = 1;
923
924 event_base_dispatch(data->base);
925
926 tt_int_op(test_ok, ==, 1);
927
928 end:
929 if (fake_listener >= 0)
930 evutil_closesocket(fake_listener);
931
932 if (bev)
933 bufferevent_free(bev);
934
935 if (close_listener_event_added)
936 event_del(&close_listener_event);
937 }
938
939 struct timeout_cb_result {
940 struct timeval read_timeout_at;
941 struct timeval write_timeout_at;
942 struct timeval last_wrote_at;
943 struct timeval last_read_at;
944 int n_read_timeouts;
945 int n_write_timeouts;
946 int total_calls;
947 };
948
949 static void
bev_timeout_read_cb(struct bufferevent * bev,void * arg)950 bev_timeout_read_cb(struct bufferevent *bev, void *arg)
951 {
952 struct timeout_cb_result *res = arg;
953 evutil_gettimeofday(&res->last_read_at, NULL);
954 }
955 static void
bev_timeout_write_cb(struct bufferevent * bev,void * arg)956 bev_timeout_write_cb(struct bufferevent *bev, void *arg)
957 {
958 struct timeout_cb_result *res = arg;
959 evutil_gettimeofday(&res->last_wrote_at, NULL);
960 }
961 static void
bev_timeout_event_cb(struct bufferevent * bev,short what,void * arg)962 bev_timeout_event_cb(struct bufferevent *bev, short what, void *arg)
963 {
964 struct timeout_cb_result *res = arg;
965 ++res->total_calls;
966
967 if ((what & (BEV_EVENT_READING|BEV_EVENT_TIMEOUT))
968 == (BEV_EVENT_READING|BEV_EVENT_TIMEOUT)) {
969 evutil_gettimeofday(&res->read_timeout_at, NULL);
970 ++res->n_read_timeouts;
971 }
972 if ((what & (BEV_EVENT_WRITING|BEV_EVENT_TIMEOUT))
973 == (BEV_EVENT_WRITING|BEV_EVENT_TIMEOUT)) {
974 evutil_gettimeofday(&res->write_timeout_at, NULL);
975 ++res->n_write_timeouts;
976 }
977 }
978
979 static void
test_bufferevent_timeouts(void * arg)980 test_bufferevent_timeouts(void *arg)
981 {
982 /* "arg" is a string containing "pair" and/or "filter". */
983 struct bufferevent *bev1 = NULL, *bev2 = NULL;
984 struct basic_test_data *data = arg;
985 int use_pair = 0, use_filter = 0;
986 struct timeval tv_w, tv_r, started_at;
987 struct timeout_cb_result res1, res2;
988
989 memset(&res1, 0, sizeof(res1));
990 memset(&res2, 0, sizeof(res2));
991
992 if (strstr((char*)data->setup_data, "pair"))
993 use_pair = 1;
994 if (strstr((char*)data->setup_data, "filter"))
995 use_filter = 1;
996
997 if (use_pair) {
998 struct bufferevent *p[2];
999 tt_int_op(0, ==, bufferevent_pair_new(data->base, 0, p));
1000 bev1 = p[0];
1001 bev2 = p[1];
1002 } else {
1003 bev1 = bufferevent_socket_new(data->base, data->pair[0], 0);
1004 bev2 = bufferevent_socket_new(data->base, data->pair[1], 0);
1005 }
1006 tt_assert(bev1);
1007 tt_assert(bev2);
1008
1009 if (use_filter) {
1010 struct bufferevent *bevf1, *bevf2;
1011 bevf1 = bufferevent_filter_new(bev1, NULL, NULL,
1012 BEV_OPT_CLOSE_ON_FREE, NULL, NULL);
1013 bevf2 = bufferevent_filter_new(bev2, NULL, NULL,
1014 BEV_OPT_CLOSE_ON_FREE, NULL, NULL);
1015 tt_assert(bevf1);
1016 tt_assert(bevf2);
1017 bev1 = bevf1;
1018 bev2 = bevf2;
1019 }
1020
1021 /* Do this nice and early. */
1022 bufferevent_disable(bev2, EV_READ);
1023
1024 /* bev1 will try to write and read. Both will time out. */
1025 evutil_gettimeofday(&started_at, NULL);
1026 tv_w.tv_sec = tv_r.tv_sec = 0;
1027 tv_w.tv_usec = 100*1000;
1028 tv_r.tv_usec = 150*1000;
1029 bufferevent_setcb(bev1, bev_timeout_read_cb, bev_timeout_write_cb,
1030 bev_timeout_event_cb, &res1);
1031 bufferevent_set_timeouts(bev1, &tv_r, &tv_w);
1032 bufferevent_write(bev1, "ABCDEFG", 7);
1033 bufferevent_enable(bev1, EV_READ|EV_WRITE);
1034
1035 /* bev2 has nothing to say, and isn't listening. */
1036 bufferevent_setcb(bev2, bev_timeout_read_cb, bev_timeout_write_cb,
1037 bev_timeout_event_cb, &res2);
1038 tv_w.tv_sec = tv_r.tv_sec = 0;
1039 tv_w.tv_usec = 200*1000;
1040 tv_r.tv_usec = 100*1000;
1041 bufferevent_set_timeouts(bev2, &tv_r, &tv_w);
1042 bufferevent_enable(bev2, EV_WRITE);
1043
1044 tv_r.tv_sec = 0;
1045 tv_r.tv_usec = 350000;
1046
1047 event_base_loopexit(data->base, &tv_r);
1048 event_base_dispatch(data->base);
1049
1050 /* XXXX Test that actually reading or writing a little resets the
1051 * timeouts. */
1052
1053 tt_want(res1.total_calls == 2);
1054 tt_want(res1.n_read_timeouts == 1);
1055 tt_want(res1.n_write_timeouts == 1);
1056 tt_want(res2.total_calls == !(use_pair && !use_filter));
1057 tt_want(res2.n_write_timeouts == !(use_pair && !use_filter));
1058 tt_want(!res2.n_read_timeouts);
1059
1060 test_timeval_diff_eq(&started_at, &res1.read_timeout_at, 150);
1061 test_timeval_diff_eq(&started_at, &res1.write_timeout_at, 100);
1062
1063 #define tt_assert_timeval_empty(tv) do { \
1064 tt_int_op((tv).tv_sec, ==, 0); \
1065 tt_int_op((tv).tv_usec, ==, 0); \
1066 } while(0)
1067 tt_assert_timeval_empty(res1.last_read_at);
1068 tt_assert_timeval_empty(res2.last_read_at);
1069 tt_assert_timeval_empty(res2.last_wrote_at);
1070 tt_assert_timeval_empty(res2.last_wrote_at);
1071 #undef tt_assert_timeval_empty
1072
1073 end:
1074 if (bev1)
1075 bufferevent_free(bev1);
1076 if (bev2)
1077 bufferevent_free(bev2);
1078 }
1079
1080 static void
trigger_failure_cb(evutil_socket_t fd,short what,void * ctx)1081 trigger_failure_cb(evutil_socket_t fd, short what, void *ctx)
1082 {
1083 TT_FAIL(("The triggered callback did not fire or the machine is really slow (try increasing timeout)."));
1084 }
1085
1086 static void
trigger_eventcb(struct bufferevent * bev,short what,void * ctx)1087 trigger_eventcb(struct bufferevent *bev, short what, void *ctx)
1088 {
1089 struct event_base *base = ctx;
1090 if (what == ~0) {
1091 TT_BLATHER(("Event successfully triggered."));
1092 event_base_loopexit(base, NULL);
1093 return;
1094 }
1095 reader_eventcb(bev, what, ctx);
1096 }
1097
1098 static void
trigger_readcb_triggered(struct bufferevent * bev,void * ctx)1099 trigger_readcb_triggered(struct bufferevent *bev, void *ctx)
1100 {
1101 TT_BLATHER(("Read successfully triggered."));
1102 n_reads_invoked++;
1103 bufferevent_trigger_event(bev, ~0, bufferevent_trigger_test_flags);
1104 }
1105
1106 static void
trigger_readcb(struct bufferevent * bev,void * ctx)1107 trigger_readcb(struct bufferevent *bev, void *ctx)
1108 {
1109 struct timeval timeout = { 30, 0 };
1110 struct event_base *base = ctx;
1111 size_t low, high, len;
1112 int expected_reads;
1113
1114 TT_BLATHER(("Read invoked on %d.", (int)bufferevent_getfd(bev)));
1115 expected_reads = ++n_reads_invoked;
1116
1117 bufferevent_setcb(bev, trigger_readcb_triggered, NULL, trigger_eventcb, ctx);
1118
1119 bufferevent_getwatermark(bev, EV_READ, &low, &high);
1120 len = evbuffer_get_length(bufferevent_get_input(bev));
1121
1122 bufferevent_setwatermark(bev, EV_READ, len + 1, 0);
1123 bufferevent_trigger(bev, EV_READ, bufferevent_trigger_test_flags);
1124 /* no callback expected */
1125 tt_int_op(n_reads_invoked, ==, expected_reads);
1126
1127 if ((bufferevent_trigger_test_flags & BEV_TRIG_DEFER_CALLBACKS) ||
1128 (bufferevent_connect_test_flags & BEV_OPT_DEFER_CALLBACKS)) {
1129 /* will be deferred */
1130 } else {
1131 expected_reads++;
1132 }
1133
1134 event_base_once(base, -1, EV_TIMEOUT, trigger_failure_cb, NULL, &timeout);
1135
1136 bufferevent_trigger(bev, EV_READ,
1137 bufferevent_trigger_test_flags | BEV_TRIG_IGNORE_WATERMARKS);
1138 tt_int_op(n_reads_invoked, ==, expected_reads);
1139
1140 bufferevent_setwatermark(bev, EV_READ, low, high);
1141 end:
1142 ;
1143 }
1144
1145 static void
test_bufferevent_trigger(void * arg)1146 test_bufferevent_trigger(void *arg)
1147 {
1148 struct basic_test_data *data = arg;
1149 struct evconnlistener *lev=NULL;
1150 struct bufferevent *bev=NULL;
1151 struct sockaddr_in localhost;
1152 struct sockaddr_storage ss;
1153 struct sockaddr *sa;
1154 ev_socklen_t slen;
1155
1156 int be_flags=BEV_OPT_CLOSE_ON_FREE;
1157 int trig_flags=0;
1158
1159 if (strstr((char*)data->setup_data, "defer")) {
1160 be_flags |= BEV_OPT_DEFER_CALLBACKS;
1161 }
1162 bufferevent_connect_test_flags = be_flags;
1163
1164 if (strstr((char*)data->setup_data, "postpone")) {
1165 trig_flags |= BEV_TRIG_DEFER_CALLBACKS;
1166 }
1167 bufferevent_trigger_test_flags = trig_flags;
1168
1169 memset(&localhost, 0, sizeof(localhost));
1170
1171 localhost.sin_port = 0; /* pick-a-port */
1172 localhost.sin_addr.s_addr = htonl(0x7f000001L);
1173 localhost.sin_family = AF_INET;
1174 sa = (struct sockaddr *)&localhost;
1175 lev = evconnlistener_new_bind(data->base, listen_cb, data->base,
1176 LEV_OPT_CLOSE_ON_FREE|LEV_OPT_REUSEABLE,
1177 16, sa, sizeof(localhost));
1178 tt_assert(lev);
1179
1180 sa = (struct sockaddr *)&ss;
1181 slen = sizeof(ss);
1182 if (regress_get_listener_addr(lev, sa, &slen) < 0) {
1183 tt_abort_perror("getsockname");
1184 }
1185
1186 tt_assert(!evconnlistener_enable(lev));
1187 bev = bufferevent_socket_new(data->base, -1, be_flags);
1188 tt_assert(bev);
1189 bufferevent_setcb(bev, trigger_readcb, NULL, trigger_eventcb, data->base);
1190
1191 bufferevent_enable(bev, EV_READ);
1192
1193 tt_want(!bufferevent_socket_connect(bev, sa, sizeof(localhost)));
1194
1195 event_base_dispatch(data->base);
1196
1197 tt_int_op(n_reads_invoked, ==, 2);
1198 end:
1199 if (lev)
1200 evconnlistener_free(lev);
1201
1202 if (bev)
1203 bufferevent_free(bev);
1204 }
1205
1206 static void
test_bufferevent_socket_filter_inactive(void * arg)1207 test_bufferevent_socket_filter_inactive(void *arg)
1208 {
1209 struct basic_test_data *data = arg;
1210 struct bufferevent *bev = NULL, *bevf = NULL;
1211
1212 bev = bufferevent_socket_new(data->base, -1, 0);
1213 tt_assert(bev);
1214 bevf = bufferevent_filter_new(bev, NULL, NULL, 0, NULL, NULL);
1215 tt_assert(bevf);
1216
1217 end:
1218 if (bevf)
1219 bufferevent_free(bevf);
1220 if (bev)
1221 bufferevent_free(bev);
1222 }
1223
1224 static void
pair_flush_eventcb(struct bufferevent * bev,short what,void * ctx)1225 pair_flush_eventcb(struct bufferevent *bev, short what, void *ctx)
1226 {
1227 int *callback_what = ctx;
1228 *callback_what = what;
1229 }
1230
1231 static void
test_bufferevent_pair_flush(void * arg)1232 test_bufferevent_pair_flush(void *arg)
1233 {
1234 struct basic_test_data *data = arg;
1235 struct bufferevent *xpair[2];
1236 struct bufferevent *bev1 = NULL;
1237 struct bufferevent *bev2 = NULL;
1238 int callback_what = 0;
1239
1240 tt_assert(0 == bufferevent_pair_new(data->base, 0, xpair));
1241 bev1 = xpair[0];
1242 bev2 = xpair[1];
1243 tt_assert(0 == bufferevent_enable(bev1, EV_WRITE));
1244 tt_assert(0 == bufferevent_enable(bev2, EV_READ));
1245
1246 bufferevent_setcb(bev2, NULL, NULL, pair_flush_eventcb, &callback_what);
1247
1248 bufferevent_flush(bev1, EV_WRITE, BEV_FINISHED);
1249
1250 event_base_loop(data->base, EVLOOP_ONCE);
1251
1252 tt_assert(callback_what == (BEV_EVENT_READING | BEV_EVENT_EOF));
1253
1254 end:
1255 if (bev1)
1256 bufferevent_free(bev1);
1257 if (bev2)
1258 bufferevent_free(bev2);
1259 }
1260
1261 struct bufferevent_filter_data_stuck {
1262 size_t header_size;
1263 size_t total_read;
1264 };
1265
1266 static void
bufferevent_filter_data_stuck_readcb(struct bufferevent * bev,void * arg)1267 bufferevent_filter_data_stuck_readcb(struct bufferevent *bev, void *arg)
1268 {
1269 struct bufferevent_filter_data_stuck *filter_data = arg;
1270 struct evbuffer *input = bufferevent_get_input(bev);
1271 size_t read_size = evbuffer_get_length(input);
1272 evbuffer_drain(input, read_size);
1273 filter_data->total_read += read_size;
1274 }
1275
1276 /**
1277 * This filter prepends header once before forwarding data.
1278 */
1279 static enum bufferevent_filter_result
bufferevent_filter_data_stuck_inputcb(struct evbuffer * src,struct evbuffer * dst,ev_ssize_t dst_limit,enum bufferevent_flush_mode mode,void * ctx)1280 bufferevent_filter_data_stuck_inputcb(
1281 struct evbuffer *src, struct evbuffer *dst, ev_ssize_t dst_limit,
1282 enum bufferevent_flush_mode mode, void *ctx)
1283 {
1284 struct bufferevent_filter_data_stuck *filter_data = ctx;
1285 static int header_inserted = 0;
1286 size_t payload_size;
1287 size_t header_size = 0;
1288
1289 if (!header_inserted) {
1290 char *header = calloc(filter_data->header_size, 1);
1291 evbuffer_add(dst, header, filter_data->header_size);
1292 free(header);
1293 header_size = filter_data->header_size;
1294 header_inserted = 1;
1295 }
1296
1297 payload_size = evbuffer_get_length(src);
1298 if (payload_size > dst_limit - header_size) {
1299 payload_size = dst_limit - header_size;
1300 }
1301
1302 tt_int_op(payload_size, ==, evbuffer_remove_buffer(src, dst, payload_size));
1303
1304 end:
1305 return BEV_OK;
1306 }
1307
1308 static void
test_bufferevent_filter_data_stuck(void * arg)1309 test_bufferevent_filter_data_stuck(void *arg)
1310 {
1311 const size_t read_high_wm = 4096;
1312 struct bufferevent_filter_data_stuck filter_data;
1313 struct basic_test_data *data = arg;
1314 struct bufferevent *xpair[2];
1315 struct bufferevent *filter = NULL;
1316
1317 int options = BEV_OPT_CLOSE_ON_FREE | BEV_OPT_DEFER_CALLBACKS;
1318
1319 char payload[4096];
1320 int payload_size = sizeof(payload);
1321
1322 memset(&filter_data, 0, sizeof(filter_data));
1323 filter_data.header_size = 20;
1324
1325 tt_assert(bufferevent_pair_new(data->base, options, xpair) == 0);
1326
1327 bufferevent_setwatermark(xpair[0], EV_READ, 0, read_high_wm);
1328 bufferevent_setwatermark(xpair[1], EV_READ, 0, read_high_wm);
1329
1330 tt_assert(
1331 filter =
1332 bufferevent_filter_new(xpair[1],
1333 bufferevent_filter_data_stuck_inputcb,
1334 NULL,
1335 options,
1336 NULL,
1337 &filter_data));
1338
1339 bufferevent_setcb(filter,
1340 bufferevent_filter_data_stuck_readcb,
1341 NULL,
1342 NULL,
1343 &filter_data);
1344
1345 tt_assert(bufferevent_enable(filter, EV_READ|EV_WRITE) == 0);
1346
1347 bufferevent_setwatermark(filter, EV_READ, 0, read_high_wm);
1348
1349 tt_assert(bufferevent_write(xpair[0], payload, sizeof(payload)) == 0);
1350
1351 event_base_dispatch(data->base);
1352
1353 tt_int_op(filter_data.total_read, ==, payload_size + filter_data.header_size);
1354 end:
1355 if (xpair[0])
1356 bufferevent_free(xpair[0]);
1357 if (filter)
1358 bufferevent_free(filter);
1359 }
1360
1361 struct testcase_t bufferevent_testcases[] = {
1362
1363 LEGACY(bufferevent, TT_ISOLATED),
1364 LEGACY(bufferevent_pair, TT_ISOLATED),
1365 LEGACY(bufferevent_flush_normal, TT_ISOLATED),
1366 LEGACY(bufferevent_flush_flush, TT_ISOLATED),
1367 LEGACY(bufferevent_flush_finished, TT_ISOLATED),
1368 LEGACY(bufferevent_pair_flush_normal, TT_ISOLATED),
1369 LEGACY(bufferevent_pair_flush_flush, TT_ISOLATED),
1370 LEGACY(bufferevent_pair_flush_finished, TT_ISOLATED),
1371 #if defined(EVTHREAD_USE_PTHREADS_IMPLEMENTED) && !defined(__SANITIZE_ADDRESS__)
1372 { "bufferevent_pair_release_lock", test_bufferevent_pair_release_lock,
1373 TT_FORK|TT_ISOLATED|TT_NEED_THREADS|TT_NEED_BASE|TT_LEGACY|TT_NO_LOGS,
1374 &basic_setup, NULL },
1375 #endif
1376 LEGACY(bufferevent_watermarks, TT_ISOLATED),
1377 LEGACY(bufferevent_pair_watermarks, TT_ISOLATED),
1378 LEGACY(bufferevent_filters, TT_ISOLATED),
1379 LEGACY(bufferevent_pair_filters, TT_ISOLATED),
1380 LEGACY(bufferevent_filters_disable, TT_ISOLATED),
1381 LEGACY(bufferevent_pair_filters_disable, TT_ISOLATED),
1382 { "bufferevent_connect", test_bufferevent_connect, TT_FORK|TT_NEED_BASE,
1383 &basic_setup, __UNCONST("") },
1384 { "bufferevent_connect_defer", test_bufferevent_connect,
1385 TT_FORK|TT_NEED_BASE, &basic_setup, __UNCONST("defer") },
1386 { "bufferevent_connect_lock", test_bufferevent_connect,
1387 TT_FORK|TT_NEED_BASE|TT_NEED_THREADS, &basic_setup, __UNCONST("lock") },
1388 { "bufferevent_connect_lock_defer", test_bufferevent_connect,
1389 TT_FORK|TT_NEED_BASE|TT_NEED_THREADS, &basic_setup,
1390 __UNCONST("defer lock") },
1391 { "bufferevent_connect_unlocked_cbs", test_bufferevent_connect,
1392 TT_FORK|TT_NEED_BASE|TT_NEED_THREADS, &basic_setup,
1393 __UNCONST("lock defer unlocked") },
1394 { "bufferevent_connect_fail", test_bufferevent_connect_fail,
1395 TT_FORK|TT_NEED_BASE, &basic_setup, NULL },
1396 { "bufferevent_timeout", test_bufferevent_timeouts,
1397 TT_FORK|TT_NEED_BASE, &basic_setup, __UNCONST("") },
1398 { "bufferevent_timeout_pair", test_bufferevent_timeouts,
1399 TT_FORK|TT_NEED_BASE, &basic_setup, __UNCONST("pair") },
1400 { "bufferevent_timeout_filter", test_bufferevent_timeouts,
1401 TT_FORK|TT_NEED_BASE, &basic_setup, __UNCONST("filter") },
1402 { "bufferevent_timeout_filter_pair", test_bufferevent_timeouts,
1403 TT_FORK|TT_NEED_BASE, &basic_setup, __UNCONST("filter pair") },
1404 { "bufferevent_trigger", test_bufferevent_trigger, TT_FORK|TT_NEED_BASE,
1405 &basic_setup, __UNCONST("") },
1406 { "bufferevent_trigger_defer", test_bufferevent_trigger,
1407 TT_FORK|TT_NEED_BASE, &basic_setup, __UNCONST("defer") },
1408 { "bufferevent_trigger_postpone", test_bufferevent_trigger,
1409 TT_FORK|TT_NEED_BASE|TT_NEED_THREADS, &basic_setup,
1410 __UNCONST("postpone") },
1411 { "bufferevent_trigger_defer_postpone", test_bufferevent_trigger,
1412 TT_FORK|TT_NEED_BASE|TT_NEED_THREADS, &basic_setup,
1413 __UNCONST("defer postpone") },
1414 #ifdef EVENT__HAVE_LIBZ
1415 LEGACY(bufferevent_zlib, TT_ISOLATED),
1416 #else
1417 { "bufferevent_zlib", NULL, TT_SKIP, NULL, NULL },
1418 #endif
1419
1420 { "bufferevent_connect_fail_eventcb_defer",
1421 test_bufferevent_connect_fail_eventcb,
1422 TT_FORK|TT_NEED_BASE, &basic_setup, (void*)BEV_OPT_DEFER_CALLBACKS },
1423 { "bufferevent_connect_fail_eventcb",
1424 test_bufferevent_connect_fail_eventcb,
1425 TT_FORK|TT_NEED_BASE, &basic_setup, NULL },
1426
1427 { "bufferevent_socket_filter_inactive",
1428 test_bufferevent_socket_filter_inactive,
1429 TT_FORK|TT_NEED_BASE, &basic_setup, NULL },
1430 { "bufferevent_pair_flush",
1431 test_bufferevent_pair_flush,
1432 TT_FORK|TT_NEED_BASE, &basic_setup, NULL },
1433 { "bufferevent_filter_data_stuck",
1434 test_bufferevent_filter_data_stuck,
1435 TT_FORK|TT_NEED_BASE, &basic_setup, NULL },
1436
1437 END_OF_TESTCASES,
1438 };
1439
1440 #define TT_IOCP (TT_FORK|TT_NEED_BASE|TT_ENABLE_IOCP)
1441 #define TT_IOCP_LEGACY (TT_ISOLATED|TT_ENABLE_IOCP)
1442 struct testcase_t bufferevent_iocp_testcases[] = {
1443 LEGACY(bufferevent, TT_IOCP_LEGACY),
1444 LEGACY(bufferevent_flush_normal, TT_ISOLATED),
1445 LEGACY(bufferevent_flush_flush, TT_ISOLATED),
1446 LEGACY(bufferevent_flush_finished, TT_ISOLATED),
1447 LEGACY(bufferevent_watermarks, TT_IOCP_LEGACY),
1448 LEGACY(bufferevent_filters, TT_IOCP_LEGACY),
1449 LEGACY(bufferevent_filters_disable, TT_IOCP_LEGACY),
1450
1451 { "bufferevent_connect", test_bufferevent_connect,
1452 TT_IOCP, &basic_setup, __UNCONST("") },
1453 { "bufferevent_connect_defer", test_bufferevent_connect,
1454 TT_IOCP, &basic_setup, __UNCONST("defer") },
1455 { "bufferevent_connect_lock", test_bufferevent_connect,
1456 TT_IOCP, &basic_setup, __UNCONST("lock") },
1457 { "bufferevent_connect_lock_defer", test_bufferevent_connect,
1458 TT_IOCP, &basic_setup, __UNCONST("defer lock") },
1459 { "bufferevent_connect_fail", test_bufferevent_connect_fail,
1460 TT_IOCP, &basic_setup, NULL },
1461 { "bufferevent_connect_nonblocking", test_bufferevent_connect,
1462 TT_IOCP, &basic_setup, __UNCONST("unset_connectex") },
1463
1464 { "bufferevent_connect_fail_eventcb_defer",
1465 test_bufferevent_connect_fail_eventcb,
1466 TT_IOCP, &basic_setup, (void*)BEV_OPT_DEFER_CALLBACKS },
1467 { "bufferevent_connect_fail_eventcb",
1468 test_bufferevent_connect_fail_eventcb, TT_IOCP, &basic_setup, NULL },
1469
1470 END_OF_TESTCASES,
1471 };
1472