On the verse of memory leaks

praxis3/key_value_pair.c

@@ -29,6 +29,7 @@ char *key_value_pair_list_to_string(key_value_pair **head){
         }
         if(!ret){
             ret = strdup(kvp2str);
+            free(kvp2str);
             kvp_it = kvp_it->next;
         }
         else{

praxis3/zmq_distributor.c

@@ -13,12 +13,15 @@ typedef struct{
     int port;
     char *data;
     TYPE type;
+    void *context;
 }THREAD_ARGS;
 
 pthread_mutex_t words_mutex;
 
 char *words_list;
 
+char *sanitize_input(char *input);
+
 char *get_file_content(FILE *fp)
 {
     if (!fp) {
@@ -84,7 +87,19 @@ char *get_file_content(FILE *fp)
 void *receiverFunc(void *args)
 {
     THREAD_ARGS *params = (THREAD_ARGS*)args;
-    void *context = zmq_ctx_new();
+    if(!params){
+        perror("[THREAD] invalid arguments");
+        return NULL;
+    }
+
+    if (params->port <= 0) {
+        fprintf(stderr, "[THREAD] Invalid port: %d\n", params->port);
+        free(params);
+        return NULL;
+    }
+
+    void *context = params->context;
+
     if(!context){
         perror("[THREAD] Failed to create context");
         free(params);
@@ -99,7 +114,7 @@ void *receiverFunc(void *args)
     }
 
     char addr[22];
-    snprintf(addr, 21, "tcp://localhost:%d", params->port);
+    snprintf(addr, 21, "tcp://localhost:%d", params->port); // BUG: FIX PORT 0
 
     int rc = zmq_connect(requester, addr);
     if (rc != 0){
@@ -111,33 +126,52 @@ void *receiverFunc(void *args)
     }
     fprintf(stderr, "[THREAD] Connected to worker: %s\n", addr);
 
-    size_t len = strlen(params->data) + 4;
+    char *data = sanitize_input(params->data);
+    if(!data){
+        perror("[THREAD] Failed to sanitize input");
+        free(params->data);
+        free(params);
+        zmq_close(requester);
+        zmq_ctx_destroy(context);
+        return NULL;
+    }
+
+    free(params->data);
+
+    size_t len = strlen(data) + 4;
     char *request = (char*)malloc(sizeof(char) * len);
     if(!request){
         perror("[THREAD] Failed to allocate memory for request");
         free(params);
+        free(data);
         zmq_close(requester);
         zmq_ctx_destroy(context);
-        return NULL;    }
+        return NULL;
+    }
+
 
     switch(params->type){
         case MAP:
-            snprintf(request, len, "map%s", params->data);
+            snprintf(request, len, "map%s", data);
             break;
         case RED:
-            snprintf(request, len, "red%s", params->data);
+            snprintf(request, len, "red%s", data);
             break;
         case RIP:
             snprintf(request, len, "rip");
             break;
         default:
             perror("[THREAD] Got unsupported request");
+            free(data);
             free(params);
             free(request);
             zmq_close(requester);
             zmq_ctx_destroy(context);
             return NULL;
     }
+
+    free(data);
+
     fprintf(stderr, "[THREAD] Send request: %s\n", request);
     if(zmq_send(requester, request, strlen(request), 0) == -1){
         perror("[THREAD] Failed to send request");
@@ -151,8 +185,8 @@ void *receiverFunc(void *args)
     free(request);
 
     fprintf(stderr, "[THREAD]: Waiting for response\n");
-    char reply[MESSAGE_LENGTH];
+    char reply[MESSAGE_LENGTH + 1];
-    int bytes_recv = zmq_recv(requester, reply, MESSAGE_LENGTH - 1, 0);
+    int bytes_recv = zmq_recv(requester, reply, MESSAGE_LENGTH, 0);
     if (bytes_recv == -1){
         perror("[THREAD] Failed to receive reply");
         free(params);
@@ -205,12 +239,15 @@ void *receiverFunc(void *args)
     pthread_mutex_unlock(&words_mutex);
 
     zmq_close(requester);
-    zmq_ctx_destroy(context);
 
     return NULL;
 }
 
 char *sanitize_input(char *input){
+    if(!input) return NULL;
+
+    int i = 0;
+
     char *ret = malloc(sizeof(char));
     if(!ret){
         perror("[MAIN - sanitize_input] Failed to allocate memory for ret");
@@ -221,21 +258,20 @@ char *sanitize_input(char *input){
     char *it = input;
     while(!isalpha(*it)) it++;
 
-    int i = 0;
     bool writeSpace = false;
     while(*it){
         if(isalpha(*it)){
             char *tmp = NULL;
             if(writeSpace){
-                tmp = realloc(ret, strlen(ret) + 2);
+                tmp = realloc(ret, i + 3);
             }
             else{
-                tmp = realloc(ret, strlen(ret) + 1);
+                tmp = realloc(ret, i + 2);
             }
             if(!tmp){
                 perror("[MAIN - sanitize_input] Failed to realloc memory for ret");
                 free(ret);
-                exit(1);
+                return NULL;
             }
             ret = tmp;
             if(writeSpace) ret[i++] = ' ';
@@ -247,11 +283,11 @@ char *sanitize_input(char *input){
             writeSpace = true;
         }
     }
-    ret[strlen(ret) + 1] = '\0';
+    ret[i] = '\0';
     return ret;
 }
 
-void map(size_t worker_count, char **worker_ports, const char *filename) {
+void map(size_t worker_count, char **worker_ports, const char *filename, void *context) {
     FILE *fp = fopen(filename, "rb");
     char *raw = get_file_content(fp);
     if(!raw){
@@ -281,13 +317,13 @@ void map(size_t worker_count, char **worker_ports, const char *filename) {
     size_t offset = 0;
 
     while (offset < total_len) {
-        size_t chunk_size = (total_len - offset >= MESSAGE_LENGTH) ? MESSAGE_LENGTH : (total_len - offset);
+        size_t chunk_size = (total_len - offset >= PAYLOAD_LENGTH + 1) ? PAYLOAD_LENGTH + 1 : (total_len - offset);
 
         if (offset + chunk_size < total_len) {
             size_t new_chunk_size = chunk_size;
             while (new_chunk_size > 0) {
                 char c = content[offset + new_chunk_size];
-                if (!isalpha((unsigned char)c)) {
+                if (!isalpha((unsigned char)c) || (offset + chunk_size < strlen(content) && (content[offset + chunk_size] == ' ' || content[offset + chunk_size] == '\0'))) {
                     break;
                 }
                 new_chunk_size--;
@@ -324,6 +360,7 @@ void map(size_t worker_count, char **worker_ports, const char *filename) {
         args->type = MAP;
 
         args->data = strndup(content + offset, chunk_size);
+        args->context = context;
 
         if (pthread_create(&threads[thread_count], NULL, receiverFunc, (void*)args) != 0) {
             perror("[MAIN - map] Failed to create thread");
@@ -336,13 +373,13 @@ void map(size_t worker_count, char **worker_ports, const char *filename) {
         worker_index = (worker_index + 1) % worker_count;
         thread_count++;
     }
+    free(content);
 
     for (size_t i = 0; i < thread_count; i++) {
         pthread_join(threads[i], NULL);
     }
 
     free(threads);
-    free(content);
 }
 
 char *extract_reduce_string(const char * toExtract){
@@ -363,7 +400,7 @@ char *extract_reduce_string(const char * toExtract){
     return extracted;
 }
 
-void reduce(char *toReduce, size_t worker_count, char **worker_ports) {
+void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void *context) {
     if (!toReduce || worker_count == 0) {
         fprintf(stderr, "[MAIN - reduce] Invalid input parameters.\n");
         return;
@@ -389,6 +426,7 @@ void reduce(char *toReduce, size_t worker_count, char **worker_ports) {
             while (new_size > 0) {
                 char c = toReduce[offset + new_size];
                 if (isdigit((unsigned char)c)) {
+                    new_size++;
                     break;
                 }
                 new_size--;
@@ -410,6 +448,7 @@ void reduce(char *toReduce, size_t worker_count, char **worker_ports) {
         args->port = atol(worker_ports[worker_index]);
         args->data = strndup(toReduce + offset, chunk_size);
         args->type = RED;
+        args->context = context;
 
         if (!args->data) {
             perror("[MAIN - reduce] Failed to allocate memory for chunk data");
@@ -437,7 +476,7 @@ void reduce(char *toReduce, size_t worker_count, char **worker_ports) {
 }
 
 
-void rip(size_t worker_count, char **worker_ports){
+void rip(size_t worker_count, char **worker_ports, void *context){
     pthread_t *threads = malloc(sizeof(pthread_t) * worker_count);
     if(!threads){
         perror("[MAIN- reduce] Failed to allocate memory for threads");
@@ -458,6 +497,7 @@ void rip(size_t worker_count, char **worker_ports){
         args->port = atol(worker_ports[i]);
         args->data = strdup("rip");
         args->type = RIP;
+        args->context = context;
 
         if (!args->data) {
             fprintf(stderr, "[MAIN - reduce] Warning: extract_reduce_string() returned NULL\n");
@@ -540,8 +580,14 @@ int main(const int argc, char **argv) {
 
     pthread_mutex_init(&words_mutex, NULL);
 
+    void *context = zmq_ctx_new();
+    if(!context){
+        perror("[MAIN] zmq_ctx_new failed");
+        exit(1);
+    }
+
     // MAP
-    map(argc - 2, argv + 2, argv[1]);
+    map(argc - 2, argv + 2, argv[1], context);
     fprintf(stderr, "MAP:\n%s\n", words_list);
 
     char *word_list_mapped = strdup(words_list);
@@ -549,7 +595,7 @@ int main(const int argc, char **argv) {
     words_list = NULL;
 
     // REDUCE
-    reduce(word_list_mapped, argc - 2, argv + 2);
+    reduce(word_list_mapped, argc - 2, argv + 2, context);
     free(word_list_mapped);
 
     char *word_list_reduced = strdup(words_list);
@@ -559,12 +605,14 @@ int main(const int argc, char **argv) {
     fprintf(stderr, "REDUCE:\n%s\n", word_list_reduced);
 
     // RIP
-    rip(argc - 2, argv + 2);
+    rip(argc - 2, argv + 2, context);
 
     // print words_list
     print_words_list(word_list_reduced);
     free(word_list_reduced);
 
     pthread_mutex_destroy(&words_mutex);
+    zmq_ctx_destroy(context);
+
     return 0;
 }

praxis3/zmq_worker.c

@@ -31,8 +31,9 @@ int main(int argc, char **argv){
     while(thread_count < worker_count){
         int *port = malloc(sizeof(int));
         *port = atoi(argv[worker_index]);
-        if(pthread_create(&threads[thread_count++], NULL, workerFunc, port) != 0){
+        if(pthread_create(&threads[thread_count++], NULL, workerFunc, (void*)port) != 0){
             perror("[MAIN] Failed to create thread");
+            free(port);
             free(threads);
             exit(1);
         }
@@ -65,8 +66,13 @@ TYPE get_request_type(const char *request){
 }
 
 void* workerFunc(void *args){
+    if(!args){
+        perror("[THREAD] Invalid arguments");
+        return NULL;
+    }
     int port = *(int*)args;
     free(args);
+
     void *context = zmq_ctx_new();
     if(!context){
         perror("[THREAD] Failed to create context");
@@ -94,10 +100,10 @@ void* workerFunc(void *args){
         key_value_pair *head = NULL;
         key_value_pair *tail = NULL;
 
-        char request[MESSAGE_LENGTH + 1];
+        char request[MESSAGE_LENGTH + 1] = {0};
         int bytes_recv = zmq_recv(responder, request, MESSAGE_LENGTH, 0);
-        if(bytes_recv <= 0){
+        if(bytes_recv < 3){ // minimal receive is TYPE
-            perror("Failed to receive request");
+            perror("Failed to receive valid request");
             continue;
         }
 
@@ -106,9 +112,11 @@ void* workerFunc(void *args){
         char endpoint[256];
         size_t endpoint_size = sizeof(endpoint);
         if (zmq_getsockopt(responder, ZMQ_LAST_ENDPOINT, endpoint, &endpoint_size) == 0) {
-            fprintf(stderr, "Got request from: %s\n", endpoint);
+            fprintf(stderr, "Got request on: %s\n", endpoint);
         }
 
+        fprintf(stderr, "Received: %s\n", request);
+
         const TYPE type = get_request_type(request);
         char *content = request + TYPE_LENGTH;
 
@@ -137,6 +145,7 @@ void* workerFunc(void *args){
                 exit(1);
                 break;
         }
+
         fprintf(stderr, "Sending content: %s\n", ret);
         if(zmq_send(responder, ret, strlen(ret), 0) < 0){
             perror("Failed to send response");
@@ -178,7 +187,7 @@ char* get_word(const char *request){
         return NULL;
 
     while(*request && isalpha(*request)){
-        char *new_word = (char*)realloc(word, strlen(word) + 2);    // realloc word size + \0 and extra char space
+        char *new_word = (char*)realloc(word, length + 2);    // realloc word size + \0 and extra char space
         if(!new_word){
             free(word);
             return NULL;
@@ -193,6 +202,7 @@ char* get_word(const char *request){
 }
 
 char* map(const char *content, key_value_pair **head, key_value_pair **tail){
+    if(!content) return NULL;
     char *content_iterator = (char*)content;
     void *content_start_address = (void*)content;
     size_t content_length = strlen(content);