diff --git a/praxis3/key_value_pair.c b/praxis3/key_value_pair.c index 34f707e..990d3f6 100644 --- a/praxis3/key_value_pair.c +++ b/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{ diff --git a/praxis3/zmq_distributor.c b/praxis3/zmq_distributor.c index 3cb433d..2aaff4b 100644 --- a/praxis3/zmq_distributor.c +++ b/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]; - int bytes_recv = zmq_recv(requester, reply, MESSAGE_LENGTH - 1, 0); + char reply[MESSAGE_LENGTH + 1]; + 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; } diff --git a/praxis3/zmq_worker.c b/praxis3/zmq_worker.c index 93a51d1..2188e46 100644 --- a/praxis3/zmq_worker.c +++ b/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){ - perror("Failed to receive request"); + if(bytes_recv < 3){ // minimal receive is TYPE + 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);