Start splitting network backends, added writev and TCP_CORK

src/network.c

@@ -2,7 +2,7 @@
 #include "network.h"
 
 /** repeats write after EINTR */
-static ssize_t net_write(int fd, void *buf, ssize_t nbyte) {
+ssize_t net_write(int fd, void *buf, ssize_t nbyte) {
 	ssize_t r;
 	while (-1 == (r = write(fd, buf, nbyte))) {
 		switch (errno) {
@@ -18,55 +18,8 @@ static ssize_t net_write(int fd, void *buf, ssize_t nbyte) {
 	return r;
 }
 
-network_status_t network_write(server *srv, connection *con, int fd, chunkqueue *cq) {
-	const ssize_t blocksize = 16*1024; /* 16k */
-	const off_t max_write = 16 * blocksize; /* 256k */
-	char *block_data;
-	off_t block_len;
-	ssize_t r;
-	off_t len = 0;
-	chunkiter ci;
-
-	do {
-		if (0 == cq->length) return NETWORK_STATUS_SUCCESS;
-
-		ci = chunkqueue_iter(cq);
-		switch (chunkiter_read(srv, con, ci, 0, blocksize, &block_data, &block_len)) {
-		case HANDLER_GO_ON:
-			break;
-		case HANDLER_WAIT_FOR_FD:
-			return len ? NETWORK_STATUS_SUCCESS : NETWORK_STATUS_WAIT_FOR_EVENT;
-		case HANDLER_ERROR:
-		default:
-			return NETWORK_STATUS_FATAL_ERROR;
-		}
-
-		if (-1 == (r = net_write(fd, block_data, block_len))) {
-			switch (errno) {
-			case EAGAIN:
-#if EWOULDBLOCK != EAGAIN
-			case EWOULDBLOCK
-#endif
-				return len ? NETWORK_STATUS_SUCCESS : NETWORK_STATUS_WAIT_FOR_EVENT;
-			case ECONNRESET:
-			case EPIPE:
-				return NETWORK_STATUS_CONNECTION_CLOSE;
-			default:
-				CON_ERROR(srv, con, "oops, write to fd=%d failed: %s (%d)", fd, strerror(errno), errno );
-				return NETWORK_STATUS_FATAL_ERROR;
-			}
-		} else if (0 == r) {
-			return len ? NETWORK_STATUS_SUCCESS : NETWORK_STATUS_WAIT_FOR_EVENT;
-		}
-		chunkqueue_skip(cq, r);
-		len += r;
-	} while (r == block_len && len < max_write);
-
-	return NETWORK_STATUS_SUCCESS;
-}
-
 /** repeats read after EINTR */
-static ssize_t net_read(int fd, void *buf, ssize_t nbyte) {
+ssize_t net_read(int fd, void *buf, ssize_t nbyte) {
 	ssize_t r;
 	while (-1 == (r = read(fd, buf, nbyte))) {
 		switch (errno) {
@@ -82,6 +35,34 @@ static ssize_t net_read(int fd, void *buf, ssize_t nbyte) {
 	return r;
 }
 
+network_status_t network_write(server *srv, connection *con, int fd, chunkqueue *cq) {
+	network_status_t res;
+#ifdef TCP_CORK
+	int corked = 0;
+#endif
+
+#ifdef TCP_CORK
+	/* Linux: put a cork into the socket as we want to combine the write() calls
+	 * but only if we really have multiple chunks
+	 */
+	if (cq->queue->length > 1) {
+		corked = 1;
+		setsockopt(fd, IPPROTO_TCP, TCP_CORK, &corked, sizeof(corked));
+	}
+#endif
+
+	res = network_backend_writev(srv, con, fd, cq);
+
+#ifdef TCP_CORK
+	if (corked) {
+		corked = 0;
+		setsockopt(fd, IPPROTO_TCP, TCP_CORK, &corked, sizeof(corked));
+	}
+#endif
+
+	return res;
+}
+
 network_status_t network_read(server *srv, connection *con, int fd, chunkqueue *cq) {
 	const ssize_t blocksize = 16*1024; /* 16k */
 	const off_t max_read = 16 * blocksize; /* 256k */

src/network.h

@@ -4,15 +4,35 @@
 #include "base.h"
 
 typedef enum {
-	NETWORK_STATUS_SUCCESS,
+	NETWORK_STATUS_SUCCESS,             /**< some IO was actually done (read/write) or cq was empty for write */
 	NETWORK_STATUS_FATAL_ERROR,
 	NETWORK_STATUS_CONNECTION_CLOSE,
-	NETWORK_STATUS_WAIT_FOR_EVENT,
-	NETWORK_STATUS_WAIT_FOR_AIO_EVENT,
-	NETWORK_STATUS_WAIT_FOR_FD,
+	NETWORK_STATUS_WAIT_FOR_EVENT,      /**< read/write returned -1 with errno=EAGAIN/EWOULDBLOCK; no real IO was done */
+	NETWORK_STATUS_WAIT_FOR_AIO_EVENT,  /**< nothing done yet, read/write will be done somewhere else */
+	NETWORK_STATUS_WAIT_FOR_FD,         /**< need a fd to open a file */
 } network_status_t;
 
+/** repeats write after EINTR */
+LI_API ssize_t net_write(int fd, void *buf, ssize_t nbyte);
+
+/** repeats read after EINTR */
+LI_API ssize_t net_read(int fd, void *buf, ssize_t nbyte);
+
 LI_API network_status_t network_write(server *srv, connection *con, int fd, chunkqueue *cq);
 LI_API network_status_t network_read(server *srv, connection *con, int fd, chunkqueue *cq);
 
+/* write backends */
+LI_API network_status_t network_backend_write(server *srv, connection *con, int fd, chunkqueue *cq);
+LI_API network_status_t network_backend_writev(server *srv, connection *con, int fd, chunkqueue *cq);
+
+#define NETWORK_FALLBACK(f) do { \
+	network_status_t res; \
+	switch(res = f(srv, con, fd, cq)) { \
+		case NETWORK_STATUS_SUCCESS: \
+			break; \
+		default: \
+			return res; \
+	} \
+} while(0)
+
 #endif

src/network_write.c

@@ -0,0 +1,49 @@
+
+#include "network.h"
+
+network_status_t network_backend_write(server *srv, connection *con, int fd, chunkqueue *cq) {
+	const ssize_t blocksize = 16*1024; /* 16k */
+	const off_t max_write = 16 * blocksize; /* 256k */
+	char *block_data;
+	off_t block_len;
+	ssize_t r;
+	off_t len = 0;
+	chunkiter ci;
+
+	do {
+		if (0 == cq->length) return NETWORK_STATUS_SUCCESS;
+
+		ci = chunkqueue_iter(cq);
+		switch (chunkiter_read(srv, con, ci, 0, blocksize, &block_data, &block_len)) {
+		case HANDLER_GO_ON:
+			break;
+		case HANDLER_WAIT_FOR_FD:
+			return len ? NETWORK_STATUS_SUCCESS : NETWORK_STATUS_WAIT_FOR_EVENT;
+		case HANDLER_ERROR:
+		default:
+			return NETWORK_STATUS_FATAL_ERROR;
+		}
+
+		if (-1 == (r = net_write(fd, block_data, block_len))) {
+			switch (errno) {
+			case EAGAIN:
+#if EWOULDBLOCK != EAGAIN
+			case EWOULDBLOCK
+#endif
+				return len ? NETWORK_STATUS_SUCCESS : NETWORK_STATUS_WAIT_FOR_EVENT;
+			case ECONNRESET:
+			case EPIPE:
+				return NETWORK_STATUS_CONNECTION_CLOSE;
+			default:
+				CON_ERROR(srv, con, "oops, write to fd=%d failed: %s (%d)", fd, strerror(errno), errno );
+				return NETWORK_STATUS_FATAL_ERROR;
+			}
+		} else if (0 == r) {
+			return len ? NETWORK_STATUS_SUCCESS : NETWORK_STATUS_WAIT_FOR_EVENT;
+		}
+		chunkqueue_skip(cq, r);
+		len += r;
+	} while (r == block_len && len < max_write);
+
+	return NETWORK_STATUS_SUCCESS;
+}

src/network_writev.c

@@ -0,0 +1,104 @@
+
+#include "network.h"
+
+#include <sys/uio.h>
+
+#ifndef UIO_MAXIOV
+# if defined(__FreeBSD__) || defined(__APPLE__) || defined(__NetBSD__)
+/* FreeBSD 4.7 defines it in sys/uio.h only if _KERNEL is specified */
+#  define UIO_MAXIOV 1024
+# elif defined(__sgi)
+/* IRIX 6.5 has sysconf(_SC_IOV_MAX) which might return 512 or bigger */
+#  define UIO_MAXIOV 512
+# elif defined(__sun)
+/* Solaris (and SunOS?) defines IOV_MAX instead */
+#  ifndef IOV_MAX
+#   define UIO_MAXIOV 16
+#  else
+#   define UIO_MAXIOV IOV_MAX
+#  endif
+# elif defined(IOV_MAX)
+#  define UIO_MAXIOV IOV_MAX
+# else
+#  error UIO_MAXIOV nor IOV_MAX are defined
+# endif
+#endif
+
+network_status_t network_backend_writev(server *srv, connection *con, int fd, chunkqueue *cq) {
+	const off_t max_write = 256 * 1024; /* 256k */
+	off_t min_cq_len, max_chunks_len, we_have;
+	ssize_t r;
+	gboolean did_write_something = FALSE;
+	chunkiter ci;
+	chunk *c;
+
+	GArray *chunks = g_array_sized_new(FALSE, TRUE, sizeof(struct iovec), UIO_MAXIOV);
+
+	/* stop if chunkqueue length gets less or equal than min_cq_len */
+	min_cq_len = cq->length - max_write;
+	if (min_cq_len < 0) min_cq_len = 0;
+
+	do {
+		if (0 == cq->length) return NETWORK_STATUS_SUCCESS;
+
+		ci = chunkqueue_iter(cq);
+
+		if (MEM_CHUNK != (c = chunkiter_chunk(ci))->type) {
+			NETWORK_FALLBACK(network_backend_write);
+			did_write_something = TRUE;
+			continue;
+		}
+
+		max_chunks_len = cq->length - min_cq_len;
+		we_have = 0;
+		do {
+			guint i = chunks->len;
+			off_t len = c->mem->len - c->offset;
+			struct iovec *v;
+			g_array_set_size(chunks, i + 1);
+			v = &g_array_index(chunks, struct iovec, i);
+			v->iov_base = c->mem->str + c->offset;
+			if (len > max_write) len = max_write;
+			v->iov_len = len;
+			we_have += len;
+		} while (we_have < max_chunks_len &&
+		         chunkiter_next(&ci) &&
+		         MEM_CHUNK == (c = chunkiter_chunk(ci))->type &&
+		         chunks->len < UIO_MAXIOV);
+
+		while (-1 == (r = writev(fd, (struct iovec*) chunks->data, chunks->len))) {
+			switch (errno) {
+			case EAGAIN:
+#if EWOULDBLOCK != EAGAIN
+			case EWOULDBLOCK
+#endif
+				g_array_free(chunks, TRUE);
+				return did_write_something ? NETWORK_STATUS_SUCCESS : NETWORK_STATUS_WAIT_FOR_EVENT;
+			case ECONNRESET:
+			case EPIPE:
+				g_array_free(chunks, TRUE);
+				return NETWORK_STATUS_CONNECTION_CLOSE;
+			case EINTR:
+				break; /* try again */
+			default:
+				g_array_free(chunks, TRUE);
+				CON_ERROR(srv, con, "oops, write to fd=%d failed: %s (%d)", fd, strerror(errno), errno );
+				return NETWORK_STATUS_FATAL_ERROR;
+			}
+		}
+		if (0 == r) {
+			g_array_free(chunks, TRUE);
+			return did_write_something ? NETWORK_STATUS_SUCCESS : NETWORK_STATUS_WAIT_FOR_EVENT;
+		}
+		chunkqueue_skip(cq, r);
+		if (r != we_have) {
+			g_array_free(chunks, TRUE);
+			return NETWORK_STATUS_SUCCESS;
+		}
+		did_write_something = TRUE;
+		g_array_set_size(chunks, 0);
+	} while (cq->length > min_cq_len);
+
+	g_array_free(chunks, TRUE);
+	return NETWORK_STATUS_SUCCESS;
+}

src/wscript

@@ -19,6 +19,7 @@ common_source='''
 	http_request_parser.rl
 	log.c
 	network.c
+	network_write.c network_writev.c
 	options.c
 	plugin.c
 	request.c