diff --git a/praxis3/key_value_pair.c b/praxis3/key_value_pair.c index 47b264a..14822cd 100644 --- a/praxis3/key_value_pair.c +++ b/praxis3/key_value_pair.c @@ -1,5 +1,7 @@ #include "key_value_pair.h" +#include +#include #include #include #include @@ -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++; + } + } +} diff --git a/praxis3/key_value_pair.h b/praxis3/key_value_pair.h index d5460d7..66d6c6b 100644 --- a/praxis3/key_value_pair.h +++ b/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 diff --git a/praxis3/zmq_distributor.c b/praxis3/zmq_distributor.c index 1f5be5e..d59654b 100644 --- a/praxis3/zmq_distributor.c +++ b/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) { - if(argc < 3){ - fprintf(stderr, "Usage: %s ... ", argv[0]); - exit(1); - } - - pthread_mutex_init(&words_mutex, NULL); - - FILE *fp = fopen(argv[1], "r"); +char *map(size_t worker_count, char **worker_ports, char *filename){ + FILE *fp = fopen(filename, "r"); char *file_content_chunks; - int worker_index = 2; - int worker_count = argc - 2; + int worker_index = 0; 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){ - perror("[MAIN] Failed to create thread"); + if(pthread_create(&threads[thread_count], NULL, receiverFunc, (void*)args) != 0){ + perror("[MAIN - map] Failed to create thread"); + free(args->data); free(args); - free(threads); - pthread_mutex_destroy(&words_mutex); - exit(1); + continue; } 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 ... ", 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; } - diff --git a/praxis3/zmq_worker.c b/praxis3/zmq_worker.c index de517f1..10b3df5 100644 --- a/praxis3/zmq_worker.c +++ b/praxis3/zmq_worker.c @@ -1,6 +1,7 @@ #include #include #include +#include #include #include @@ -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 *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 *key_value_pair_list_to_string(key_value_pair **head){ 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(); - return NULL; + key_value_pair *head = 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; }