diff --git a/include/myclib/hashmap/myhashmap.c b/include/myclib/hashmap/myhashmap.c index 4208413..ddaf5cb 100644 --- a/include/myclib/hashmap/myhashmap.c +++ b/include/myclib/hashmap/myhashmap.c @@ -1,9 +1,12 @@ #include "myhashmap.h" +#include #include #include #include +static size_t mcl_get_mutex(mcl_hashmap *hashmap, size_t hash) { return hash % hashmap->num_locks; } + static size_t mcl_get_bucket_index(mcl_hashmap *hashmap, void *key) { unsigned int hash = hashmap->hash_fn(key); return hash % MYCLIB_HASHMAP_SIZE; @@ -62,6 +65,28 @@ mcl_hashmap *mcl_hm_init(mcl_hash_fn *hash_fn, mcl_equal_fn *equal_fn, mcl_free_ hashmap->key_size = key_size; hashmap->value_size = value_size; + hashmap->num_locks = 64; + hashmap->locks = malloc(sizeof(pthread_mutex_t) * hashmap->num_locks); + if (hashmap->locks == NULL) { + free(hashmap); + + return NULL; + } + + int ret; + for (size_t i = 0; i < hashmap->num_locks; ++i) { + ret = pthread_mutex_init(&(hashmap->locks[i]), NULL); + if (ret != 0) { + /* Mutex failed */ + for (size_t j = 0; j < i; ++j) { + pthread_mutex_destroy(&(hashmap->locks[j])); + } + + free(hashmap->locks); + free(hashmap); + } + } + memset(hashmap->map, 0, sizeof(hashmap->map)); return hashmap; @@ -91,15 +116,35 @@ void mcl_hm_free(mcl_hashmap *hashmap) { } } + /* Free the mutex */ + for (size_t i = 0; i < hashmap->num_locks; ++i) { + pthread_mutex_destroy(&(hashmap->locks[i])); + } + free(hashmap->locks); + /* Free the hash map structure itself */ free(hashmap); } +void mcl_hm_free_bucket(mcl_bucket *bucket) { + if (bucket == NULL) { + return; + } + + free(bucket->key); + free(bucket->value); + free(bucket); +} + bool mcl_hm_set(mcl_hashmap *hashmap, void *key, void *value) { if (hashmap == NULL || key == NULL || value == NULL) { return false; } + size_t mutex_id = mcl_get_mutex(hashmap, hashmap->hash_fn(key)); + pthread_mutex_t *mutex = &(hashmap->locks[mutex_id]); + pthread_mutex_lock(mutex); + mcl_bucket *prev; mcl_bucket *existing = mcl_find_bucket(hashmap, key, &prev); @@ -111,9 +156,14 @@ bool mcl_hm_set(mcl_hashmap *hashmap, void *key, void *value) { existing->value = malloc(hashmap->value_size); if (existing->value == NULL) { + pthread_mutex_unlock(mutex); + return false; } + memcpy(existing->value, value, hashmap->value_size); + pthread_mutex_unlock(mutex); + return true; } @@ -125,6 +175,8 @@ bool mcl_hm_set(mcl_hashmap *hashmap, void *key, void *value) { /* First bucket is empty, use it */ bucket->key = malloc(hashmap->key_size); if (bucket->key == NULL) { + pthread_mutex_unlock(mutex); + return false; } @@ -132,24 +184,32 @@ bool mcl_hm_set(mcl_hashmap *hashmap, void *key, void *value) { if (bucket->value == NULL) { free(bucket->key); bucket->key = NULL; + pthread_mutex_unlock(mutex); + return false; } memcpy(bucket->key, key, hashmap->key_size); memcpy(bucket->value, value, hashmap->value_size); bucket->next = NULL; + pthread_mutex_unlock(mutex); + return true; } /* Create new bucket and insert at head of collision chain */ mcl_bucket *new_bucket = malloc(sizeof(mcl_bucket)); if (new_bucket == NULL) { + pthread_mutex_unlock(mutex); + return false; } new_bucket->key = malloc(hashmap->key_size); if (new_bucket->key == NULL) { free(new_bucket); + pthread_mutex_unlock(mutex); + return false; } @@ -157,6 +217,8 @@ bool mcl_hm_set(mcl_hashmap *hashmap, void *key, void *value) { if (new_bucket->value == NULL) { free(new_bucket->key); free(new_bucket); + pthread_mutex_unlock(mutex); + return false; } @@ -164,17 +226,61 @@ bool mcl_hm_set(mcl_hashmap *hashmap, void *key, void *value) { memcpy(new_bucket->value, value, hashmap->value_size); new_bucket->next = bucket->next; bucket->next = new_bucket; + pthread_mutex_unlock(mutex); return true; } +static mcl_bucket *mcl_get_bucket_copy(mcl_bucket *from, size_t key_size, size_t value_size) { + mcl_bucket *copy = malloc(sizeof(mcl_bucket)); + if (copy == NULL) { + return NULL; + } + memcpy(copy, from, sizeof(mcl_bucket)); + + copy->key = malloc(key_size); + if (copy->key == NULL) { + free(copy); + + return NULL; + } + memcpy(copy->key, from->key, key_size); + + copy->value = malloc(value_size); + if (copy->value == NULL) { + free(copy->key); + free(copy); + + return NULL; + } + memcpy(copy->value, from->value, value_size); + + return copy; +} + mcl_bucket *mcl_hm_get(mcl_hashmap *hashmap, void *key) { if (hashmap == NULL || key == NULL) { return NULL; } + size_t mutex_id = mcl_get_mutex(hashmap, hashmap->hash_fn(key)); + pthread_mutex_t *mutex = &(hashmap->locks[mutex_id]); + pthread_mutex_lock(mutex); + mcl_bucket *prev; - return mcl_find_bucket(hashmap, key, &prev); + mcl_bucket *found = mcl_find_bucket(hashmap, key, &prev); + + if (found) { + mcl_bucket *copy = mcl_get_bucket_copy(found, hashmap->key_size, hashmap->value_size); + + pthread_mutex_unlock(mutex); + + return copy; + } + + pthread_mutex_unlock(mutex); + + return NULL; } bool mcl_hm_remove(mcl_hashmap *hashmap, void *key) { @@ -182,10 +288,16 @@ bool mcl_hm_remove(mcl_hashmap *hashmap, void *key) { return false; } + size_t mutex_id = mcl_get_mutex(hashmap, hashmap->hash_fn(key)); + pthread_mutex_t *mutex = &(hashmap->locks[mutex_id]); + pthread_mutex_lock(mutex); + mcl_bucket *prev; mcl_bucket *to_remove = mcl_find_bucket(hashmap, key, &prev); if (to_remove == NULL) { + pthread_mutex_unlock(mutex); + return false; } @@ -214,5 +326,7 @@ bool mcl_hm_remove(mcl_hashmap *hashmap, void *key) { free(to_remove); } + pthread_mutex_unlock(mutex); + return true; } diff --git a/include/myclib/hashmap/myhashmap.h b/include/myclib/hashmap/myhashmap.h index b1e546f..3826603 100644 --- a/include/myclib/hashmap/myhashmap.h +++ b/include/myclib/hashmap/myhashmap.h @@ -1,6 +1,7 @@ #ifndef MYCLIB_HASHMAP_H #define MYCLIB_HASHMAP_H +#include #include #include @@ -63,6 +64,8 @@ typedef struct mcl_hashmap_t { size_t key_size; /**< Size in bytes of the key */ size_t value_size; /**< Size in bytes of the value */ mcl_bucket map[MYCLIB_HASHMAP_SIZE]; /**< Array of bucket chains */ + pthread_mutex_t *locks; /**< Mutex array */ + size_t num_locks; /**< Number of mutex */ } mcl_hashmap; /** @@ -93,6 +96,13 @@ mcl_hashmap *mcl_hm_init(mcl_hash_fn *hash_fn, mcl_equal_fn *equal_fn, mcl_free_ */ void mcl_hm_free(mcl_hashmap *hashmap); +/** + * @brief Free a bucket returned by mcl_hm_get() + * + * @param[in] bucket Pointer to the bucket to free + */ +void mcl_hm_free_bucket(mcl_bucket *bucket); + /** * @brief Insert or update a key-value pair in the hash map * @@ -115,7 +125,7 @@ bool mcl_hm_set(mcl_hashmap *hashmap, void *key, void *value); * * @param[in] hashmap Pointer to the hash map * @param[in] key Pointer to the key to search for - * @return Pointer to the found bucket, or NULL if not found or on invalid input + * @return Pointer to the copy of the bucket, to avoid race conditions, or NULL if not found or on invalid input */ mcl_bucket *mcl_hm_get(mcl_hashmap *hashmap, void *key); diff --git a/include/myclib/queue/myqueue.c b/include/myclib/queue/myqueue.c new file mode 100644 index 0000000..49fe73e --- /dev/null +++ b/include/myclib/queue/myqueue.c @@ -0,0 +1,89 @@ +#include "myqueue.h" + +#include +#include + +mcl_queue *mcl_queue_init(size_t queue_size, size_t elem_size) { + mcl_queue *queue = malloc(sizeof(mcl_queue)); + if (queue == NULL) { + return NULL; + } + + queue->buffer = malloc(queue_size * elem_size); + if (queue->buffer == NULL) { + free(queue); + + return NULL; + } + + queue->front = 0; + queue->rear = 0; + queue->size = 0; + queue->capacity = queue_size; + queue->elem_size = elem_size; + + return queue; +} + +int mcl_queue_push(mcl_queue *queue, const void *elem) { + if (queue->size == queue->capacity) { + /* Queue full */ + return -1; + } + + /* Copy the elem in the buffer */ + void *dest = (void *)queue->buffer + (queue->rear * queue->elem_size); + memcpy(dest, elem, queue->elem_size); + + queue->size++; + queue->rear = (queue->rear + 1) % queue->capacity; + + return 0; +} + +int mcl_queue_pop(mcl_queue *queue, void *out_elem) { + if (queue->size == 0) { + /* Queue empty */ + return -1; + } + + void *src = (void *)queue->buffer + (queue->front * queue->elem_size); + memcpy(out_elem, src, queue->elem_size); + + queue->front = (queue->front + 1) % queue->capacity; + queue->size--; + + return 0; +} + +void *mcl_queue_get_front(mcl_queue *queue) { + if (queue->size == 0) { + return NULL; + } + + return (void *)queue->buffer + (queue->front * queue->elem_size); +} + +void *mcl_queue_get_rear(mcl_queue *queue) { + if (queue->size == 0) { + return NULL; + } + + size_t rear_index; + if (queue->rear == 0) { + rear_index = queue->capacity - 1; + } else { + rear_index = queue->rear - 1; + } + + return (void *)queue->buffer + (rear_index * queue->elem_size); +} + +void mcl_queue_free(mcl_queue *queue) { + if (queue == NULL) { + return; + } + + free(queue->buffer); + free(queue); +} diff --git a/include/myclib/queue/myqueue.h b/include/myclib/queue/myqueue.h new file mode 100644 index 0000000..42d4bfd --- /dev/null +++ b/include/myclib/queue/myqueue.h @@ -0,0 +1,68 @@ +#ifndef MYCLIB_QUEUE_H +#define MYCLIB_QUEUE_H + +#include + +/** + * @brief A generic circular queue (ring buffer). + */ +typedef struct mcl_queue_t { + size_t front; /**< Index of the first element (read position). */ + size_t rear; /**< Index where the next element will be written (write position). */ + size_t size; /**< Current number of elements in the queue. */ + size_t capacity; /**< Maximum number of elements the queue can hold. */ + size_t elem_size; /**< Size in bytes of each element in the queue. */ + void *buffer; /**< Pointer to the memory buffer that stores the elements. */ +} mcl_queue; + +/** + * @brief Create and initialize a new queue. + * + * @param queue_size Number of elements the queue can hold. + * @param elem_size Size (in bytes) of each element in the queue. + * @return Pointer to the new queue, or NULL if allocation fails. + */ +mcl_queue *mcl_queue_init(size_t queue_size, size_t elem_size); + +/** + * @brief Add an element to the queue. + * + * @param queue Pointer to the queue. + * @param elem Pointer to the element to insert. + * @return 0 on success, -1 if the queue is full. + */ +int mcl_queue_push(mcl_queue *queue, const void *elem); + +/** + * @brief Remove an element from the queue. + * + * @param queue Pointer to the queue. + * @param out_elem Pointer where the removed element will be copied. + * @return 0 on success, -1 if the queue is empty. + */ +int mcl_queue_pop(mcl_queue *queue, void *out_elem); + +/** + * @brief Get a pointer to the front element of the queue (oldest one). + * + * @param queue Pointer to the queue. + * @return Pointer to the front element, or NULL if the queue is empty. + */ +void *mcl_queue_get_front(mcl_queue *queue); + +/** + * @brief Get a pointer to the rear element of the queue (most recently added). + * + * @param queue Pointer to the queue. + * @return Pointer to the rear element, or NULL if the queue is empty. + */ +void *mcl_queue_get_rear(mcl_queue *queue); + +/** + * @brief Free all memory used by the queue. + * + * @param queue Pointer to the queue to free. + */ +void mcl_queue_free(mcl_queue *queue); + +#endif diff --git a/include/myclib/string/mystring.c b/include/myclib/string/mystring.c index 2e979fc..fbc3400 100644 --- a/include/myclib/string/mystring.c +++ b/include/myclib/string/mystring.c @@ -1,17 +1,26 @@ #include "mystring.h" #include +#include #include #include mcl_string *mcl_string_new(const char *text, long initial_capacity) { - if (!text) { + if (text == NULL) { return NULL; } /* Allocate string struct */ mcl_string *str = malloc(sizeof(mcl_string)); - if (!str) { + if (str == NULL) { + return NULL; + } + + /* Init pthread mutex */ + int ret = pthread_mutex_init(&str->lock, NULL); + if (ret != 0) { + free(str); + return NULL; } @@ -20,6 +29,8 @@ mcl_string *mcl_string_new(const char *text, long initial_capacity) { size_t capacity = initial_capacity; if (capacity != -1 && capacity - 1 < str->size) { + free(str); + return NULL; } @@ -31,7 +42,7 @@ mcl_string *mcl_string_new(const char *text, long initial_capacity) { /* Allocate data buffer */ str->data = malloc(sizeof(char) * str->capacity); - if (!str->data) { + if (str->data == NULL) { free(str); return NULL; @@ -46,13 +57,21 @@ mcl_string *mcl_string_new(const char *text, long initial_capacity) { } int mcl_string_append(mcl_string *string, const char *text) { - if (!string || !text) { + if (string == NULL || text == NULL) { + return -1; + } + + /* Lock resource */ + int ret = pthread_mutex_lock(&string->lock); + if (ret != 0) { return -1; } /* Handle empty case */ size_t text_len = strlen(text); if (text_len == 0) { + pthread_mutex_unlock(&string->lock); + return 0; } @@ -64,6 +83,8 @@ int mcl_string_append(mcl_string *string, const char *text) { /* Reallocate the buffer */ void *new_data = realloc(string->data, sizeof(char) * new_capacity); if (!new_data) { + pthread_mutex_unlock(&string->lock); + return -1; } @@ -79,11 +100,22 @@ int mcl_string_append(mcl_string *string, const char *text) { string->size = new_size; string->data[string->size] = '\0'; + /* Unlock resource */ + ret = pthread_mutex_unlock(&string->lock); + if (ret != 0) { + return -1; + } + return 0; } void mcl_string_free(mcl_string *string) { - if (!string) { + if (string == NULL) { + return; + } + + int ret = pthread_mutex_lock(&string->lock); + if (ret != 0) { return; } @@ -91,29 +123,59 @@ void mcl_string_free(mcl_string *string) { free(string->data); } + pthread_mutex_unlock(&string->lock); + pthread_mutex_destroy(&string->lock); + free(string); } -size_t mcl_string_length(const mcl_string *string) { - if (!string) { +size_t mcl_string_length(mcl_string *string) { + if (string == NULL) { return 0; } - return string->size; -} - -size_t mcl_string_capacity(const mcl_string *string) { - if (!string) { + int ret = pthread_mutex_lock(&string->lock); + if (ret != 0) { return 0; } - return string->capacity; + size_t len = string->size; + + pthread_mutex_unlock(&string->lock); + + return len; } -const char *mcl_string_cstr(const mcl_string *string) { - if (!string || !string->data) { +size_t mcl_string_capacity(mcl_string *string) { + if (string == NULL) { + return 0; + } + + int ret = pthread_mutex_lock(&string->lock); + if (ret != 0) { + return 0; + } + + size_t cap = string->capacity; + + pthread_mutex_unlock(&string->lock); + + return cap; +} + +const char *mcl_string_cstr(mcl_string *string) { + if (string == NULL || string->data == NULL) { return ""; } - return string->data; + int ret = pthread_mutex_lock(&string->lock); + if (ret != 0) { + return NULL; + } + + char *data = string->data; + + pthread_mutex_unlock(&string->lock); + + return data; } diff --git a/include/myclib/string/mystring.h b/include/myclib/string/mystring.h index f6daa54..b13d40b 100644 --- a/include/myclib/string/mystring.h +++ b/include/myclib/string/mystring.h @@ -1,25 +1,27 @@ #ifndef MYCLIB_STRING_H #define MYCLIB_STRING_H +#include #include /** - * @brief String structure + * @brief Thread-safe dynamic string structure */ typedef struct mcl_string_t { - size_t size; /**< Length of the string (excluding null terminator) */ - size_t capacity; /**< Total allocated capacity */ - char *data; /**< Pointer to the string data */ + size_t size; /**< Current length of the string (excluding null terminator) */ + size_t capacity; /**< Allocated capacity */ + char *data; /**< Pointer to string data */ + pthread_mutex_t lock; /**< Mutex for thread safety */ } mcl_string; /** - * @brief Create a new string + * @brief Create a new string initialized with given text * - * @param text The text to initialize from - * @param initial_capacity The initial capacity, pass -1 to retrieve it from the text - * @return Pointer to the new string, or NULL on failure + * @param text Initial text (cannot be NULL) + * @param initial_capacity Initial buffer capacity; pass -1 to auto-calculate + * @return Pointer to new string, or NULL on failure * - * @note The caller is responsible for freeing the returned string with mcl_string_free() and to pass the right inital_capacity + * @note Caller must free the string with mcl_string_free() */ mcl_string *mcl_string_new(const char *text, long initial_capacity); @@ -27,47 +29,45 @@ mcl_string *mcl_string_new(const char *text, long initial_capacity); * @brief Append text to an existing string * * @param string The string to append to - * @param text The string to append (can be empty) + * @param text Text to append (can be empty) * @return 0 on success, -1 on failure * - * @note If it fails, the original string remains unchanged + * @note Original string remains unchanged if append fails */ int mcl_string_append(mcl_string *string, const char *text); /** - * @brief Free a string + * @brief Free a string and its resources * - * @param string The string to free - * - * @note This function is safe to call with NULL pointers + * @param string String to free (NULL is safe) */ void mcl_string_free(mcl_string *string); /** * @brief Get the current length of the string * - * @param string The string to query - * @return The length of the string, or 0 if string is NULL + * @param string String to query + * @return Length of string, or 0 if NULL */ -size_t mcl_string_length(const mcl_string *string); +size_t mcl_string_length(mcl_string *string); /** - * @brief Get the current capacity of the string + * @brief Get the current capacity of the string buffer * - * @param string The string to query - * @return The capacity of the string buffer, or 0 if string is NULL + * @param string String to query + * @return Capacity, or 0 if NULL */ -size_t mcl_string_capacity(const mcl_string *string); +size_t mcl_string_capacity(mcl_string *string); /** - * @brief Get a read-only string representation + * @brief Get a read-only C-string pointer * - * @param string The string to access - * @return Pointer to null-terminated string data, or empty string "" if string is NULL + * @param string String to access + * @return Null-terminated string pointer, or "" if NULL * - * @warning The returned pointer should not be modified directly and may become - * invalid after any modification operation on the string + * @warning Do not modify returned pointer. + * Pointer may become invalid after string modifications. */ -const char *mcl_string_cstr(const mcl_string *string); +const char *mcl_string_cstr(mcl_string *string); #endif /* MYCLIB_STRING_H */ diff --git a/include/server/server.h b/include/server/server.h index 828827b..986c3fb 100644 --- a/include/server/server.h +++ b/include/server/server.h @@ -44,11 +44,19 @@ typedef enum cws_server_ret_t { CWS_SERVER_REQUEST_TOO_LARGE, } cws_server_ret; +/* TODO: use last_activity as keep-alive */ typedef struct cws_client_t { struct sockaddr_storage addr; time_t last_activity; } cws_client; +typedef struct cws_pthread_data_t { + int client_fd; + int epfd; + cws_config *config; + mcl_hashmap *clients; +} cws_pthread_data; + /** * @brief Setups hints object * @@ -114,6 +122,6 @@ int cws_server_accept_client(int sockfd, struct sockaddr_storage *their_sa, sock void cws_server_close_client(int epfd, int client_fd, mcl_hashmap *hashmap); cws_server_ret cws_server_handle_new_client(int sockfd, int epfd, mcl_hashmap *clients); -cws_server_ret cws_server_handle_client_data(int client_fd, int epfd, mcl_hashmap *clients, cws_config *config); +void *cws_server_handle_client_data(void *arg); #endif diff --git a/include/server/threadpool.h b/include/server/threadpool.h new file mode 100644 index 0000000..b23b51c --- /dev/null +++ b/include/server/threadpool.h @@ -0,0 +1,34 @@ +#ifndef CWS_THREADPOOL_H +#define CWS_THREADPOOL_H + +#include "myclib/hashmap/myhashmap.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_threadpool_t { + cws_task queue[CWS_MAX_QUEUE_TASKS]; + int front, rear; + + pthread_mutex_t lock; + pthread_cond_t cond; + pthread_t threads[CWS_MAX_THREADS]; + + int shutdown; +} cws_threadpool; + +cws_threadpool *cws_threadpool_init(); +cws_server_ret cws_threadpool_add_task(cws_threadpool *pool, cws_task *task); +cws_server_ret cws_threadpool_worker(cws_threadpool *pool); +cws_server_ret cws_threadpool_destroy(cws_threadpool *pool); + +#endif diff --git a/src/http/http.c b/src/http/http.c index 1f0d9fb..f60b516 100644 --- a/src/http/http.c +++ b/src/http/http.c @@ -64,7 +64,7 @@ cws_http *cws_http_parse(mcl_string *request_str, int sockfd, cws_config *config return NULL; } - //CWS_LOG_DEBUG("method: %s", pch); + // CWS_LOG_DEBUG("method: %s", pch); int ret = cws_http_parse_method(request, pch); if (ret < 0) { @@ -85,7 +85,7 @@ cws_http *cws_http_parse(mcl_string *request_str, int sockfd, cws_config *config return NULL; } - //CWS_LOG_DEBUG("location: %s", pch); + // CWS_LOG_DEBUG("location: %s", pch); mcl_string_append(request->location, pch); mcl_string_append(request->location_path, config->www); @@ -95,7 +95,7 @@ cws_http *cws_http_parse(mcl_string *request_str, int sockfd, cws_config *config } else { mcl_string_append(request->location_path, mcl_string_cstr(request->location)); } - //CWS_LOG_DEBUG("location path: %s", mcl_string_cstr(request->location_path)); + // CWS_LOG_DEBUG("location path: %s", mcl_string_cstr(request->location_path)); /* Parse HTTP version */ pch = strtok_r(NULL, " \r\n", &saveptr); @@ -105,7 +105,7 @@ cws_http *cws_http_parse(mcl_string *request_str, int sockfd, cws_config *config return NULL; } - //CWS_LOG_DEBUG("version: %s", pch); + // CWS_LOG_DEBUG("version: %s", pch); mcl_string_append(request->http_version, pch); /* Parse headers until a \r\n */ diff --git a/src/server/server.c b/src/server/server.c index e707bdf..0fa6abc 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -83,6 +83,7 @@ cws_server_ret cws_server_start(cws_config *config) { } cws_server_ret cws_server_loop(int sockfd, cws_config *config) { + /* Make the server loop multi-thread */ 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(cws_client)); if (!clients) { return CWS_SERVER_HASHMAP_INIT; @@ -116,9 +117,7 @@ cws_server_ret cws_server_loop(int sockfd, cws_config *config) { } while (cws_server_run) { - CWS_LOG_DEBUG("Waiting for epoll events..."); int nfds = epoll_wait(epfd, revents, CWS_SERVER_EPOLL_MAXEVENTS, CWS_SERVER_EPOLL_TIMEOUT); - CWS_LOG_DEBUG("epoll_wait returned %d events", nfds); if (nfds == 0) { /* TODO: Check for inactive clients */ @@ -132,10 +131,27 @@ cws_server_ret cws_server_loop(int sockfd, cws_config *config) { CWS_LOG_DEBUG("Handle new client: %d", ret); } } else { - ret = cws_server_handle_client_data(revents[i].data.fd, epfd, clients, config); - if (ret != CWS_SERVER_OK) { - CWS_LOG_DEBUG("Handle client data: %d", ret); + cws_pthread_data *thread_data = malloc(sizeof(cws_pthread_data)); + if (!thread_data) { + continue; } + pthread_t client_thread; + + int client_fd = revents[i].data.fd; + thread_data->client_fd = client_fd; + thread_data->clients = clients; + thread_data->config = config; + thread_data->epfd = epfd; + + ret = pthread_create(&client_thread, NULL, cws_server_handle_client_data, thread_data); + CWS_LOG_DEBUG("Started client thread for fd: %d", client_fd); + if (ret != 0) { + free(thread_data); + cws_server_close_client(epfd, client_fd, clients); + continue; + } + + pthread_detach(client_thread); } } } @@ -173,7 +189,14 @@ cws_server_ret cws_server_handle_new_client(int sockfd, int epfd, mcl_hashmap *c return CWS_SERVER_OK; } -cws_server_ret cws_server_handle_client_data(int client_fd, int epfd, mcl_hashmap *clients, cws_config *config) { +void *cws_server_handle_client_data(void *arg) { + cws_pthread_data *thread_data = (cws_pthread_data *)arg; + + int client_fd = thread_data->client_fd; + int epfd = thread_data->epfd; + mcl_hashmap *clients = thread_data->clients; + cws_config *config = thread_data->config; + char tmp_data[1024]; memset(tmp_data, 0, sizeof(tmp_data)); char ip[INET_ADDRSTRLEN] = {0}; @@ -187,8 +210,9 @@ cws_server_ret cws_server_handle_client_data(int client_fd, int epfd, mcl_hashma if (total_bytes > CWS_SERVER_MAX_REQUEST_SIZE) { mcl_string_free(data); cws_server_close_client(epfd, client_fd, clients); + free(thread_data); - return CWS_SERVER_REQUEST_TOO_LARGE; + return NULL; } mcl_string_append(data, tmp_data); memset(tmp_data, 0, sizeof(tmp_data)); @@ -199,8 +223,9 @@ cws_server_ret cws_server_handle_client_data(int client_fd, int epfd, mcl_hashma CWS_LOG_ERROR("recv(): %s", strerror(errno)); mcl_string_free(data); cws_server_close_client(epfd, client_fd, clients); + free(thread_data); - return CWS_SERVER_CLIENT_DISCONNECTED_ERROR; + return NULL; } /* Retrieve client ip */ @@ -210,8 +235,9 @@ cws_server_ret cws_server_handle_client_data(int client_fd, int epfd, mcl_hashma mcl_string_free(data); cws_epoll_del(epfd, client_fd); close(client_fd); + free(thread_data); - return CWS_SERVER_CLIENT_NOT_FOUND; + return NULL; } cws_client *client_info = (cws_client *)client->value; cws_utils_get_client_ip(&client_info->addr, ip); @@ -222,8 +248,9 @@ cws_server_ret cws_server_handle_client_data(int client_fd, int epfd, mcl_hashma CWS_LOG_INFO("Client (%s) disconnected", ip); mcl_string_free(data); cws_server_close_client(epfd, client_fd, clients); + free(thread_data); - return CWS_SERVER_CLIENT_DISCONNECTED; + return NULL; } /* Parse HTTP request */ @@ -233,19 +260,23 @@ cws_server_ret cws_server_handle_client_data(int client_fd, int epfd, mcl_hashma if (request == NULL) { CWS_LOG_INFO("Client (%s) disconnected (request NULL)", ip); cws_server_close_client(epfd, client_fd, clients); + free(thread_data); - return CWS_SERVER_HTTP_PARSE_ERROR; + return NULL; } int keepalive = cws_http_send_resource(request); cws_http_free(request); /* Only close connection if not keep-alive */ + /* TODO: fix */ if (keepalive <= 0) { cws_server_close_client(epfd, client_fd, clients); } - return CWS_SERVER_OK; + free(thread_data); + + return NULL; } cws_server_ret cws_epoll_add(int epfd, int sockfd, uint32_t events) { @@ -298,6 +329,8 @@ int cws_server_accept_client(int sockfd, struct sockaddr_storage *their_sa, sock } void cws_server_close_client(int epfd, int client_fd, mcl_hashmap *hashmap) { - cws_epoll_del(epfd, client_fd); - mcl_hm_remove(hashmap, &client_fd); + if (fcntl(client_fd, F_GETFD) != -1) { + cws_epoll_del(epfd, client_fd); + mcl_hm_remove(hashmap, &client_fd); + } } diff --git a/src/server/threadpool.c b/src/server/threadpool.c new file mode 100644 index 0000000..f80e7ef --- /dev/null +++ b/src/server/threadpool.c @@ -0,0 +1,49 @@ +#include "server/threadpool.h" + +#include + +cws_threadpool *cws_threadpool_init() { + cws_threadpool *pool = malloc(sizeof(cws_threadpool)); + if (pool == NULL) { + return NULL; + } + + int ret; + + ret = pthread_mutex_init(&pool->lock, NULL); + if (ret != 0) { + free(pool); + return NULL; + } + + ret = pthread_cond_init(&pool->cond, NULL); + if (ret != 0) { + free(pool); + return NULL; + } + + pool->front = 0; + pool->rear = 0; + pool->shutdown = 0; + + return pool; +} + +cws_server_ret cws_threadpool_add_task(cws_threadpool *pool, cws_task *task) { + pthread_mutex_lock(&pool->lock); + + /* ? */ + + pthread_mutex_unlock(&pool->lock); + + return CWS_SERVER_OK; +} + +cws_server_ret cws_threadpool_worker(cws_threadpool *pool) { return CWS_SERVER_OK; } + +cws_server_ret cws_threadpool_destroy(cws_threadpool *pool) { + pthread_mutex_destroy(&pool->lock); + pthread_cond_destroy(&pool->cond); + + return CWS_SERVER_OK; +}