francesco/cws
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

move to nginx-like worker eventloop

Browse Source
This commit is contained in:
Francesco
2025-08-04 20:52:54 +02:00
parent 11bb7070da
commit a250346f22
8 changed files with 278 additions and 455 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
include/server/server.h
View File

@@ -4,7 +4,6 @@
#include <netdb.h>
#include <signal.h>
#include <sys/socket.h>
#include <time.h>
#include "myclib/hashmap/myhashmap.h"
#include "utils/config.h"
@@ -44,70 +43,13 @@ typedef enum cws_server_ret_t {
CWS_SERVER_REQUEST_TOO_LARGE,
CWS_SERVER_THREADPOOL_ERROR,
CWS_SERVER_EPOLL_CREATE_ERROR,
CWS_SERVER_WORKER_ERROR,
} cws_server_ret;
typedef struct cws_client_info_t {
time_t last_activity;
int client_fd;
} cws_client_info;
/**
* @brief Runs the server
*
* @param[in] config The server's config
*/
cws_server_ret cws_server_start(cws_config *config);
/**
* @brief Main server loop
*
* @param[in,out] sockfd Socket of the commincation endpoint
*/
cws_server_ret cws_server_loop(int server_fd, cws_config *config);
/**
* @brief Adds a file descriptor to the interest list
*
* @param[in] epfd epoll file descriptor
* @param[in] sockfd The file descriptor to watch
* @param[in] events The events to follow
*/
cws_server_ret cws_epoll_add(int epfd, int sockfd, uint32_t events);
/**
* @brief Removes a file descriptor from the interest list
*
* @param[in] epfd epoll file descriptor
* @param[in] sockfd The file descriptor to remove
*/
cws_server_ret cws_epoll_del(int epfd, int sockfd);
/**
* @brief Makes a file descriptor non-blocking
*
* @param[in] sockfd The file descriptor to make non-blocking
*/
int cws_server_handle_new_client(int server_fd, mcl_hashmap *clients);
int cws_server_accept_client(int server_fd, struct sockaddr_storage *their_sa, socklen_t *theirsa_size);
cws_server_ret cws_fd_set_nonblocking(int sockfd);
/**
* @brief Handles the new client
*
* @param[in] sockfd Server's file descriptor
* @param[out] their_sa Populates the struct with client's information
* @param[in] theirsa_size Size of the struct
*/
int cws_server_accept_client(int server_fd, struct sockaddr_storage *their_sa, socklen_t *theirsa_size);
/**
* @brief Disconnect a client
*
* @param[in] epfd Epoll file descriptor
* @param[in] client_fd Client file descriptor
* @param[in] clients Clients hash map
*/
void cws_server_close_client(int epfd, int client_fd, mcl_hashmap *clients);
int cws_server_handle_new_client(int server_fd, int epfd, mcl_hashmap *clients);
void *cws_server_handle_client_data(void *arg);
#endif

37
include/server/threadpool.h
View File

@@ -1,37 +0,0 @@
#ifndef CWS_THREADPOOL_H
#define CWS_THREADPOOL_H
#include "myclib/queue/myqueue.h"
#include "server.h"
#include "utils/config.h"
#define CWS_MAX_QUEUE_TASKS 16
#define CWS_MAX_THREADS 16
typedef struct cws_task_t {
int client_fd;
int epfd;
mcl_hashmap *clients;
cws_config *config;
} cws_task;
typedef struct cws_thread_task_t {
void (*function)(void *);
void *arg;
} cws_thread_task;
typedef struct cws_threadpool_t {
mcl_queue *queue;
pthread_mutex_t lock;
pthread_cond_t notify;
size_t threads_num;
pthread_t *threads;
int shutdown;
} cws_threadpool;
cws_threadpool *cws_threadpool_init(size_t threads_num, size_t queue_size);
cws_server_ret cws_threadpool_submit(cws_threadpool *pool, cws_thread_task *task);
void *cws_threadpool_worker(void *arg);
void cws_threadpool_destroy(cws_threadpool *pool);
#endif

27
include/server/worker.h Normal file
View File

@@ -0,0 +1,27 @@
#ifndef CWS_WORKER_H
#define CWS_WORKER_H
#include <pthread.h>
#include "myclib/hashmap/myhashmap.h"
#include "server/server.h"
typedef struct cws_worker_t {
int epfd;
int pipefd[2];
size_t clients_num;
pthread_t thread;
mcl_hashmap *clients;
cws_config *config;
} cws_worker;
cws_worker **cws_worker_init(size_t workers_num, mcl_hashmap *clients, cws_config *config);
void cws_worker_free(cws_worker **workers, size_t workers_num);
void *cws_worker_loop(void *arg);
void cws_server_close_client(int epfd, int client_fd, mcl_hashmap *clients);
cws_server_ret cws_epoll_add(int epfd, int sockfd, uint32_t events);
cws_server_ret cws_epoll_del(int epfd, int sockfd);
cws_server_ret cws_server_handle_client_data(int epfd, int client_fd, mcl_hashmap *clients, cws_config *config);
#endif

11
src/http/http.c
View File

@@ -208,7 +208,7 @@ void cws_http_send_response(cws_http *request, cws_http_status status) {
}
int cws_http_send_resource(cws_http *request) {
int keepalive = 0;
int keepalive = 1;
FILE *file = fopen(mcl_string_cstr(request->location_path), "rb");
if (file == NULL) {
@@ -249,12 +249,12 @@ int cws_http_send_resource(cws_http *request) {
return 0;
}
/* Check for keep-alive connection */
char conn[32] = "close";
/* Check for keep-alive */
char conn[32] = "keep-alive";
mcl_bucket *connection = mcl_hm_get(request->headers, "Connection");
if (connection && strcmp((char *)connection->value, "keep-alive") == 0) {
strcpy(conn, "keep-alive");
keepalive = 1;
keepalive = 0;
}
mcl_hm_free_bucket(connection);
@@ -324,12 +324,13 @@ void cws_http_send_simple_html(cws_http *request, cws_http_status status, char *
title, description);
size_t body_len = strlen(body);
char conn[32] = "close";
char conn[32] = "keep-alive";
mcl_bucket *connection = mcl_hm_get(request->headers, "Connection");
if (connection) {
strncpy(conn, (char *)connection->value, sizeof(conn) - 1);
conn[sizeof(conn) - 1] = '\0';
}
mcl_hm_free_bucket(connection);
char *response = NULL;
size_t response_len = cws_http_response_builder(&response, "HTTP/1.1", status, "text/html", conn, body, body_len);

2
src/meson.build
View File

@@ -1,3 +1,3 @@
server = files('main.c', 'server/server.c', 'server/threadpool.c')
server = files('main.c', 'server/server.c', 'server/worker.c')
server += files('utils/utils.c', 'utils/config.c')
server += files('http/http.c')

250
src/server/server.c
View File

@@ -5,14 +5,12 @@
#include <fcntl.h>
#include <netinet/in.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/types.h>
#include <unistd.h>
#include "http/http.h"
#include "server/threadpool.h"
#include "server/worker.h"
#include "utils/colors.h"
#include "utils/utils.h"
@@ -83,130 +81,41 @@ cws_server_ret cws_server_start(cws_config *config) {
return CWS_SERVER_OK;
}
static cws_server_ret cws_server_setup_epoll(int sockfd, int *epfd_out) {
int epfd = epoll_create1(0);
if (epfd == -1) {
CWS_LOG_ERROR("epoll_create(): %s", strerror(errno));
return CWS_SERVER_EPOLL_CREATE_ERROR;
}
cws_server_ret ret;
ret = cws_fd_set_nonblocking(sockfd);
if (ret != CWS_SERVER_OK) {
close(epfd);
return ret;
}
ret = cws_epoll_add(epfd, sockfd, EPOLLIN | EPOLLET);
if (ret != CWS_SERVER_OK) {
close(epfd);
return ret;
}
*epfd_out = epfd;
return CWS_SERVER_OK;
}
cws_server_ret cws_server_loop(int server_fd, cws_config *config) {
cws_server_ret ret;
int epfd;
ret = cws_server_setup_epoll(server_fd, &epfd);
if (ret != CWS_SERVER_OK) {
return ret;
}
mcl_hashmap *clients = mcl_hm_init(my_int_hash_fn, my_int_equal_fn, my_int_free_key_fn, my_str_free_fn, sizeof(int), sizeof(char) * INET_ADDRSTRLEN);
if (!clients) {
if (clients == NULL) {
return CWS_SERVER_HASHMAP_INIT;
}
struct epoll_event *revents = malloc(CWS_SERVER_EPOLL_MAXEVENTS * sizeof(struct epoll_event));
if (!revents) {
size_t workers_num = 4;
size_t workers_index = 0;
cws_worker **workers = cws_worker_init(workers_num, clients, config);
if (workers == NULL) {
mcl_hm_free(clients);
close(epfd);
return CWS_SERVER_MALLOC_ERROR;
}
cws_threadpool *pool = cws_threadpool_init(4, 16);
if (pool == NULL) {
mcl_hm_free(clients);
close(epfd);
return CWS_SERVER_THREADPOOL_ERROR;
return CWS_SERVER_WORKER_ERROR;
}
int client_fd = 0;
while (cws_server_run) {
int nfds = epoll_wait(epfd, revents, CWS_SERVER_EPOLL_MAXEVENTS, CWS_SERVER_EPOLL_TIMEOUT);
if (nfds == 0) {
/* TODO: Check for inactive clients */
continue;
}
for (int i = 0; i < nfds; ++i) {
if (revents[i].data.fd == server_fd) {
/* Handle new client */
int client_fd = cws_server_handle_new_client(server_fd, epfd, clients);
client_fd = cws_server_handle_new_client(server_fd, clients);
if (client_fd < 0) {
continue;
}
cws_epoll_add(epfd, client_fd, EPOLLIN | EPOLLET);
cws_fd_set_nonblocking(client_fd);
mcl_hm_set(clients, &client_fd, "test");
} else {
/* Handle client data */
int client_fd = revents[i].data.fd;
cws_task *task = malloc(sizeof(cws_task));
if (!task) {
cws_server_close_client(epfd, client_fd, clients);
continue;
/* Add client to worker */
int random = 10;
mcl_hm_set(clients, &client_fd, &random);
write(workers[workers_index]->pipefd[1], &client_fd, sizeof(int));
workers_index = (workers_index + 1) % workers_num;
}
task->client_fd = client_fd;
task->config = config;
task->epfd = epfd;
task->clients = clients;
cws_thread_task *ttask = malloc(sizeof(cws_thread_task));
if (!ttask) {
free(task);
continue;
}
ttask->arg = task;
ttask->function = (void *)cws_server_handle_client_data;
ret = cws_threadpool_submit(pool, ttask);
if (ret != CWS_SERVER_OK) {
cws_server_close_client(epfd, client_fd, clients);
free(task);
}
free(ttask);
}
}
}
/* Clean up everything */
free(revents);
close(epfd);
cws_worker_free(workers, workers_num);
mcl_hm_free(clients);
cws_threadpool_destroy(pool);
return CWS_SERVER_OK;
}
int cws_server_handle_new_client(int server_fd, int epfd, mcl_hashmap *clients) {
int cws_server_handle_new_client(int server_fd, mcl_hashmap *clients) {
struct sockaddr_storage their_sa;
socklen_t theirsa_size = sizeof their_sa;
char ip[INET_ADDRSTRLEN];
@@ -222,106 +131,16 @@ int cws_server_handle_new_client(int server_fd, int epfd, mcl_hashmap *clients)
return client_fd;
}
static size_t cws_read_data(int sockfd, mcl_string **str) {
size_t total_bytes = 0;
ssize_t bytes_read;
int cws_server_accept_client(int server_fd, struct sockaddr_storage *their_sa, socklen_t *theirsa_size) {
const int client_fd = accept(server_fd, (struct sockaddr *)their_sa, theirsa_size);
if (*str == NULL) {
*str = mcl_string_new("", 4096);
if (client_fd == -1) {
if (errno != EWOULDBLOCK) {
CWS_LOG_ERROR("accept(): %s", strerror(errno));
}
}
char tmp[4096];
memset(tmp, 0, sizeof(tmp));
while (1) {
bytes_read = recv(sockfd, tmp, sizeof(tmp), 0);
if (bytes_read == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
break;
}
CWS_LOG_ERROR("recv(): %s", strerror(errno));
return -1;
} else if (bytes_read == 0) {
return -1;
}
total_bytes += bytes_read;
mcl_string_append(*str, tmp);
}
return total_bytes;
}
void *cws_server_handle_client_data(void *arg) {
cws_task *task = (cws_task *)arg;
int client_fd = task->client_fd;
int epfd = task->epfd;
mcl_hashmap *clients = task->clients;
cws_config *config = task->config;
/* Read data from socket */
mcl_string *data = NULL;
size_t total_bytes = cws_read_data(client_fd, &data);
if (total_bytes <= 0) {
if (data) {
mcl_string_free(data);
}
cws_server_close_client(epfd, client_fd, clients);
free(task);
return NULL;
}
/* Parse HTTP request */
cws_http *request = cws_http_parse(data, client_fd, config);
mcl_string_free(data);
if (request == NULL) {
cws_server_close_client(epfd, client_fd, clients);
free(task);
return NULL;
}
int keepalive = cws_http_send_resource(request);
cws_http_free(request);
if (!keepalive) {
CWS_LOG_DEBUG("keep-alive: %d, closing fd: %d", keepalive, client_fd);
cws_server_close_client(epfd, client_fd, clients);
}
free(task);
return NULL;
}
cws_server_ret cws_epoll_add(int epfd, int sockfd, uint32_t events) {
struct epoll_event event;
event.events = events;
event.data.fd = sockfd;
const int status = epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &event);
if (status != 0) {
CWS_LOG_ERROR("epoll_ctl_add(): %s", strerror(errno));
return CWS_SERVER_EPOLL_ADD_ERROR;
}
return CWS_SERVER_OK;
}
cws_server_ret cws_epoll_del(int epfd, int sockfd) {
const int status = epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL);
if (status != 0) {
CWS_LOG_ERROR("epoll_ctl_del(): %s", strerror(errno));
return CWS_SERVER_EPOLL_DEL_ERROR;
}
return CWS_SERVER_OK;
return client_fd;
}
cws_server_ret cws_fd_set_nonblocking(int sockfd) {
@@ -334,26 +153,3 @@ cws_server_ret cws_fd_set_nonblocking(int sockfd) {
return CWS_SERVER_OK;
}
int cws_server_accept_client(int server_fd, struct sockaddr_storage *their_sa, socklen_t *theirsa_size) {
const int client_fd = accept(server_fd, (struct sockaddr *)their_sa, theirsa_size);
if (client_fd == -1) {
if (errno != EWOULDBLOCK) {
CWS_LOG_ERROR("accept(): %s", strerror(errno));
}
return -1;
}
return client_fd;
}
void cws_server_close_client(int epfd, int client_fd, mcl_hashmap *clients) {
/* TODO: fix race conditions */
mcl_bucket *client = mcl_hm_get(clients, &client_fd);
if (client) {
cws_epoll_del(epfd, client_fd);
mcl_hm_remove(clients, &client_fd);
mcl_hm_free_bucket(client);
}
}

124
src/server/threadpool.c
View File

@@ -1,124 +0,0 @@
#include "server/threadpool.h"
#include <stdlib.h>
#include <string.h>
cws_threadpool *cws_threadpool_init(size_t threads_num, size_t queue_size) {
/* Allocate threadpool */
cws_threadpool *pool = malloc(sizeof(cws_threadpool));
if (pool == NULL) {
return NULL;
}
memset(pool, 0, sizeof(cws_threadpool));
int ret;
ret = pthread_mutex_init(&pool->lock, NULL);
if (ret != 0) {
free(pool);
return NULL;
}
ret = pthread_cond_init(&pool->notify, NULL);
if (ret != 0) {
pthread_mutex_destroy(&pool->lock);
free(pool);
return NULL;
}
pool->queue = mcl_queue_init(queue_size, sizeof(cws_thread_task));
if (pool->queue == NULL) {
pthread_mutex_destroy(&pool->lock);
pthread_cond_destroy(&pool->notify);
free(pool);
return NULL;
}
pool->threads_num = threads_num;
pool->threads = malloc(sizeof(pthread_t) * threads_num);
if (pool->threads == NULL) {
mcl_queue_free(pool->queue);
pthread_cond_destroy(&pool->notify);
pthread_mutex_destroy(&pool->lock);
free(pool);
return NULL;
}
for (size_t i = 0; i < threads_num; ++i) {
pthread_create(&pool->threads[i], NULL, cws_threadpool_worker, pool);
}
return pool;
}
cws_server_ret cws_threadpool_submit(cws_threadpool *pool, cws_thread_task *task) {
cws_thread_task tt = {
.function = task->function,
.arg = task->arg,
};
pthread_mutex_lock(&pool->lock);
int res = mcl_queue_push(pool->queue, &tt);
if (res == 0) {
pthread_cond_signal(&pool->notify);
}
pthread_mutex_unlock(&pool->lock);
return res;
}
void *cws_threadpool_worker(void *arg) {
cws_threadpool *pool = (cws_threadpool *)arg;
cws_thread_task task = {
.arg = NULL,
.function = NULL,
};
int ret;
while (1) {
/* Loop until notify */
ret = pthread_mutex_lock(&pool->lock);
if (ret != 0) {
continue;
}
while (!pool->shutdown && pool->queue->size == 0) {
pthread_cond_wait(&pool->notify, &pool->lock);
}
if (pool->shutdown && pool->queue->size == 0) {
pthread_mutex_unlock(&pool->lock);
break;
}
mcl_queue_pop(pool->queue, &task);
pthread_mutex_unlock(&pool->lock);
task.function(task.arg);
}
return NULL;
}
void cws_threadpool_destroy(cws_threadpool *pool) {
pthread_mutex_lock(&pool->lock);
pool->shutdown = 1;
pthread_cond_broadcast(&pool->notify);
pthread_mutex_unlock(&pool->lock);
for (size_t i = 0; i < pool->threads_num; ++i) {
pthread_join(pool->threads[i], NULL);
}
pthread_mutex_destroy(&pool->lock);
pthread_cond_destroy(&pool->notify);
mcl_queue_free(pool->queue);
free(pool->threads);
free(pool);
}

218
src/server/worker.c Normal file
View File

@@ -0,0 +1,218 @@
#include "server/worker.h"
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <unistd.h>
#include "http/http.h"
#include "utils/colors.h"
static int cws_worker_setup_epoll(cws_worker *worker) {
worker->epfd = epoll_create1(0);
if (worker->epfd == -1) {
return -1;
}
cws_server_ret ret;
ret = cws_fd_set_nonblocking(worker->pipefd[0]);
if (ret != CWS_SERVER_OK) {
close(worker->epfd);
return -1;
}
ret = cws_epoll_add(worker->epfd, worker->pipefd[0], EPOLLIN);
if (ret != CWS_SERVER_OK) {
close(worker->epfd);
close(worker->pipefd[0]);
}
return 0;
}
cws_worker **cws_worker_init(size_t workers_num, mcl_hashmap *clients, cws_config *config) {
cws_worker **workers = malloc(sizeof(cws_worker) * workers_num);
if (workers == NULL) {
return NULL;
}
memset(workers, 0, sizeof(cws_worker) * workers_num);
for (size_t i = 0; i < workers_num; ++i) {
workers[i] = malloc(sizeof(cws_worker));
if (workers[i] == NULL) {
for (size_t j = 0; j < i; ++j) {
free(workers[j]);
return NULL;
}
free(workers);
}
memset(workers[i], 0, sizeof(cws_worker));
workers[i]->config = config;
/* Communicate though threads */
pipe(workers[i]->pipefd);
int ret = cws_worker_setup_epoll(workers[i]);
if (ret == -1) {
for (size_t j = 0; j < i; ++j) {
free(workers[j]);
return NULL;
}
free(workers);
}
}
for (size_t i = 0; i < workers_num; ++i) {
pthread_create(&workers[i]->thread, NULL, cws_worker_loop, workers[i]);
}
return workers;
}
void cws_worker_free(cws_worker **workers, size_t workers_num) {
for (size_t i = 0; i < workers_num; ++i) {
pthread_join(workers[i]->thread, NULL);
free(workers[i]);
}
free(workers);
}
void *cws_worker_loop(void *arg) {
cws_worker *worker = arg;
struct epoll_event events[32];
int nfds;
while (cws_server_run) {
nfds = epoll_wait(worker->epfd, events, 32, 1000);
if (nfds == 0) {
continue;
}
for (size_t i = 0; i < nfds; ++i) {
if (events[i].data.fd == worker->pipefd[0]) {
/* Handle new client */
int client_fd;
read(worker->pipefd[0], &client_fd, sizeof(int));
CWS_LOG_DEBUG("Data from main, add client: %d", client_fd);
cws_fd_set_nonblocking(client_fd);
cws_epoll_add(worker->epfd, client_fd, EPOLLIN | EPOLLET);
} else {
/* Handle client data */
int client_fd = events[i].data.fd;
CWS_LOG_DEBUG("Data from client (thread: %ld)", worker->thread);
cws_server_handle_client_data(worker->epfd, client_fd, worker->clients, worker->config);
}
}
}
return NULL;
}
void cws_server_close_client(int epfd, int client_fd, mcl_hashmap *clients) {
mcl_bucket *client = mcl_hm_get(clients, &client_fd);
if (client) {
cws_epoll_del(epfd, client_fd);
mcl_hm_remove(clients, &client_fd);
mcl_hm_free_bucket(client);
}
}
cws_server_ret cws_epoll_add(int epfd, int sockfd, uint32_t events) {
struct epoll_event event;
event.events = events;
event.data.fd = sockfd;
const int status = epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &event);
if (status != 0) {
CWS_LOG_ERROR("epoll_ctl_add(): %s", strerror(errno));
return CWS_SERVER_EPOLL_ADD_ERROR;
}
return CWS_SERVER_OK;
}
cws_server_ret cws_epoll_del(int epfd, int sockfd) {
const int status = epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL);
if (status != 0) {
CWS_LOG_ERROR("epoll_ctl_del(): %s", strerror(errno));
return CWS_SERVER_EPOLL_DEL_ERROR;
}
return CWS_SERVER_OK;
}
static size_t cws_read_data(int sockfd, mcl_string **str) {
size_t total_bytes = 0;
ssize_t bytes_read;
if (*str == NULL) {
*str = mcl_string_new("", 4096);
}
char tmp[4096];
memset(tmp, 0, sizeof(tmp));
while (1) {
bytes_read = recv(sockfd, tmp, sizeof(tmp), 0);
if (bytes_read == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
break;
}
CWS_LOG_ERROR("recv(): %s", strerror(errno));
return -1;
} else if (bytes_read == 0) {
return -1;
}
total_bytes += bytes_read;
mcl_string_append(*str, tmp);
}
return total_bytes;
}
cws_server_ret cws_server_handle_client_data(int epfd, int client_fd, mcl_hashmap *clients, cws_config *config) {
/* Read data from socket */
mcl_string *data = NULL;
size_t total_bytes = cws_read_data(client_fd, &data);
if (total_bytes <= 0) {
if (data) {
mcl_string_free(data);
}
cws_server_close_client(epfd, client_fd, clients);
return CWS_SERVER_CLIENT_DISCONNECTED_ERROR;
}
/* Parse HTTP request */
cws_http *request = cws_http_parse(data, client_fd, config);
mcl_string_free(data);
if (request == NULL) {
cws_server_close_client(epfd, client_fd, clients);
return CWS_SERVER_HTTP_PARSE_ERROR;
}
int keepalive = cws_http_send_resource(request);
cws_http_free(request);
if (!keepalive) {
cws_server_close_client(epfd, client_fd, clients);
}
return CWS_SERVER_OK;
}