Added some functions

praxis3/zmq_distributor.c

@@ -6,8 +6,20 @@
 #include <string.h>
 #include <regex.h>
 #include <zmq.h>
+
+#include "key_value_pair.h"
 #include "transfer_defines.h"
 
+typedef struct{
+    int port;
+    char *data;
+    TYPE type;
+}THREAD_ARGS;
+
+pthread_mutex_t words_mutex;
+
+char *words_list;
+
 char *get_file_content(FILE *fp)
 {
     char *buffer = (char*)malloc(sizeof(char) * MESSAGE_LENGTH);
@@ -48,11 +60,131 @@ char *get_file_content(FILE *fp)
     return buffer;
 }
 
-pthread_mutex_t words_mutex;
-
-void *receiver()
+void *receiverFunc(void *args)
 {
+    THREAD_ARGS *params = (THREAD_ARGS*)args;
+    void *context = zmq_ctx_new();
+    if(!context){
+        perror("[THREAD] Failed to create context");
+        free(params);
+        return NULL;
+    }
+    void *requester = zmq_socket(context, ZMQ_REQ);
+    if(!requester){
+        perror("[THREAD] Failed to create socket");
+        free(params);
+        zmq_ctx_destroy(context);
+        return NULL;
+    }
 
+    char addr[22];
+    snprintf(addr, 21, "tcp://localhost:%d", params->port);
+
+    int rc = zmq_connect(requester, addr);
+    if (rc != 0){
+        perror("Failed to connect");
+        free(params);
+        zmq_close(requester);
+        zmq_ctx_destroy(context);
+        return NULL;
+    }
+
+    size_t len = strlen(params->data) + 4;
+    char *request = (char*)malloc(sizeof(char) * len);
+    if(!request){
+        perror("[THREAD] Failed to allocate memory for request");
+        free(params);
+        zmq_close(requester);
+        zmq_ctx_destroy(context);
+        return NULL;    }
+
+    switch(params->type){
+        case MAP:
+            snprintf(request, len, "map%s", params->data);
+            break;
+        case RED:
+            snprintf(request, len, "red%s", params->data);
+            break;
+        case RIP:
+            snprintf(request, len, "rip");
+            break;
+        default:
+            perror("[THREAD] Got unsupported request");
+            free(params);
+            free(request);
+            zmq_close(requester);
+            zmq_ctx_destroy(context);
+            return NULL;
+    }
+
+    if(zmq_send(requester, request, strlen(request), 0) == -1){
+        perror("[THREAD] Failed to send request");
+        free(params);
+        free(request);
+        zmq_close(requester);
+        zmq_ctx_destroy(context);
+        return NULL;
+    }
+
+    free(request);
+
+    char reply[MESSAGE_LENGTH];
+    int bytes_recv = zmq_recv(requester, reply, MESSAGE_LENGTH - 1, 0);
+    if (bytes_recv == -1){
+        perror("[THREAD] Failed to receive reply");
+        free(params);
+        zmq_close(requester);
+        zmq_ctx_destroy(context);
+        return NULL;
+    }
+
+    reply[bytes_recv] = '\0';
+
+    if(strcmp(reply, "rip") == 0){
+        free(params);
+        zmq_close(requester);
+        zmq_ctx_destroy(context);
+        return NULL;
+    }
+
+    pthread_mutex_lock(&words_mutex);
+
+    len = strlen(reply);
+    if(!words_list){
+        words_list = (char*)malloc(len + 1);
+        if(!words_list){
+            perror("[THREAD] Failed to allocate memory for words_list");
+            free(params);
+            pthread_mutex_unlock(&words_mutex);
+            zmq_close(requester);
+            zmq_ctx_destroy(context);
+            return NULL;
+        }
+    }
+    else{
+        char *tmp = realloc(words_list , sizeof(char) * (strlen(words_list) + len + 1));
+        if(tmp == NULL){
+            perror("[THREAD] Failed to realloc memory for words_list");
+            free(params);
+            pthread_mutex_unlock(&words_mutex);
+            zmq_close(requester);
+            zmq_ctx_destroy(context);
+            return NULL;
+
+        }
+        words_list = tmp;
+    }
+
+    strcat(words_list, reply);
+
+    free(params);
+
+    pthread_mutex_unlock(&words_mutex);
+
+    zmq_close(requester);
+    zmq_ctx_destroy(context);
+
+    return NULL;
 }
 
 int main(const int argc, char **argv) {
@@ -61,29 +193,58 @@ int main(const int argc, char **argv) {
         exit(1);
     }
 
-    void *context = zmq_ctx_new();
-    void *responder = zmq_socket(context, ZMQ_PULL);
-    int rc = zmq_bind(responder, "tcp://localhost:5555");
-
-    if (rc != 0)
-    {
-        perror("Failed to bind");
-        exit(1);
-    }
-
     pthread_mutex_init(&words_mutex, NULL);
 
     FILE *fp = fopen(argv[1], "r");
     char *file_content_chunks;
     int worker_index = 1;
-    while ((file_content_chunks = get_file_content(fp)) != NULL)
+    int worker_count = argc - 1;
+    int thread_count = 0;
+
+    pthread_t *threads = (pthread_t*)malloc(sizeof(pthread_t) * worker_count);
+
+    while ((file_content_chunks = get_file_content(fp)) != NULL && thread_count < worker_count)
     {
-        // TODO: send to worker
+
+        pthread_t receiver;
+        THREAD_ARGS *args = (THREAD_ARGS*)malloc(sizeof(THREAD_ARGS));
+        if(!args){
+            perror("[MAIN] Failed to allocate memory for args");
+            free(threads);
+            pthread_mutex_destroy(&words_mutex);
+            exit(1);
+        }
+        args->port = atol(argv[worker_index]);
+        args->data = strdup(file_content_chunks);
+        args->type = MAP;
+
+        if(pthread_create(&threads[thread_count++], NULL, receiverFunc, (void*)args) != 0){
+            perror("[MAIN] Failed to create thread");
+            free(args);
+            free(threads);
+            pthread_mutex_destroy(&words_mutex);
+            exit(1);
+        }
 
         free(file_content_chunks);
+        worker_index++;
     }
+
     fclose(fp);
 
+    // wait for workers to finish
+    for(int i = 0; i < thread_count; i++){
+        pthread_join(threads[i], NULL);
+    }
+
+    free(threads);
+
+    printf("MAP:\n%s\n\n", words_list);
+
+    // TODO: REDUCE
+
+    // TODO: RIP
+
     return 0;
 }