checkpoint work on big bufferevent refactoring

svn:r1095

6 files changed, 216 insertions, 81 deletions

diff --git a/test/Makefile.am b/test/Makefile.am
index f178879c..a691f30b 100644
--- a/test/Makefile.am
+++ b/test/Makefile.am
@@ -20,6 +20,7 @@ test_time_LDADD = ../libevent_core.la
 
 regress_SOURCES = regress.c regress_buffer.c regress_http.c regress_dns.c \
 	regress_rpc.c regress.gen.c regress.gen.h regress_et.c \
+	regress_bufferevent.c \
 	regress_util.c tinytest.c regress_main.c \
 	$(regress_pthread_SOURCES) $(regress_zlib_SOURCES)
 if PTHREADS
diff --git a/test/regress.c b/test/regress.c
index 7ef31d8a..8d3493e8 100644
--- a/test/regress.c
+++ b/test/regress.c
@@ -61,6 +61,7 @@
 #include "event2/tag.h"
 #include "event2/buffer.h"
 #include "event2/bufferevent.h"
+#include "event2/bufferevent_compat.h"
 #include "event2/bufferevent_struct.h"
 #include "event2/util.h"
 #include "event-internal.h"
@@ -150,7 +151,7 @@ multiple_write_cb(int fd, short event, void *arg)
 	}
 
 	woff += len;
-
+        
 	if (woff >= sizeof(wbuf)) {
 		shutdown(fd, SHUT_WR);
 		if (usepersist)
@@ -1098,8 +1099,9 @@ readcb(struct bufferevent *bev, void *arg)
 
 		bufferevent_disable(bev, EV_READ);
 
-		if (EVBUFFER_LENGTH(evbuf) == 8333)
+		if (EVBUFFER_LENGTH(evbuf) == 8333) {
 			test_ok++;
+                }
 
 		evbuffer_free(evbuf);
 	}
@@ -1108,8 +1110,9 @@ readcb(struct bufferevent *bev, void *arg)
 static void
 writecb(struct bufferevent *bev, void *arg)
 {
-	if (EVBUFFER_LENGTH(bev->output) == 0)
+	if (EVBUFFER_LENGTH(bev->output) == 0) {
 		test_ok++;
+        }
 }
 
 static void
@@ -1118,7 +1121,8 @@ errorcb(struct bufferevent *bev, short what, void *arg)
 	test_ok = -2;
 }
 
-static void
+/*XXXX move to regress_bufferevent.c; make static again. */
+void
 test_bufferevent(void)
 {
 	struct bufferevent *bev1, *bev2;
@@ -1179,8 +1183,11 @@ wm_readcb(struct bufferevent *bev, void *arg)
 static void
 wm_writecb(struct bufferevent *bev, void *arg)
 {
-	if (EVBUFFER_LENGTH(bev->output) == 0)
+        assert(EVBUFFER_LENGTH(bev->output) <= 100);
+	if (EVBUFFER_LENGTH(bev->output) == 0) {
+                evbuffer_drain(bev->output, EVBUFFER_LENGTH(bev->output));
 		test_ok++;
+        }
 }
 
 static void
@@ -1189,7 +1196,8 @@ wm_errorcb(struct bufferevent *bev, short what, void *arg)
 	test_ok = -2;
 }
 
-static void
+/*XXXX move to regress_bufferevent.c; make static again. */
+void
 test_bufferevent_watermarks(void)
 {
 	struct bufferevent *bev1, *bev2;
@@ -1205,21 +1213,29 @@ test_bufferevent_watermarks(void)
 	bufferevent_enable(bev2, EV_READ);
 
 	for (i = 0; i < sizeof(buffer); i++)
-		buffer[i] = i;
+		buffer[i] = (char)i;
 
 	bufferevent_write(bev1, buffer, sizeof(buffer));
 
 	/* limit the reading on the receiving bufferevent */
 	bufferevent_setwatermark(bev2, EV_READ, 10, 20);
 
+        /* Tell the sending bufferevent not to notify us till it's down to
+           100 bytes. */
+        bufferevent_setwatermark(bev1, EV_WRITE, 100, 2000);
+
 	event_dispatch();
 
+	tt_int_op(test_ok, ==, 2);
+
+        /* The write callback drained all the data from outbuf, so we
+         * should have removed the write event... */
+        tt_assert(!event_pending(&bev2->ev_write, EV_WRITE, NULL));
+
+end:
 	bufferevent_free(bev1);
 	bufferevent_free(bev2);
 
-	if (test_ok != 2)
-		test_ok = 0;
-
 	cleanup_test();
 }
 
@@ -1231,7 +1247,7 @@ test_bufferevent_watermarks(void)
 
 static enum bufferevent_filter_result
 bufferevent_input_filter(struct evbuffer *src, struct evbuffer *dst,
-    enum bufferevent_filter_state state, void *ctx)
+    ssize_t lim, enum bufferevent_flush_mode state, void *ctx)
 {
 	const unsigned char *buffer;
 	int i;
@@ -1253,7 +1269,7 @@ bufferevent_input_filter(struct evbuffer *src, struct evbuffer *dst,
 
 static enum bufferevent_filter_result
 bufferevent_output_filter(struct evbuffer *src, struct evbuffer *dst,
-    enum bufferevent_filter_state state, void *ctx)
+    ssize_t lim, enum bufferevent_flush_mode state, void *ctx)
 {
 	const unsigned char *buffer;
 	int i;
@@ -1268,41 +1284,39 @@ bufferevent_output_filter(struct evbuffer *src, struct evbuffer *dst,
 	return (BEV_OK);
 }
 
-static void
+
+/*XXXX move to regress_bufferevent.c; make static again. */
+void
 test_bufferevent_filters(void)
 {
 	struct bufferevent *bev1, *bev2;
-	struct bufferevent_filter *finput, *foutput;
 	char buffer[8333];
 	int i;
 
+        test_ok = 0;
 	setup_test("Bufferevent Filters: ");
 
-	bev1 = bufferevent_new(pair[0], NULL, writecb, errorcb, NULL);
-	bev2 = bufferevent_new(pair[1], readcb, NULL, errorcb, NULL);
-
-	bufferevent_disable(bev1, EV_READ);
-	bufferevent_enable(bev2, EV_READ);
+	bev1 = bufferevent_socket_new(NULL, pair[0], 0);
+	bev2 = bufferevent_socket_new(NULL, pair[1], 0);
 
 	for (i = 0; i < sizeof(buffer); i++)
 		buffer[i] = i;
 
-	/* insert some filters */
-	finput = bufferevent_filter_new(
-		NULL, NULL,bufferevent_input_filter, NULL);
-	foutput = bufferevent_filter_new(
-		NULL, NULL, bufferevent_output_filter, NULL);
+	bev1 = bufferevent_filter_new(bev1, NULL, bufferevent_output_filter,
+				      0, NULL, NULL);
 
-	bufferevent_filter_insert(bev1, BEV_OUTPUT, foutput);
-	bufferevent_filter_insert(bev2, BEV_INPUT, finput);
+	bev2 = bufferevent_filter_new(bev2, bufferevent_input_filter,
+				      NULL, 0, NULL, NULL);
+	bufferevent_setcb(bev1, NULL, writecb, errorcb, NULL);
+	bufferevent_setcb(bev2, readcb, NULL, errorcb, NULL);
 
+	bufferevent_disable(bev1, EV_READ);
+	bufferevent_enable(bev2, EV_READ);
+	/* insert some filters */
 	bufferevent_write(bev1, buffer, sizeof(buffer));
 
 	event_dispatch();
 
-	bufferevent_filter_remove(bev1, BEV_OUTPUT, foutput);
-	bufferevent_filter_free(foutput);
-
 	bufferevent_free(bev1);
 	bufferevent_free(bev2);
 
@@ -1630,10 +1644,6 @@ struct testcase_t legacy_testcases[] = {
         LEGACY(persistent_timeout, TT_FORK|TT_NEED_BASE),
         LEGACY(priorities, TT_FORK|TT_NEED_BASE),
 
-        LEGACY(bufferevent, TT_ISOLATED),
-        LEGACY(bufferevent_watermarks, TT_ISOLATED),
-        LEGACY(bufferevent_filters, TT_ISOLATED),
-
         LEGACY(free_active_base, TT_FORK|TT_NEED_BASE),
         LEGACY(event_base_new, TT_FORK|TT_NEED_SOCKETPAIR),
 
diff --git a/test/regress.h b/test/regress.h
index 196d3896..7cc3ca86 100644
--- a/test/regress.h
+++ b/test/regress.h
@@ -34,8 +34,15 @@ extern "C" {
 #include "tinytest.h"
 #include "tinytest_macros.h"
 
+/*XXX move these to regress_bufferevent.c when bufferevent changes are
+ * merged. */
+void test_bufferevent(void);
+void test_bufferevent_watermarks(void);
+void test_bufferevent_filters(void);
+
 extern struct testcase_t legacy_testcases[];
 extern struct testcase_t evbuffer_testcases[];
+extern struct testcase_t bufferevent_testcases[];
 extern struct testcase_t util_testcases[];
 extern struct testcase_t signal_testcases[];
 extern struct testcase_t http_testcases[];
diff --git a/test/regress_bufferevent.c b/test/regress_bufferevent.c
new file mode 100644
index 00000000..afa714c8
--- /dev/null
+++ b/test/regress_bufferevent.c
@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) 2003-2007 Niels Provos <provos@citi.umich.edu>
+ * Copyright (c) 2007-2009 Niels Provos and Nick Mathewson
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ * 3. The name of the author may not be used to endorse or promote products
+ *    derived from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
+ * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
+ * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
+ * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
+ * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
+ * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
+ * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifdef WIN32
+#include <winsock2.h>
+#include <windows.h>
+#endif
+
+#ifdef HAVE_CONFIG_H
+#include "event-config.h"
+#endif
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#ifdef _EVENT_HAVE_SYS_TIME_H
+#include <sys/time.h>
+#endif
+#include <sys/queue.h>
+#ifndef WIN32
+#include <sys/socket.h>
+#include <sys/wait.h>
+#include <signal.h>
+#include <unistd.h>
+#include <netdb.h>
+#endif
+#include <fcntl.h>
+#include <signal.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+#include <errno.h>
+#include <assert.h>
+
+#include "event2/event.h"
+#include "event2/event_struct.h"
+#include "event2/event_compat.h"
+#include "event2/tag.h"
+#include "event2/buffer.h"
+#include "event2/bufferevent.h"
+#include "event2/bufferevent_compat.h"
+#include "event2/bufferevent_struct.h"
+#include "event2/util.h"
+
+#include "bufferevent-internal.h"
+
+#include "regress.h"
+
+struct testcase_t bufferevent_testcases[] = {
+
+        LEGACY(bufferevent, TT_ISOLATED),
+        LEGACY(bufferevent_watermarks, TT_ISOLATED),
+        LEGACY(bufferevent_filters, TT_ISOLATED),
+
+        END_OF_TESTCASES,
+};
diff --git a/test/regress_main.c b/test/regress_main.c
index d0c4a288..a6dac356 100644
--- a/test/regress_main.c
+++ b/test/regress_main.c
@@ -189,6 +189,7 @@ struct testgroup_t testgroups[] = {
 		{ "evbuffer/", evbuffer_testcases },
         { "signal/", signal_testcases },
         { "util/", util_testcases },
+        { "bufferevent/", bufferevent_testcases },
         { "http/", http_testcases },
         { "dns/", dns_testcases },
         { "rpc/", rpc_testcases },
diff --git a/test/regress_zlib.c b/test/regress_zlib.c
index 0e7d3081..4e4af464 100644
--- a/test/regress_zlib.c
+++ b/test/regress_zlib.c
@@ -56,22 +56,17 @@
 
 void regress_zlib(void);
 
-static int test_ok;
+static int infilter_calls;
+static int outfilter_calls;
+static int readcb_finished;
+static int writecb_finished;
+static int errorcb_invoked;
 
 /*
  * Zlib filters
  */
 
 static void
-zlib_deflate_init(void *ctx)
-{
-	z_streamp p = ctx;
-
-	memset(p, 0, sizeof(z_stream));
-	assert(deflateInit(p, Z_DEFAULT_COMPRESSION) == Z_OK);
-}
-
-static void
 zlib_deflate_free(void *ctx)
 {
 	z_streamp p = ctx;
@@ -80,20 +75,25 @@ zlib_deflate_free(void *ctx)
 }
 
 static void
-zlib_inflate_init(void *ctx)
+zlib_inflate_free(void *ctx)
 {
 	z_streamp p = ctx;
 
-	memset(p, 0, sizeof(z_stream));
-	assert(inflateInit(p) == Z_OK);
+	assert(inflateEnd(p) == Z_OK);
 }
 
-static void
-zlib_inflate_free(void *ctx)
+static int
+getstate(enum bufferevent_flush_mode state)
 {
-	z_streamp p = ctx;
-
-	assert(inflateEnd(p) == Z_OK);
+	switch (state) {
+	case BEV_FINISHED:
+		return Z_FINISH;
+	case BEV_FLUSH:
+		return Z_SYNC_FLUSH;
+	case BEV_NORMAL:
+	default:
+		return Z_NO_FLUSH;
+	}
 }
 
 /*
@@ -103,7 +103,7 @@ zlib_inflate_free(void *ctx)
  */
 static enum bufferevent_filter_result
 zlib_input_filter(struct evbuffer *src, struct evbuffer *dst,
-    enum bufferevent_filter_state state, void *ctx)
+    ssize_t lim, enum bufferevent_flush_mode state, void *ctx)
 {
 	int nread, nwrite;
 	int res;
@@ -119,9 +119,7 @@ zlib_input_filter(struct evbuffer *src, struct evbuffer *dst,
 		p->avail_out = 4096;
 
 		/* we need to flush zlib if we got a flush */
-		res = inflate(p, state == BEV_FLUSH ?
-		    Z_FINISH : Z_NO_FLUSH);
-		assert(res == Z_OK || res == Z_STREAM_END);
+		res = inflate(p, getstate(state));
 
 		/* let's figure out how much was compressed */
 		nread = evbuffer_get_contiguous_space(src) - p->avail_in;
@@ -129,16 +127,27 @@ zlib_input_filter(struct evbuffer *src, struct evbuffer *dst,
 
 		evbuffer_drain(src, nread);
 		evbuffer_commit_space(dst, nwrite);
+
+		if (res==Z_BUF_ERROR) {
+			/* We're out of space, or out of decodeable input.
+			   Only if nwrite == 0 assume the latter.
+			 */
+			if (nwrite == 0)
+				return BEV_NEED_MORE;
+		} else {
+			assert(res == Z_OK || res == Z_STREAM_END);
+		}
+
 	} while (EVBUFFER_LENGTH(src) > 0);
 
-	test_ok++;
+        ++infilter_calls;
 
 	return (BEV_OK);
 }
 
 static enum bufferevent_filter_result
 zlib_output_filter(struct evbuffer *src, struct evbuffer *dst,
-    enum bufferevent_filter_state state, void *ctx)
+    ssize_t lim, enum bufferevent_flush_mode state, void *ctx)
 {
 	int nread, nwrite;
 	int res;
@@ -153,9 +162,9 @@ zlib_output_filter(struct evbuffer *src, struct evbuffer *dst,
 		p->next_out = evbuffer_reserve_space(dst, 4096);
 		p->avail_out = 4096;
 
+
 		/* we need to flush zlib if we got a flush */
-		res = deflate(p, state == BEV_FLUSH ? Z_FINISH : Z_NO_FLUSH);
-		assert(res == Z_OK || res == Z_STREAM_END);
+		res = deflate(p, getstate(state));
 
 		/* let's figure out how much was compressed */
 		nread = evbuffer_get_contiguous_space(src) - p->avail_in;
@@ -163,9 +172,20 @@ zlib_output_filter(struct evbuffer *src, struct evbuffer *dst,
 
 		evbuffer_drain(src, nread);
 		evbuffer_commit_space(dst, nwrite);
+
+		if (res==Z_BUF_ERROR) {
+			/* We're out of space, or out of decodeable input.
+			   Only if nwrite == 0 assume the latter.
+			 */
+			if (nwrite == 0)
+				return BEV_NEED_MORE;
+		} else {
+			assert(res == Z_OK || res == Z_STREAM_END);
+		}
+
 	} while (EVBUFFER_LENGTH(src) > 0);
 
-	test_ok++;
+	++outfilter_calls;
 
 	return (BEV_OK);
 }
@@ -186,8 +206,9 @@ readcb(struct bufferevent *bev, void *arg)
 
 		bufferevent_disable(bev, EV_READ);
 
-		if (EVBUFFER_LENGTH(evbuf) == 8333)
-			test_ok++;
+		if (EVBUFFER_LENGTH(evbuf) == 8333) {
+			++readcb_finished;
+                }
 
 		evbuffer_free(evbuf);
 	}
@@ -196,26 +217,29 @@ readcb(struct bufferevent *bev, void *arg)
 static void
 writecb(struct bufferevent *bev, void *arg)
 {
-	if (EVBUFFER_LENGTH(bufferevent_get_output(bev)) == 0)
-		test_ok++;
+	if (EVBUFFER_LENGTH(bufferevent_get_output(bev)) == 0) {
+		++writecb_finished;
+        }
 }
 
 static void
 errorcb(struct bufferevent *bev, short what, void *arg)
 {
-	test_ok = -2;
+	errorcb_invoked = 1;
 }
 
 static void
 test_bufferevent_zlib(void)
 {
-	struct bufferevent *bev1, *bev2;
-	struct bufferevent_filter *finput, *foutput;
+	struct bufferevent *bev1, *bev2, *bev1_orig, *bev2_orig;
 	char buffer[8333];
 	z_stream z_input, z_output;
-	int i, pair[2];
+	int i, pair[2], r;
+        int test_ok;
+
+	infilter_calls = outfilter_calls = readcb_finished = writecb_finished
+            = errorcb_invoked = 0;
 
-	test_ok = 0;
 	fprintf(stdout, "Testing Zlib Filter: ");
 
 	if (evutil_socketpair(AF_UNIX, SOCK_STREAM, 0, pair) == -1) {
@@ -226,21 +250,27 @@ test_bufferevent_zlib(void)
 	evutil_make_socket_nonblocking(pair[0]);
 	evutil_make_socket_nonblocking(pair[1]);
 
-	bev1 = bufferevent_new(pair[0], readcb, writecb, errorcb, NULL);
-	bev2 = bufferevent_new(pair[1], readcb, writecb, errorcb, NULL);
+	bev1_orig = bev1 = bufferevent_socket_new(NULL, pair[0], 0);
+	bev2_orig = bev2 = bufferevent_socket_new(NULL, pair[1], 0);
+
+	memset(&z_output, 0, sizeof(z_output));
+	r = deflateInit(&z_output, Z_DEFAULT_COMPRESSION);
+	assert(r == Z_OK);
+	memset(&z_input, 0, sizeof(z_input));
+	r = inflateInit(&z_input);
 
 	/* initialize filters */
-	finput = bufferevent_filter_new(
-		zlib_inflate_init, zlib_inflate_free,
-		zlib_input_filter, &z_input);
-	bufferevent_filter_insert(bev2, BEV_INPUT, finput);
+	bev1 = bufferevent_filter_new(bev1, NULL, zlib_output_filter, 0,
+				      zlib_deflate_free, &z_output);
+	bev2 = bufferevent_filter_new(bev2, zlib_input_filter,
+				      NULL, 0, zlib_inflate_free, &z_input);
+	bufferevent_setcb(bev1, readcb, writecb, errorcb, NULL);
+	bufferevent_setcb(bev2, readcb, writecb, errorcb, NULL);
 
-	foutput = bufferevent_filter_new(
-		zlib_deflate_init, zlib_deflate_free,
-		zlib_output_filter, &z_output);
-	bufferevent_filter_insert(bev1, BEV_OUTPUT, foutput);
 
 	bufferevent_disable(bev1, EV_READ);
+	bufferevent_enable(bev1, EV_WRITE);
+
 	bufferevent_enable(bev2, EV_READ);
 
 	for (i = 0; i < sizeof(buffer); i++)
@@ -251,14 +281,21 @@ test_bufferevent_zlib(void)
 	bufferevent_write(bev1, buffer + 1800, sizeof(buffer) - 1800);
 
 	/* we are done writing - we need to flush everything */
-	bufferevent_trigger_filter(bev1, NULL, BEV_OUTPUT, BEV_FLUSH);
+	bufferevent_flush(bev1, EV_WRITE, BEV_FINISHED);
 
 	event_dispatch();
 
 	bufferevent_free(bev1);
 	bufferevent_free(bev2);
 
-	if (test_ok != 6) {
+
+        test_ok = infilter_calls &&
+            outfilter_calls &&
+            readcb_finished &&
+            writecb_finished &&
+            !errorcb_invoked;
+
+        if (! test_ok) {
 		fprintf(stdout, "FAILED: %d\n", test_ok);
 		exit(1);
 	}