* Some performance improvements to IO dispatcher (do not drain the whole buffer after a single line readed)

* Fix smtp data input

6 files changed, 50 insertions, 75 deletions

diff --git a/src/buffer.c b/src/buffer.c
index 5eb2c81d1..899b42fee 100644
--- a/src/buffer.c
+++ b/src/buffer.c
@@ -313,31 +313,44 @@ read_buffers (int fd, rspamd_io_dispatcher_t * d, gboolean skip_read)
 	c = d->in_buf->data->begin;
 	b = c;
 	r = 0;
-
+	
 	switch (d->policy) {
 	case BUFFER_LINE:
+		/** Variables:
+		* b - begin of line
+		* r - current position in buffer
+		* *len - length of remaining buffer
+		* c - pointer to current position (buffer->begin + r)
+		* res - result string 
+		*/
 		while (r < *len) {
 			if (*c == '\n') {
 				res.begin = b;
-				res.len = r;
-				if (r != 0 && *(c - 1) == '\r') {
-					res.len--;
+				res.len = c - b;
+				/* Strip EOL */
+				if (d->strip_eol) {
+					if (r != 0 && *(c - 1) == '\r') {
+						res.len--;
+					}
+				}
+				else {
+					/* Include EOL in reply */
+					res.len ++;
 				}
+				/* Set new begin of line */
+				b = c + 1;
+				/* Call callback for a line */
 				if (d->read_callback) {
 					if (!d->read_callback (&res, d->user_data)) {
 						return;
 					}
-					/* Move remaining string to begin of buffer (draining) */
-					/* Reinit pointers as buffer may be changed */
-					len = &d->in_buf->data->len;
-					pos = &d->in_buf->pos;
-					memmove (d->in_buf->data->begin, c + 1, *len - r - 1);
-					b = d->in_buf->data->begin;
-					c = b;
-					*len -= r + 1;
-					*pos = b + *len;
-					r = 0;
 					if (d->policy != saved_policy) {
+						/* Drain buffer as policy is changed */
+						len = &d->in_buf->data->len;
+						pos = &d->in_buf->pos;
+						memmove (d->in_buf->data->begin, b, c - b + 1);
+						*len = c - b + 1;
+						*pos = d->in_buf->data->begin + *len;
 						debug_ip (d->peer_addr, "policy changed during callback, restart buffer's processing");
 						read_buffers (fd, d, TRUE);
 						return;
@@ -347,6 +360,12 @@ read_buffers (int fd, rspamd_io_dispatcher_t * d, gboolean skip_read)
 			r++;
 			c++;
 		}
+		/* Now drain buffer */
+		len = &d->in_buf->data->len;
+		pos = &d->in_buf->pos;
+		memmove (d->in_buf->data->begin, b, c - b);
+		*len = c - b;
+		*pos = d->in_buf->data->begin + *len;
 		break;
 	case BUFFER_CHARACTER:
 		r = d->nchars;
@@ -462,6 +481,7 @@ rspamd_create_dispatcher (int fd, enum io_policy policy,
 	new->write_callback = write_cb;
 	new->err_callback = err_cb;
 	new->user_data = user_data;
+	new->strip_eol = TRUE;
 
 	new->ev = memory_pool_alloc0 (new->pool, sizeof (struct event));
 	new->fd = fd;
@@ -514,6 +534,7 @@ rspamd_set_dispatcher_policy (rspamd_io_dispatcher_t * d, enum io_policy policy,
 				d->in_buf->data = tmp;
 				d->in_buf->pos = d->in_buf->data->begin + t;
 			}
+			d->strip_eol = TRUE;
 		}
 	}
 
diff --git a/src/buffer.h b/src/buffer.h
index 9f3897d1c..cbb6c190a 100644
--- a/src/buffer.h
+++ b/src/buffer.h
@@ -50,6 +50,7 @@ typedef struct rspamd_io_dispatcher_s {
 	size_t file_size;
 	int sendfile_fd;
 	gboolean in_sendfile;											/**< whether buffer is in sendfile mode */
+	gboolean strip_eol;												/**< strip or not line ends in BUFFER_LINE policy */
 #ifndef HAVE_SENDFILE
 	void *map;
 #endif
diff --git a/src/smtp.c b/src/smtp.c
index c7e9dc428..7860f1ac5 100644
--- a/src/smtp.c
+++ b/src/smtp.c
@@ -405,8 +405,6 @@ static                          gboolean
 smtp_read_socket (f_str_t * in, void *arg)
 {
 	struct smtp_session            *session = arg;
-	char                           *p;
-	gboolean                        do_write;
 
 	switch (session->state) {
 		case SMTP_STATE_RESOLVE_REVERSE:
@@ -429,58 +427,17 @@ smtp_read_socket (f_str_t * in, void *arg)
 			if (in->len == 0) {
 				return TRUE;
 			}
-			p = in->begin + in->len;
-			do_write = TRUE;
-			if (in->len > sizeof (session->data_end)) {
-				/* New data is more than trailer buffer */
-				if (session->data_idx != 0 && write (session->temp_fd, session->data_end, session->data_idx) != session->data_idx) {
-					msg_err ("cannot write to temp file: %s", strerror (errno));
-					session->error = SMTP_ERROR_FILE;
-					session->state = SMTP_STATE_CRITICAL_ERROR;
-					rspamd_dispatcher_write (session->dispatcher, session->error, 0, FALSE, TRUE);
-					destroy_session (session->s);
-					return FALSE;
-				}
-				memcpy (session->data_end, p - sizeof (session->data_end), sizeof (session->data_end));
-				session->data_idx = 5;
-			}
-			else if (session->data_idx + in->len < sizeof (session->data_end)){
-				/* New data is less than trailer buffer plus index */
-				memcpy (session->data_end + session->data_idx, in->begin, in->len);
-				session->data_idx += in->len;
-				do_write = FALSE;
+			if (in->len == 3 && memcmp (in->begin, DATA_END_TRAILER, in->len) == 0) {
+				return process_smtp_data (session);
 			}
-			else {
-				/* Save remaining bytes */
-				if (session->data_idx != 0 && write (session->temp_fd, session->data_end, session->data_idx) != session->data_idx) {
-					msg_err ("cannot write to temp file: %s", strerror (errno));
-					session->error = SMTP_ERROR_FILE;
-					session->state = SMTP_STATE_CRITICAL_ERROR;
-					rspamd_dispatcher_write (session->dispatcher, session->error, 0, FALSE, TRUE);
-					destroy_session (session->s);
-					return FALSE;
-				}
-				/* Move bytes */
-				session->data_idx = sizeof (session->data_end) - in->len;
-				memmove (session->data_end, session->data_end + (sizeof (session->data_end) - in->len) + 1, sizeof (session->data_end) - in->len);
-				memcpy (session->data_end + session->data_idx, in->begin, in->len);
-				session->data_idx = 5;
-			}
-			if (do_write) {
-				if (session->data_idx < in->len) {
-					if (in->len - session->data_idx != 0 && 
-							write (session->temp_fd, in->begin, in->len - session->data_idx) != in->len - session->data_idx) {
-						msg_err ("cannot write to temp file: %s", strerror (errno));
-						session->error = SMTP_ERROR_FILE;
-						session->state = SMTP_STATE_CRITICAL_ERROR;
-						rspamd_dispatcher_write (session->dispatcher, session->error, 0, FALSE, TRUE);
-						destroy_session (session->s);
-						return FALSE;
-					}
-				}
-				if (memcmp (session->data_end, DATA_END_TRAILER, sizeof (session->data_end)) == 0) {
-					return process_smtp_data (session);
-				}
+
+			if (write (session->temp_fd, in->begin, in->len) != in->len) {
+				msg_err ("cannot write to temp file: %s", strerror (errno));
+				session->error = SMTP_ERROR_FILE;
+				session->state = SMTP_STATE_CRITICAL_ERROR;
+				rspamd_dispatcher_write (session->dispatcher, session->error, 0, FALSE, TRUE);
+				destroy_session (session->s);
+				return FALSE;
 			}
 			break;
 		case SMTP_STATE_WAIT_UPSTREAM:
diff --git a/src/smtp.h b/src/smtp.h
index 598526a6b..83d89c5aa 100644
--- a/src/smtp.h
+++ b/src/smtp.h
@@ -83,9 +83,6 @@ struct smtp_session {
 
 	struct smtp_upstream *upstream;
 
-	char data_end[5];
-	char data_idx;
-
 	gboolean resolved;
 	gboolean esmtp;
 };
diff --git a/src/smtp_proto.c b/src/smtp_proto.c
index 0eef4ec17..b80573276 100644
--- a/src/smtp_proto.c
+++ b/src/smtp_proto.c
@@ -380,7 +380,7 @@ smtp_upstream_write_socket (void *arg)
 	
 	if (session->upstream_state == SMTP_STATE_IN_SENDFILE) {
 		session->upstream_state = SMTP_STATE_END;
-		return rspamd_dispatcher_write (session->upstream_dispatcher, DATA_END_TRAILER, sizeof (DATA_END_TRAILER) - 1, FALSE, TRUE);
+		return rspamd_dispatcher_write (session->upstream_dispatcher, CRLF DATA_END_TRAILER, sizeof (CRLF DATA_END_TRAILER) - 1, FALSE, TRUE);
 	}
 
 	return TRUE;
@@ -567,9 +567,8 @@ smtp_upstream_read_socket (f_str_t * in, void *arg)
 				rspamd_dispatcher_restore (session->dispatcher);
 				rspamd_dispatcher_write (session->dispatcher, session->error, 0, FALSE, TRUE);
 				rspamd_dispatcher_pause (session->upstream_dispatcher);
-				rspamd_set_dispatcher_policy (session->dispatcher, BUFFER_ANY, 0);
-				session->data_idx = 0;
-				memset (session->data_end, 0, sizeof (session->data_end));
+				rspamd_set_dispatcher_policy (session->dispatcher, BUFFER_LINE, 0);
+				session->dispatcher->strip_eol = FALSE;
 				return TRUE;
 			}
 			break;
diff --git a/src/smtp_proto.h b/src/smtp_proto.h
index e850e4a5d..f451f3abf 100644
--- a/src/smtp_proto.h
+++ b/src/smtp_proto.h
@@ -15,7 +15,7 @@
 #define SMTP_ERROR_OK "250 Requested mail action okay, completed" CRLF
 #define SMTP_ERROR_DATA_OK "354 Start mail input; end with <CRLF>.<CRLF>" CRLF
 
-#define DATA_END_TRAILER CRLF "." CRLF
+#define DATA_END_TRAILER "." CRLF
 
 
 struct smtp_command {