server.c: a basic multithreaded server to handle connectionsHEADmain

Signed-off-by: Anna (navi) Figueiredo Gomes <navi@vlhl.dev>

5 files changed, 167 insertions, 80 deletions

diff --git a/src/buffer.c b/src/buffer.c
index 0759f5d..011d1b4 100644
--- a/src/buffer.c
+++ b/src/buffer.c
@@ -59,7 +59,7 @@ void buf_append(struct buffer *buf, const char *str) {
 
 void buf_printf(struct buffer *buf, char *fmt, ...) {
 	va_list ap;
-	size_t len;
+	ssize_t len;
 
 	va_start(ap, fmt);
 	len = vsnprintf(buf->data, buf->size, fmt, ap);
diff --git a/src/http.c b/src/http.c
index 793f257..99204ec 100644
--- a/src/http.c
+++ b/src/http.c
@@ -97,10 +97,7 @@ static enum parse_result parse_request(struct request *req, const char *request)
 
 	line = strtok_r(copy, "\r\n", &saveptr);
 
-	path = malloc(strlen(line) + 1);
-	req->method = malloc(strlen(line) + 1);
-
-	if (sscanf(line, "%s %s HTTP/%s", req->method, path, req->http_version) != 3) {
+	if (sscanf(line, "%ms %ms HTTP/%s", &req->method, &path, req->http_version) != 3) {
 		free(copy);
 		free(path);
 		free(req->method);
diff --git a/src/main.c b/src/main.c
index 09cf045..8429797 100644
--- a/src/main.c
+++ b/src/main.c
@@ -1,7 +1,5 @@
 #include <stdio.h>
 #include <cjson/cJSON.h>
-#include <sys/socket.h>
-#include <netinet/in.h>
 #include <unistd.h>
 #include <stdlib.h>
 #include <stdbool.h>
@@ -9,9 +7,11 @@
 
 #include "buffer.h"
 #include "http.h"
+#include "server.h"
 
 static struct reply *handle(struct request *req, void *data) {
 	(void)data;
+
 	struct reply *ret = malloc(sizeof(struct reply));
 	cJSON *obj = cJSON_CreateObject();
 	cJSON_AddStringToObject(obj, "path", req->path);
@@ -33,43 +33,6 @@ static struct reply *handle(struct request *req, void *data) {
 int main(int argc, char **argv) {
 	(void)argc;
 	(void)argv;
-	int server_fd = 0;
-	int client_fd = 0;
-	struct sockaddr_in addr = {
-		.sin_family = AF_INET,
-		.sin_addr.s_addr = INADDR_ANY,
-		.sin_port = htons(45748),
-	};
-
-	int addrlen = sizeof(addr);
-
-	if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
-		perror("socket failed");
-		return EXIT_FAILURE;
-	}
-
-	int opt = 1;
-
-	if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
-		perror("setsockopt failed");
-		return EXIT_FAILURE;
-	}
-
-	if (bind(server_fd, (struct sockaddr*)&addr, addrlen) < 0) {
-		perror("bind failed");
-		return EXIT_FAILURE;
-	}
-
-	if (listen(server_fd, 16) < 0) {
-		perror("listen failed");
-		return EXIT_FAILURE;
-	}
-
-	bool running = true;
-	ssize_t rlen;
-	char buf[1025];
-	struct buffer *request, *reply;
-	struct reply *rpy;
 
 	struct http *http = http_init();
 
@@ -77,40 +40,7 @@ int main(int argc, char **argv) {
 	http_register_handler(http, "GET", "/nyaa", NULL, &handle);
 	http_register_handler(http, "POST", "/nyaa", NULL, &handle);
 
-	while (running) {
-		if ((client_fd = accept(server_fd,
-						(struct sockaddr*)&addr, (socklen_t*)&addrlen)) < 0) {
-			perror("accept failed");
-			return EXIT_FAILURE;
-		}
-		
-		if ((rlen = read(client_fd, buf, 1024)) < 0) {
-			perror("read failed");
-			return EXIT_FAILURE;
-		}
-
-		buf[rlen] = '\0';
-		request = buf_new(buf);
-
-		while (rlen >= 1024) {
-			if ((rlen = read(client_fd, buf, 1024)) < 0) {
-				perror("read failed");
-				return EXIT_FAILURE;
-			}
-			buf[rlen] = '\0';
-			buf_append(request, buf);
-		}
-
-		printf("%s\n", request->data);
+	struct server *server = server_init(http);
 
-		rpy = http_handle_request(http, request);
-		reply = http_build_reply(rpy);
-		free(rpy);
-
-		write(client_fd, reply->data, reply->size);
-
-		buf_del(&request);
-
-		close(client_fd);
-	}
+	server_poll(server);
 }
diff --git a/src/server.c b/src/server.c
new file mode 100644
index 0000000..66c1ac5
--- /dev/null
+++ b/src/server.c
@@ -0,0 +1,160 @@
+#include <assert.h>
+#include <stdbool.h>
+#include <pthread.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+
+#include "server.h"
+#include "buffer.h"
+#include "http.h"
+
+static bool enqueue(struct fd_queue *queue, int fd) {
+	if (queue->count == MAX_FD_QUEUE) {
+		return false;
+	}
+
+	queue->queue[queue->back] = fd;
+	queue->count++;
+	queue->back = (queue->back + 1) % MAX_FD_QUEUE;
+
+	return true;
+}
+
+static int dequeue(struct fd_queue *queue) {
+	if (queue->count == 0) {
+		return -1;
+	}
+
+	int fd = queue->queue[queue->front];
+	queue->count--;
+	queue->front = (queue->front + 1) % MAX_FD_QUEUE;
+
+	return fd;
+}
+
+static void *handle_connection(void *data) {
+	assert(data);
+	struct thread_data *tdata = (struct thread_data*)data;
+	int fd = -1;
+	char buf[BUFSIZ + 1];
+	ssize_t rlen;
+	struct buffer *request_buf, *reply_buf;
+	struct reply *reply;
+
+	while(true) {
+		pthread_mutex_lock(&tdata->lock);
+		fd = dequeue(&tdata->queue);
+		if (fd == -1) {
+			pthread_cond_wait(&tdata->cond, &tdata->lock);
+			fd = dequeue(&tdata->queue);
+		}
+		pthread_mutex_unlock(&tdata->lock);
+
+		printf("got fd %d\n", fd);
+
+		if ((rlen = read(fd, buf, BUFSIZ)) < 0) {
+			perror("read failed");
+			close(fd);
+			continue;
+		}
+
+		buf[rlen] = '\0';
+		request_buf = buf_new(buf);
+		
+		while (rlen >= BUFSIZ) {
+			if ((rlen = read(fd, buf, BUFSIZ)) < 0) {
+				perror("read failed");
+				close(fd);
+				continue;
+			}
+			buf[rlen] = '\0';
+			buf_append(request_buf, buf);
+		}
+
+		reply = http_handle_request(tdata->http, request_buf);
+		reply_buf = http_build_reply(reply);
+		free(reply);
+
+		write(fd, reply_buf->data, reply_buf->size);
+
+		buf_del(&reply_buf);
+		buf_del(&request_buf);
+
+		close(fd);
+	}
+}
+
+struct server *server_init(struct http *http) {
+	struct server *server = malloc(sizeof(struct server));
+
+	server->server_fd = socket(AF_INET, SOCK_STREAM, 0);
+	if (server->server_fd < 0) {
+		perror("socket failure");
+		return NULL;
+	}
+
+	struct sockaddr_in addr = {
+		.sin_family = AF_INET,
+		.sin_addr.s_addr = INADDR_ANY,
+		.sin_port = htons(45748),
+	};
+
+	int addrlen = sizeof(addr);
+	int opt = 1;
+
+	if (setsockopt(server->server_fd, SOL_SOCKET, SO_REUSEADDR,
+				&opt, sizeof(opt)) < 0) {
+		perror("setsockopt failed");
+		return NULL;
+	}
+
+	if (bind(server->server_fd, (struct sockaddr*)&addr, addrlen) < 0) {
+		perror("bind failed");
+		return NULL;
+	}
+
+	if (listen(server->server_fd, 16) < 0) {
+		perror("listen failed");
+		return NULL;
+	}
+
+	server->addr = addr;
+	server->addrlen = addrlen;
+	server->thread_pool_size = THREAD_POOL_NUM;
+	server->data.http = http;
+	server->pool = calloc(server->thread_pool_size, sizeof(pthread_t));
+	pthread_mutex_init(&server->data.lock, NULL);
+	pthread_cond_init(&server->data.cond, NULL);
+	server->data.queue.back = -1;
+	server->data.queue.front = -1;
+	server->data.queue.count = 0;
+
+	for (size_t i = 0; i < server->thread_pool_size; i++) {
+		pthread_create(&server->pool[i], NULL, handle_connection, &server->data);
+	}
+
+	return server;
+}
+
+void server_poll(struct server *server) {
+	int client_fd;
+
+	while (true) {
+		if ((client_fd = accept(server->server_fd, (struct sockaddr*)&server->addr,
+						(socklen_t*)&server->addrlen)) < 0) {
+			perror("accept failed");
+			return;
+		}
+
+		printf("connection accepted on fd %d\n", client_fd);
+
+		pthread_mutex_lock(&server->data.lock);
+		enqueue(&server->data.queue, client_fd);
+		pthread_cond_signal(&server->data.cond);
+		pthread_mutex_unlock(&server->data.lock);
+
+	}
+}
diff --git a/src/util.c b/src/util.c
index e6b7977..616fd8a 100644
--- a/src/util.c
+++ b/src/util.c
@@ -8,7 +8,7 @@
 
 int vasprintf(char **dest, const char *fmt, va_list ap) {
 	assert(*dest == NULL);
-	char *buf = NULL; size_t len, size;
+	char *buf = NULL; ssize_t len, size;
 	va_list ap2;
 
 	size = 4096;