Fix asynchronous IO API.

Write test case for aio.

src/aio_event.c

@@ -35,6 +35,7 @@
 #define MAX_AIO_EV        32768
 
 struct io_cbdata {
+	gint fd;
 	rspamd_aio_cb cb;
 	gsize len;
 	gpointer buf;
@@ -58,33 +59,53 @@ typedef enum io_iocb_cmd {
 	IO_CMD_NOOP = 6,
 } io_iocb_cmd_t;
 
-struct io_iocb_common {
-	void *buf;
-	unsigned __pad1;
-	long nbytes;
-	unsigned __pad2;
-	long long offset;
-	long long __pad3;
-    unsigned flags;
-    unsigned resfd;
-}; /* result code is the amount read or -'ve errno */
+#if defined(__LITTLE_ENDIAN)
+#define PADDED(x,y)     x, y
+#elif defined(__BIG_ENDIAN)
+#define PADDED(x,y)     y, x
+#else
+#error edit for your odd byteorder.
+#endif
+
+/*
+ * we always use a 64bit off_t when communicating
+ * with userland.  its up to libraries to do the
+ * proper padding and aio_error abstraction
+ */
 
 struct iocb {
-	void *data;
-	unsigned key;
-	short aio_lio_opcode;
-	short aio_reqprio;
-	int aio_fildes;
-	union {
-		struct io_iocb_common c;
-	} u;
+	/* these are internal to the kernel/libc. */
+	guint64 aio_data; /* data to be returned in event's data */
+	guint32 PADDED(aio_key, aio_reserved1);
+	/* the kernel sets aio_key to the req # */
+
+	/* common fields */
+	guint16 aio_lio_opcode; /* see IOCB_CMD_ above */
+	gint16 aio_reqprio;
+	guint32 aio_fildes;
+
+	guint64 aio_buf;
+	guint64 aio_nbytes;
+	guint64 aio_offset;
+
+	/* extra parameters */
+	guint64 aio_reserved2; /* TODO: use this for a (struct sigevent *) */
+
+	/* flags for the "struct iocb" */
+	guint32 aio_flags;
+
+	/*
+	 * if the IOCB_FLAG_RESFD flag of "aio_flags" is set, this is an
+	 * eventfd to signal AIO readiness to
+	 */
+	guint32 aio_resfd;
 };
 
 struct io_event {
-	uint64_t  data;  /* the data field from the iocb */
-	uint64_t  obj;   /* what iocb this event came from */
-	int64_t   res;   /* result code for this event */
-	int64_t   res2;  /* secondary result */
+	guint64  data;  /* the data field from the iocb */
+	guint64  obj;   /* what iocb this event came from */
+	gint64   res;   /* result code for this event */
+	gint64   res2;  /* secondary result */
 };
 
 /* Linux specific io calls */
@@ -178,7 +199,7 @@ rspamd_eventfdcb (gint fd, gshort what, gpointer ud)
 			for (i = 0; i < done; i ++) {
 				ev_data = (struct io_cbdata *) (uintptr_t) event[i].data;
 				/* Call this callback */
-				ev_data->cb (event[i].res, ev_data->len, ev_data->buf, ev_data->ud);
+				ev_data->cb (ev_data->fd, event[i].res, ev_data->len, ev_data->buf, ev_data->ud);
 			}
 		}
 		else if (done == 0) {
@@ -221,6 +242,7 @@ rspamd_aio_init (struct event_base *base)
 		else {
 			event_set (&new->eventfd_ev, new->event_fd, EV_READ|EV_PERSIST, rspamd_eventfdcb, new);
 			event_base_set (new->base, &new->eventfd_ev);
+			event_add (&new->eventfd_ev, NULL);
 			if (io_setup (MAX_AIO_EV, &new->io_ctx) == -1) {
 				msg_err ("io_setup failed: %s", strerror (errno));
 				close (new->event_fd);
@@ -278,18 +300,19 @@ rspamd_aio_read (gint fd, gpointer buf, gsize len, struct aio_context *ctx, rspa
 		cbdata->buf = buf;
 		cbdata->len = len;
 		cbdata->ud = ud;
+		cbdata->fd = fd;
 
 		iocb[0] = alloca (sizeof (struct iocb));
 		memset (iocb[0], 0, sizeof (struct iocb));
 		iocb[0]->aio_fildes = fd;
 		iocb[0]->aio_lio_opcode = IO_CMD_PREAD;
 		iocb[0]->aio_reqprio = 0;
-		iocb[0]->u.c.buf = buf;
-		iocb[0]->u.c.nbytes = len;
-		iocb[0]->u.c.offset = 0;
-		iocb[0]->u.c.flags |= (1 << 0) /* IOCB_FLAG_RESFD */;
-		iocb[0]->u.c.resfd = ctx->event_fd;
-		iocb[0]->data = cbdata;
+		iocb[0]->aio_buf = (guint64)((uintptr_t)buf);
+		iocb[0]->aio_nbytes = len;
+		iocb[0]->aio_offset = 0;
+		iocb[0]->aio_flags |= (1 << 0) /* IOCB_FLAG_RESFD */;
+		iocb[0]->aio_resfd = ctx->event_fd;
+		iocb[0]->aio_data = (guint64)((uintptr_t)cbdata);
 
 		/* Iocb is copied to kernel internally, so it is safe to put it on stack */
 		if (io_submit (ctx->io_ctx, 1, iocb) == 1) {
@@ -311,10 +334,10 @@ rspamd_aio_read (gint fd, gpointer buf, gsize len, struct aio_context *ctx, rspa
 blocking:
 		r = read (fd, buf, len);
 		if (r >= 0) {
-			cb (0, r, buf, ud);
+			cb (fd, 0, r, buf, ud);
 		}
 		else {
-			cb (r, -1, buf, ud);
+			cb (fd, r, -1, buf, ud);
 		}
 	}
 
@@ -339,18 +362,19 @@ rspamd_aio_write (gint fd, gpointer buf, gsize len, struct aio_context *ctx, rsp
 		cbdata->buf = buf;
 		cbdata->len = len;
 		cbdata->ud = ud;
+		cbdata->fd = fd;
 
 		iocb[0] = alloca (sizeof (struct iocb));
 		memset (iocb[0], 0, sizeof (struct iocb));
 		iocb[0]->aio_fildes = fd;
 		iocb[0]->aio_lio_opcode = IO_CMD_PWRITE;
 		iocb[0]->aio_reqprio = 0;
-		iocb[0]->u.c.buf = buf;
-		iocb[0]->u.c.nbytes = len;
-		iocb[0]->u.c.offset = 0;
-		iocb[0]->u.c.flags |= (1 << 0) /* IOCB_FLAG_RESFD */;
-		iocb[0]->u.c.resfd = ctx->event_fd;
-		iocb[0]->data = cbdata;
+		iocb[0]->aio_buf = (guint64)((uintptr_t)buf);
+		iocb[0]->aio_nbytes = len;
+		iocb[0]->aio_offset = 0;
+		iocb[0]->aio_flags |= (1 << 0) /* IOCB_FLAG_RESFD */;
+		iocb[0]->aio_resfd = ctx->event_fd;
+		iocb[0]->aio_data = (guint64)((uintptr_t)cbdata);
 
 		/* Iocb is copied to kernel internally, so it is safe to put it on stack */
 		if (io_submit (ctx->io_ctx, 1, iocb) == 1) {
@@ -372,10 +396,10 @@ rspamd_aio_write (gint fd, gpointer buf, gsize len, struct aio_context *ctx, rsp
 blocking:
 		r = write (fd, buf, len);
 		if (r >= 0) {
-			cb (0, r, buf, ud);
+			cb (fd, 0, r, buf, ud);
 		}
 		else {
-			cb (r, -1, buf, ud);
+			cb (fd, r, -1, buf, ud);
 		}
 	}
 

src/aio_event.h

@@ -35,7 +35,7 @@ struct aio_context;
 /**
  * Callback for notifying
  */
-typedef void (*rspamd_aio_cb) (gint res, gsize len, gpointer data, gpointer ud);
+typedef void (*rspamd_aio_cb) (gint fd, gint res, gsize len, gpointer data, gpointer ud);
 
 /**
  * Initialize aio with specified event base

test/CMakeLists.txt

@@ -5,7 +5,8 @@ SET(TESTSRC		rspamd_expression_test.c
 				rspamd_fuzzy_test.c
 				rspamd_test_suite.c
 				rspamd_url_test.c
-				rspamd_dns_test.c)
+				rspamd_dns_test.c
+				rspamd_async_test.c)
 
 ADD_EXECUTABLE(rspamd-test EXCLUDE_FROM_ALL ${TESTSRC})
 SET_TARGET_PROPERTIES(rspamd-test PROPERTIES LINKER_LANGUAGE C)

test/rspamd_async_test.c

@@ -0,0 +1,87 @@
+/* Copyright (c) 2011, Vsevolod Stakhov
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *       * Redistributions of source code must retain the above copyright
+ *         notice, this list of conditions and the following disclaimer.
+ *       * 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.
+ *
+ * THIS SOFTWARE IS PROVIDED ''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 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.
+ */
+
+#include "config.h"
+#include "tests.h"
+#include "main.h"
+#include "aio_event.h"
+
+
+extern struct event_base *base;
+
+static void
+aio_read_cb (gint fd, gint res, gsize len, gpointer data, gpointer ud)
+{
+	guchar *p = data;
+	guint i;
+
+	g_assert (res != -1);
+
+	g_assert (len == BUFSIZ);
+	for (i = 0; i < len; i ++) {
+		g_assert (p[i] == 0xef);
+	}
+
+	event_base_loopbreak (base);
+}
+
+static void
+aio_write_cb (gint fd, gint res, gsize len, gpointer data, gpointer ud)
+{
+	struct aio_context *aio_ctx = ud;
+	static gchar testbuf[BUFSIZ];
+
+	g_assert (res != -1);
+
+	g_assert (rspamd_aio_read (fd, testbuf, sizeof (testbuf), aio_ctx, aio_read_cb, aio_ctx) != -1);
+}
+
+void
+rspamd_async_test_func ()
+{
+	struct aio_context *aio_ctx;
+	gchar *tmpfile;
+	static gchar testbuf[BUFSIZ];
+	gint fd, afd, ret;
+
+	aio_ctx = rspamd_aio_init (base);
+
+	g_assert (aio_ctx != NULL);
+
+	fd = g_file_open_tmp ("raXXXXXX", &tmpfile, NULL);
+	g_assert (fd != -1);
+
+	afd = rspamd_aio_open (aio_ctx, tmpfile, O_RDWR);
+	g_assert (fd != -1);
+
+	/* Write some data */
+	memset (testbuf, 0xef, sizeof (testbuf));
+	ret = rspamd_aio_write (afd, testbuf, sizeof (testbuf), aio_ctx, aio_write_cb, aio_ctx);
+	g_assert (ret != -1);
+
+	event_base_loop (base, 0);
+
+	close (afd);
+	close (fd);
+	unlink (tmpfile);
+}

test/rspamd_dns_test.c

@@ -69,6 +69,7 @@ rspamd_dns_test_func ()
 	struct config_file *cfg;
 	memory_pool_t *pool;
 	struct rspamd_async_session *s;
+	struct in_addr ina;
 
 	cfg = (struct config_file *)g_malloc (sizeof (struct config_file));
 	bzero (cfg, sizeof (struct config_file));
@@ -78,15 +79,16 @@ rspamd_dns_test_func ()
 
 	pool = memory_pool_new (memory_pool_get_size ());
 
-	event_init ();
 	s = new_async_session (pool, session_fin, NULL, NULL, NULL);
 
 	resolver = dns_resolver_init (base, cfg);
 
+	ina.s_addr = inet_addr ("81.19.70.3");
+
 	requests ++;
 	g_assert (make_dns_request (resolver, s, pool, test_dns_cb, NULL, DNS_REQUEST_A, "google.com"));
 	requests ++;
-	g_assert (make_dns_request (resolver, s, pool, test_dns_cb, NULL, DNS_REQUEST_PTR, inet_addr ("81.19.70.3")));
+	g_assert (make_dns_request (resolver, s, pool, test_dns_cb, NULL, DNS_REQUEST_PTR, &ina));
 	requests ++;
 	g_assert (make_dns_request (resolver, s, pool, test_dns_cb, NULL, DNS_REQUEST_MX, "rambler.ru"));
 	requests ++;

test/rspamd_test_suite.c

@@ -63,6 +63,7 @@ main (int argc, char **argv)
 	g_test_add_func ("/rspamd/expression", rspamd_expression_test_func);
 	g_test_add_func ("/rspamd/statfile", rspamd_statfile_test_func);
 	g_test_add_func ("/rspamd/dns", rspamd_dns_test_func);
+	g_test_add_func ("/rspamd/aio", rspamd_async_test_func);
 
 	g_test_run ();
 

test/tests.h

@@ -26,4 +26,7 @@ void rspamd_statfile_test_func ();
 /* DNS resolving */
 void rspamd_dns_test_func ();
 
+/* Async IO */
+void rspamd_async_test_func ();
+
 #endif