#define MAX_AIO_EV 32768
struct io_cbdata {
+ gint fd;
rspamd_aio_cb cb;
gsize len;
gpointer buf;
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 */
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) {
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);
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) {
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);
}
}
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) {
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);
}
}
--- /dev/null
+/* 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);
+}
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));
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 ++;
projects / rspamd.git / commitdiff