use linux aio io-submit

2 files changed, 135 insertions, 60 deletions

diff --git a/Makefile b/Makefile
index 4203428..b1946df 100644
--- a/Makefile
+++ b/Makefile
@@ -1,5 +1,6 @@
 ALL=hittpd
 OBJ=hittpd.o http-parser/http_parser.o
+LDLIBS=-laio
 
 CFLAGS=-g -O2 -Wall -Wno-switch -Wextra -Wwrite-strings
 CPPFLAGS=-Ihttp-parser
diff --git a/hittpd.c b/hittpd.c
index bbc94ac..bc4128d 100644
--- a/hittpd.c
+++ b/hittpd.c
@@ -26,6 +26,7 @@
 
 #ifdef __linux__
 #include <sys/sendfile.h>
+#include <libaio.h>
 #endif
 #include <sys/socket.h>
 #include <sys/stat.h>
@@ -719,51 +720,71 @@ static http_parser_settings settings = {
 	.on_url = on_url,
 };
 
-struct pollfd client[MAX_CLIENTS];
+struct iocb client[MAX_CLIENTS];
+struct io_event events[MAX_CLIENTS];
+io_context_t ctx = 0;
+
 struct http_parser parsers[MAX_CLIENTS];
 struct conn_data datas[MAX_CLIENTS];
 
 void
 close_connection(int i)
 {
-	if (client[i].fd >= 0)
-		close(client[i].fd);
-	client[i].fd = -1;
+	struct http_parser *p = events[i].data;
+	struct conn_data *data = p->data;
 
-	free(datas[i].buf);
-	free(datas[i].path);
-	free(datas[i].ims);
-	free(datas[i].host);
+	if (events[i].obj->aio_fildes >= 0) {
+		printf("close(%d)\n", events[i].obj->aio_fildes);
 
-	datas[i] = (struct conn_data){ 0 };
+		close(events[i].obj->aio_fildes);
+}
+	events[i].obj->aio_fildes = -1;
+	events[i].obj->aio_lio_opcode = IO_CMD_NOOP;
+
+	free(data->buf);
+	free(data->path);
+	free(data->ims);
+	free(data->host);
+
+	*data = (struct conn_data){ 0 };
 }
 
 void
 finish_response(int i)
 {
-	if (datas[i].stream_fd >= 0)
-		close(datas[i].stream_fd);
-	datas[i].stream_fd = -1;
+	struct http_parser *p = events[i].data;
+	struct conn_data *data = p->data;
+
+	if (data->stream_fd >= 0)
+		close(data->stream_fd);
+	data->stream_fd = -1;
 
-	free(datas[i].buf);
-	free(datas[i].path);
-	free(datas[i].ims);
-	free(datas[i].host);
+	free(data->buf);
+	free(data->path);
+	free(data->ims);
+	free(data->host);
 
-	datas[i].buf = 0;
-	datas[i].path = 0;
-	datas[i].ims = 0;
-	datas[i].host = 0;
+	data->buf = 0;
+	data->path = 0;
+	data->ims = 0;
+	data->host = 0;
 
-	client[i].events = POLLRDNORM;
+	events[i].obj->aio_lio_opcode = IO_CMD_POLL;
+	events[i].obj->u.poll.events = POLLIN;
 
-	if (parsers[i].flags & F_CONNECTION_CLOSE)
+	if (p->flags & F_CONNECTION_CLOSE) {
 		close_connection(i);
-	else if (parsers[i].flags & F_CONNECTION_KEEP_ALIVE)
+		return;
+	}
+	else if (p->flags & F_CONNECTION_KEEP_ALIVE)
 		;
-	else if ((parsers[i].http_major == 1 && parsers[i].http_minor == 0) ||
-	    parsers[i].http_major == 0)
+	else if ((p->http_major == 1 && p->http_minor == 0) ||
+	    p->http_major == 0) {
 		close_connection(i);    // HTTP 1.0 default
+		return;
+	}
+
+	io_submit(ctx, 1, &events[i].obj);
 }
 
 void
@@ -771,7 +792,7 @@ accept_client(int i, int fd)
 {
 	fcntl(fd, F_SETFL, O_NONBLOCK);
 
-	client[i].fd = fd;
+	client[i].aio_fildes = fd;
 
 	http_parser_init(&parsers[i], HTTP_REQUEST);
 	datas[i] = (struct conn_data){ 0 };
@@ -782,14 +803,19 @@ accept_client(int i, int fd)
 
 	parsers[i].data = &datas[i];
 
-	client[i].events = POLLRDNORM;
+	client[i].aio_lio_opcode = IO_CMD_POLL;
+	client[i].u.poll.events = POLLIN;
+	client[i].data = &parsers[i];
+
+	io_poll(ctx, &client[i], (io_callback_t)&parsers[i], fd, POLLIN);
 }
 
 void
 write_client(int i)
 {
-	struct conn_data *data = &datas[i];
-	int sockfd = client[i].fd;
+	struct http_parser *p = events[i].data;
+	struct conn_data *data = p->data;
+	int sockfd = events[i].obj->aio_fildes;
 	ssize_t w = 0;
 
 	if (data->stream_fd >= 0) {
@@ -814,14 +840,18 @@ write_client(int i)
 		    &(data->off), data->last - data->off);
 		if (w == 0 || data->off == data->last)
 			finish_response(i);
+		else
+			io_submit(ctx, 1, &events[i].obj);
 #endif
 	} else if (data->buf) {
 		if (data->off == data->last) {
 			finish_response(i);
 		} else {
 			w = write(sockfd, data->buf, data->last - data->off);
-			if (w > 0)
+			if (w > 0) {
 				data->off += w;
+				io_submit(ctx, 1, &events[i].obj);
+			}
 		}
 	} else {
 		finish_response(i);
@@ -838,8 +868,9 @@ write_client(int i)
 void
 read_client(int i)
 {
-	struct conn_data *data = &datas[i];
-	int sockfd = client[i].fd;
+	struct http_parser *p = events[i].data;
+	struct conn_data *data = p->data;
+	int sockfd = events[i].obj->aio_fildes;
 	ssize_t n;
 	char buf[1024];
 
@@ -855,23 +886,25 @@ read_client(int i)
 	} else if (n == 0) {
 		close_connection(i);
 	} else {
-		http_parser_execute(&parsers[i], &settings, buf, n);
+		http_parser_execute(p, &settings, buf, n);
 
-		if (parsers[i].http_errno) {
+		if (p->http_errno) {
 			printf("err=%s\n",
-			    http_errno_name(parsers[i].http_errno));
+			    http_errno_name(p->http_errno));
 			close_connection(i);
 		} else {
 			// switch to write mode when needed
 			if (data->state == SENDING) {
-				client[i].events = POLLRDNORM | POLLWRNORM;
+				events[i].obj->u.poll.events = POLLIN | POLLOUT;
 				data->off = data->first;
 
-				if (parsers[i].method == HTTP_HEAD)
+				if (p->method == HTTP_HEAD) {
 					finish_response(i);
+					return;
+				}
 			}
+			io_submit(ctx, 1, &events[i].obj);
 		}
-
 	}
 }
 
@@ -924,7 +957,7 @@ main(int argc, char *argv[])
 	sigaction(SIGINT, &act, 0);
 	sigaction(SIGTERM, &act, 0);
 
-	int i, maxi, listenfd, sockfd;
+	int i, listenfd, sockfd;
 	int nready;
 	int r = 0;
 
@@ -988,16 +1021,35 @@ main(int argc, char *argv[])
 		exit(111);
 	}
 
-	client[0].fd = listenfd;
-	client[0].events = POLLRDNORM;
+	r = io_setup(MAX_CLIENTS, &ctx);
+	if (r < 0) {
+		perror("io_setup");
+		exit(111);
+	}
+
+	//client[0].aio_fildes = listenfd;
+	//client[0].aio_lio_opcode = IO_CMD_POLL;
+	//client[0].u.poll.events = POLLIN;
 
-	for (i = 1; i < MAX_CLIENTS; i++)
-		client[i].fd = -1;  /* -1 indicates available entry */
+	io_poll(ctx, &client[0], 0, listenfd, POLLIN);
 
-	maxi = 0; /* max index into client[] array */
+	for (i = 1; i < MAX_CLIENTS; i++) {
+		client[i].aio_fildes = -1;  /* -1 indicates available entry */
+		client[i].aio_lio_opcode = IO_CMD_NOOP;
+	}
+
+//	r = io_submit(ctx, MAX_CLIENTS, clientp);
+//?	if (r != MAX_CLIENTS) {
+	if (r < 0) {
+		perror("io_submit");
+		exit(111);
+	}
 
 	while (!stop) {
-		nready = poll(client, maxi + 1, maxi ? TIMEOUT*1000 : -1);
+//		nready = poll(client, maxi + 1, maxi ? TIMEOUT*1000 : -1);
+
+		nready = io_getevents(ctx, 1, MAX_CLIENTS, events, 0);
+		// xx TIMEOUT
 
 		if (nready < 0) {
 			if (errno == EINTR) {
@@ -1010,6 +1062,7 @@ main(int argc, char *argv[])
 
 		time_t now = time(0);
 
+#if 0
 		if (nready == 0) {
 			// clear timeouted
 			for (i = 1; i <= maxi; i++)
@@ -1043,11 +1096,13 @@ main(int argc, char *argv[])
 
 			maxi = j;
 		}
+#endif
 
+#if 0
 		if (client[0].revents & POLLRDNORM) {
 			/* new client connection */
 			for (i = 1; i < MAX_CLIENTS; i++)
-				if (client[i].fd < 0) {
+				if (client[i].aio_fildes < 0) {
 					struct sockaddr_in6 cliaddr;
 					socklen_t clilen = sizeof cliaddr;
 					int connfd = accept(listenfd,
@@ -1062,28 +1117,47 @@ main(int argc, char *argv[])
 			if (--nready <= 0)
 				continue; /* no more readable descriptors */
 		}
-		for (i = 1; i <= maxi; i++) { /* check all clients for data */
-			if ((sockfd = client[i].fd) < 0)
+#endif
+		for (i = 0; i < nready; i++) { /* check all clients for data */
+			if ((sockfd = events[i].obj->aio_fildes) < 0)
+				continue;
+
+			if (sockfd == listenfd) { // accept
+				for (i = 1; i < MAX_CLIENTS; i++)
+					if (client[i].aio_fildes < 0) {
+						struct sockaddr_in6 cliaddr;
+						socklen_t clilen = sizeof cliaddr;
+						int connfd = accept(listenfd,
+						    (struct sockaddr *)&cliaddr, &clilen);
+						accept_client(i, connfd);
+						int r = io_poll(ctx, &client[0], 0, listenfd, POLLIN);
+						printf("accept=%d reaccept=%d %s\n",
+						    connfd, r, strerror(-r));
+						break;
+					}
+				if (i == MAX_CLIENTS)
+					printf("too many clients\n");
 				continue;
+			}
+
+			if (events[i].obj->u.poll.events & POLLOUT) {
+				struct http_parser *p = events[i].data;
+				struct conn_data *data = p->data;
 
-			if (client[i].revents & POLLWRNORM) {
-				if (datas[i].state != SENDING) {
-					client[i].events = POLLRDNORM;
+				if (data->state != SENDING) {
+					printf("foo\n");
+					events[i].obj->u.poll.events = POLLRDNORM;
 					continue;
 				}
 
 				write_client(i);
-				datas[i].deadline = now + TIMEOUT;
+				data->deadline = now + TIMEOUT;
+			} else if (events[i].obj->u.poll.events & POLLIN) {
+				struct http_parser *p = events[i].data;
+				struct conn_data *data = p->data;
 
-				if (--nready <= 0)
-					break; /* no more readable descriptors */
-			}
-			else if (client[i].revents & (POLLRDNORM | POLLERR)) {
 				read_client(i);
-				datas[i].deadline = now + TIMEOUT;
-
-				if (--nready <= 0)
-					break; /* no more readable descriptors */
+				data->deadline = now + TIMEOUT;
 			}
 		}
 	}