Mirrors
/
rspamd
mirror of https://github.com/rspamd/rspamd.git
3
0
Fork 0
Code Releases Activity
Browse Source

* Significate performance improving by vectorizing IO output (about 4 times for kv storage).

rspamd-0.5
Vsevolod Stakhov 14 years ago
parent
c750ae5859
commit
48e621e0c0
3 changed files with 98 additions and 27 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 116
      src/buffer.c
  2. 2
      src/buffer.h
  3. 7
      src/kvstorage_server.c

116
src/buffer.c
View File

@ -177,23 +177,98 @@ sendfile_callback (rspamd_io_dispatcher_t *d)
static gboolean
write_buffers (gint fd, rspamd_io_dispatcher_t * d, gboolean is_delayed)
{
GList *cur;
GList *cur = NULL, *tmp;
GError *err = NULL;
rspamd_buffer_t *buf;
ssize_t r;
/* Fix order */
if (d->out_buffers) {
d->out_buffers = g_list_reverse (d->out_buffers);
}
cur = g_list_first (d->out_buffers);
while (cur) {
buf = (rspamd_buffer_t *) cur->data;
if (BUFREMAIN (buf) == 0) {
/* Skip empty buffers */
cur = g_list_next (cur);
continue;
struct iovec *iov;
guint i, len, blen;
len = g_queue_get_length (d->out_buffers);
if (len > 1) {
/* IOV version */
cur = d->out_buffers->tail;
iov = g_slice_alloc (len * sizeof (struct iovec));
i = 0;
while (cur) {
buf = cur->data;
blen = BUFREMAIN (buf);
if (blen > 0) {
iov[i].iov_base = buf->pos;
iov[i].iov_len = blen;
}
else {
iov[i].iov_base = NULL;
iov[i].iov_len = 0;
}
i ++;
cur = g_list_previous (cur);
}
/* Now try to write the whole vector */
r = writev (fd, iov, len);
if (r == -1 && errno != EAGAIN) {
g_slice_free1 (len * sizeof (struct iovec), iov);
if (d->err_callback) {
err = g_error_new (G_DISPATCHER_ERROR, errno, "%s", strerror (errno));
d->err_callback (err, d->user_data);
return FALSE;
}
}
else if (r > 0) {
/* Find pos inside buffers */
cur = d->out_buffers->tail;
i = 0;
while (cur) {
buf = cur->data;
blen = BUFREMAIN (buf);
if (r >= blen) {
tmp = cur;
cur = g_list_previous (cur);
/* Mark this buffer as read */
g_queue_delete_link (d->out_buffers, tmp);
r -= blen;
}
else {
/* This buffer was not written completely */
buf->pos += r;
break;
}
i ++;
cur = g_list_previous (cur);
}
g_slice_free1 (len * sizeof (struct iovec), iov);
if (cur != 0) {
/* Wait for other event */
event_del (d->ev);
event_set (d->ev, fd, EV_WRITE, dispatcher_cb, (void *)d);
event_base_set (d->ev_base, d->ev);
event_add (d->ev, d->tv);
return TRUE;
}
}
else if (r == 0) {
/* Got EOF while we wait for data */
g_slice_free1 (len * sizeof (struct iovec), iov);
if (d->err_callback) {
err = g_error_new (G_DISPATCHER_ERROR, EOF, "got EOF");
d->err_callback (err, d->user_data);
return FALSE;
}
}
else if (r == -1 && errno == EAGAIN) {
g_slice_free1 (len * sizeof (struct iovec), iov);
debug_ip("partially write data, retry");
/* Wait for other event */
event_del (d->ev);
event_set (d->ev, fd, EV_WRITE, dispatcher_cb, (void *)d);
event_base_set (d->ev_base, d->ev);
event_add (d->ev, d->tv);
return TRUE;
}
}
else if (len == 1) {
/* Single write version */
buf = d->out_buffers->head->data;
r = write (fd, buf->pos, BUFREMAIN (buf));
if (r == -1 && errno != EAGAIN) {
if (d->err_callback) {
@ -207,7 +282,7 @@ write_buffers (gint fd, rspamd_io_dispatcher_t * d, gboolean is_delayed)
if (BUFREMAIN (buf) != 0) {
/* Continue with this buffer */
debug_ip("wrote %z bytes of %z", r, buf->data->len);
continue;
return write_buffers (fd, d, is_delayed);
}
}
else if (r == 0) {
@ -227,13 +302,11 @@ write_buffers (gint fd, rspamd_io_dispatcher_t * d, gboolean is_delayed)
event_add (d->ev, d->tv);
return TRUE;
}
cur = g_list_next (cur);
}
if (cur == NULL) {
/* Disable write event for this time */
g_list_free (d->out_buffers);
d->out_buffers = NULL;
g_queue_clear (d->out_buffers);
debug_ip("all buffers were written successfully");
@ -472,7 +545,7 @@ dispatcher_cb (gint fd, short what, void *arg)
sendfile_callback (d);
}
else {
if (d->out_buffers == NULL) {
if (g_queue_get_length (d->out_buffers) == 0) {
event_del (d->ev);
event_set (d->ev, fd, EV_READ | EV_PERSIST, dispatcher_cb, (void *)d);
event_base_set (d->ev_base, d->ev);
@ -526,6 +599,7 @@ rspamd_create_dispatcher (struct event_base *base, gint fd, enum io_policy polic
new->ev = memory_pool_alloc0 (new->pool, sizeof (struct event));
new->fd = fd;
new->ev_base = base;
new->out_buffers = g_queue_new ();
event_set (new->ev, fd, EV_WRITE, dispatcher_cb, (void *)new);
event_base_set (new->ev_base, new->ev);
@ -540,9 +614,7 @@ rspamd_remove_dispatcher (rspamd_io_dispatcher_t * dispatcher)
if (dispatcher != NULL) {
event_del (dispatcher->ev);
memory_pool_delete (dispatcher->pool);
if (dispatcher->out_buffers) {
g_list_free (dispatcher->out_buffers);
}
g_queue_free (dispatcher->out_buffers);
g_free (dispatcher);
}
}
@ -609,7 +681,7 @@ rspamd_dispatcher_write (rspamd_io_dispatcher_t * d, void *data, size_t len, gbo
newbuf->pos = newbuf->data->begin;
newbuf->data->len = len;
d->out_buffers = g_list_prepend (d->out_buffers, newbuf);
g_queue_push_head (d->out_buffers, newbuf);
if (!delayed) {
debug_ip("plan write event");

2
src/buffer.h
View File

@ -33,7 +33,7 @@ typedef struct rspamd_buffer_s {
typedef struct rspamd_io_dispatcher_s {
rspamd_buffer_t *in_buf; /**< input buffer */
GList *out_buffers; /**< out buffers chain */
GQueue *out_buffers; /**< out buffers chain */
struct timeval *tv; /**< io timeout */
struct event *ev; /**< libevent io event */
memory_pool_t *pool; /**< where to store data */

7
src/kvstorage_server.c
View File

@ -355,10 +355,10 @@ kvstorage_read_socket (f_str_t * in, void *arg)
r = rspamd_snprintf (outbuf, sizeof (outbuf), "VALUE %s %ud %ud" CRLF,
elt->key, elt->flags, elt->size);
if (!rspamd_dispatcher_write (session->dispather, outbuf,
r, FALSE, FALSE)) {
r, TRUE, FALSE)) {
return FALSE;
}
if (!rspamd_dispatcher_write (session->dispather, elt->data, elt->size, FALSE, TRUE)) {
if (!rspamd_dispatcher_write (session->dispather, elt->data, elt->size, TRUE, TRUE)) {
return FALSE;
}
return rspamd_dispatcher_write (session->dispather, CRLF "END" CRLF,
@ -443,9 +443,9 @@ thr_accept_socket (gint fd, short what, void *arg)
g_static_mutex_unlock (thr->accept_mtx);
return;
}
g_static_mutex_unlock (thr->accept_mtx);
/* Check for EAGAIN */
if (nfd == 0) {
g_static_mutex_unlock (thr->accept_mtx);
return;
}
@ -465,7 +465,6 @@ thr_accept_socket (gint fd, short what, void *arg)
memcpy (&session->client_addr, &su.s4.sin_addr,
sizeof (struct in_addr));
}
g_static_mutex_unlock (thr->accept_mtx);
}
/**

Write Preview
Loading…
Cancel
Save