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