[!T] Added reduce

praxis3/key_value_pair.c

@@ -1,5 +1,7 @@
 #include "key_value_pair.h"
 
+#include <ctype.h>
+#include <stdbool.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -130,3 +132,39 @@ void remove_key_value_pair(key_value_pair *kv, key_value_pair **head, key_value_
         iterator = iterator->next;
     }
 }
+
+void extract_key_value_pairs(char *red_str, key_value_pair **head, key_value_pair **tail){
+    char *it = red_str;
+    bool reading_value = false;
+    char *word_start = it;
+    size_t word_length = 0;
+    size_t value_length = 0;
+
+    while(*it){
+        if(isalpha(*it) && reading_value){
+            char key[word_length + 1];
+            strncpy(key, word_start, word_length);
+            char value[16];
+            snprintf(value, 16, "%d", (int)value_length);
+
+            key_value_pair *kv = init_key_value_pair(key, value);
+            add_key_value_pair(kv, head, tail);
+
+            // reset
+            word_length = 0;
+            value_length = 0;
+            word_start = it;
+            reading_value = false;
+            continue;
+        }
+        if(isalpha(*it)){
+            word_length++;
+            it++;
+        }
+        else{
+            reading_value = true; // reassigning true is faster than if checks every iteration
+            word_length++;
+            it++;
+        }
+    }
+}

praxis3/key_value_pair.h

@@ -19,4 +19,6 @@ void free_key_value_pairs(key_value_pair **head, key_value_pair **tail);
 int add_key_value_pair(key_value_pair *kv, key_value_pair **head, key_value_pair **tail);
 void remove_key_value_pair(key_value_pair *kv, key_value_pair **head, key_value_pair **tail);
 
+void extract_key_value_pairs(char *red_str, key_value_pair **head, key_value_pair **tail);
+
 #endif //KEY_VALUE_PAIR_H

praxis3/zmq_distributor.c

@@ -117,7 +117,7 @@ void *receiverFunc(void *args)
             zmq_ctx_destroy(context);
             return NULL;
     }
-    printf("[THREAD] Sent request: %s\n", request);
+    printf("[THREAD] Send request: %s\n", request);
     if(zmq_send(requester, request, strlen(request), 0) == -1){
         perror("[THREAD] Failed to send request");
         free(params);
@@ -189,47 +189,42 @@ void *receiverFunc(void *args)
     return NULL;
 }
 
-int main(const int argc, char **argv) {
+char *map(size_t worker_count, char **worker_ports, char *filename){
-    if(argc < 3){
+    FILE *fp = fopen(filename, "r");
-        fprintf(stderr, "Usage: %s <file.txt> <worker port 1> <worker port 2> ... <worker port n>", argv[0]);
-        exit(1);
-    }
-
-    pthread_mutex_init(&words_mutex, NULL);
-
-    FILE *fp = fopen(argv[1], "r");
     char *file_content_chunks;
-    int worker_index = 2;
+    int worker_index = 0;
-    int worker_count = argc - 2;
     int thread_count = 0;
 
     pthread_t *threads = (pthread_t*)malloc(sizeof(pthread_t) * worker_count);
+    if(!threads){
+        perror("[MAIN - map] Failed to allocate memory for threads");
+        pthread_mutex_destroy(&words_mutex);
+        exit(1);
+    }
 
     while (thread_count < worker_count && (file_content_chunks = get_file_content(fp)) != NULL)
     {
-
-        pthread_t receiver;
         THREAD_ARGS *args = (THREAD_ARGS*)malloc(sizeof(THREAD_ARGS));
         if(!args){
-            perror("[MAIN] Failed to allocate memory for args");
+            perror("[MAIN - map] Failed to allocate memory for args");
             free(threads);
             pthread_mutex_destroy(&words_mutex);
             exit(1);
         }
-        args->port = atol(argv[worker_index]);
+        args->port = atol(worker_ports[worker_index]);
         args->data = strdup(file_content_chunks);
         args->type = MAP;
 
-        if(pthread_create(&threads[thread_count++], NULL, receiverFunc, (void*)args) != 0){
+        if(pthread_create(&threads[thread_count], NULL, receiverFunc, (void*)args) != 0){
-            perror("[MAIN] Failed to create thread");
+            perror("[MAIN - map] Failed to create thread");
+            free(args->data);
             free(args);
-            free(threads);
+            continue;
-            pthread_mutex_destroy(&words_mutex);
-            exit(1);
         }
 
         free(file_content_chunks);
         worker_index++;
+        thread_count++;
     }
 
     fclose(fp);
@@ -240,13 +235,101 @@ int main(const int argc, char **argv) {
     }
 
     free(threads);
+    return words_list;
+}
 
-    printf("MAP:\n%s\n\n", words_list);
+char *extract_reduce_string(const char * toExtract){
+    if(!toExtract) return NULL;
+    size_t toExtract_len = strlen(toExtract);
+    size_t max_len = toExtract_len < MESSAGE_LENGTH ? toExtract_len : MESSAGE_LENGTH;
+
+    char *end_ptr = (char*)toExtract + max_len - 1;
+    while(end_ptr > toExtract && isalpha(*end_ptr)) end_ptr--;
+
+    size_t copy_length = end_ptr - toExtract + 1;
+
+    char *extracted = strndup(toExtract, copy_length);
+    if(!extracted){
+        perror("[MAIN - reduce - extract-reduce-string] Failed to allocate memory for extracted string");
+        return NULL;
+    }
+    return extracted;
+}
+
+char *reduce(char *toReduce, size_t worker_count, char **worker_ports){
+    pthread_t *threads = malloc(sizeof(pthread_t) * worker_count);
+    if(!threads){
+        perror("[MAIN- reduce] Failed to allocate memory for threads");
+        pthread_mutex_destroy(&words_mutex);
+        exit(1);
+    }
+
+    key_value_pair *head, *tail;
+
+    extract_key_value_pairs(toReduce, &head, &tail);
+
+    size_t thread_count = 0;
+    key_value_pair *kv_it = head;
+
+    char *extraction_ptr = toReduce;
+
+    for(int i = 0; kv_it && i < worker_count; i++){
+        THREAD_ARGS *args = (THREAD_ARGS*)malloc(sizeof(THREAD_ARGS));
+        if(!args){
+            perror("[MAIN - reduce] Failed to allocate memory for args");
+            free(threads);
+            pthread_mutex_destroy(&words_mutex);
+            exit(1);
+        }
+        args->port = atol(worker_ports[i]);
+        args->data = extract_reduce_string(extraction_ptr);
+        args->type = RED;
+
+        if (!args->data) {
+            fprintf(stderr, "[MAIN - reduce] Warning: extract_reduce_string() returned NULL\n");
+            free(args);
+            continue;
+        }
+
+        extraction_ptr = toReduce + strlen(args->data) + 1;
+
+        if(pthread_create(&threads[thread_count], NULL, receiverFunc, (void*)args) != 0){
+            perror("[MAIN - reduce] Failed to create thread");
+            free(args->data);   // NULL frees are safe
+            free(args);
+            continue;
+        }
+        kv_it = kv_it->next;
+        thread_count++;
+    }
+
+    // wait for workers to finish
+    for(int i = 0; i < thread_count; i++){
+        pthread_join(threads[i], NULL);
+    }
+
+    free(threads);
+    return words_list;
+}
+
+int main(const int argc, char **argv) {
+    if(argc < 3){
+        fprintf(stderr, "Usage: %s <file.txt> <worker port 1> <worker port 2> ... <worker port n>", argv[0]);
+        exit(1);
+    }
+
+    pthread_mutex_init(&words_mutex, NULL);
+
+    // MAP
+    char *word_list_mapped = map(argc - 2, argv + 2, argv[1]);
+    printf("MAP:\n%s\n", word_list_mapped);
 
     // TODO: REDUCE
+    char *word_list_reduced = reduce(word_list_mapped, argc - 2, argv + 2);
+    free(word_list_mapped);
+    printf("REDUCE:\n%s\n", word_list_reduced);
 
     // TODO: RIP
 
     return 0;
 }
-

praxis3/zmq_worker.c

@@ -1,6 +1,7 @@
 #include <zmq.h>
 #include <ctype.h>
 #include <pthread.h>
+#include <stdbool.h>
 #include <stdlib.h>
 #include <string.h>
 
@@ -110,7 +111,7 @@ void* workerFunc(void *args){
         const TYPE type = get_request_type(request);
         char *content = request + TYPE_LENGTH;
 
-        char *ret;
+        char *ret = NULL;
 
         switch(type){
             case MAP:
@@ -142,6 +143,7 @@ void* workerFunc(void *args){
             zmq_ctx_destroy(context);
             exit(1);
         }
+        free(ret);  // NULL free is safe
     } //loop
 }
 
@@ -188,22 +190,7 @@ char* get_word(const char *request){
     return word;
 }
 
-char* map(const char *content, key_value_pair **head, key_value_pair **tail){
+char *key_value_pair_list_to_string(key_value_pair **head){
-    char *content_iterator = (char*)content;
-    void *content_start_address = (void*)content;
-    size_t content_length = strlen(content);
-
-    while(content_iterator != NULL && content_iterator < content_start_address + content_length){
-        char *word = get_word(content_iterator);
-        if(word != NULL){
-            if(strcmp(word, "") != 0){
-                add_word(word, head, tail);
-            }
-            content_iterator += strlen(word) + 1;   // skip current read word + one delimiter char
-            free(word);
-        }
-    }
-
     key_value_pair *kvp_it = *head;
 
     char *ret = NULL;
@@ -232,18 +219,44 @@ char* map(const char *content, key_value_pair **head, key_value_pair **tail){
             kvp_it = kvp_it->next;
         }
     }
+    return ret;
+}
+
+char* map(const char *content, key_value_pair **head, key_value_pair **tail){
+    char *content_iterator = (char*)content;
+    void *content_start_address = (void*)content;
+    size_t content_length = strlen(content);
+
+    while(content_iterator != NULL && content_iterator < content_start_address + content_length){
+        char *word = get_word(content_iterator);
+        if(word != NULL){
+            if(strcmp(word, "") != 0){
+                add_word(word, head, tail);
+            }
+            content_iterator += strlen(word) + 1;   // skip current read word + one delimiter char
+            free(word);
+        }
+    }
+
+    char *ret = key_value_pair_list_to_string(head);
 
     if(!ret){
         perror("Could not extract key value pairs");
         return NULL;
     }
-
     free_key_value_pairs(head, tail);
 
     return ret;
 }
 
 char* reduce(char *content){
-    // free_key_value_pairs();
+    key_value_pair *head = NULL;
-    return NULL;
+    key_value_pair *tail = NULL;
+
+    extract_key_value_pairs(content, &head, &tail);
+    char *ret = key_value_pair_list_to_string(&head);
+
+    free_key_value_pairs(&head, &tail);
+
+    return ret;
 }