added thread pool on epoll server.

Author: felipeagger

Makefile

@@ -1,5 +1,5 @@
 CC = clang
-CFLAGS = -static -lm -Wall -Wextra -O2
+CFLAGS = -lm -Wall -Wextra -O2 -static
 
 # files
 SOURCES = simple-server.c epoll-server.c uring-server.c
@@ -13,7 +13,7 @@ simple: simple-server.c
 #	$(CC) $(CFLAGS) -o $@ $< -DFILE1_DEFINE
 
 epoll: epoll-server.c
-	$(CC) -o epoll-server epoll-server.c request-handle.c cpu-bound.c $(CFLAGS)
+	$(CC) -o epoll-server epoll-server.c threadpool.c request-handle.c cpu-bound.c $(CFLAGS) -pthread
 #	$(CC) $(CFLAGS) -o $@ $< -DFILE2_DEFINE -DADDITIONAL_FLAG
 
 uring: uring-server.c

epoll-server.c

@@ -1,18 +1,22 @@
 #include <sys/epoll.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
+#include <arpa/inet.h>
 #include <unistd.h>
 #include <fcntl.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <signal.h>
 #include <unistd.h>
+#include <errno.h>
 #include "request-handle.h"
+#include "threadpool.h"
 
 #define PORT 8080
-#define MAX_EVENTS 10
-#define BUFFER_SIZE 1024
+#define MAX_EVENTS 1000
+#define WORKERS 3
+#define BUFFER_SIZE 2048
 
 int server_fd = -1; // global server socket
 
@@ -50,9 +54,11 @@ void set_nonblocking(int sock) {
 }
 
 int main() {
+    int opt = 1;
     int epoll_fd;
     struct sockaddr_in server_addr;
     struct epoll_event event, events[MAX_EVENTS];
+    ThreadPool pool;
 
     setup_signal_handler();
 
@@ -63,6 +69,13 @@ int main() {
         exit(EXIT_FAILURE);
     }
 
+    // Configurando a opção SO_REUSEADDR 
+    if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt))) { 
+        perror("Erro ao configurar socket");
+        close(server_fd);
+        exit(EXIT_FAILURE);
+    }
+
     // Configurar endereço e bind
     server_addr.sin_family = AF_INET;
     server_addr.sin_port = htons(PORT);
@@ -89,7 +102,7 @@ int main() {
     }
 
     // Adicionar o socket do servidor ao epoll
-    event.events = EPOLLIN; // Monitorar para leitura
+    event.events = EPOLLIN | EPOLLET; // Monitorar para leitura
     event.data.fd = server_fd;
     if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_fd, &event) < 0) {
         perror("Erro ao adicionar ao epoll");
@@ -98,6 +111,9 @@ int main() {
         exit(EXIT_FAILURE);
     }
 
+    //iniciar workers thread_pool
+    thread_pool_init(&pool, WORKERS, MAX_EVENTS);
+
     printf("epoll server listening on port %d\n", PORT);
 
     while (1) {
@@ -106,29 +122,56 @@ int main() {
             if (events[i].data.fd == server_fd) {
                 // Aceitar nova conexão
                 int client_fd = accept(server_fd, NULL, NULL);
+                if (client_fd == -1) {
+                    if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                        // Nenhuma conexão pendente 
+                        continue;
+                    } else {
+                        perror("Erro ao aceitar conexão");
+                        continue;
+                    }
+                }
                 set_nonblocking(client_fd);
                 event.events = EPOLLIN | EPOLLET;
                 event.data.fd = client_fd;
                 epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client_fd, &event);
-                printf("Nova conexão aceita\n");
+                //printf("Nova conexão aceita\n");
             } else {
                 // process request
                 char buffer[BUFFER_SIZE];
                 int bytes_read = read(events[i].data.fd, buffer, BUFFER_SIZE - 1);
 
                 if (bytes_read < 0) {
-                    perror("Erro ao ler do cliente");
+                    if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                        continue;
+                    } else {
+                        perror("Erro ao ler do cliente");
+                        close(events[i].data.fd);
+                        epoll_ctl(epoll_fd, EPOLL_CTL_DEL, events[i].data.fd, NULL);
+                        continue;
+                    }
+                } else if (bytes_read == 0) {
+                    // Conexão fechada pelo cliente
                     close(events[i].data.fd);
+                    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, events[i].data.fd, NULL);
                     continue;
                 }
 
                 buffer[bytes_read] = '\0';
 
-                handle_request(events[i].data.fd, buffer);
+                TaskArgs *args = malloc(sizeof(TaskArgs));
+                args->client_fd = events[i].data.fd;
+                args->buffer = malloc(BUFFER_SIZE * sizeof(char));
+                strncpy(args->buffer, buffer, BUFFER_SIZE - 1);
+                args->buffer[BUFFER_SIZE - 1] = '\0';
+
+                //handle_request(events[i].data.fd, buffer); //sync
+                thread_pool_add(&pool, handle_request, args); //async
             }
         }
     }
 
     close(server_fd);
+    close(epoll_fd);
     return 0;
 }

request-handle.c

@@ -5,13 +5,14 @@
 #include <sys/socket.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
-#include <signal.h>
+//#include <signal.h>
 #include <string.h>
 #include <unistd.h>
 #include "cpu-bound.h"
 #include "request-handle.h"
+#include "threadpool.h"
 
-#define BUF_SIZE 1024
+#define BUF_SIZE 2048
 
 
 char* getParamQueryString(char buffer[BUF_SIZE], const char *param_name) {
@@ -51,11 +52,11 @@ char* getParamQueryString(char buffer[BUF_SIZE], const char *param_name) {
     return NULL;
 }
 
-void handle_request(int client_fd, char *buffer) {
-    printf("Requisição recebida:\n%s\n", buffer);
+void handle_request(TaskArgs* args) {
+    //printf("Requisição recebida:\n%s\n", args->buffer);
 
     //params
-    char *timeout = getParamQueryString(buffer, "timeout");
+    char *timeout = getParamQueryString(args->buffer, "timeout");
     if (timeout == NULL) {
         timeout = "10"; // ms
     }
@@ -68,32 +69,35 @@ void handle_request(int client_fd, char *buffer) {
             "{\"message\": \"request completed\"}";
 
     // Routing
-    if (strstr(buffer, "GET /api/cpu-bound") == buffer) {
+    if (strstr(args->buffer, "GET /api/cpu-bound") == args->buffer) {
 
         simulateCPU(atoi(timeout));
 
-        write(client_fd, response, strlen(response));
+        write(args->client_fd, response, strlen(response));
 
-    } else if (strstr(buffer, "GET /api/io-bound") == buffer) {
+    } else if (strstr(args->buffer, "GET /api/io-bound") == args->buffer) {
 
         usleep(atoi(timeout) * 1000);
 
-        write(client_fd, response, strlen(response));
+        write(args->client_fd, response, strlen(response));
 
-    } else if (strstr(buffer, "GET /health") == buffer) {
+    } else if (strstr(args->buffer, "GET /health") == args->buffer) {
 
         // response 200
         char *resp = "HTTP/1.1 200 OK\r\nContent-Length: 8\r\n\r\nHealthy!";
-        write(client_fd, resp, strlen(resp));
+        write(args->client_fd, resp, strlen(resp));
 
     } else {
 
         // response 404
         char *resp = "HTTP/1.1 404 Not Found\r\nContent-Length: 16\r\n\r\nRoute not found!";
-        write(client_fd, resp, strlen(resp));
+        write(args->client_fd, resp, strlen(resp));
     }
 
     //free(timeout);
     //free(resp_size);
-    close(client_fd);
+    if (strstr(args->buffer, "Connection: keep-alive") == NULL) {
+        close(args->client_fd);
+    }
+    free(args);
 }

request-handle.h

@@ -1,7 +1,8 @@
 #ifndef REQUEST_HANDLE_H
 #define REQUEST_HANDLE_H
 
+#include "threadpool.h"
 
-void handle_request(int client_fd, char *buffer);
+void handle_request(TaskArgs* args); //(int client_fd, char *buffer);
 
 #endif

simple-server.c

@@ -8,6 +8,7 @@
 #include <signal.h>
 #include <unistd.h>
 #include "request-handle.h"
+#include "threadpool.h"
 
 #define PORT 8080
 #define BACKLOG 10
@@ -102,7 +103,13 @@ int main() {
 
         buffer[bytes_read] = '\0';
 
-        handle_request(client_fd, buffer);
+        TaskArgs *args = malloc(sizeof(TaskArgs));
+        args->client_fd = client_fd;
+        args->buffer = malloc(BUF_SIZE * sizeof(char));
+        strncpy(args->buffer, buffer, BUF_SIZE - 1);
+        args->buffer[BUF_SIZE - 1] = '\0';
+
+        handle_request(args);
     }
 
     close(server_fd);

threadpool.c

@@ -0,0 +1,101 @@
+#include <pthread.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <unistd.h>
+
+
+typedef struct {
+    int client_fd;
+    char *buffer;
+} TaskArgs;
+
+typedef void (*FunctionPointer)(TaskArgs*);
+
+typedef struct {
+    FunctionPointer function;  // Ponteiro para a função
+    void* argument;           // Argumento da função
+} Task;
+
+typedef struct {
+    Task* tasks;              // Fila de tarefas
+    int task_count;           // Número de tarefas na fila
+    int head, tail;           // Ponteiros para início e fim da fila
+    int queue_size;           // Tamanho da fila
+
+    pthread_mutex_t lock;     // Mutex para sincronização
+    pthread_cond_t notify;    // Condição para notificação
+    int shutdown;             // Flag para finalizar o pool
+} ThreadPool;
+
+
+int thread_pool_add(ThreadPool* pool, FunctionPointer TaskFunc, void* argument) {
+    pthread_mutex_lock(&pool->lock);
+
+    if (pool->task_count == pool->queue_size) {
+        pthread_mutex_unlock(&pool->lock);
+        return -1;  // Fila cheia
+    }
+
+    Task task = { TaskFunc, argument };
+    pool->tasks[pool->tail] = task;
+    pool->tail = (pool->tail + 1) % pool->queue_size;
+    pool->task_count++;
+
+    pthread_cond_signal(&pool->notify);
+    pthread_mutex_unlock(&pool->lock);
+    return 0;
+}
+
+void* thread_loop(void* arg) {
+    ThreadPool* pool = (ThreadPool*)arg;
+
+    while (1) {
+        pthread_mutex_lock(&pool->lock);
+
+        while (pool->task_count == 0 && !pool->shutdown) {
+            pthread_cond_wait(&pool->notify, &pool->lock);
+        }
+
+        if (pool->shutdown) {
+            pthread_mutex_unlock(&pool->lock);
+            pthread_exit(NULL);
+        }
+
+        Task task = pool->tasks[pool->head];
+        pool->head = (pool->head + 1) % pool->queue_size;
+        pool->task_count--;
+
+        pthread_mutex_unlock(&pool->lock);
+
+        // Executa a tarefa
+        task.function(task.argument);
+    }
+}
+
+void thread_pool_init(ThreadPool* pool, int thread_count, int queue_size) {
+    pool->tasks = (Task*)malloc(sizeof(Task) * queue_size);
+    pool->task_count = 0;
+    pool->queue_size = queue_size;
+    pool->head = pool->tail = 0;
+    pool->shutdown = 0;
+
+    pthread_mutex_init(&pool->lock, NULL);
+    pthread_cond_init(&pool->notify, NULL);
+
+    for (int i = 0; i < thread_count; i++) {
+        pthread_t thread;
+        pthread_create(&thread, NULL, thread_loop, (void*)pool);
+        pthread_detach(thread);
+    }
+}
+
+void thread_pool_destroy(ThreadPool* pool) {
+    pthread_mutex_lock(&pool->lock);
+    pool->shutdown = 1;
+    pthread_cond_broadcast(&pool->notify);
+    pthread_mutex_unlock(&pool->lock);
+
+    free(pool->tasks);
+    pthread_mutex_destroy(&pool->lock);
+    pthread_cond_destroy(&pool->notify);
+}

threadpool.h

@@ -0,0 +1,36 @@
+#ifndef THREAD_POOL_H
+#define THREAD_POOL_H
+
+#include <pthread.h>
+
+typedef struct {
+    int client_fd;
+    char *buffer;
+} TaskArgs;
+
+typedef void (*FunctionPointer)(TaskArgs*);
+
+typedef struct {
+    FunctionPointer function;  // Ponteiro para a função
+    void* argument;           // Argumento da função
+} Task;
+
+// Estrutura para o pool de threads
+typedef struct {
+    Task* tasks;              // Fila de tarefas
+    int task_count;           // Número de tarefas na fila
+    int head, tail;           // Ponteiros para início e fim da fila
+    int queue_size;           // Tamanho da fila
+
+    pthread_mutex_t lock;     // Mutex para sincronização
+    pthread_cond_t notify;    // Condição para notificação
+    int shutdown;             // Flag para finalizar o pool
+} ThreadPool;
+
+
+//int thread_pool_add(ThreadPool* pool, void (*function)(void*), void* argument);
+int thread_pool_add(ThreadPool* pool, FunctionPointer TaskFunc, void* argument);
+void thread_pool_init(ThreadPool* pool, int thread_count, int queue_size);
+void thread_pool_destroy(ThreadPool* pool);
+
+#endif